From d397aa8f7d28e97eb8ce1e1ca02304107e8c639c Mon Sep 17 00:00:00 2001 From: Nicolas Jouanin Date: Wed, 1 Jul 2015 22:18:27 +0200 Subject: [PATCH] HBMQTT-2 #comment Fix unit tests --- tests/mqtt/test_publish.py | 32 +++++++++++++++++++++++++------- 1 file changed, 25 insertions(+), 7 deletions(-) diff --git a/tests/mqtt/test_publish.py b/tests/mqtt/test_publish.py index 910c1c5..a24df8a 100644 --- a/tests/mqtt/test_publish.py +++ b/tests/mqtt/test_publish.py @@ -10,22 +10,40 @@ class PublishPacketTest(unittest.TestCase): def setUp(self): self.loop = asyncio.new_event_loop() - def test_from_stream(self): - data = b'\x3f\x09\x00\x05topic\x00\x0a0123456789' + def test_from_stream_qos_0(self): + data = b'\x31\x11\x00\x05topic0123456789' + stream = asyncio.StreamReader(loop=self.loop) + stream.feed_data(data) + stream.feed_eof() + message = self.loop.run_until_complete(PublishPacket.from_stream(stream)) + self.assertEqual(message.variable_header.topic_name, 'topic') + self.assertEqual(message.variable_header.packet_id, None) + self.assertFalse((message.fixed_header.flags >> 1) & 0x03) + self.assertTrue(message.fixed_header.flags & 0x01) + self.assertTrue(message.payload.data, b'0123456789') + + def test_from_stream_qos_2(self): + data = b'\x37\x13\x00\x05topic\x00\x0a0123456789' stream = asyncio.StreamReader(loop=self.loop) stream.feed_data(data) stream.feed_eof() message = self.loop.run_until_complete(PublishPacket.from_stream(stream)) self.assertEqual(message.variable_header.topic_name, 'topic') self.assertEqual(message.variable_header.packet_id, 10) - self.assertEqual(message.fixed_header.qos, 0x03) - self.assertTrue(message.fixed_header.dup_flag) - self.assertTrue(message.fixed_header.retain_flag) + self.assertTrue((message.fixed_header.flags >> 1) & 0x03) + self.assertTrue(message.fixed_header.flags & 0x01) self.assertTrue(message.payload.data, b'0123456789') - def test_to_stream(self): + def test_to_stream_no_packet_id(self): + variable_header = PublishVariableHeader('topic', None) + payload = PublishPayload(b'0123456789') + publish = PublishPacket(variable_header=variable_header, payload=payload) + out = publish.to_bytes() + self.assertEqual(out, b'\x30\x11\x00\x05topic0123456789') + + def test_to_stream_packet(self): variable_header = PublishVariableHeader('topic', 10) payload = PublishPayload(b'0123456789') publish = PublishPacket(variable_header=variable_header, payload=payload) out = publish.to_bytes() - self.assertEqual(out, b'\x30\x13\x00\x05topic\x00\x0a0123456789') \ No newline at end of file + self.assertEqual(out, b'\x30\x13\x00\x05topic\00\x0a0123456789')