particle-os/debian-koji
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

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

Browse source
This commit is contained in:
Mike McLean 2017-08-24 15:06:11 -04:00
parent bce4e28ed8
commit fcfa6789c4
1 changed files with 30 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

30
tests/test_plugins/test_protonmsg.py
View file

@ -170,6 +170,7 @@ send_timeout = 60
""")
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 @@ send_timeout = 60
""")
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 @@ send_timeout = 60
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]