From 7831bf39ed4ab14f733b7913918eae78875c6103 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Przemys=C5=82aw=20Fierek?= Date: Tue, 17 Jan 2017 07:04:56 +0100 Subject: [PATCH] Add the default factory option to the ICMP body factory. (#1159) --- tests/scripts/thread-cert/config.py | 3 ++- tests/scripts/thread-cert/ipv6.py | 21 +++++++++++++----- tests/scripts/thread-cert/test_ipv6.py | 28 ++++++++++++------------ tests/scripts/thread-cert/test_lowpan.py | 2 +- 4 files changed, 32 insertions(+), 22 deletions(-) diff --git a/tests/scripts/thread-cert/config.py b/tests/scripts/thread-cert/config.py index 5563f68b8..d8de3b074 100644 --- a/tests/scripts/thread-cert/config.py +++ b/tests/scripts/thread-cert/config.py @@ -209,7 +209,8 @@ def create_default_ipv6_icmp_body_factories(): return { ipv6.ICMP_DESTINATION_UNREACHABLE: ipv6.ICMPv6DestinationUnreachableFactory(), ipv6.ICMP_ECHO_REQUEST: ipv6.ICMPv6EchoBodyFactory(), - ipv6.ICMP_ECHO_RESPONSE: ipv6.ICMPv6EchoBodyFactory() + ipv6.ICMP_ECHO_RESPONSE: ipv6.ICMPv6EchoBodyFactory(), + "default": ipv6.BytesPayloadFactory() } diff --git a/tests/scripts/thread-cert/ipv6.py b/tests/scripts/thread-cert/ipv6.py index 640823224..4065868f4 100644 --- a/tests/scripts/thread-cert/ipv6.py +++ b/tests/scripts/thread-cert/ipv6.py @@ -1029,7 +1029,16 @@ class ICMPv6Factory(PacketFactory): return self._body_factories[_type] except KeyError: - raise RuntimeError("Could not find factory to parse ICMP body. Unsupported ICMP type: {}".format(_type)) + if "default" not in self._body_factories: + raise RuntimeError("Could not find specialized factory to parse ICMP body. " + "Unsupported ICMP type: {}".format(_type)) + + default_factory = self._body_factories["default"] + + print("Could not find specialized factory to parse ICMP body. " + "Take the default one: {}".format(type(default_factory))) + + return default_factory def parse(self, data, message_info): header = ICMPv6Header.from_bytes(data) @@ -1049,9 +1058,9 @@ class ICMPv6EchoBodyFactory(PacketFactory): return ICMPv6EchoBody.from_bytes(data) -class UDPBytesPayload(ConvertibleToBytes, BuildableFromBytes): +class BytesPayload(ConvertibleToBytes, BuildableFromBytes): - """ Class representing payload of UDP datagram. """ + """ Class representing bytes payload. """ def __init__(self, data): self.data = data @@ -1067,12 +1076,12 @@ class UDPBytesPayload(ConvertibleToBytes, BuildableFromBytes): return len(self.data) -class UDPBytesPayloadFactory(PacketFactory): +class BytesPayloadFactory(PacketFactory): - """ Factory that produces payload of UDP datagram. """ + """ Factory that produces bytes payload. """ def parse(self, data, message_info): - return UDPBytesPayload(data.read()) + return BytesPayload(data.read()) class ICMPv6EchoBody(ConvertibleToBytes, BuildableFromBytes): diff --git a/tests/scripts/thread-cert/test_ipv6.py b/tests/scripts/thread-cert/test_ipv6.py index 8aa65b349..d51fdf706 100755 --- a/tests/scripts/thread-cert/test_ipv6.py +++ b/tests/scripts/thread-cert/test_ipv6.py @@ -37,9 +37,9 @@ from ipaddress import ip_address from ipv6 import ICMPv6Header, UDPHeader, IPv6Header, IPv6PacketFactory, UDPDatagram, \ UDPDatagramFactory, ICMPv6Factory, HopByHopFactory, MPLOptionFactory, ICMPv6, HopByHopOptionHeader, HopByHopOption, \ - HopByHop, MPLOption, HopByHopFactory, IPv6Packet, ICMPv6EchoBody, UDPBytesPayload, ICMPv6EchoBodyFactory, \ + HopByHop, MPLOption, HopByHopFactory, IPv6Packet, ICMPv6EchoBody, BytesPayload, ICMPv6EchoBodyFactory, \ UpperLayerProtocol, UDPHeaderFactory, HopByHopOptionsFactory, ICMPv6DestinationUnreachableFactory, \ - UDPBytesPayloadFactory, ICMPv6DestinationUnreachable, UdpBasedOnSrcDstPortsPayloadFactory + BytesPayloadFactory, ICMPv6DestinationUnreachable, UdpBasedOnSrcDstPortsPayloadFactory import common @@ -519,7 +519,7 @@ class TestIPv6Packet(unittest.TestCase): hop_limit=255) udp_dgram = UDPDatagram(UDPHeader(src_port=19788, dst_port=19788), - UDPBytesPayload(bytearray([0x00, 0x15, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + BytesPayload(bytearray([0x00, 0x15, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x01, 0x09, 0x01, 0x01, 0x0b, 0x03, 0x04, 0xc6, 0x69, 0x73, 0x51, 0x0e, 0x01, 0x80, 0x12, 0x02, 0x00, 0x01, 0xde, 0xad, 0xbe, 0xef]))) @@ -684,7 +684,7 @@ class TestUDPDatagram(unittest.TestCase): payload_length = len(payload) + 8 # UDP length consists of UDP header length and payload length udp_header = UDPHeader(src_port, dst_port, payload_length, checksum) - udp_payload = UDPBytesPayload(payload) + udp_payload = BytesPayload(payload) udp_dgram = UDPDatagram(udp_header, udp_payload) # WHEN @@ -1019,7 +1019,7 @@ class TestUdpBasedOnSrcDstPortsPayloadFactory(unittest.TestCase): factory = UdpBasedOnSrcDstPortsPayloadFactory( src_dst_port_based_payload_factories={ - message_info.src_port: UDPBytesPayloadFactory() + message_info.src_port: BytesPayloadFactory() }) # WHEN @@ -1038,7 +1038,7 @@ class TestUdpBasedOnSrcDstPortsPayloadFactory(unittest.TestCase): factory = UdpBasedOnSrcDstPortsPayloadFactory( src_dst_port_based_payload_factories={ - message_info.dst_port: UDPBytesPayloadFactory() + message_info.dst_port: BytesPayloadFactory() }) # WHEN @@ -1077,7 +1077,7 @@ class TestUDPDatagramFactory(unittest.TestCase): (payload_length >> 8), (payload_length & 0xFF), (checksum >> 8), (checksum & 0xFF)]) + payload - factory = UDPDatagramFactory(UDPHeaderFactory(), UDPBytesPayloadFactory()) + factory = UDPDatagramFactory(UDPHeaderFactory(), BytesPayloadFactory()) # WHEN udp_dgram = factory.parse(io.BytesIO(data), any_message_info()) @@ -1105,7 +1105,7 @@ class TestUDPDatagramFactory(unittest.TestCase): (payload_length >> 8), (payload_length & 0xFF), (checksum >> 8), (checksum & 0xFF)]) + payload - factory = UDPDatagramFactory(UDPHeaderFactory(), UDPBytesPayloadFactory()) + factory = UDPDatagramFactory(UDPHeaderFactory(), BytesPayloadFactory()) # WHEN udp_dgram = factory.parse(io.BytesIO(data), message_info) @@ -1152,14 +1152,14 @@ class TestICMPv6Factory(unittest.TestCase): self.assertRaises(RuntimeError, factory.parse, io.BytesIO(data), any_message_info()) -class TestUDPBytesPayload(unittest.TestCase): +class TestBytesPayload(unittest.TestCase): - def test_should_create_UDPBytesPayload_when_from_bytes_class_method_is_called(self): + def test_should_create_BytesPayload_when_from_bytes_class_method_is_called(self): # GIVEN data = any_data() # WHEN - actual = UDPBytesPayload.from_bytes(data) + actual = BytesPayload.from_bytes(data) # THEN self.assertEqual(data, actual.data) @@ -1167,7 +1167,7 @@ class TestUDPBytesPayload(unittest.TestCase): def test_should_return_exactly_the_same_data_as_passed_to_constructor_when_to_bytes_method_is_called(self): # GIVEN data = any_data() - payload = UDPBytesPayload(data) + payload = BytesPayload(data) # WHEN actual = payload.to_bytes() @@ -1175,10 +1175,10 @@ class TestUDPBytesPayload(unittest.TestCase): # THEN self.assertEqual(data, actual) - def test_should_return_the_same_length_as_data_passed_to_constructor_when_len_is_called_on_UDPBytesPayload_object(self): + def test_should_return_the_same_length_as_data_passed_to_constructor_when_len_is_called_on_BytesPayload_object(self): # GIVEN data = any_data() - payload = UDPBytesPayload(data) + payload = BytesPayload(data) # WHEN actual = len(payload) diff --git a/tests/scripts/thread-cert/test_lowpan.py b/tests/scripts/thread-cert/test_lowpan.py index 774ee3167..3e4d13919 100755 --- a/tests/scripts/thread-cert/test_lowpan.py +++ b/tests/scripts/thread-cert/test_lowpan.py @@ -48,7 +48,7 @@ def create_default_lowpan_parser(context_manager): ulpf={ 17: ipv6.UDPDatagramFactory( udp_header_factory=ipv6.UDPHeaderFactory(), - udp_payload_factory=ipv6.UDPBytesPayloadFactory()), + udp_payload_factory=ipv6.BytesPayloadFactory()), 58: ipv6.ICMPv6Factory( body_factories=config.create_default_ipv6_icmp_body_factories() )