#538 protonmsg plugin: test mode
Merged 2 years ago by mikem. Opened 2 years ago by mikem.

@@ -261,6 +261,14 @@ 

              conf.readfp(conffile)

          CONFIG = conf

      urls = CONFIG.get('broker', 'urls').split()

+     test_mode = False

+     if CONFIG.has_option('broker', 'test_mode'):

+         test_mode = CONFIG.getboolean('broker', 'test_mode')

+     if test_mode:

+         log.debug('test mode: skipping send to urls: %r', urls)

+         for msg in msgs:

+             log.debug('test mode: skipped msg: %r', msg)

+         return

      for url in sorted(urls, key=lambda k: random.random()):

          container = Container(TimeoutHandler(url, msgs, CONFIG))

          container.run()

@@ -170,6 +170,7 @@ 

  """)

          conf.flush()

          protonmsg.CONFIG_FILE = conf.name

+         protonmsg.CONFIG = None

          protonmsg.send_queued_msgs('postCommit')

          log = getLogger.return_value

          self.assertEqual(log.debug.call_count, 2)

@@ -193,6 +194,7 @@ 

  """)

          conf.flush()

          protonmsg.CONFIG_FILE = conf.name

+         protonmsg.CONFIG = None

          def clear_msgs():

              del context.protonmsg_msgs[:]

          Container.return_value.run.side_effect = clear_msgs

@@ -202,6 +204,34 @@ 

          self.assertTrue(log.debug.args[0][0].startswith('all msgs sent'))

          self.assertEqual(log.error.call_count, 0)

  

+     @patch('protonmsg.Container')

+     @patch('logging.getLogger')

+     def test_send_queued_msgs_test_mode(self, getLogger, Container):

+         context.protonmsg_msgs = [('test.topic', {'testheader': 1}, 'test body')]

+         conf = tempfile.NamedTemporaryFile()

+         conf.write("""[broker]

+ urls = amqps://broker1.example.com:5671 amqps://broker2.example.com:5671

+ cert = /etc/koji-hub/plugins/client.pem

+ cacert = /etc/koji-hub/plugins/ca.pem

+ topic_prefix = koji

+ connect_timeout = 10

+ send_timeout = 60

+ test_mode = on

+ """)

+         conf.flush()

+         protonmsg.CONFIG_FILE = conf.name

+         protonmsg.CONFIG = None

+         def clear_msgs():

+             del context.protonmsg_msgs[:]

+         Container.return_value.run.side_effect = clear_msgs

+         protonmsg.send_queued_msgs('postCommit')

+         Container.assert_not_called()

+         log = getLogger.return_value

+         self.assertEqual(log.debug.call_count, len(context.protonmsg_msgs) + 1)

+         self.assertTrue(log.debug.args[0][0].startswith('all msgs sent'))

+         self.assertEqual(log.error.call_count, 0)

+ 

+ 

  class TestTimeoutHandler(unittest.TestCase):

      def setUp(self):

          confdata = StringIO("""[broker]

add a test mode so that protonmsg can be enabled in test environments that lack an actual message bus

:thumbsup:
test pass on my local

1 new commit added

  • unit test: get protonmsg coverage back up to 100%, ensure test config is read
2 years ago

rebased

2 years ago

rebased

2 years ago

Commit 2395ea2 fixes this pull-request

Pull-Request has been merged by mikem@redhat.com

2 years ago