espressif/openthread
1
0
Fork 0
mirror of https://github.com/espressif/openthread.git synced 2026-08-07 03:07:47 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add the default factory option to the ICMP body factory. (#1159)

Browse Source
This commit is contained in:
Przemysław Fierek
2017-01-16 22:04:56 -08:00
committed by Jonathan Hui
parent 9b7d423f52
commit 7831bf39ed
4 changed files with 32 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/scripts/thread-cert/config.py
+2 -1
View File
@@ -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()
}
tests/scripts/thread-cert/ipv6.py
+15 -6
View File
@@ -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):
tests/scripts/thread-cert/test_ipv6.py
+14 -14
View File
@@ -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)
tests/scripts/thread-cert/test_lowpan.py
+1 -1
View File
@@ -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()
)