diff --git a/tests/handlers/packet.py b/tests/handlers/packet.py deleted file mode 100644 index 5842b67..0000000 --- a/tests/handlers/packet.py +++ /dev/null @@ -1,57 +0,0 @@ -# Copyright (c) 2015 Nicolas JOUANIN -# -# See the file license.txt for copying permission. -import unittest -import asyncio - -from hbmqtt.handlers.packet import PacketHandler -from hbmqtt.messages.packet import PacketType, MQTTHeader -from hbmqtt.errors import MQTTException - - -class TestPacketHandler(unittest.TestCase): - def setUp(self): - self.loop = asyncio.new_event_loop() - - def test_read_packet_header(self): - packet_handler = PacketHandler() - stream = asyncio.StreamReader(loop=self.loop) - stream.feed_data(b'\x10\x7f') - header = self.loop.run_until_complete(packet_handler.read_packet_header(stream)) - self.assertEqual(header.message_type, PacketType.CONNECT) - self.assertFalse(header.flags & 0x08) - self.assertEqual((header.flags & 0x06) >> 1, 0) - self.assertFalse(header.flags & 0x01) - self.assertEqual(header.remaining_length, 127) - - def test_decode_ok_with_length(self): - stream = asyncio.StreamReader(loop=self.loop) - stream.feed_data(b'\x10\xff\xff\xff\x7f') - header = self.loop.run_until_complete(PacketHandler.read_packet_header(stream)) - self.assertEqual(header.packet_type, PacketType.CONNECT) - self.assertFalse(header.flags & 0x08) - self.assertEqual((header.flags & 0x06) >> 1, 0) - self.assertFalse(header.flags & 0x01) - self.assertEqual(header.remaining_length, 268435455) - - def test_decode_reserved(self): - stream = asyncio.StreamReader(loop=self.loop) - stream.feed_data(b'\x0f\x7f') - with self.assertRaises(MQTTException): - self.loop.run_until_complete(PacketHandler.read_packet_header(stream)) - - def test_decode_ko_with_length(self): - stream = asyncio.StreamReader(loop=self.loop) - stream.feed_data(b'\x10\xff\xff\xff\xff\x7f') - with self.assertRaises(MQTTException): - self.loop.run_until_complete(PacketHandler.read_packet_header(stream)) - - def test_encode(self): - header = MQTTHeader(PacketType.CONNECT, 0x00, 0) - data = self.loop.run_until_complete(PacketHandler._encode_fixed_header(header)) - self.assertEqual(data, b'\x10\x00') - - def test_encode_2(self): - header = MQTTHeader(PacketType.CONNECT, 0x00, 268435455) - data = self.loop.run_until_complete(PacketHandler._encode_fixed_header(header)) - self.assertEqual(data, b'\x10\xff\xff\xff\x7f')