diff --git a/tests/scripts/thread-cert/lowpan.py b/tests/scripts/thread-cert/lowpan.py index b5827ac2b..15d0ca6ff 100644 --- a/tests/scripts/thread-cert/lowpan.py +++ b/tests/scripts/thread-cert/lowpan.py @@ -849,7 +849,11 @@ class LowpanMeshHeaderFactory: is_short_originator_address = bool(data_byte & 0x20) is_short_final_destination_address = bool(data_byte & 0x10) - hops_left = (data_byte & 0x0f) + if (data_byte & 0x0f) != 0x0f: + hops_left = (data_byte & 0x0f) + else: + hops_left = ord(data.read(1)) + originator_address = self._parse_address(data, is_short_originator_address) final_destination_address = self._parse_address(data, is_short_final_destination_address) @@ -1091,3 +1095,7 @@ class LowpanParser(object): elif self._is_iphc(first_byte): return self._handle_iphc_header(data, message_info) + + else: + raise RuntimeError("Unsupported header type: 0x{:02x}".format(first_byte)) + diff --git a/tests/scripts/thread-cert/test_lowpan.py b/tests/scripts/thread-cert/test_lowpan.py index 3e4d13919..fe4fcde21 100755 --- a/tests/scripts/thread-cert/test_lowpan.py +++ b/tests/scripts/thread-cert/test_lowpan.py @@ -264,6 +264,42 @@ class TestLowpanIPHC(unittest.TestCase): class TestLowpanParser(unittest.TestCase): + def test_should_parse_6lowpan_packet_with_mesh_header_that_contains_hop_limit_storred_on_two_bytes_when_decompress_method_is_called(self): + # GIVEN + lowpan_packet = bytearray([0xbf, 0x13, 0x90, 0x00, 0x48, 0x01, 0x7c, 0x77, + 0x3f, 0xf2, 0xbf, 0xc0, 0x00, 0x24, 0xb1, 0x62, + 0x44, 0x02, 0xf0, 0xba, 0x0d, 0xff, 0x04, 0x01, + 0x00, 0x02, 0x02, 0x08, 0x00, 0x07, 0x09, 0x50, + 0x20, 0x00, 0x20, 0x00, 0x08, 0x00, 0x00, 0x00]) + + ipv6_packet = bytearray([0x60, 0x00, 0x00, 0x00, 0x00, 0x21, 0x11, 0x3f, + 0xfd, 0x00, 0x0d, 0xb8, 0x00, 0x00, 0x00, 0x00, + 0x00, 0x00, 0x00, 0xff, 0xfe, 0x00, 0x90, 0x00, + 0xfd, 0x00, 0x0d, 0xb8, 0x00, 0x00, 0x00, 0x00, + 0x00, 0x00, 0x00, 0xff, 0xfe, 0x00, 0x48, 0x01, + 0xf0, 0xbf, 0xc0, 0x00, 0x00, 0x21, 0xe2, 0xdd, + 0x62, 0x44, 0x02, 0xf0, 0xba, 0x0d, 0xff, 0x04, + 0x01, 0x00, 0x02, 0x02, 0x08, 0x00, 0x07, 0x09, + 0x50, 0x20, 0x00, 0x20, 0x00, 0x08, 0x00, 0x00, + 0x00]) + + message_info = common.MessageInfo() + message_info.source_mac_address = common.MacAddress.from_eui64( + bytearray([0x00, 0x99, 0x99, 0xff, 0xfe, 0x22, 0x11, 0x00])) + message_info.destination_mac_address = common.MacAddress.from_eui64( + bytearray([0x34, 0x29, 0x96, 0xff, 0xfe, 0xac, 0xff, 0x17])) + + context_manager = lowpan.ContextManager() + context_manager[0] = lowpan.Context(prefix="fd00:db8::/64") + + parser = create_default_lowpan_parser(context_manager) + + # WHEN + actual_ipv6_packet = parser.parse(io.BytesIO(lowpan_packet), message_info) + + # THEN + self.assertEqual(ipv6_packet, actual_ipv6_packet.to_bytes()) + def test_should_parse_6lowpan_packet_with_not_compressed_udp_and_without_hop_by_hop_extension_header_when_decompress_method_is_called(self): # GIVEN lowpan_packet = bytearray([0x7a, 0x33, 0x11, 0x16, 0x33, 0x16, 0x34, 0x00,