mirror
/
amqtt
kopia lustrzana https://github.com/Yakifo/amqtt
1
0
Forkuj 0
Kod Zgłoszenia Packages Projekty Wydania Wiki Aktywność

HBMQTT-2 #comment Fix unit tests

pull/8/head
Nicolas Jouanin 2015-07-01 22:18:27 +02:00
rodzic b3e18adefe
commit d397aa8f7d
1 zmienionych plików z 25 dodań i 7 usunięć
Pokaż statystyki Ściągnij plik aktualizacji Ściągnij plik porównania Expand all files Collapse all files

32
tests/mqtt/test_publish.py
Wyświetl plik

@ -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')
self.assertEqual(out, b'\x30\x13\x00\x05topic\00\x0a0123456789')