# -*- coding: utf-8 -*- """ unit tests for the sqs_events engine """ # Import Python libs from __future__ import absolute_import, print_function, unicode_literals # Import Salt Libs import salt.engines.sqs_events as sqs_events # Import Salt Testing Libs from tests.support.mixins import LoaderModuleMockMixin from tests.support.mock import MagicMock, patch from tests.support.unit import TestCase, skipIf @skipIf(sqs_events.HAS_BOTO is False, "The boto library is not installed") class EngineSqsEventTestCase(TestCase, LoaderModuleMockMixin): """ Test cases for salt.engine.sqs_events """ def setup_loader_modules(self): patcher = patch("salt.engines.sqs_events.boto.sqs") self.mock_sqs = patcher.start() self.addCleanup(patcher.stop) self.addCleanup(delattr, self, "mock_sqs") return {sqs_events: {}} def sample_msg(self): fake_msg = MagicMock() fake_msg.get_body.return_value = "This is a test message" fake_msg.delete.return_value = True return fake_msg # 'present' function tests: 1 def test_no_queue_present(self): """ Test to ensure the SQS engine logs a warning when queue not present """ with patch("salt.engines.sqs_events.log") as mock_logging: with patch("time.sleep", return_value=None) as mock_sleep: q = None q_name = "mysqs" mock_fire = MagicMock(return_value=True) sqs_events._process_queue(q, q_name, mock_fire) self.assertTrue(mock_logging.warning.called) self.assertFalse(self.mock_sqs.queue.Queue().get_messages.called) def test_minion_message_fires(self): """ Test SQS engine correctly gets and fires messages on minion """ msgs = [self.sample_msg(), self.sample_msg()] self.mock_sqs.queue.Queue().get_messages.return_value = msgs q = self.mock_sqs.queue.Queue() q_name = "mysqs" mock_event = MagicMock(return_value=True) mock_fire = MagicMock(return_value=True) with patch.dict(sqs_events.__salt__, {"event.send": mock_event}): sqs_events._process_queue(q, q_name, mock_fire) self.assertTrue(self.mock_sqs.queue.Queue().get_messages.called) self.assertTrue(all(x.delete.called for x in msgs)) def test_master_message_fires(self): """ Test SQS engine correctly gets and fires messages on master """ msgs = [self.sample_msg(), self.sample_msg()] self.mock_sqs.queue.Queue().get_messages.return_value = msgs q = self.mock_sqs.queue.Queue() q_name = "mysqs" mock_fire = MagicMock(return_value=True) sqs_events._process_queue(q, q_name, mock_fire) self.assertTrue(self.mock_sqs.queue.Queue().get_messages.called, len(msgs)) self.assertTrue(mock_fire.called, len(msgs))