diff --git a/.travis/before_install.sh b/.travis/before_install.sh index 3295a938c..40bac8c03 100755 --- a/.travis/before_install.sh +++ b/.travis/before_install.sh @@ -97,8 +97,9 @@ cd /tmp || die # Packages used by sniffer sudo -H pip install pycryptodome==3.4.3 || die + sudo -H pip install enum34 || die pip install pycryptodome==3.4.3 || die - + pip install enum34 || die } [ $TRAVIS_OS_NAME != osx ] || { diff --git a/tests/scripts/Makefile.am b/tests/scripts/Makefile.am index 984656b2e..17ed5fd43 100644 --- a/tests/scripts/Makefile.am +++ b/tests/scripts/Makefile.am @@ -121,6 +121,7 @@ EXTRA_DIST = \ thread-cert/Cert_9_2_16_ActivePendingPartition.py \ thread-cert/Cert_9_2_17_Orphan.py \ thread-cert/Cert_9_2_18_RollBackActiveTimestamp.py \ + thread-cert/coap.py \ thread-cert/common.py \ thread-cert/config.py \ thread-cert/ipv6.py \ @@ -130,9 +131,19 @@ EXTRA_DIST = \ thread-cert/mle.py \ thread-cert/net_crypto.py \ thread-cert/network_data.py \ + thread-cert/network_layer.py \ thread-cert/node.py \ thread-cert/node_cli.py \ thread-cert/sniffer.py \ + thread-cert/test_coap.py \ + thread-cert/test_common.py \ + thread-cert/test_crypto.py \ + thread-cert/test_ipv6.py \ + thread-cert/test_lowpan.py \ + thread-cert/test_mac802154.py \ + thread-cert/test_mle.py \ + thread-cert/test_network_data.py \ + thread-cert/test_network_layer.py \ $(NULL) if OPENTHREAD_BUILD_TESTS @@ -142,6 +153,15 @@ if OPENTHREAD_BUILD_TESTS check_PROGRAMS = check_SCRIPTS = \ + thread-cert/test_coap.py \ + thread-cert/test_common.py \ + thread-cert/test_crypto.py \ + thread-cert/test_ipv6.py \ + thread-cert/test_lowpan.py \ + thread-cert/test_mac802154.py \ + thread-cert/test_mle.py \ + thread-cert/test_network_data.py \ + thread-cert/test_network_layer.py \ $(NULL) if OPENTHREAD_TESTS_SUBSET1 diff --git a/tests/scripts/thread-cert/Cert_5_1_01_RouterAttach.py b/tests/scripts/thread-cert/Cert_5_1_01_RouterAttach.py index 515cdc7b9..ce18dfaee 100755 --- a/tests/scripts/thread-cert/Cert_5_1_01_RouterAttach.py +++ b/tests/scripts/thread-cert/Cert_5_1_01_RouterAttach.py @@ -32,6 +32,7 @@ import unittest import config import mle +import network_layer import node LEADER = 1 @@ -136,6 +137,20 @@ class Cert_5_1_01_RouterAttach(unittest.TestCase): msg.assertMleMessageContainsOptionalTlv(mle.Route64) msg.assertMleMessageContainsOptionalTlv(mle.AddressRegistration) + # 6 - Router + msg = router_messages.next_coap_message("0.02") + msg.assertCoapMessageRequestUriPath("/a/as") + msg.assertCoapMessageContainsTlv(network_layer.MacExtendedAddress) + msg.assertCoapMessageContainsTlv(network_layer.Status) + + # 7 - Leader + msg = leader_messages.next_coap_message("2.04") + msg.assertCoapMessageContainsTlv(network_layer.Status) + msg.assertCoapMessageContainsOptionalTlv(network_layer.RouterMask) + + status_tlv = msg.get_coap_message_tlv(network_layer.Status) + self.assertEqual(network_layer.StatusValues.SUCCESS, status_tlv.status) + # 8 - Router msg = router_messages.next_mle_message(mle.CommandType.LINK_REQUEST) msg.assertMleMessageContainsTlv(mle.SourceAddress) diff --git a/tests/scripts/thread-cert/Cert_5_1_02_ChildAddressTimeout.py b/tests/scripts/thread-cert/Cert_5_1_02_ChildAddressTimeout.py index 2409b8287..5f0a6dc27 100755 --- a/tests/scripts/thread-cert/Cert_5_1_02_ChildAddressTimeout.py +++ b/tests/scripts/thread-cert/Cert_5_1_02_ChildAddressTimeout.py @@ -123,12 +123,20 @@ class Cert_5_1_02_ChildAddressTimeout(unittest.TestCase): # 1 - All leader_messages.next_mle_message(mle.CommandType.ADVERTISEMENT) + router1_messages.next_mle_message(mle.CommandType.PARENT_REQUEST) leader_messages.next_mle_message(mle.CommandType.PARENT_RESPONSE) + router1_messages.next_mle_message(mle.CommandType.CHILD_ID_REQUEST) leader_messages.next_mle_message(mle.CommandType.CHILD_ID_RESPONSE) + msg = router1_messages.next_coap_message("0.02") + msg.assertCoapMessageRequestUriPath("/a/as") + + msg = leader_messages.next_coap_message("2.04") + router1_messages.next_mle_message(mle.CommandType.ADVERTISEMENT) + ed_messages.next_mle_message(mle.CommandType.PARENT_REQUEST) router1_messages.next_mle_message(mle.CommandType.PARENT_RESPONSE) ed_messages.next_mle_message(mle.CommandType.CHILD_ID_REQUEST) @@ -139,6 +147,26 @@ class Cert_5_1_02_ChildAddressTimeout(unittest.TestCase): sed_messages.next_mle_message(mle.CommandType.CHILD_ID_REQUEST) router1_messages.next_mle_message(mle.CommandType.CHILD_ID_RESPONSE) + # 3 - Leader + msg = leader_messages.next_coap_message("0.02") + msg.assertCoapMessageRequestUriPath("/a/aq") + + msg = leader_messages.next_coap_message("0.02") + msg.assertCoapMessageRequestUriPath("/a/aq") + + # 4 - Router1 + msg = router1_messages.does_not_contain_coap_message() + + # 6 - Leader + msg = leader_messages.next_coap_message("0.02") + msg.assertCoapMessageRequestUriPath("/a/aq") + + msg = leader_messages.next_coap_message("0.02") + msg.assertCoapMessageRequestUriPath("/a/aq") + + # 7 - Router1 + msg = router1_messages.does_not_contain_coap_message() + if __name__ == '__main__': unittest.main() diff --git a/tests/scripts/thread-cert/Cert_5_1_03_RouterAddressReallocation.py b/tests/scripts/thread-cert/Cert_5_1_03_RouterAddressReallocation.py index 3820937bc..6628f6d9b 100755 --- a/tests/scripts/thread-cert/Cert_5_1_03_RouterAddressReallocation.py +++ b/tests/scripts/thread-cert/Cert_5_1_03_RouterAddressReallocation.py @@ -32,6 +32,7 @@ import unittest import config import mle +import network_layer import node LEADER = 1 @@ -110,14 +111,45 @@ class Cert_5_1_03_RouterAddressReallocation(unittest.TestCase): router1_messages.next_mle_message(mle.CommandType.PARENT_REQUEST) leader_messages.next_mle_message(mle.CommandType.PARENT_RESPONSE) + router1_messages.next_mle_message(mle.CommandType.CHILD_ID_REQUEST) leader_messages.next_mle_message(mle.CommandType.CHILD_ID_RESPONSE) + msg = router1_messages.next_coap_message("0.02") + msg.assertCoapMessageRequestUriPath("/a/as") + + msg = leader_messages.next_coap_message("2.04") + + router1_messages.next_mle_message(mle.CommandType.ADVERTISEMENT) + router2_messages.next_mle_message(mle.CommandType.PARENT_REQUEST) leader_messages.next_mle_message(mle.CommandType.PARENT_RESPONSE) router1_messages.next_mle_message(mle.CommandType.PARENT_RESPONSE) + router2_messages.next_mle_message(mle.CommandType.CHILD_ID_REQUEST) + # Leader or Router1 can be parent of Router2 + if leader_messages.contains_mle_message(mle.CommandType.CHILD_ID_RESPONSE): + leader_messages.next_mle_message(mle.CommandType.CHILD_ID_RESPONSE) + + msg = router2_messages.next_coap_message("0.02") + msg.assertCoapMessageRequestUriPath("/a/as") + + msg = leader_messages.next_coap_message("2.04") + + elif router1_messages.contains_mle_message(mle.CommandType.CHILD_ID_RESPONSE): + router1_messages.next_mle_message(mle.CommandType.CHILD_ID_RESPONSE) + + msg = router2_messages.next_coap_message("0.02") + msg.assertCoapMessageRequestUriPath("/a/as") + + msg = router1_messages.next_coap_message("0.02") + msg.assertCoapMessageRequestUriPath("/a/as") + + msg = leader_messages.next_coap_message("2.04") + + msg = router1_messages.next_coap_message("2.04") + # 5 - Router1 # Router1 make two attempts to reconnect to its current Partition. for _ in range(2): @@ -158,6 +190,16 @@ class Cert_5_1_03_RouterAddressReallocation(unittest.TestCase): msg.assertMleMessageContainsTlv(mle.TlvRequest) msg.assertMleMessageDoesNotContainTlv(mle.AddressRegistration) + # 8 - Router1 + msg = router1_messages.next_coap_message("0.02") + msg.assertCoapMessageRequestUriPath("/a/as") + msg.assertCoapMessageContainsTlv(network_layer.MacExtendedAddress) + msg.assertCoapMessageContainsOptionalTlv(network_layer.Rloc16) + msg.assertCoapMessageContainsTlv(network_layer.Status) + + # 8 - Router2 + msg = router2_messages.next_coap_message("2.04") + if __name__ == '__main__': unittest.main() diff --git a/tests/scripts/thread-cert/Cert_5_1_04_RouterAddressReallocation.py b/tests/scripts/thread-cert/Cert_5_1_04_RouterAddressReallocation.py index 13b793b49..a60c5f25b 100755 --- a/tests/scripts/thread-cert/Cert_5_1_04_RouterAddressReallocation.py +++ b/tests/scripts/thread-cert/Cert_5_1_04_RouterAddressReallocation.py @@ -32,6 +32,7 @@ import unittest import config import mle +import network_layer import node LEADER = 1 @@ -109,14 +110,45 @@ class Cert_5_1_04_RouterAddressReallocation(unittest.TestCase): router1_messages.next_mle_message(mle.CommandType.PARENT_REQUEST) leader_messages.next_mle_message(mle.CommandType.PARENT_RESPONSE) + router1_messages.next_mle_message(mle.CommandType.CHILD_ID_REQUEST) leader_messages.next_mle_message(mle.CommandType.CHILD_ID_RESPONSE) + msg = router1_messages.next_coap_message("0.02") + msg.assertCoapMessageRequestUriPath("/a/as") + + msg = leader_messages.next_coap_message("2.04") + + router1_messages.next_mle_message(mle.CommandType.ADVERTISEMENT) + router2_messages.next_mle_message(mle.CommandType.PARENT_REQUEST) leader_messages.next_mle_message(mle.CommandType.PARENT_RESPONSE) router1_messages.next_mle_message(mle.CommandType.PARENT_RESPONSE) + router2_messages.next_mle_message(mle.CommandType.CHILD_ID_REQUEST) + # Leader or Router1 can be parent of Router2 + if leader_messages.contains_mle_message(mle.CommandType.CHILD_ID_RESPONSE): + leader_messages.next_mle_message(mle.CommandType.CHILD_ID_RESPONSE) + + msg = router2_messages.next_coap_message("0.02") + msg.assertCoapMessageRequestUriPath("/a/as") + + msg = leader_messages.next_coap_message("2.04") + + elif router1_messages.contains_mle_message(mle.CommandType.CHILD_ID_RESPONSE): + router1_messages.next_mle_message(mle.CommandType.CHILD_ID_RESPONSE) + + msg = router2_messages.next_coap_message("0.02") + msg.assertCoapMessageRequestUriPath("/a/as") + + msg = router1_messages.next_coap_message("0.02") + msg.assertCoapMessageRequestUriPath("/a/as") + + msg = leader_messages.next_coap_message("2.04") + + msg = router1_messages.next_coap_message("2.04") + # 5 - Router1 # Router1 make two attempts to reconnect to its current Partition. for _ in range(2): @@ -171,6 +203,14 @@ class Cert_5_1_04_RouterAddressReallocation(unittest.TestCase): msg.assertMleMessageContainsOptionalTlv(mle.Route64) msg.assertMleMessageContainsOptionalTlv(mle.AddressRegistration) + # 10 - Router1 + msg = router1_messages.next_coap_message("2.04") + msg.assertCoapMessageContainsTlv(network_layer.Status) + msg.assertCoapMessageContainsOptionalTlv(network_layer.RouterMask) + + status_tlv = msg.get_coap_message_tlv(network_layer.Status) + self.assertEqual(network_layer.StatusValues.SUCCESS, status_tlv.status) + if __name__ == '__main__': unittest.main() diff --git a/tests/scripts/thread-cert/Cert_5_1_05_RouterAddressTimeout.py b/tests/scripts/thread-cert/Cert_5_1_05_RouterAddressTimeout.py index 95e6381bd..c967f3782 100755 --- a/tests/scripts/thread-cert/Cert_5_1_05_RouterAddressTimeout.py +++ b/tests/scripts/thread-cert/Cert_5_1_05_RouterAddressTimeout.py @@ -32,6 +32,7 @@ import unittest import config import mle +import network_layer import node LEADER = 1 @@ -112,18 +113,39 @@ class Cert_5_1_05_RouterAddressTimeout(unittest.TestCase): router1_messages.next_mle_message(mle.CommandType.PARENT_REQUEST) router1_messages.next_mle_message(mle.CommandType.CHILD_ID_REQUEST) + msg = router1_messages.next_coap_message("0.02") + msg.assertCoapMessageRequestUriPath("/a/as") + # 4 - Leader leader_messages.next_mle_message(mle.CommandType.PARENT_RESPONSE) leader_messages.next_mle_message(mle.CommandType.CHILD_ID_RESPONSE) + msg = leader_messages.next_coap_message("2.04") + msg.assertCoapMessageContainsTlv(network_layer.Status) + msg.assertCoapMessageContainsOptionalTlv(network_layer.RouterMask) + + status_tlv = msg.get_coap_message_tlv(network_layer.Status) + self.assertEqual(network_layer.StatusValues.SUCCESS, status_tlv.status) + # 6 - Router1 router1_messages.next_mle_message(mle.CommandType.LINK_REQUEST) router1_messages.next_mle_message(mle.CommandType.PARENT_REQUEST) router1_messages.next_mle_message(mle.CommandType.CHILD_ID_REQUEST) + msg = router1_messages.next_coap_message("0.02") + msg.assertCoapMessageRequestUriPath("/a/as") + # 7 - Leader leader_messages.next_mle_message(mle.CommandType.PARENT_RESPONSE) leader_messages.next_mle_message(mle.CommandType.CHILD_ID_RESPONSE) + msg = leader_messages.next_coap_message("2.04") + msg.assertCoapMessageContainsTlv(network_layer.Status) + msg.assertCoapMessageContainsOptionalTlv(network_layer.RouterMask) + + status_tlv = msg.get_coap_message_tlv(network_layer.Status) + self.assertEqual(network_layer.StatusValues.SUCCESS, status_tlv.status) + + if __name__ == '__main__': unittest.main() diff --git a/tests/scripts/thread-cert/Cert_5_1_06_RemoveRouterId.py b/tests/scripts/thread-cert/Cert_5_1_06_RemoveRouterId.py index 291adad67..7ff710173 100755 --- a/tests/scripts/thread-cert/Cert_5_1_06_RemoveRouterId.py +++ b/tests/scripts/thread-cert/Cert_5_1_06_RemoveRouterId.py @@ -32,6 +32,7 @@ import unittest import config import mle +import network_layer import node LEADER = 1 @@ -100,6 +101,11 @@ class Cert_5_1_06_RemoveRouterId(unittest.TestCase): router1_messages.next_mle_message(mle.CommandType.CHILD_ID_REQUEST) leader_messages.next_mle_message(mle.CommandType.CHILD_ID_RESPONSE) + msg = router1_messages.next_coap_message("0.02") + msg.assertCoapMessageRequestUriPath("/a/as") + + msg = leader_messages.next_coap_message("2.04") + # 3 - Router1 router1_messages.next_mle_message(mle.CommandType.PARENT_REQUEST) leader_messages.next_mle_message(mle.CommandType.PARENT_RESPONSE) @@ -107,6 +113,17 @@ class Cert_5_1_06_RemoveRouterId(unittest.TestCase): router1_messages.next_mle_message(mle.CommandType.CHILD_ID_REQUEST) msg = leader_messages.next_mle_message(mle.CommandType.CHILD_ID_RESPONSE) msg.assertSentToNode(self.nodes[ROUTER1]) + + msg = router1_messages.next_coap_message(code="0.02", uri_path="/a/as") + msg.assertCoapMessageContainsTlv(network_layer.MacExtendedAddress) + msg.assertCoapMessageContainsTlv(network_layer.Status) + + msg = leader_messages.next_coap_message("2.04") + msg.assertCoapMessageContainsTlv(network_layer.Status) + msg.assertCoapMessageContainsOptionalTlv(network_layer.RouterMask) + + status_tlv = msg.get_coap_message_tlv(network_layer.Status) + self.assertEqual(network_layer.StatusValues.SUCCESS, status_tlv.status) if __name__ == '__main__': diff --git a/tests/scripts/thread-cert/Cert_5_1_10_RouterAttachLinkQuality.py b/tests/scripts/thread-cert/Cert_5_1_10_RouterAttachLinkQuality.py index 97fae4179..93416364f 100755 --- a/tests/scripts/thread-cert/Cert_5_1_10_RouterAttachLinkQuality.py +++ b/tests/scripts/thread-cert/Cert_5_1_10_RouterAttachLinkQuality.py @@ -110,6 +110,29 @@ class Cert_5_1_10_RouterAttachLinkQuality(unittest.TestCase): # 1 - Leader, Router1, Router2 leader_messages.next_mle_message(mle.CommandType.ADVERTISEMENT) + + router1_messages.next_mle_message(mle.CommandType.PARENT_REQUEST) + leader_messages.next_mle_message(mle.CommandType.PARENT_RESPONSE) + + router1_messages.next_mle_message(mle.CommandType.CHILD_ID_REQUEST) + leader_messages.next_mle_message(mle.CommandType.CHILD_ID_RESPONSE) + + msg = router1_messages.next_coap_message("0.02") + msg.assertCoapMessageRequestUriPath("/a/as") + + msg = leader_messages.next_coap_message("2.04") + + router2_messages.next_mle_message(mle.CommandType.PARENT_REQUEST) + leader_messages.next_mle_message(mle.CommandType.PARENT_RESPONSE) + + router2_messages.next_mle_message(mle.CommandType.CHILD_ID_REQUEST) + leader_messages.next_mle_message(mle.CommandType.CHILD_ID_RESPONSE) + + msg = router2_messages.next_coap_message("0.02") + msg.assertCoapMessageRequestUriPath("/a/as") + + msg = leader_messages.next_coap_message("2.04") + router1_messages.next_mle_message(mle.CommandType.ADVERTISEMENT) router2_messages.next_mle_message(mle.CommandType.ADVERTISEMENT) diff --git a/tests/scripts/thread-cert/Cert_5_1_11_REEDAttachLinkQuality.py b/tests/scripts/thread-cert/Cert_5_1_11_REEDAttachLinkQuality.py index a87edcb68..9b8a55721 100755 --- a/tests/scripts/thread-cert/Cert_5_1_11_REEDAttachLinkQuality.py +++ b/tests/scripts/thread-cert/Cert_5_1_11_REEDAttachLinkQuality.py @@ -105,12 +105,30 @@ class Cert_5_1_11_REEDAttachLinkQuality(unittest.TestCase): self.assertEqual(self.nodes[REED].get_state(), 'router') leader_messages = self.sniffer.get_messages_sent_by(LEADER) - router1_messages = self.sniffer.get_messages_sent_by(ROUTER1) reed_messages = self.sniffer.get_messages_sent_by(REED) + router1_messages = self.sniffer.get_messages_sent_by(ROUTER1) router2_messages = self.sniffer.get_messages_sent_by(ROUTER2) # 1 - Leader. REED1, Router2 leader_messages.next_mle_message(mle.CommandType.ADVERTISEMENT) + + reed_messages.next_mle_message(mle.CommandType.PARENT_REQUEST) + leader_messages.next_mle_message(mle.CommandType.PARENT_RESPONSE) + + reed_messages.next_mle_message(mle.CommandType.CHILD_ID_REQUEST) + leader_messages.next_mle_message(mle.CommandType.CHILD_ID_RESPONSE) + + router2_messages.next_mle_message(mle.CommandType.PARENT_REQUEST) + leader_messages.next_mle_message(mle.CommandType.PARENT_RESPONSE) + + router2_messages.next_mle_message(mle.CommandType.CHILD_ID_REQUEST) + leader_messages.next_mle_message(mle.CommandType.CHILD_ID_RESPONSE) + + msg = router2_messages.next_coap_message("0.02") + msg.assertCoapMessageRequestUriPath("/a/as") + + msg = leader_messages.next_coap_message("2.04") + reed_messages.next_mle_message(mle.CommandType.ADVERTISEMENT) router2_messages.next_mle_message(mle.CommandType.ADVERTISEMENT) diff --git a/tests/scripts/thread-cert/Cert_5_1_13_RouterReset.py b/tests/scripts/thread-cert/Cert_5_1_13_RouterReset.py index 57fc7bc13..c9ec20dd5 100755 --- a/tests/scripts/thread-cert/Cert_5_1_13_RouterReset.py +++ b/tests/scripts/thread-cert/Cert_5_1_13_RouterReset.py @@ -99,6 +99,11 @@ class Cert_5_1_13_RouterReset(unittest.TestCase): router1_messages.next_mle_message(mle.CommandType.CHILD_ID_REQUEST) leader_messages.next_mle_message(mle.CommandType.CHILD_ID_RESPONSE) + msg = router1_messages.next_coap_message("0.02") + msg.assertCoapMessageRequestUriPath("/a/as") + + msg = leader_messages.next_coap_message("2.04") + router1_messages.next_mle_message(mle.CommandType.LINK_REQUEST) msg = leader_messages.next_mle_message_of_one_of_command_types(mle.CommandType.LINK_ACCEPT_AND_REQUEST, mle.CommandType.LINK_ACCEPT) diff --git a/tests/scripts/thread-cert/coap.py b/tests/scripts/thread-cert/coap.py new file mode 100644 index 000000000..d626024fa --- /dev/null +++ b/tests/scripts/thread-cert/coap.py @@ -0,0 +1,387 @@ +#!/usr/bin/python +# +# Copyright (c) 2016, The OpenThread Authors. +# All rights reserved. +# +# Redistribution and use in source and binary forms, with or without +# modification, are permitted provided that the following conditions are met: +# 1. Redistributions of source code must retain the above copyright +# notice, this list of conditions and the following disclaimer. +# 2. Redistributions in binary form must reproduce the above copyright +# notice, this list of conditions and the following disclaimer in the +# documentation and/or other materials provided with the distribution. +# 3. Neither the name of the copyright holder nor the +# names of its contributors may be used to endorse or promote products +# derived from this software without specific prior written permission. +# +# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" +# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE +# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE +# ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE +# LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR +# CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF +# SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS +# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN +# CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) +# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE +# POSSIBILITY OF SUCH DAMAGE. +# + +import collections +import io +import struct + +from binascii import hexlify +from enum import IntEnum + + +class CoapMessageType(IntEnum): + CON = 0 # Confirmable + NON = 1 # Non-confirmable + ACK = 2 # Acknowledgement + RST = 3 # Reset + + +class CoapOptionsTypes(IntEnum): + IF_MATCH = 1 + URI_HOST = 3 + ETAG = 4 + IF_NOT_MATCH = 5 + URI_PORT = 7 + LOCATION_PATH = 8 + URI_PATH = 11 + CONTENT_FORMAT = 12 + MAX_AGE = 14 + URI_QUERY = 15 + ACCEPT = 17 + LOCATION_QUERY = 20 + PROXY_URI = 35 + PROXY_SCHEME = 39 + SIZE1 = 60 + + +class CoapOptionHeader(object): + + """ Class representing CoAP optiona header. """ + + def __init__(self, delta, length): + self._delta = delta + self._length = length + + @property + def delta(self): + return self._delta + + @property + def length(self): + return self._length + + @property + def is_payload_marker(self): + return self.delta == 0xf and self.length == 0xf + + @classmethod + def _read_extended_value(cls, data, value): + if value == 13: + return ord(data.read(1)) + 13 + elif value == 14: + data.read(1) + return ord(data.read(1)) + 269 + else: + return value + + @classmethod + def from_bytes(cls, data): + initial_byte = ord(data.read(1)) + + delta = (initial_byte >> 4) & 0xf + length = initial_byte & 0xf + + delta = cls._read_extended_value(data, delta) + length = cls._read_extended_value(data, length) + + return cls(delta, length) + + +class CoapOption(object): + + """ Class representing CoAP option. """ + + def __init__(self, _type, value): + self._type = _type + self._value = value + + @property + def type(self): + return self._type + + @property + def value(self): + return self._value + + def __repr__(self): + return "CoapOption(type={}, value={})".format(self.type, hexlify(self.value)) + + +class CoapOptionsFactory(object): + + """ Factory that produces CoAP options. """ + + def parse(self, data, message_info): + options = [] + + _type = 0 + while data.tell() < len(data.getvalue()): + option_header = CoapOptionHeader.from_bytes(data) + if option_header.is_payload_marker: + break + + _type += option_header.delta + value = data.read(option_header.length) + + option = CoapOption(_type, value) + options.append(option) + + return options + + +class CoapCode(object): + + """ Class representing CoAP code. """ + + def __init__(self, code): + self._code = code + + @property + def code(self): + return self._code + + @property + def _class(self): + return (self.code >> 5) & 0x7 + + @property + def detail(self): + return self.code & 0x1f + + @classmethod + def from_class_and_detail(cls, _class, detail): + return cls(((_class & 0x7) << 5) | (detail & 0x1f)) + + @classmethod + def from_dotted(cls, dotted_str): + _class, detail = dotted_str.split(".") + return cls.from_class_and_detail(int(_class), int(detail)) + + def is_equal_dotted(self, dotted_code): + other = self.from_dotted(dotted_code) + return self.code == other.code + + @property + def dotted(self): + return ".".join(["{:01d}".format(self._class), "{:02d}".format(self.detail)]) + + def __eq__(self, other): + if isinstance(other, int): + return self.code == other + + elif isinstance(other, str): + return self.is_equal_dotted(other) + + elif isinstance(other, self.__class__): + return self.code == other.code + + else: + raise TypeError("Could not compare {} and {}".format(type(self), type(other))) + + def __repr__(self): + return self.dotted + + +class CoapMessage(object): + + """ Class representing CoAP message. """ + + def __init__(self, version, _type, code, message_id, token, options, payload, uri_path=None): + self._version = version + self._type = _type + self._code = code + self._message_id = message_id + self._token = token + self._options = options + self._payload = payload + self._uri_path = uri_path + + @property + def version(self): + return self._version + + @property + def type(self): + return self._type + + @property + def code(self): + return self._code + + @property + def message_id(self): + return self._message_id + + @property + def token(self): + return self._token + + @property + def tkl(self): + return len(self._token) + + @property + def options(self): + return self._options + + @property + def payload(self): + return self._payload + + @property + def uri_path(self): + return self._uri_path + + def __repr__(self): + options_str = ", ".join([repr(opt) for opt in self.options]) + return "CoapMessage(version={}, type={}, code={}, message_id={}, token={}, options=[{}], payload={}, uri-path='{}')".format( + self.version, CoapMessageType.name[self.type], self.code, self.message_id, hexlify(self.token), + options_str, self.payload, self.uri_path) + + +class CoapMessageProxy(object): + + """ Proxy class of CoAP message. + + The main idea behind this class is to delay parsing payload. Due to architecture of the existing solution + it is possible to process confirmation message before a request message. In such case it is not possible + to get URI path to get proper payload parser. + """ + + def __init__(self, coap_message, message_info, mid_to_uri_path_binder, uri_path_based_payload_factories): + self._coap_message = coap_message + self._message_info = message_info + self._mid_to_uri_path_binder = mid_to_uri_path_binder + self._uri_path_based_payload_factories = uri_path_based_payload_factories + + @property + def version(self): + return self._coap_message.version + + @property + def type(self): + return self._coap_message.type + + @property + def code(self): + return self._coap_message.code + + @property + def message_id(self): + return self._coap_message.message_id + + @property + def token(self): + return self._coap_message.token + + @property + def tkl(self): + return self._coap_message.tkl + + @property + def options(self): + return self._coap_message.options + + @property + def payload(self): + try: + binded_uri_path = self._mid_to_uri_path_binder.get_uri_path_for(self.message_id, self.token) + + factory = self._uri_path_based_payload_factories[binded_uri_path] + + return factory.parse(io.BytesIO(self._coap_message.payload), self._message_info) + + except RuntimeError: + return self._coap_message.payload + + @property + def uri_path(self): + return self._coap_message.uri_path + + def __repr__(self): + options_str = ", ".join([repr(opt) for opt in self.options]) + return "CoapMessageProxy(version={}, type={}, code={}, message_id={}, token={}, options=[{}], payload={}, uri-path='{}')".format( + self.version, self.type, self.code, self.message_id, hexlify(self.token), + options_str, self.payload, self.uri_path) + + +class CoapMessageIdToUriPathBinder: + + """ Class binds message id and token with URI path. """ + + def __init__(self): + self._uri_path_binds = collections.defaultdict(collections.defaultdict) + + def add_uri_path_for(self, message_id, token, uri_path): + self._uri_path_binds[message_id][hexlify(token)] = uri_path + + def get_uri_path_for(self, message_id, token): + try: + return self._uri_path_binds[message_id][hexlify(token)] + except KeyError: + raise RuntimeError("Could not find URI PATH for message_id: {} and token: {}".format( + message_id, hexlify(token))) + + +class CoapMessageFactory(object): + + """ Factory that produces CoAP messages. """ + + def __init__(self, options_factory, uri_path_based_payload_factories, message_id_to_uri_path_binder): + self._options_factory = options_factory + self._uri_path_based_payload_factories = uri_path_based_payload_factories + self._mid_to_uri_path_binder = message_id_to_uri_path_binder + + def _uri_path_from(self, options): + uri_path_options = [] + + for option in options: + if option.type == CoapOptionsTypes.URI_PATH: + uri_path_options.append(option.value.decode("utf-8")) + + if not uri_path_options: + return None + + return "/" + "/".join(uri_path_options) + + def _parse_initial_byte(self, data, message_info): + initial_byte = ord(data.read(1)) + + version = (initial_byte >> 6) & 0x3 + _type = CoapMessageType((initial_byte >> 4) & 0x3) + token_length = initial_byte & 0xf + + return version, _type, token_length + + def parse(self, data, message_info): + version, _type, token_length = self._parse_initial_byte(data, message_info) + + code = CoapCode(ord(data.read(1))) + message_id = struct.unpack(">H", data.read(2))[0] + token = data.read(token_length) + + options = self._options_factory.parse(data, message_info) + + uri_path = self._uri_path_from(options) + if uri_path is not None: + self._mid_to_uri_path_binder.add_uri_path_for(message_id, token, uri_path) + + coap_message = CoapMessage(version, _type, code, message_id, token, options, data.read(), uri_path) + + return CoapMessageProxy(coap_message, message_info, self._mid_to_uri_path_binder, self._uri_path_based_payload_factories) diff --git a/tests/scripts/thread-cert/common.py b/tests/scripts/thread-cert/common.py index 3e5f6f388..1afc66613 100644 --- a/tests/scripts/thread-cert/common.py +++ b/tests/scripts/thread-cert/common.py @@ -27,18 +27,18 @@ # POSSIBILITY OF SUCH DAMAGE. # -from binascii import hexlify - -import ipaddress import struct import sys +from binascii import hexlify +from enum import IntEnum -def enum(*sequential, **named): - enums = dict(list(zip(sequential, list(range(len(sequential))))), **named) - names = dict((value, key) for key, value in list(enums.items())) - enums['name'] = names - return type('Enum', (), enums) +import ipaddress + + +def expect_the_same_class(self, other): + if not isinstance(other, self.__class__): + raise TypeError("Expected the same class. Got {} and {}".format(type(self), type(other))) class MessageInfo(object): @@ -56,6 +56,9 @@ class MessageInfo(object): self._source_ipv6 = None self._destination_ipv6 = None + self._src_port = None + self._dst_port = None + self.stable = None self.payload_length = 0 @@ -84,16 +87,34 @@ class MessageInfo(object): def destination_ipv6(self, value): self._destination_ipv6 = self._convert_value_to_ip_address(value) + @property + def src_port(self): + return self._src_port -class MacAddress(object): + @src_port.setter + def src_port(self, value): + self._src_port = value + @property + def dst_port(self): + return self._dst_port + + @dst_port.setter + def dst_port(self, value): + self._dst_port = value + + +class MacAddressType(IntEnum): SHORT = 0 LONG = 1 + +class MacAddress(object): + def __init__(self, mac_address, _type, big_endian=True): - if _type == self.SHORT: + if _type == MacAddressType.SHORT: length = 2 - elif _type == self.LONG: + elif _type == MacAddressType.LONG: length = 8 if not big_endian: @@ -106,10 +127,6 @@ class MacAddress(object): def type(self): return self._type - @property - def type_str(self): - return "SHORT" if self.type == self.SHORT else "LONG" - @property def mac_address(self): return self._mac_address @@ -119,9 +136,9 @@ class MacAddress(object): return struct.unpack(">H", self._mac_address)[0] def convert_to_iid(self): - if self._type == self.SHORT: + if self._type == MacAddressType.SHORT: return bytearray([0x00, 0x00, 0x00, 0xff, 0xfe, 0x00]) + self._mac_address[:2] - elif self._type == self.LONG: + elif self._type == MacAddressType.LONG: return bytearray([self._mac_address[0] ^ 0x02]) + self._mac_address[1:] else: raise RuntimeError("Could not convert to IID. Invalid MAC address type: {}".format(self._type)) @@ -131,7 +148,7 @@ class MacAddress(object): if not isinstance(eui64, bytearray): raise RuntimeError("Could not create MAC address from EUI64. Invalid data type: {}".format(type(eui64))) - return cls(eui64, MacAddress.LONG) + return cls(eui64, MacAddressType.LONG) @classmethod def from_rloc16(cls, rloc16, big_endian=True): @@ -142,10 +159,10 @@ class MacAddress(object): else: raise RuntimeError("Could not create MAC address from RLOC16. Invalid data type: {}".format(type(rloc16))) - return cls(mac_address, MacAddress.SHORT) + return cls(mac_address, MacAddressType.SHORT) def __eq__(self, other): return (self.type == other.type) and (self.mac_address == other.mac_address) def __repr__(self): - return "MacAddress(mac_address=b'{}', type={})".format(hexlify(self.mac_address), self.type_str) + return "MacAddress(mac_address=b'{}', type={})".format(hexlify(self.mac_address), MacAddressType(self._type)) diff --git a/tests/scripts/thread-cert/config.py b/tests/scripts/thread-cert/config.py index df7895d9f..5563f68b8 100644 --- a/tests/scripts/thread-cert/config.py +++ b/tests/scripts/thread-cert/config.py @@ -27,27 +27,28 @@ # POSSIBILITY OF SUCH DAMAGE. # +import coap import ipv6 import lowpan import message import mle import net_crypto import network_data +import network_layer import sniffer - DEFAULT_MASTER_KEY = bytearray([0x00, 0x11, 0x22, 0x33, 0x44, 0x55, 0x66, 0x77, 0x88, 0x99, 0xaa, 0xbb, 0xcc, 0xdd, 0xee, 0xff]) def create_default_network_data_prefix_sub_tlvs_factories(): return { - 0: network_data.HasRouteFactory( + network_data.TlvType.HAS_ROUTE: network_data.HasRouteFactory( routes_factory=network_data.RoutesFactory( route_factory=network_data.RouteFactory()) ), - 2: network_data.BorderRouterFactory(), - 3: network_data.LowpanIdFactory() + network_data.TlvType.BORDER_ROUTER: network_data.BorderRouterFactory(), + network_data.TlvType.LOWPAN_ID: network_data.LowpanIdFactory() } @@ -58,7 +59,7 @@ def create_default_network_data_prefix_sub_tlvs_factory(): def create_default_network_data_service_sub_tlvs_factories(): return { - 6: network_data.ServerFactory() + network_data.TlvType.SERVER: network_data.ServerFactory() } @@ -69,10 +70,10 @@ def create_default_network_data_service_sub_tlvs_factory(): def create_default_network_data_tlvs_factories(): return { - 1: network_data.PrefixFactory( + network_data.TlvType.PREFIX: network_data.PrefixFactory( sub_tlvs_factory=create_default_network_data_prefix_sub_tlvs_factory() ), - 5: network_data.ServiceFactory( + network_data.TlvType.SERVICE: network_data.ServiceFactory( sub_tlvs_factory=create_default_network_data_service_sub_tlvs_factory() ) } @@ -141,6 +142,46 @@ def create_default_mle_message_factory(master_key): crypto_engine=create_default_mle_crypto_engine(master_key)) +def create_deafult_network_tlvs_factories(): + return { + network_layer.TlvType.TARGET_EID: network_layer.TargetEidFactory(), + network_layer.TlvType.MAC_EXTENDED_ADDRESS: network_layer.MacExtendedAddressFactory(), + network_layer.TlvType.RLOC16: network_layer.Rloc16Factory(), + network_layer.TlvType.ML_EID: network_layer.MlEidFactory(), + network_layer.TlvType.STATUS: network_layer.StatusFactory(), + network_layer.TlvType.TIME_SINCE_LAST_TRANSACTION: network_layer.TimeSinceLastTransactionFactory(), + network_layer.TlvType.ROUTER_MASK: network_layer.RouterMaskFactory(), + network_layer.TlvType.ND_OPTION: network_layer.NdOptionFactory(), + network_layer.TlvType.ND_DATA: network_layer.NdDataFactory(), + network_layer.TlvType.THREAD_NETWORK_DATA: network_layer.ThreadNetworkDataFactory(create_default_network_data_tlvs_factory), + + # Routing information are distributed in a Thread network by MLE Routing TLV + # which is in fact MLE Route64 TLV. Thread specificaton v1.1. - Chapter 5.20 + network_layer.TlvType.MLE_ROUTING: create_default_mle_tlv_route64_factory() + } + + +def create_default_network_tlvs_factory(): + return network_layer.NetworkLayerTlvsFactory( + tlvs_factories=create_deafult_network_tlvs_factories()) + + +def create_default_uri_path_based_payload_factories(): + network_layer_tlvs_factory = create_default_network_tlvs_factory() + + return { + "/a/as": network_layer_tlvs_factory, + "/a/aq": network_layer_tlvs_factory, + "/a/an": network_layer_tlvs_factory + } + + +def create_default_coap_message_factory(): + return coap.CoapMessageFactory(options_factory=coap.CoapOptionsFactory(), + uri_path_based_payload_factories=create_default_uri_path_based_payload_factories(), + message_id_to_uri_path_binder=coap.CoapMessageIdToUriPathBinder()) + + def create_default_ipv6_hop_by_hop_options_factories(): return { 109: ipv6.MPLOptionFactory() @@ -152,35 +193,33 @@ def create_default_ipv6_hop_by_hop_options_factory(): options_factories=create_default_ipv6_hop_by_hop_options_factories()) -def create_default_ipv6_udp_dst_port_factories(master_key): +def create_default_based_on_src_dst_ports_udp_payload_factory(master_key): mle_message_factory = create_default_mle_message_factory(master_key) + coap_message_factory = create_default_coap_message_factory() - return { - 19788: mle_message_factory, - - # TODO: Improve CoAP support - 61631: ipv6.UDPBytesPayloadFactory(), - 49152: ipv6.UDPBytesPayloadFactory(), - 49153: ipv6.UDPBytesPayloadFactory(), - 49154: ipv6.UDPBytesPayloadFactory() - } + return ipv6.UdpBasedOnSrcDstPortsPayloadFactory( + src_dst_port_based_payload_factories={ + 19788: mle_message_factory, + 61631: coap_message_factory + } + ) def create_default_ipv6_icmp_body_factories(): return { - 0: ipv6.ICMPv6DestinationUnreachableFactory(), - 128: ipv6.ICMPv6EchoBodyFactory(), - 129: ipv6.ICMPv6EchoBodyFactory() + ipv6.ICMP_DESTINATION_UNREACHABLE: ipv6.ICMPv6DestinationUnreachableFactory(), + ipv6.ICMP_ECHO_REQUEST: ipv6.ICMPv6EchoBodyFactory(), + ipv6.ICMP_ECHO_RESPONSE: ipv6.ICMPv6EchoBodyFactory() } def create_default_ipv6_upper_layer_factories(master_key): return { - 17: ipv6.UDPDatagramFactory( + ipv6.IPV6_NEXT_HEADER_UDP: ipv6.UDPDatagramFactory( udp_header_factory=ipv6.UDPHeaderFactory(), - dst_port_factories=create_default_ipv6_udp_dst_port_factories(master_key) + udp_payload_factory=create_default_based_on_src_dst_ports_udp_payload_factory(master_key) ), - 58: ipv6.ICMPv6Factory( + ipv6.IPV6_NEXT_HEADER_ICMP: ipv6.ICMPv6Factory( body_factories=create_default_ipv6_icmp_body_factories() ) } @@ -188,7 +227,7 @@ def create_default_ipv6_upper_layer_factories(master_key): def create_default_lowpan_extension_headers_factories(): return { - 0: lowpan.LowpanHopByHopFactory( + ipv6.IPV6_NEXT_HEADER_HOP_BY_HOP: lowpan.LowpanHopByHopFactory( hop_by_hop_options_factory=create_default_ipv6_hop_by_hop_options_factory() ) } @@ -196,7 +235,7 @@ def create_default_lowpan_extension_headers_factories(): def create_default_ipv6_extension_headers_factories(): return { - 0: ipv6.HopByHopFactory( + ipv6.IPV6_NEXT_HEADER_HOP_BY_HOP: ipv6.HopByHopFactory( hop_by_hop_options_factory=create_default_ipv6_hop_by_hop_options_factory()) } diff --git a/tests/scripts/thread-cert/ipv6.py b/tests/scripts/thread-cert/ipv6.py index 505f078b5..961959ab9 100644 --- a/tests/scripts/thread-cert/ipv6.py +++ b/tests/scripts/thread-cert/ipv6.py @@ -53,6 +53,7 @@ UPPER_LAYER_PROTOCOLS = [ ] # ICMP Protocol codes +ICMP_DESTINATION_UNREACHABLE = 0 ICMP_ECHO_REQUEST = 128 ICMP_ECHO_RESPONSE = 129 @@ -92,11 +93,12 @@ class PacketFactory(object): """ Interface for classes that produce objects from data. """ - def parse(self, data): + def parse(self, data, message_info): """ Convert data to object. Args: - data (bytes) + data (BytesIO) + message_info (MessageInfo) """ raise NotImplementedError @@ -442,7 +444,7 @@ class IPv6Packet(ConvertibleToBytes): return ipv6_packet def __repr__(self): - return "IPv6Packet(\n\theader={})".format(self.ipv6_header) + return "IPv6Packet(header={}, upper_layer_protocol={})".format(self.ipv6_header, self.upper_layer_protocol) class UDPHeader(ConvertibleToBytes, BuildableFromBytes): @@ -953,34 +955,56 @@ class UDPHeaderFactory: return UDPHeader.from_bytes(data) +class UdpBasedOnSrcDstPortsPayloadFactory: + + # TODO: Unittests + + """ Factory that produces UDP payload. """ + + def __init__(self, src_dst_port_based_payload_factories): + """ + Args: + src_dst_port_based_payload_factories (PacketFactory): Factories parse UDP payload based on source or destination port. + """ + self._factories = src_dst_port_based_payload_factories + + def parse(self, data, message_info): + factory = None + + if message_info.dst_port in self._factories: + factory = self._factories[message_info.dst_port] + + if message_info.src_port in self._factories: + factory = self._factories[message_info.src_port] + + if factory is None: + raise RuntimeError("Could not find factory to build UDP payload.") + + return factory.parse(data, message_info) + + class UDPDatagramFactory(PacketFactory): + # TODO: Unittests + """ Factory that produces UDP datagrams. """ - def __init__(self, udp_header_factory, dst_port_factories=None): - """ - Args: - dst_port_factories (PacketFactory): Factories parse UDP payload based on destination port. - """ + def __init__(self, udp_header_factory, udp_payload_factory): self._udp_header_factory = udp_header_factory - self._dst_port_factories = dst_port_factories if dst_port_factories is not None else {} - - def _get_payload_factory(self, dst_port): - try: - return self._dst_port_factories[dst_port] - - except KeyError: - raise RuntimeError( - "Could not find factory to parse UDP datagram payload based on destination port: {}".format(dst_port)) + self._udp_payload_factory = udp_payload_factory def parse(self, data, message_info): - udp_header = self._udp_header_factory.parse(data, message_info) + header = self._udp_header_factory.parse(data, message_info) - factory = self._get_payload_factory(udp_header.dst_port) + # Update message payload length: UDP header (8B) + payload length + message_info.payload_length += len(header) + (len(data.getvalue()) - data.tell()) - message_info.payload_length += len(udp_header) + (len(data.getvalue()) - data.tell()) + message_info.src_port = header.src_port + message_info.dst_port = header.dst_port - return UDPDatagram(udp_header, factory.parse(data, message_info)) + payload = self._udp_payload_factory.parse(data, message_info) + + return UDPDatagram(header, payload) class ICMPv6Factory(PacketFactory): diff --git a/tests/scripts/thread-cert/mac802154.py b/tests/scripts/thread-cert/mac802154.py index 6a0d4c4ef..57aaf70d2 100644 --- a/tests/scripts/thread-cert/mac802154.py +++ b/tests/scripts/thread-cert/mac802154.py @@ -35,7 +35,7 @@ import io import struct import config -from common import MacAddress, MessageInfo +from common import MacAddress, MacAddressType, MessageInfo from net_crypto import AuxiliarySecurityHeader, CryptoEngine, MacCryptoMaterialCreator @@ -200,7 +200,7 @@ class MacFrame: message_info.aux_sec_hdr_bytes = aux_sec_hdr_bytes message_info.nonpayload_fields = non_payload_fields message_info.mhr_bytes = mhr_bytes - if src_address.type == MacAddress.SHORT: + if src_address.type == MacAddressType.SHORT: message_info.source_mac_address = DeviceDescriptors.get_extended(src_address).mac_address else: message_info.source_mac_address = src_address.mac_address @@ -213,10 +213,10 @@ class MacFrame: def _parse_address(self, data, mode): if mode == MacHeader.AddressMode.SHORT: - return MacAddress(data.read(2), MacAddress.SHORT, big_endian=False) + return MacAddress(data.read(2), MacAddressType.SHORT, big_endian=False) if mode == MacHeader.AddressMode.EXTENDED: - return MacAddress(data.read(8), MacAddress.LONG, big_endian=False) + return MacAddress(data.read(8), MacAddressType.LONG, big_endian=False) else: return None diff --git a/tests/scripts/thread-cert/message.py b/tests/scripts/thread-cert/message.py index b771c3ad5..287317981 100644 --- a/tests/scripts/thread-cert/message.py +++ b/tests/scripts/thread-cert/message.py @@ -32,19 +32,22 @@ import ipaddress import struct import sys +import coap import common import ipv6 import lowpan import mac802154 import mle +from enum import IntEnum -MessageType = common.enum("MLE", - "COAP", - "ICMP", - "ACK", - "BEACON", - "DATA") +class MessageType(IntEnum): + MLE = 0 + COAP = 1 + ICMP = 2 + ACK = 3 + BEACON = 4 + DATA = 5 class Message(object): @@ -54,6 +57,7 @@ class Message(object): self._channel = None self._mac_header = None self._ipv6_packet = None + self._coap = None self._mle = None self._icmp = None @@ -62,8 +66,7 @@ class Message(object): self._type = MessageType.MLE self._mle = udp_datagram.payload - # TODO: Initial version doesn't support CoAP - elif isinstance(udp_datagram.payload, ipv6.UDPBytesPayload): + elif isinstance(udp_datagram.payload, (coap.CoapMessage, coap.CoapMessageProxy)): self._type = MessageType.COAP self._coap = udp_datagram.payload @@ -117,14 +120,14 @@ class Message(object): self._ipv6_packet = value self._extract_upper_layer_protocol(value.upper_layer_protocol) + @property + def coap(self): + return self._coap + @property def mle(self): return self._mle - @mle.setter - def mle(self, value): - self._mle = value - @property def icmp(self): return self._icmp @@ -183,6 +186,50 @@ class Message(object): print("MleMessage doesn't contain optional TLV: {}".format(tlv_class_type)) + def get_coap_message_tlv(self, tlv_class_type): + if self.type != MessageType.COAP: + raise ValueError("Invalid message type. Expected CoAP message.") + + for tlv in self.coap.payload: + if isinstance(tlv, tlv_class_type): + return tlv + + def assertCoapMessageContainsTlv(self, tlv_class_type): + if self.type != MessageType.COAP: + raise ValueError("Invalid message type. Expected CoAP message.") + + contains_tlv = False + for tlv in self.coap.payload: + if isinstance(tlv, tlv_class_type): + contains_tlv = True + break + + assert(contains_tlv == True) + + def assertCoapMessageContainsOptionalTlv(self, tlv_class_type): + if self.type != MessageType.COAP: + raise ValueError("Invalid message type. Expected CoAP message.") + + contains_tlv = False + for tlv in self.coap.payload: + if isinstance(tlv, tlv_class_type): + contains_tlv = True + break + + print("CoapMessage doesn't contain optional TLV: {}".format(tlv_class_type)) + + def assertCoapMessageRequestUriPath(self, uri_path): + if self.type != MessageType.COAP: + raise ValueError("Invalid message type. Expected CoAP message.") + + assert(uri_path == self.coap.uri_path) + + def assertCoapMessageCode(self, code): + if self.type != MessageType.COAP: + raise ValueError("Invalid message type. Expected CoAP message.") + + assert(code == self.coap.code) + def assertSentToNode(self, node): sent_to_node = False dst_addr = self.ipv6_packet.ipv6_header.destination_address @@ -215,6 +262,30 @@ class MessagesSet(object): def messages(self): return self._messages + def next_coap_message(self, code, uri_path=None, assert_enabled=True): + message = None + + while self.messages: + m = self.messages.pop(0) + + if m.type != MessageType.COAP: + continue + + if uri_path is not None and m.coap.uri_path != uri_path: + continue + + else: + if not m.coap.code.is_equal_dotted(code): + continue + + message = m + break + + if assert_enabled: + assert message is not None, "Could not find CoapMessage with code: {}".format(code) + + return message + def next_mle_message(self, command_type, assert_enabled=True): message = self.next_mle_message_of_one_of_command_types(command_type,) @@ -255,6 +326,15 @@ class MessagesSet(object): return False + def does_not_contain_coap_message(self): + for m in self.messages: + if m.type != MessageType.COAP: + continue + + return False + + return True + class MessageFactory: diff --git a/tests/scripts/thread-cert/mle.py b/tests/scripts/thread-cert/mle.py index 4ad06f7df..92827f872 100644 --- a/tests/scripts/thread-cert/mle.py +++ b/tests/scripts/thread-cert/mle.py @@ -36,52 +36,56 @@ from binascii import hexlify import common import network_data -CommandType = common.enum(LINK_REQUEST=0, - LINK_ACCEPT=1, - LINK_ACCEPT_AND_REQUEST=2, - LINK_REJECT=3, - ADVERTISEMENT=4, - UPDATE=5, - UPDATE_REQUEST=6, - DATA_REQUEST=7, - DATA_RESPONSE=8, - PARENT_REQUEST=9, - PARENT_RESPONSE=10, - CHILD_ID_REQUEST=11, - CHILD_ID_RESPONSE=12, - CHILD_UPDATE_REQUEST=13, - CHILD_UPDATE_RESPONSE=14, - ANNOUNCE=15, - DISCOVERY_REQUEST=16, - DISCOVERY_RESPONSE=17 - ) +from enum import IntEnum -TlvType = common.enum(SOURCE_ADDRESS=0, - MODE=1, - TIMEOUT=2, - CHALLENGE=3, - RESPONSE=4, - LINK_LAYER_FRAME_COUNTER=5, - MLE_FRAME_COUNTER=8, - ROUTE64=9, - ADDRESS16=10, - LEADER_DATA=11, - NETWORK_DATA=12, - TLV_REQUEST=13, - SCAN_MASK=14, - CONNECTIVITY=15, - LINK_MARGIN=16, - STATUS=17, - VERSION=18, - ADDRESS_REGISTRATION=19, - CHANNEL=20, - PANID=21, - ACTIVE_TIMESTAMP=22, - PENDING_TIMESTAMP=23, - ACTIVE_OPERATIONAL_DATASET=24, - PENDING_OPERATIONAL_DATASET=25, - THREAD_DISCOVERY=26 - ) + +class CommandType(IntEnum): + LINK_REQUEST = 0 + LINK_ACCEPT = 1 + LINK_ACCEPT_AND_REQUEST = 2 + LINK_REJECT = 3 + ADVERTISEMENT = 4 + UPDATE = 5 + UPDATE_REQUEST = 6 + DATA_REQUEST = 7 + DATA_RESPONSE = 8 + PARENT_REQUEST = 9 + PARENT_RESPONSE = 10 + CHILD_ID_REQUEST = 11 + CHILD_ID_RESPONSE = 12 + CHILD_UPDATE_REQUEST = 13 + CHILD_UPDATE_RESPONSE = 14 + ANNOUNCE = 15 + DISCOVERY_REQUEST = 16 + DISCOVERY_RESPONSE = 17 + + +class TlvType(IntEnum): + SOURCE_ADDRESS = 0 + MODE = 1 + TIMEOUT = 2 + CHALLENGE = 3 + RESPONSE = 4 + LINK_LAYER_FRAME_COUNTER = 5 + MLE_FRAME_COUNTER = 8 + ROUTE64 = 9 + ADDRESS16 = 10 + LEADER_DATA = 11 + NETWORK_DATA = 12 + TLV_REQUEST = 13 + SCAN_MASK = 14 + CONNECTIVITY = 15 + LINK_MARGIN = 16 + STATUS = 17 + VERSION = 18 + ADDRESS_REGISTRATION = 19 + CHANNEL = 20 + PANID = 21 + ACTIVE_TIMESTAMP = 22 + PENDING_TIMESTAMP = 23 + ACTIVE_OPERATIONAL_DATASET = 24 + PENDING_OPERATIONAL_DATASET = 25 + THREAD_DISCOVERY = 26 class SourceAddress(object): @@ -94,8 +98,7 @@ class SourceAddress(object): return self._address def __eq__(self, other): - if not isinstance(other, self.__class__): - raise TypeError("Could not compare {} and {}".format(type(self), type(other))) + common.expect_the_same_class(self, other) return self.address == other.address @@ -135,11 +138,12 @@ class Mode(object): return self._network_data def __eq__(self, other): - if not isinstance(other, self.__class__): - raise TypeError("Could not compare {} and {}".format(type(self), type(other))) + common.expect_the_same_class(self, other) - return self.receiver == other.receiver and self.secure == other.secure and \ - self.device_type == other.device_type and self.network_data == other.network_data + return self.receiver == other.receiver and \ + self.secure == other.secure and \ + self.device_type == other.device_type and \ + self.network_data == other.network_data def __repr__(self): return "Mode(receiver={}, secure={}, device_type={}, network_data={})".format(self.receiver, @@ -169,8 +173,7 @@ class Timeout(object): return self._timeout def __eq__(self, other): - if not isinstance(other, self.__class__): - raise TypeError("Could not compare {} and {}".format(type(self), type(other))) + common.expect_the_same_class(self, other) return self.timeout == other.timeout @@ -195,13 +198,12 @@ class Challenge(object): return self._challenge def __eq__(self, other): - if not isinstance(other, self.__class__): - raise TypeError("Could not compare {} and {}".format(type(self), type(other))) + common.expect_the_same_class(self, other) return self.challenge == other.challenge def __repr__(self): - return "Challenge(challenge=b'{}')".format(hexlify(self.challenge)) + return "Challenge(challenge={})".format(hexlify(self.challenge)) class ChallengeFactory: @@ -221,13 +223,12 @@ class Response(object): return self._response def __eq__(self, other): - if not isinstance(other, Response): - raise TypeError("Could not compare {} and {}".format(type(self), type(other))) + common.expect_the_same_class(self, other) return self.response == other.response def __repr__(self): - return "Response(response=b'{}')".format(hexlify(self.response)) + return "Response(response={})".format(hexlify(self.response)) class ResponseFactory: @@ -247,8 +248,7 @@ class LinkLayerFrameCounter(object): return self._frame_counter def __eq__(self, other): - if not isinstance(other, self.__class__): - raise TypeError("Could not compare {} and {}".format(type(self), type(other))) + common.expect_the_same_class(self, other) return self.frame_counter == other.frame_counter @@ -273,8 +273,7 @@ class MleFrameCounter(object): return self._frame_counter def __eq__(self, other): - if not isinstance(other, self.__class__): - raise TypeError("Could not compare {} and {}".format(type(self), type(other))) + common.expect_the_same_class(self, other) return self.frame_counter == other.frame_counter @@ -309,8 +308,7 @@ class LinkQualityAndRouteData(object): return self._route def __eq__(self, other): - if not isinstance(other, self.__class__): - raise TypeError("Could not compare {} and {}".format(type(self), type(other))) + common.expect_the_same_class(self, other) return self.output == other.output and self.input == other.input and self.route == other.route @@ -348,10 +346,10 @@ class Route64(object): return self._link_quality_and_route_data def __eq__(self, other): - if not isinstance(other, self.__class__): - raise TypeError("Could not compare {} and {}".format(type(self), type(other))) + common.expect_the_same_class(self, other) - return self.id_sequence == other.id_sequence and self.router_id_mask == other.router_id_mask and \ + return self.id_sequence == other.id_sequence and \ + self.router_id_mask == other.router_id_mask and \ self.link_quality_and_route_data == other.link_quality_and_route_data def __repr__(self): @@ -387,8 +385,7 @@ class Address16(object): return self._address def __eq__(self, other): - if not isinstance(other, self.__class__): - raise TypeError("Could not compare {} and {}".format(type(self), type(other))) + common.expect_the_same_class(self, other) return self.address == other.address @@ -433,8 +430,7 @@ class LeaderData(object): return self._leader_router_id def __eq__(self, other): - if not isinstance(other, self.__class__): - raise TypeError("Could not compare {} and {}".format(type(self), type(other))) + common.expect_the_same_class(self, other) return self.partition_id == other.partition_id and \ self.weighting == other.weighting and \ @@ -468,8 +464,7 @@ class NetworkData(object): return self._tlvs def __eq__(self, other): - if not isinstance(other, self.__class__): - raise TypeError("Could not compare {} and {}".format(type(self), type(other))) + common.expect_the_same_class(self, other) return self.tlvs == other.tlvs @@ -498,8 +493,7 @@ class TlvRequest(object): return self._tlvs def __eq__(self, other): - if not isinstance(other, self.__class__): - raise TypeError("Could not compare {} and {}".format(type(self), type(other))) + common.expect_the_same_class(self, other) return self.tlvs == other.tlvs @@ -530,8 +524,7 @@ class ScanMask(object): return self._end_device def __eq__(self, other): - if not isinstance(other, self.__class__): - raise TypeError("Could not compare {} and {}".format(type(self), type(other))) + common.expect_the_same_class(self, other) return self.router == other.router and self.end_device == other.end_device @@ -607,8 +600,7 @@ class Connectivity(object): return self._sed_datagram_count def __eq__(self, other): - if not isinstance(other, self.__class__): - raise TypeError("Could not compare {} and {}".format(type(self), type(other))) + common.expect_the_same_class(self, other) return self.pp == other.pp and \ self.link_quality_3 == other.link_quality_3 and \ @@ -621,7 +613,7 @@ class Connectivity(object): self.sed_datagram_count == other.sed_datagram_count def __repr__(self): - return "Connectivity(pp={}, \ + return r"Connectivity(pp={}, \ link_quality_3={}, \ link_quality_2={}, \ link_quality_1={}, \ @@ -683,8 +675,7 @@ class LinkMargin(object): return self._link_margin def __eq__(self, other): - if not isinstance(other, self.__class__): - raise TypeError("Could not compare {} and {}".format(type(self), type(other))) + common.expect_the_same_class(self, other) return self.link_margin == other.link_margin @@ -709,8 +700,7 @@ class Status(object): return self._status def __eq__(self, other): - if not isinstance(other, self.__class__): - raise TypeError("Could not compare {} and {}".format(type(self), type(other))) + common.expect_the_same_class(self, other) return self.status == other.status @@ -735,8 +725,7 @@ class Version(object): return self._version def __eq__(self, other): - if not isinstance(other, self.__class__): - raise TypeError("Could not compare {} and {}".format(type(self), type(other))) + common.expect_the_same_class(self, other) return self.version == other.version @@ -761,13 +750,12 @@ class AddressFull(object): return self._ipv6_address def __eq__(self, other): - if not isinstance(other, self.__class__): - raise TypeError("Could not compare {} and {}".format(type(self), type(other))) + common.expect_the_same_class(self, other) return self.ipv6_address == other.ipv6_address def __repr__(self): - return "AddressFull(ipv6_address=b'{}'')".format(hexlify(self.ipv6_address)) + return "AddressFull(ipv6_address={}')".format(hexlify(self.ipv6_address)) class AddressFullFactory: @@ -793,13 +781,12 @@ class AddressCompressed(object): return self._iid def __eq__(self, other): - if not isinstance(other, self.__class__): - raise TypeError("Could not compare {} and {}".format(type(self), type(other))) + common.expect_the_same_class(self, other) return self.cid == other.cid and self.iid == other.iid def __repr__(self): - return "AddressCompressed(cid={}, iid=b'{}'')".format(self.cid, hexlify(self.iid)) + return "AddressCompressed(cid={}, iid={}')".format(self.cid, hexlify(self.iid)) class AddressCompressedFactory: @@ -820,8 +807,7 @@ class AddressRegistration(object): return self._addresses def __eq__(self, other): - if not isinstance(other, self.__class__): - raise TypeError("Could not compare {} and {}".format(type(self), type(other))) + common.expect_the_same_class(self, other) return self.addresses == other.addresses @@ -866,8 +852,7 @@ class Channel(object): return self._channel def __eq__(self, other): - if not isinstance(other, self.__class__): - raise TypeError("Could not compare {} and {}".format(type(self), type(other))) + common.expect_the_same_class(self, other) return self.channel_page == other.channel_page and \ self.channel == other.channel @@ -894,8 +879,7 @@ class PanId: return self._pan_id def __eq__(self, other): - if not isinstance(other, self.__class__): - raise TypeError("Could not compare {} and {}".format(type(self), type(other))) + common.expect_the_same_class(self, other) return self.pan_id == other.pan_id @@ -930,8 +914,7 @@ class ActiveTimestamp(object): return self._u def __eq__(self, other): - if not isinstance(other, self.__class__): - raise TypeError("Could not compare {} and {}".format(type(self), type(other))) + common.expect_the_same_class(self, other) return self.timestamp_seconds == other.timestamp_seconds and \ self.timestamp_ticks == other.timestamp_ticks and \ @@ -974,8 +957,7 @@ class PendingTimestamp(object): return self._u def __eq__(self, other): - if not isinstance(other, self.__class__): - raise TypeError("Could not compare {} and {}".format(type(self), type(other))) + common.expect_the_same_class(self, other) return self.timestamp_seconds == other.timestamp_seconds and \ self.timestamp_ticks == other.timestamp_ticks and \ @@ -1053,7 +1035,7 @@ class MleCommand(object): def __repr__(self): tlvs_str = ", ".join(["{}".format(tlv) for tlv in self.tlvs]) - return "MleCommand(type={}, tlvs=[{}])".format(self.type, tlvs_str) + return "MleCommand(type={}, tlvs=[{}])".format(self.type.name, tlvs_str) class MleCommandFactory: @@ -1078,7 +1060,7 @@ class MleCommandFactory: raise KeyError("Could not find TLV factory. Unsupported TLV type: {}".format(_type)) def _parse_tlv(self, data, message_info): - _type = ord(data.read(1)) + _type = TlvType(ord(data.read(1))) length = self._get_length(data) value = data.read(length) @@ -1087,7 +1069,7 @@ class MleCommandFactory: return factory.parse(io.BytesIO(value), message_info) def parse(self, data, message_info): - cmd_type = ord(data.read(1)) + cmd_type = CommandType(ord(data.read(1))) tlvs = [] while data.tell() < len(data.getvalue()): diff --git a/tests/scripts/thread-cert/network_data.py b/tests/scripts/thread-cert/network_data.py index 6258f6d70..37309ae2e 100644 --- a/tests/scripts/thread-cert/network_data.py +++ b/tests/scripts/thread-cert/network_data.py @@ -32,6 +32,19 @@ import math import struct from binascii import hexlify +from enum import IntEnum + +import common + + +class TlvType(IntEnum): + HAS_ROUTE = 0 + PREFIX = 1 + BORDER_ROUTER = 2 + LOWPAN_ID = 3 + COMMISSIONING = 4 + SERVICE = 5 + SERVER = 6 class NetworkData(object): @@ -92,8 +105,7 @@ class Route(object): return self._prf def __eq__(self, other): - if not isinstance(other, self.__class__): - raise TypeError("Could not compare {} and {}".format(type(self), type(other))) + common.expect_the_same_class(self, other) return self.border_router_16 == other.border_router_16 and self.prf == other.prf @@ -139,8 +151,7 @@ class HasRoute(NetworkData): return self._routes def __eq__(self, other): - if not isinstance(other, self.__class__): - raise TypeError("Could not compare {} and {}".format(type(self), type(other))) + common.expect_the_same_class(self, other) return self.routes == other.routes @@ -186,8 +197,7 @@ class Prefix(NetworkData): return self._sub_tlvs def __eq__(self, other): - if not isinstance(other, self.__class__): - raise TypeError("Could not compare {} and {}".format(type(self), type(other))) + common.expect_the_same_class(self, other) return self.domain_id == other.domain_id and \ self.prefix_length == other.prefix_length and \ @@ -277,8 +287,7 @@ class BorderRouter(NetworkData): return self._n def __eq__(self, other): - if not isinstance(other, self.__class__): - raise TypeError("Could not compare {} and {}".format(type(self), type(other))) + common.expect_the_same_class(self, other) return self.border_router_16 == other.border_router_16 and \ self.prf == other.prf and \ @@ -336,8 +345,7 @@ class LowpanId(NetworkData): return self._context_length def __eq__(self, other): - if not isinstance(other, self.__class__): - raise TypeError("Could not compare {} and {}".format(type(self), type(other))) + common.expect_the_same_class(self, other) return self.c == other.c and \ self.cid == other.cid and \ @@ -411,8 +419,7 @@ class Service(NetworkData): return self._sub_tlvs def __eq__(self, other): - if not isinstance(other, self.__class__): - raise TypeError("Could not compare {} and {}".format(type(self), type(other))) + common.expect_the_same_class(self, other) return self.t == other.t and \ self.id == other.id and \ @@ -468,14 +475,13 @@ class Server(NetworkData): return self._server_data def __eq__(self, other): - if not isinstance(other, self.__class__): - raise TypeError("Could not compare {} and {}".format(type(self), type(other))) + common.expect_the_same_class(self, other) return self.server_16 == other.server_16 and \ self.server_data == other.server_data def __repr__(self): - return "LowpanId(stable={}, server_16={}, server_data=\"{}\")".format( + return "LowpanId(stable={}, server_16={}, server_data=b'{}')".format( self.stable, self.server_16, hexlify(self.server_data)) diff --git a/tests/scripts/thread-cert/network_layer.py b/tests/scripts/thread-cert/network_layer.py new file mode 100644 index 000000000..ef52c3aea --- /dev/null +++ b/tests/scripts/thread-cert/network_layer.py @@ -0,0 +1,322 @@ +#!/usr/bin/python +# +# Copyright (c) 2016, The OpenThread Authors. +# All rights reserved. +# +# Redistribution and use in source and binary forms, with or without +# modification, are permitted provided that the following conditions are met: +# 1. Redistributions of source code must retain the above copyright +# notice, this list of conditions and the following disclaimer. +# 2. Redistributions in binary form must reproduce the above copyright +# notice, this list of conditions and the following disclaimer in the +# documentation and/or other materials provided with the distribution. +# 3. Neither the name of the copyright holder nor the +# names of its contributors may be used to endorse or promote products +# derived from this software without specific prior written permission. +# +# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" +# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE +# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE +# ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE +# LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR +# CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF +# SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS +# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN +# CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) +# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE +# POSSIBILITY OF SUCH DAMAGE. +# + +import io +import struct + +from binascii import hexlify +from enum import IntEnum + +import common + + +class TlvType(IntEnum): + TARGET_EID = 0 + MAC_EXTENDED_ADDRESS = 1 + RLOC16 = 2 + ML_EID = 3 + STATUS = 4 + TIME_SINCE_LAST_TRANSACTION = 6 + ROUTER_MASK = 7 + ND_OPTION = 8 + ND_DATA = 9 + THREAD_NETWORK_DATA = 10 + MLE_ROUTING = 11 + + +class StatusValues(IntEnum): + SUCCESS = 0 + NO_ADDRESS_AVAILABLE = 1 + TOO_FEW_ROUTERS = 2 + HAVE_CHILD_ID_REQUEST = 3 + PARENT_PARTITION_CHANGE = 4 + + +class TargetEid(object): + + def __init__(self, eid): + self._eid = eid + + @property + def eid(self): + return self._eid + + def __eq__(self, other): + common.expect_the_same_class(self, other) + return self.eid == other.eid + + def __repr__(self): + return "TargetEid(eid={})".format(hexlify(self.eid)) + + +class TargetEidFactory(object): + + def parse(self, data, message_info): + eid = bytearray(data.read(16)) + + return TargetEid(eid) + + +class MacExtendedAddress(object): + + def __init__(self, mac_address): + self._mac_address = mac_address + + @property + def mac_address(self): + return self._mac_address + + def __eq__(self, other): + common.expect_the_same_class(self, other) + return self.mac_address == other.mac_address + + def __repr__(self): + return "MacExtendedAddress(mac_address={})".format(hexlify(self.mac_address)) + + +class MacExtendedAddressFactory(object): + + def parse(self, data, message_info): + mac_address = bytearray(data.read(8)) + + return MacExtendedAddress(mac_address) + + +class Rloc16(object): + + def __init__(self, rloc16): + self._rloc16 = rloc16 + + @property + def rloc16(self): + return self._rloc16 + + def __eq__(self, other): + common.expect_the_same_class(self, other) + return self.rloc16 == other.rloc16 + + def __repr__(self): + return "Rloc16(rloc16={})".format(hex(self.rloc16)) + + +class Rloc16Factory(object): + + def parse(self, data, message_info): + rloc16 = struct.unpack(">H", data.read(2))[0] + + return Rloc16(rloc16) + + +class MlEid(object): + + def __init__(self, ml_eid): + self._ml_eid = ml_eid + + @property + def ml_eid(self): + return self._ml_eid + + def __eq__(self, other): + common.expect_the_same_class(self, other) + return self.ml_eid == other.ml_eid + + def __repr__(self): + return "MlEid(ml_eid={})".format(hexlify(self.ml_eid)) + + +class MlEidFactory(object): + + def parse(self, data, message_info): + ml_eid = bytearray(data.read(8)) + + return MlEid(ml_eid) + + +class Status(object): + + def __init__(self, status): + self._status = status + + @property + def status(self): + return self._status + + def __eq__(self, other): + common.expect_the_same_class(self, other) + return self.status == other.status + + def __repr__(self): + return "Status(status={})".format(self.status) + + +class StatusFactory(object): + + def parse(self, data, message_info): + status = StatusValues(ord(data.read(1))) + + return Status(status) + + +class TimeSinceLastTransaction(object): + + def __init__(self, seconds): + self._seconds = seconds + + @property + def seconds(self): + return self._seconds + + def __eq__(self, other): + common.expect_the_same_class(self, other) + return self.seconds == other.seconds + + def __repr__(self): + return "TimeSinceLastTransaction(seconds={})".format(self.seconds) + + +class TimeSinceLastTransactionFactory(object): + + def parse(self, data, message_info): + seconds = struct.unpack(">L", data.read(4))[0] + + return TimeSinceLastTransaction(seconds) + + +class RouterMask(object): + + def __init__(self, id_sequence, router_id_mask): + self._id_sequence = id_sequence + self._router_id_mask = router_id_mask + + @property + def id_sequence(self): + return self._id_sequence + + @property + def router_id_mask(self): + return self._router_id_mask + + def __eq__(self, other): + common.expect_the_same_class(self, other) + return self.id_sequence == other.id_sequence and self.router_id_mask == other.router_id_mask + + def __repr__(self): + return "RouterMask(id_sequence={}, router_id_mask={})".format(self.id_sequence, hex(self.router_id_mask)) + + +class RouterMaskFactory(object): + + def parse(self, data, message_info): + id_sequence = ord(data.read(1)) + router_id_mask = struct.unpack(">Q", data.read(8))[0] + + return RouterMask(id_sequence, router_id_mask) + + +class NdOption(object): + + def __init__(self, options): + self._options = options + + @property + def options(self): + return self._options + + def __eq__(self, other): + common.expect_the_same_class(self, other) + return self.options == other.options + + def __repr__(self): + return "NdOption(options=[{}])".format(", ".join([str(opt) for opt in self.options])) + + +class NdOptionFactory(object): + + def parse(self, data, message_info): + options = [opt for opt in bytearray(data.read())] + return NdOption(options) + + +class NdData(object): + # TODO: Not implemented yet + pass + + +class NdDataFactory(object): + # TODO: Not implemented yet + + def parse(self, data, message_info): + raise NotImplementedError("TODO: Not implemented yet") + + +class ThreadNetworkData(object): + + def __init__(self, tlvs): + self._tlvs = tlvs + + @property + def tlvs(self): + return self._tlvs + + def __eq__(self, other): + common.expect_the_same_class(self, other) + return self.tlvs == other.tlvs + + def __repr__(self): + return "ThreadNetworkData(tlvs=[{}])".format(", ".join([str(tlv) for tlv in self.tlvs])) + + +class ThreadNetworkDataFactory(object): + + def __init__(self, network_data_tlvs_factory): + self._network_data_tlvs_factory = network_data_tlvs_factory + + def parse(self, data, message_info): + tlvs = self._network_data_tlvs_factory.parse(data, message_info) + return ThreadNetworkData(tlvs) + + +class NetworkLayerTlvsFactory(object): + + def __init__(self, tlvs_factories): + self._tlvs_factories = tlvs_factories + + def parse(self, data, message_info): + tlvs = [] + + while data.tell() < len(data.getvalue()): + _type = ord(data.read(1)) + length = ord(data.read(1)) + + factory = self._tlvs_factories[_type] + tlv = factory.parse(io.BytesIO(data.read(length)), message_info) + + tlvs.append(tlv) + + return tlvs diff --git a/tests/scripts/thread-cert/sniffer.py b/tests/scripts/thread-cert/sniffer.py index 77c667801..5b51b6ff5 100644 --- a/tests/scripts/thread-cert/sniffer.py +++ b/tests/scripts/thread-cert/sniffer.py @@ -49,7 +49,7 @@ class Sniffer: logger = logging.getLogger("sniffer.Sniffer") - POLL_TIMEOUT = 0.11 + POLL_TIMEOUT = 0.10 RECV_BUFFER_SIZE = 4096 diff --git a/tests/scripts/thread-cert/test_coap.py b/tests/scripts/thread-cert/test_coap.py new file mode 100755 index 000000000..552b725ae --- /dev/null +++ b/tests/scripts/thread-cert/test_coap.py @@ -0,0 +1,524 @@ +#!/usr/bin/python +# +# Copyright (c) 2016, The OpenThread Authors. +# All rights reserved. +# +# Redistribution and use in source and binary forms, with or without +# modification, are permitted provided that the following conditions are met: +# 1. Redistributions of source code must retain the above copyright +# notice, this list of conditions and the following disclaimer. +# 2. Redistributions in binary form must reproduce the above copyright +# notice, this list of conditions and the following disclaimer in the +# documentation and/or other materials provided with the distribution. +# 3. Neither the name of the copyright holder nor the +# names of its contributors may be used to endorse or promote products +# derived from this software without specific prior written permission. +# +# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" +# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE +# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE +# ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE +# LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR +# CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF +# SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS +# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN +# CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) +# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE +# POSSIBILITY OF SUCH DAMAGE. +# + +import io +import random +import string +import unittest + +import coap +import config + + +def any_delta(): + return random.getrandbits(4) + + +def any_coap_option_type(): + return random.getrandbits(4) + + +def any_value(): + return random.getrandbits(8) + + +def any_4bits_value_different_than_13_and_14(): + value = None + while value is None: + value = random.getrandbits(4) + if value == 13 or value == 14: + value = None + + return value + + +def any_4bits_value_lower_or_equal_than_12(): + value = None + while value is None: + value = random.getrandbits(4) + if value > 12: + value = None + + return value + + +def any_bytearray(length): + return bytearray([random.getrandbits(8) for _ in range(length)]) + + +def any_version(): + return random.getrandbits(2) + + +def any_type(): + return random.getrandbits(2) + + +def any_code(): + return random.getrandbits(8) + + +def any_message_id(): + return random.getrandbits(16) + + +def any_token(): + length = random.randint(0, 8) + return bytearray([random.getrandbits(8) for _ in range(length)]) + + +def any_options(): + return [] + + +def any_payload(length=None): + length = length if length is not None else random.randint(0, 64) + return bytearray([random.getrandbits(8) for _ in range(length)]) + + +def any_uri_path(): + return "/" + random.choice(string.ascii_lowercase) + + +class TestCoapMessageOptionHeader(unittest.TestCase): + + def test_should_return_passed_on_value_when_read_extended_value_is_called_with_value_different_than_13_and_14(self): + # GIVEN + value = any_4bits_value_different_than_13_and_14() + + # WHEN + actual_value = coap.CoapOptionHeader._read_extended_value(None, value) + + # THEN + self.assertEqual(value, actual_value) + + def test_should_return_value_stored_in_first_byte_plus_13_when_read_extended_value_is_called_with_value_equal_13(self): + # GIVEN + value = 13 + extended_value = any_value() + + data = io.BytesIO(bytearray([extended_value])) + + # WHEN + actual_value = coap.CoapOptionHeader._read_extended_value(data, value) + + # THEN + self.assertEqual(extended_value + 13, actual_value) + + def test_should_return_value_stored_in_first_byte_plus_269_when_read_extended_value_is_called_with_value_equal_14(self): + # GIVEN + value = 14 + extended_value = any_value() + + data = io.BytesIO(bytearray([any_value(), extended_value])) + + # WHEN + actual_value = coap.CoapOptionHeader._read_extended_value(data, value) + + # THEN + self.assertEqual(extended_value + 269, actual_value) + + def test_should_create_CoapOptionHeader_when_from_bytes_classmethod_is_called(self): + # GIVEN + delta = any_4bits_value_different_than_13_and_14() + length = any_4bits_value_different_than_13_and_14() + + data = bytearray([delta << 4 | length]) + + # WHEN + option_header = coap.CoapOptionHeader.from_bytes(io.BytesIO(data)) + + # THEN + self.assertEqual(delta, option_header.delta) + self.assertEqual(length, option_header.length) + + def test_should_return_True_when_is_payload_marker_property_called_with_delta_and_length_equal_15(self): + # GIVEN + delta = 15 + length = 15 + + data = bytearray([delta << 4 | length]) + + # WHEN + option_header = coap.CoapOptionHeader.from_bytes(io.BytesIO(data)) + + # THEN + self.assertTrue(option_header.is_payload_marker) + + +class TestCoapOption(unittest.TestCase): + + def test_should_return_type_value_when_type_property_is_called(self): + # GIVEN + _type = any_coap_option_type() + + coap_opt = coap.CoapOption(_type, any_value()) + + # WHEN + actual_type = coap_opt.type + + # THEN + self.assertEqual(_type, actual_type) + + def test_should_return_value_value_when_value_property_is_called(self): + # GIVEN + value = any_value() + + coap_opt = coap.CoapOption(any_coap_option_type(), value) + + # WHEN + actual_value = coap_opt.value + + # THEN + self.assertEqual(value, actual_value) + + +class TestCoapOptionsFactory(unittest.TestCase): + + def test_should_create_list_of_CoapOption_from_bytearray_when_parse_method_is_called(self): + # GIVEN + delta = any_4bits_value_lower_or_equal_than_12() + length = any_4bits_value_lower_or_equal_than_12() + value = any_bytearray(length) + + data = bytearray([delta << 4 | length]) + value + + factory = coap.CoapOptionsFactory() + + # WHEN + coap_options = factory.parse(io.BytesIO(data), None) + + # THEN + self.assertEqual(1, len(coap_options)) + self.assertEqual(delta, coap_options[0].type) + self.assertEqual(value, coap_options[0].value) + + +class TestCoapCode(unittest.TestCase): + + def test_should_return_code_value_when_code_property_is_called(self): + # GIVEN + code = any_code() + + code_obj = coap.CoapCode(code) + + # WHEN + actual_code = code_obj.code + + # THEN + self.assertEqual(code, actual_code) + + def test_should_return_class_value_when_class_property_is_called(self): + # GIVEN + code = any_code() + + code_obj = coap.CoapCode(code) + + # WHEN + actual_class = code_obj._class + + # THEN + self.assertEqual((code >> 5) & 0x7, actual_class) + + def test_should_return_detail_value_when_detail_property_is_called(self): + # GIVEN + code = any_code() + + code_obj = coap.CoapCode(code) + + # WHEN + actual_detail = code_obj.detail + + # THEN + self.assertEqual(code & 0x1f, actual_detail) + + def test_should_return_dotted_value_when_dotted_property_is_called(self): + # GIVEN + code = any_code() + + code_obj = coap.CoapCode(code) + + # WHEN + actual_dotted = code_obj.dotted + + # THEN + _class, detail = actual_dotted.split(".") + self.assertEqual(code, (int(_class) << 5) | int(detail)) + + def test_should_create_CoapCode_when_from_class_and_detail_classmethod_is_called(self): + # GIVEN + code = any_code() + + _class = (code >> 5) & 0x7 + detail = code & 0x1f + + # WHEN + actual_coap_obj = coap.CoapCode.from_class_and_detail(_class, detail) + + # THEN + self.assertEqual(code, actual_coap_obj.code) + + def test_should_create_CoapCode_when_from_dotted_string_classmethod_is_called(self): + # GIVEN + code = any_code() + + code_obj = coap.CoapCode(code) + + # WHEN + actual_coap_obj = coap.CoapCode.from_dotted(code_obj.dotted) + + # THEN + self.assertEqual(code, actual_coap_obj.code) + + +class TestCoapMessage(unittest.TestCase): + + def test_should_return_version_value_when_version_property_is_called(self): + # GIVEN + version = any_version() + + coap_message = coap.CoapMessage(version, any_type(), any_code(), any_message_id(), + any_token(), any_options(), any_payload()) + + # WHEN + actual_version = coap_message.version + + # THEN + self.assertEqual(version, actual_version) + + def test_should_return_type_value_when_type_property_is_called(self): + # GIVEN + _type = any_type() + + coap_message = coap.CoapMessage(any_version(), _type, any_code(), any_message_id(), + any_token(), any_options(), any_payload()) + + # WHEN + actual_type = coap_message.type + + # THEN + self.assertEqual(_type, actual_type) + + def test_should_return_code_value_when_code_property_is_called(self): + # GIVEN + code = any_code() + + coap_message = coap.CoapMessage(any_version(), any_type(), code, any_message_id(), + any_token(), any_options(), any_payload()) + + # WHEN + actual_code = coap_message.code + + # THEN + self.assertEqual(code, actual_code) + + def test_should_return_message_id_value_when_message_id_property_is_called(self): + # GIVEN + message_id = any_message_id() + + coap_message = coap.CoapMessage(any_version(), any_type(), any_code(), message_id, + any_token(), any_options(), any_payload()) + + # WHEN + actual_message_id = coap_message.message_id + + # THEN + self.assertEqual(message_id, actual_message_id) + + def test_should_return_token_value_when_token_property_is_called(self): + # GIVEN + token = any_token() + + coap_message = coap.CoapMessage(any_version(), any_type(), any_code(), any_message_id(), + token, any_options(), any_payload()) + + # WHEN + actual_token = coap_message.token + + # THEN + self.assertEqual(token, actual_token) + + def test_should_return_tkl_value_when_tkl_property_is_called(self): + # GIVEN + token = any_token() + + coap_message = coap.CoapMessage(any_version(), any_type(), any_code(), any_message_id(), + token, any_options(), any_payload()) + + # WHEN + actual_tkl = coap_message.tkl + + # THEN + self.assertEqual(len(token), actual_tkl) + + def test_should_return_options_value_when_options_property_is_called(self): + # GIVEN + options = any_options() + + coap_message = coap.CoapMessage(any_version(), any_type(), any_code(), any_message_id(), + any_token(), options, any_payload()) + + # WHEN + actual_options = coap_message.options + + # THEN + self.assertEqual(options, actual_options) + + def test_should_return_payload_value_when_payload_property_is_called(self): + # GIVEN + payload = any_payload() + + coap_message = coap.CoapMessage(any_version(), any_type(), any_code(), any_message_id(), + any_token(), any_options(), payload) + + # WHEN + actual_payload = coap_message.payload + + # THEN + self.assertEqual(payload, actual_payload) + + def test_should_return_uri_path_value_when_uri_path_property_is_called(self): + # GIVEN + uri_path = any_uri_path() + + coap_message = coap.CoapMessage(any_version(), any_type(), any_code(), any_message_id(), + any_token(), any_options(), any_payload(), uri_path) + + # WHEN + actual_uri_path = coap_message.uri_path + + # THEN + self.assertEqual(uri_path, actual_uri_path) + + +class TestCoapMessageIdToUriPathBinder(unittest.TestCase): + + def test_should_add_uri_path_to_binds_when_add_uri_path_for_method_is_called(self): + # GIVEN + message_id = any_message_id() + token = any_token() + uri_path = any_uri_path() + + binder = coap.CoapMessageIdToUriPathBinder() + + # WHEN + binder.add_uri_path_for(message_id, token, uri_path) + + # THEN + self.assertEqual(uri_path, binder.get_uri_path_for(message_id, token)) + + def test_should_raise_KeyError_when_get_uri_path_for_is_called_but_it_is_not_present_in_database(self): + # GIVEN + message_id = any_message_id() + token = any_token() + uri_path = any_uri_path() + + binder = coap.CoapMessageIdToUriPathBinder() + + # THEN + self.assertRaises(RuntimeError, binder.get_uri_path_for, message_id, token) + + +class TestCoapMessageFactory(unittest.TestCase): + + def _create_dummy_payload_factory(self): + class DummyPayloadFactory: + + def parse(self, data, message_info): + return data.read() + + return DummyPayloadFactory() + + def _create_coap_message_factory(self): + return coap.CoapMessageFactory( + options_factory=coap.CoapOptionsFactory(), + uri_path_based_payload_factories={ + "/a/as": self._create_dummy_payload_factory() + }, + message_id_to_uri_path_binder=coap.CoapMessageIdToUriPathBinder()) + + def test_should_create_CoapMessage_from_solicit_request_data_when_parse_method_is_called(self): + # GIVEN + data = bytearray([0x42, 0x02, 0x00, 0xbd, 0x65, 0xee, 0xb1, 0x61, + 0x02, 0x61, 0x73, 0xff, 0x01, 0x08, 0x16, 0x6e, + 0x0a, 0x00, 0x00, 0x00, 0x00, 0x02, 0x04, 0x01, + 0x02]) + + factory = self._create_coap_message_factory() + + # WHEN + coap_message = factory.parse(io.BytesIO(data), None) + + # THEN + self.assertEqual(1, coap_message.version) + self.assertEqual(0, coap_message.type) + self.assertEqual(2, coap_message.tkl) + self.assertEqual(2, coap_message.code) + self.assertEqual(189, coap_message.message_id) + self.assertEqual(bytearray([0x65, 0xee]), coap_message.token) + self.assertEqual("a", coap_message.options[0].value.decode("utf-8")) + self.assertEqual("as", coap_message.options[1].value.decode("utf-8")) + self.assertEqual("/a/as", coap_message.uri_path) + self.assertEqual(bytearray([0x01, 0x08, 0x16, 0x6e, 0x0a, 0x00, 0x00, 0x00, + 0x00, 0x02, 0x04, 0x01, 0x02]), coap_message.payload) + + def test_should_create_CoapMessage_from_solicit_response_data_when_parse_method_is_called(self): + # GIVEN + data = bytearray([0x62, 0x44, 0x00, 0xbd, 0x65, 0xee, 0xff, 0x04, + 0x01, 0x00, 0x02, 0x02, 0x00, 0x00, 0x07, 0x09, + 0x76, 0x80, 0x00, 0x01, 0x00, 0x00, 0x00, 0x00, + 0x00]) + + mid_binder = coap.CoapMessageIdToUriPathBinder() + mid_binder.add_uri_path_for(189, bytearray([0x65, 0xee]), "/a/as") + + factory = coap.CoapMessageFactory( + options_factory=coap.CoapOptionsFactory(), + uri_path_based_payload_factories={ + "/a/as": self._create_dummy_payload_factory() + }, + message_id_to_uri_path_binder=mid_binder) + + # WHEN + coap_message = factory.parse(io.BytesIO(data), None) + + # THEN + self.assertEqual(1, coap_message.version) + self.assertEqual(2, coap_message.type) + self.assertEqual(2, coap_message.tkl) + self.assertEqual("2.04", coap_message.code) + self.assertEqual(189, coap_message.message_id) + self.assertEqual(bytearray([0x65, 0xee]), coap_message.token) + self.assertEqual(None, coap_message.uri_path) + self.assertEqual(bytearray([0x04, 0x01, 0x00, 0x02, 0x02, 0x00, 0x00, 0x07, + 0x09, 0x76, 0x80, 0x00, 0x01, 0x00, 0x00, 0x00, + 0x00, 0x00]), coap_message.payload) + +if __name__ == "__main__": + unittest.main() diff --git a/tests/scripts/thread-cert/test_common.py b/tests/scripts/thread-cert/test_common.py old mode 100644 new mode 100755 index 4fcd969a4..c65d9df3d --- a/tests/scripts/thread-cert/test_common.py +++ b/tests/scripts/thread-cert/test_common.py @@ -115,7 +115,7 @@ class TestMacAddress(unittest.TestCase): mac_address = common.MacAddress.from_eui64(eui64) # THEN - self.assertEqual(common.MacAddress.LONG, mac_address.type) + self.assertEqual(common.MacAddressType.LONG, mac_address.type) self.assertEqual(eui64, mac_address.mac_address) def test_should_create_MacAddress_from_rloc16_int_when_from_rloc16_classmethod_is_called(self): @@ -126,7 +126,7 @@ class TestMacAddress(unittest.TestCase): mac_address = common.MacAddress.from_rloc16(int(rloc16)) # THEN - self.assertEqual(common.MacAddress.SHORT, mac_address.type) + self.assertEqual(common.MacAddressType.SHORT, mac_address.type) self.assertEqual(struct.pack(">H", rloc16), mac_address.mac_address) def test_should_create_MacAddress_from_rloc16_bytearray_when_from_rloc16_classmethod_is_called(self): @@ -137,7 +137,7 @@ class TestMacAddress(unittest.TestCase): mac_address = common.MacAddress.from_rloc16(rloc16) # THEN - self.assertEqual(common.MacAddress.SHORT, mac_address.type) + self.assertEqual(common.MacAddressType.SHORT, mac_address.type) self.assertEqual(rloc16, mac_address.mac_address) def test_should_convert_short_MacAddress_to_iid_when_convert_method_is_called(self): diff --git a/tests/scripts/thread-cert/test_crypto.py b/tests/scripts/thread-cert/test_crypto.py old mode 100644 new mode 100755 diff --git a/tests/scripts/thread-cert/test_ipv6.py b/tests/scripts/thread-cert/test_ipv6.py old mode 100644 new mode 100755 index b6832e0f6..e931feedb --- a/tests/scripts/thread-cert/test_ipv6.py +++ b/tests/scripts/thread-cert/test_ipv6.py @@ -38,8 +38,8 @@ 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, \ - UpperLayerProtocol, UDPHeaderFactory, HopByHopOptionsFactory, \ - UDPBytesPayloadFactory, ICMPv6DestinationUnreachable + UpperLayerProtocol, UDPHeaderFactory, HopByHopOptionsFactory, ICMPv6DestinationUnreachableFactory, \ + UDPBytesPayloadFactory, ICMPv6DestinationUnreachable, UdpBasedOnSrcDstPortsPayloadFactory import common @@ -420,6 +420,13 @@ class TestUDPHeader(unittest.TestCase): # THEN self.assertEqual(8, udp_header_length) + def test_should_return_17_when_type_property_is_called(self): + # GIVEN + udp_header = UDPHeader(any_port(), any_port(), any_payload_length(), any_checksum()) + + # THEN + self.assertEqual(17, udp_header.type) + class TestICMPv6Header(unittest.TestCase): @@ -609,6 +616,7 @@ class TestIPv6PacketFactory(unittest.TestCase): # WHEN ipv6_packet = ipv6_factory.parse(io.BytesIO(ipv6_packet_bytes), any_message_info()) + ipv6_packet._validate_checksum() # THEN self.assertEqual('fd00:1234:4555::ff:fe00:1800', ipv6_packet.ipv6_header.source_address.compressed) @@ -993,6 +1001,60 @@ class TestMPLOptionFactory(unittest.TestCase): self.assertEqual(mpl_opt.seed_id, seed_id) +class TestUdpBasedOnSrcDstPortsPayloadFactory(unittest.TestCase): + + def test_should_create_payload_from_data_when_src_port_factory_is_defined_and_parse_method_is_called(self): + # GIVEN + data = any_data() + + message_info = common.MessageInfo() + message_info.src_port = any_port() + message_info.dst_port = any_port() + + factory = UdpBasedOnSrcDstPortsPayloadFactory( + src_dst_port_based_payload_factories={ + message_info.src_port: UDPBytesPayloadFactory() + }) + + # WHEN + actual_data = factory.parse(io.BytesIO(data), message_info) + + # THEN + self.assertEqual(data, actual_data.data) + + def test_should_create_payload_from_data_when_dst_port_factory_is_defined_and_parse_method_is_called(self): + # GIVEN + data = any_data() + + message_info = common.MessageInfo() + message_info.src_port = any_port() + message_info.dst_port = any_port() + + factory = UdpBasedOnSrcDstPortsPayloadFactory( + src_dst_port_based_payload_factories={ + message_info.dst_port: UDPBytesPayloadFactory() + }) + + # WHEN + actual_data = factory.parse(io.BytesIO(data), message_info) + + # THEN + self.assertEqual(data, actual_data.data) + + def test_should_raise_RuntimeError_when_parse_method_is_called_but_required_factory_is_not_defined(self): + # GIVEN + data = any_data() + + message_info = common.MessageInfo() + message_info.src_port = any_port() + message_info.dst_port = any_port() + + factory = UdpBasedOnSrcDstPortsPayloadFactory(src_dst_port_based_payload_factories={}) + + # THEN + self.assertRaises(RuntimeError, factory.parse, io.BytesIO(data), message_info) + + class TestUDPDatagramFactory(unittest.TestCase): def test_should_produce_UDPDatagram_from_bytes_when_to_bytes_method_is_called_with_data(self): @@ -1009,7 +1071,7 @@ class TestUDPDatagramFactory(unittest.TestCase): (payload_length >> 8), (payload_length & 0xFF), (checksum >> 8), (checksum & 0xFF)]) + payload - factory = UDPDatagramFactory(UDPHeaderFactory(), {dst_port: UDPBytesPayloadFactory()}) + factory = UDPDatagramFactory(UDPHeaderFactory(), UDPBytesPayloadFactory()) # WHEN udp_dgram = factory.parse(io.BytesIO(data), any_message_info()) @@ -1021,6 +1083,31 @@ class TestUDPDatagramFactory(unittest.TestCase): self.assertEqual(udp_dgram.header.checksum, checksum) self.assertEqual(udp_dgram.payload.data, payload) + def test_should_set_src_and_dst_port_in_message_info_when_parse_method_is_called(self): + # GIVEN + message_info = any_message_info() + + src_port = any_port() + dst_port = any_port() + checksum = any_checksum() + + payload = any_payload() + payload_length = len(payload) + len(UDPHeader(0, 0)) + + data = bytearray([(src_port >> 8), (src_port & 0xFF), + (dst_port >> 8), (dst_port & 0xFF), + (payload_length >> 8), (payload_length & 0xFF), + (checksum >> 8), (checksum & 0xFF)]) + payload + + factory = UDPDatagramFactory(UDPHeaderFactory(), UDPBytesPayloadFactory()) + + # WHEN + udp_dgram = factory.parse(io.BytesIO(data), message_info) + + # THEN + self.assertEqual(src_port, message_info.src_port) + self.assertEqual(dst_port, message_info.dst_port) + class TestICMPv6Factory(unittest.TestCase): @@ -1208,22 +1295,39 @@ class TestICMPv6DestinationUnreachable(unittest.TestCase): io.BytesIO(bytearray(struct.pack(">I", unused)) + data)) +class TestICMPv6DestinationUnreachableFactory(unittest.TestCase): + + def test_should_create_ICMPv6DestinationUnreachable_when_parse_method_is_called(self): + # GIVEN + icmp_data = any_data() + + factory = ICMPv6DestinationUnreachableFactory() + + data = bytearray([0x00, 0x00, 0x00, 0x00]) + icmp_data + + # WHEN + icmpv6_dest_unreachable=factory.parse(io.BytesIO(data), any_message_info()) + + # THEN + self.assertEqual(icmp_data, icmpv6_dest_unreachable.data) + + class TestUDPHeaderFactory(unittest.TestCase): def test_should_create_UDPHeader_when_to_bytes_method_is_called(self): # GIVEN - factory = UDPHeaderFactory() + factory=UDPHeaderFactory() - src_port = any_port() - dst_port = any_port() - payload_length = any_payload_length() - checksum = any_checksum() + src_port=any_port() + dst_port=any_port() + payload_length=any_payload_length() + checksum=any_checksum() - data = struct.pack("!H", src_port) + struct.pack("!H", dst_port) + \ + data=struct.pack("!H", src_port) + struct.pack("!H", dst_port) + \ struct.pack("!H", payload_length) + struct.pack("!H", checksum) # WHEN - udp_header = factory.parse(io.BytesIO(data), any_message_info()) + udp_header=factory.parse(io.BytesIO(data), any_message_info()) # THEN self.assertEqual(src_port, udp_header.src_port) @@ -1242,12 +1346,12 @@ class TestHopByHopOptionsFactory(unittest.TestCase): def parse(self, data, message_info): return data.read() - factory = HopByHopOptionsFactory(options_factories={2: DummyOptionFactory()}) + factory=HopByHopOptionsFactory(options_factories = {2: DummyOptionFactory()}) - data = bytearray([0x02, 0x03, 0x11, 0x22, 0x33, 0x01, 0x00]) + data=bytearray([0x02, 0x03, 0x11, 0x22, 0x33, 0x01, 0x00]) # WHEN - actual_options = factory.parse(io.BytesIO(data), any_message_info()) + actual_options=factory.parse(io.BytesIO(data), any_message_info()) # THEN self.assertEqual(1, len(actual_options)) diff --git a/tests/scripts/thread-cert/test_lowpan.py b/tests/scripts/thread-cert/test_lowpan.py old mode 100644 new mode 100755 index ec9270ba0..774ee3167 --- a/tests/scripts/thread-cert/test_lowpan.py +++ b/tests/scripts/thread-cert/test_lowpan.py @@ -39,10 +39,6 @@ import lowpan def create_default_lowpan_parser(context_manager): - dst_port_factories = { - 5684: ipv6.UDPBytesPayloadFactory() - } - return lowpan.LowpanParser( lowpan_mesh_header_factory=lowpan.LowpanMeshHeaderFactory(), lowpan_decompressor=config.create_default_lowpan_decompressor(context_manager), @@ -52,7 +48,7 @@ def create_default_lowpan_parser(context_manager): ulpf={ 17: ipv6.UDPDatagramFactory( udp_header_factory=ipv6.UDPHeaderFactory(), - dst_port_factories=dst_port_factories), + udp_payload_factory=ipv6.UDPBytesPayloadFactory()), 58: ipv6.ICMPv6Factory( body_factories=config.create_default_ipv6_icmp_body_factories() ) @@ -2242,8 +2238,8 @@ class TestLowpanMeshHeaderFactory(unittest.TestCase): originator_address = any_mac_address() final_destination_address = any_mac_address() - v = int(originator_address.type == common.MacAddress.SHORT) - f = int(final_destination_address.type == common.MacAddress.SHORT) + v = int(originator_address.type == common.MacAddressType.SHORT) + f = int(final_destination_address.type == common.MacAddressType.SHORT) mesh_header_data = bytearray([(2 << 6) | (v << 5) | (f << 4) | hops_left]) + \ originator_address.mac_address + final_destination_address.mac_address diff --git a/tests/scripts/thread-cert/test_mac802154.py b/tests/scripts/thread-cert/test_mac802154.py old mode 100644 new mode 100755 index 44718f05d..7cc42570e --- a/tests/scripts/thread-cert/test_mac802154.py +++ b/tests/scripts/thread-cert/test_mac802154.py @@ -30,7 +30,7 @@ import io import unittest -from common import MacAddress +from common import MacAddress, MacAddressType import mac802154 longaddrs = bytearray([0x61, 0xcc, 0x00, 0xce, 0xfa]) @@ -170,7 +170,7 @@ class TestMacParser(unittest.TestCase): def test_should_decrypt_data_frame(self): mac802154.DeviceDescriptors.add(0x2001, MacAddress(bytearray([0x16, 0x6e, 0x0a, 0x00, 0x00, 0x00, 0x00, 0x07]), - MacAddress.LONG)) + MacAddressType.LONG)) frame = mac802154.MacFrame() frame.parse(io.BytesIO(bytearray([0x69, 0x98, 0x68, # FC, seq diff --git a/tests/scripts/thread-cert/test_mle.py b/tests/scripts/thread-cert/test_mle.py old mode 100644 new mode 100755 diff --git a/tests/scripts/thread-cert/test_network_data.py b/tests/scripts/thread-cert/test_network_data.py old mode 100644 new mode 100755 index 895376237..105dedc1a --- a/tests/scripts/thread-cert/test_network_data.py +++ b/tests/scripts/thread-cert/test_network_data.py @@ -1,3 +1,32 @@ +#!/usr/bin/python +# +# Copyright (c) 2016, The OpenThread Authors. +# All rights reserved. +# +# Redistribution and use in source and binary forms, with or without +# modification, are permitted provided that the following conditions are met: +# 1. Redistributions of source code must retain the above copyright +# notice, this list of conditions and the following disclaimer. +# 2. Redistributions in binary form must reproduce the above copyright +# notice, this list of conditions and the following disclaimer in the +# documentation and/or other materials provided with the distribution. +# 3. Neither the name of the copyright holder nor the +# names of its contributors may be used to endorse or promote products +# derived from this software without specific prior written permission. +# +# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" +# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE +# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE +# ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE +# LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR +# CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF +# SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS +# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN +# CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) +# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE +# POSSIBILITY OF SUCH DAMAGE. +# + import io import math import random diff --git a/tests/scripts/thread-cert/test_network_layer.py b/tests/scripts/thread-cert/test_network_layer.py new file mode 100755 index 000000000..32387abc3 --- /dev/null +++ b/tests/scripts/thread-cert/test_network_layer.py @@ -0,0 +1,504 @@ +#!/usr/bin/python +# +# Copyright (c) 2016, The OpenThread Authors. +# All rights reserved. +# +# Redistribution and use in source and binary forms, with or without +# modification, are permitted provided that the following conditions are met: +# 1. Redistributions of source code must retain the above copyright +# notice, this list of conditions and the following disclaimer. +# 2. Redistributions in binary form must reproduce the above copyright +# notice, this list of conditions and the following disclaimer in the +# documentation and/or other materials provided with the distribution. +# 3. Neither the name of the copyright holder nor the +# names of its contributors may be used to endorse or promote products +# derived from this software without specific prior written permission. +# +# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" +# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE +# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE +# ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE +# LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR +# CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF +# SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS +# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN +# CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) +# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE +# POSSIBILITY OF SUCH DAMAGE. +# + +import io +import random +import struct +import unittest + +import common +import network_layer + + +def any_eid(): + return bytearray([random.getrandbits(8) for _ in range(16)]) + + +def any_mac_extended_address(): + return bytearray([random.getrandbits(8) for _ in range(8)]) + + +def any_rloc16(): + return random.getrandbits(16) + + +def any_ml_eid(): + return bytearray([random.getrandbits(8) for _ in range(8)]) + + +def any_status(): + return random.getrandbits(1) + + +def any_seconds(): + return random.getrandbits(32) + + +def any_id_sequence(): + return random.getrandbits(8) + + +def any_router_id_mask(): + return random.getrandbits(64) + + +def any_options(count=None): + count = count if count is not None else random.randint(0, 255) + return [random.getrandbits(8) for _ in range(count)] + + +def any_tlv_data(length=None): + _type = random.getrandbits(8) + length = length if length is not None else random.getrandbits(8) + value = bytearray([random.getrandbits(8) for _ in range(length)]) + + return bytearray([_type, length]) + value + + +def any_tlvs_data(count=None): + count = count if count is not None else random.randint(0, 16) + data = bytearray() + + for _ in range(count): + data += any_tlv_data(random.randint(1, 15)) + + return data + + +class TestTargetEid(unittest.TestCase): + + def test_should_return_eid_value_when_eid_property_is_called(self): + # GIVEN + eid = any_eid() + + target_eid = network_layer.TargetEid(eid) + + # WHEN + actual_eid = target_eid.eid + + # THEN + self.assertEqual(eid, actual_eid) + + def test_should_return_True_when_try_to_equal_two_the_same_type_objects_with_the_same_values(self): + # GIVEN + eid = any_eid() + + target_eid = network_layer.TargetEid(eid) + + # THEN + self.assertEqual(target_eid, network_layer.TargetEid(eid)) + + +class TestTargetEidFactory(unittest.TestCase): + + def test_should_create_TargetEid_from_bytearray_when_parse_method_is_called(self): + # GIVEN + eid = any_eid() + + factory = network_layer.TargetEidFactory() + + # WHEN + target_eid = factory.parse(io.BytesIO(eid), common.MessageInfo()) + + # THEN + self.assertTrue(isinstance(target_eid, network_layer.TargetEid)) + self.assertEqual(eid, target_eid.eid) + + +class TestMacExtendedAddress(unittest.TestCase): + + def test_should_return_mac_address_value_when_mac_address_property_is_called(self): + # GIVEN + mac_address = any_mac_extended_address() + + mac_extended_address = network_layer.MacExtendedAddress(mac_address) + + # WHEN + actual_mac_address = mac_extended_address.mac_address + + # THEN + self.assertEqual(mac_address, actual_mac_address) + + def test_should_return_True_when_try_to_equal_two_the_same_type_objects_with_the_same_values(self): + # GIVEN + mac_address = any_mac_extended_address() + + mac_extended_address = network_layer.MacExtendedAddress(mac_address) + + # THEN + self.assertEqual(mac_extended_address, network_layer.MacExtendedAddress(mac_address)) + + +class TestMacExtendedAddressFactory(unittest.TestCase): + + def test_should_create_MacExtendedAddress_from_bytearray_when_parse_method_is_called(self): + # GIVEN + mac_address = any_mac_extended_address() + + factory = network_layer.MacExtendedAddressFactory() + + # WHEN + mac_extended_address = factory.parse(io.BytesIO(mac_address), common.MessageInfo()) + + # THEN + self.assertTrue(isinstance(mac_extended_address, network_layer.MacExtendedAddress)) + self.assertEqual(mac_address, mac_extended_address.mac_address) + + +class TestRloc16(unittest.TestCase): + + def test_should_return_rloc16_value_when_rloc16_property_is_called(self): + # GIVEN + rloc16 = any_rloc16() + + rloc16_obj = network_layer.Rloc16(rloc16) + + # WHEN + actual_rloc16 = rloc16_obj.rloc16 + + # THEN + self.assertEqual(rloc16, actual_rloc16) + + def test_should_return_True_when_try_to_equal_two_the_same_type_objects_with_the_same_values(self): + # GIVEN + rloc16 = any_rloc16() + + rloc16_obj = network_layer.Rloc16(rloc16) + + # THEN + self.assertEqual(rloc16_obj, network_layer.Rloc16(rloc16)) + + +class TestRloc16Factory(unittest.TestCase): + + def test_should_create_Rloc16_from_bytearray_when_parse_method_is_called(self): + # GIVEN + rloc16 = any_rloc16() + + factory = network_layer.Rloc16Factory() + + data = bytearray(struct.pack(">H", rloc16)) + + # WHEN + rloc16_obj = factory.parse(io.BytesIO(data), common.MessageInfo()) + + # THEN + self.assertTrue(isinstance(rloc16_obj, network_layer.Rloc16)) + self.assertEqual(rloc16, rloc16_obj.rloc16) + + +class TestMlEid(unittest.TestCase): + + def test_should_return_ml_eid_value_when_ml_eid_property_is_called(self): + # GIVEN + ml_eid = any_ml_eid() + + ml_eid_obj = network_layer.MlEid(ml_eid) + + # WHEN + actual_ml_eid = ml_eid_obj.ml_eid + + # THEN + self.assertEqual(ml_eid, actual_ml_eid) + + def test_should_return_True_when_try_to_equal_two_the_same_type_objects_with_the_same_values(self): + # GIVEN + ml_eid = any_ml_eid() + + ml_eid_obj = network_layer.MlEid(ml_eid) + + # THEN + self.assertEqual(ml_eid_obj, network_layer.MlEid(ml_eid)) + + +class TestMlEidFactory(unittest.TestCase): + + def test_should_create_MlEid_from_bytearray_when_parse_method_is_called(self): + # GIVEN + ml_eid = any_ml_eid() + + factory = network_layer.MlEidFactory() + + # WHEN + ml_eid_obj = factory.parse(io.BytesIO(ml_eid), common.MessageInfo()) + + # THEN + self.assertTrue(isinstance(ml_eid_obj, network_layer.MlEid)) + self.assertEqual(ml_eid, ml_eid_obj.ml_eid) + + +class TestStatus(unittest.TestCase): + + def test_should_return_status_value_when_status_property_is_called(self): + # GIVEN + status = any_status() + + status_obj = network_layer.Status(status) + + # WHEN + actual_status = status_obj.status + + # THEN + self.assertEqual(status, actual_status) + + def test_should_return_True_when_try_to_equal_two_the_same_type_objects_with_the_same_values(self): + # GIVEN + status = any_status() + + status_obj = network_layer.Status(status) + + # THEN + self.assertEqual(status_obj, network_layer.Status(status)) + + +class TestStatusFactory(unittest.TestCase): + + def test_should_create_Status_from_bytearray_when_parse_method_is_called(self): + # GIVEN + status = any_status() + + factory = network_layer.StatusFactory() + + data = bytearray([status]) + + # WHEN + status_obj = factory.parse(io.BytesIO(data), common.MessageInfo()) + + # THEN + self.assertTrue(isinstance(status_obj, network_layer.Status)) + self.assertEqual(status, status_obj.status) + + +class TestTimeSinceLastTransaction(unittest.TestCase): + + def test_should_return_seconds_value_when_seconds_property_is_called(self): + # GIVEN + seconds = any_seconds() + + time_since_last_transaction = network_layer.TimeSinceLastTransaction(seconds) + + # WHEN + actual_seconds = time_since_last_transaction.seconds + + # THEN + self.assertEqual(seconds, actual_seconds) + + def test_should_return_True_when_try_to_equal_two_the_same_type_objects_with_the_same_values(self): + # GIVEN + seconds = any_seconds() + + time_since_last_transaction = network_layer.TimeSinceLastTransaction(seconds) + + # THEN + self.assertEqual(time_since_last_transaction, network_layer.TimeSinceLastTransaction(seconds)) + + +class TestTimeSinceLastTransactionFactory(unittest.TestCase): + + def test_should_create_TimeSinceLastTransaction_from_bytearray_when_parse_method_is_called(self): + # GIVEN + seconds = any_seconds() + + factory = network_layer.TimeSinceLastTransactionFactory() + + data = bytearray(struct.pack(">L", seconds)) + + # WHEN + time_since_last_transaction = factory.parse(io.BytesIO(data), common.MessageInfo()) + + # THEN + self.assertTrue(isinstance(time_since_last_transaction, network_layer.TimeSinceLastTransaction)) + self.assertEqual(seconds, time_since_last_transaction.seconds) + + +class TestRouterMask(unittest.TestCase): + + def test_should_return_id_sequence_value_when_id_sequence_property_is_called(self): + # GIVEN + id_sequence = any_id_sequence() + + router_mask = network_layer.RouterMask(id_sequence, any_router_id_mask()) + + # WHEN + actual_id_sequence = router_mask.id_sequence + + # THEN + self.assertEqual(id_sequence, actual_id_sequence) + + def test_should_return_router_id_mask_value_when_router_id_mask_property_is_called(self): + # GIVEN + router_id_mask = any_router_id_mask() + + router_mask = network_layer.RouterMask(any_id_sequence(), router_id_mask) + + # WHEN + actual_router_id_mask = router_mask.router_id_mask + + # THEN + self.assertEqual(router_id_mask, actual_router_id_mask) + + def test_should_return_True_when_try_to_equal_two_the_same_type_objects_with_the_same_values(self): + # GIVEN + id_sequence = any_id_sequence() + router_id_mask = any_router_id_mask() + + router_mask = network_layer.RouterMask(id_sequence, router_id_mask) + + # THEN + self.assertEqual(router_mask, network_layer.RouterMask(id_sequence, router_id_mask)) + + +class TestRouterMaskFactory(unittest.TestCase): + + def test_should_create_RouterMask_from_bytearray_when_parse_method_is_called(self): + # GIVEN + id_sequence = any_id_sequence() + router_id_mask = any_router_id_mask() + + factory = network_layer.RouterMaskFactory() + + data = bytearray([id_sequence]) + struct.pack(">Q", router_id_mask) + + # WHEN + router_mask = factory.parse(io.BytesIO(data), common.MessageInfo()) + + # THEN + self.assertTrue(isinstance(router_mask, network_layer.RouterMask)) + self.assertEqual(id_sequence, router_mask.id_sequence) + self.assertEqual(router_id_mask, router_mask.router_id_mask) + + +class TestNdOption(unittest.TestCase): + + def test_should_return_options_value_when_options_property_is_called(self): + # GIVEN + options = any_options() + + nd_option = network_layer.NdOption(options) + + # WHEN + actual_options = nd_option.options + + # THEN + self.assertEqual(options, actual_options) + + def test_should_return_True_when_try_to_equal_two_the_same_type_objects_with_the_same_values(self): + # GIVEN + options = any_options() + + nd_option = network_layer.NdOption(options) + + # THEN + self.assertEqual(nd_option, network_layer.NdOption(options)) + + +class TestNdOptionFactory(unittest.TestCase): + + def test_should_create_NdOption_from_bytearray_when_parse_method_is_called(self): + # GIVEN + options = any_options() + + factory = network_layer.NdOptionFactory() + + data = bytearray(options) + + # WHEN + nd_option = factory.parse(io.BytesIO(data), common.MessageInfo()) + + # THEN + self.assertTrue(isinstance(nd_option, network_layer.NdOption)) + self.assertEqual(options, nd_option.options) + + +class TestThreadNetworkData(unittest.TestCase): + + def test_should_return_options_value_when_options_property_is_called(self): + # GIVEN + tlvs = any_tlvs_data() + + thread_network_data = network_layer.ThreadNetworkData(tlvs) + + # WHEN + actual_tlvs = thread_network_data.tlvs + + # THEN + self.assertEqual(tlvs, actual_tlvs) + + def test_should_return_True_when_try_to_equal_two_the_same_type_objects_with_the_same_values(self): + # GIVEN + tlvs = any_tlvs_data() + + thread_network_data = network_layer.ThreadNetworkData(tlvs) + + # THEN + self.assertEqual(thread_network_data, network_layer.ThreadNetworkData(tlvs)) + + +class TestThreadNetworkDataFactory(unittest.TestCase): + + def test_should_create_ThreadNetworkData_from_bytearray_when_parse_method_is_called(self): + # GIVEN + tlvs = any_tlvs_data() + + class DummyNetworkDataTlvsFactory: + + def parse(self, data, message_info): + return bytearray(data.read()) + + factory = network_layer.ThreadNetworkDataFactory(DummyNetworkDataTlvsFactory()) + + # WHEN + thread_network_data = factory.parse(io.BytesIO(tlvs), common.MessageInfo()) + + # THEN + self.assertTrue(isinstance(thread_network_data, network_layer.ThreadNetworkData)) + self.assertEqual(tlvs, thread_network_data.tlvs) + + +class TestNetworkLayerTlvsFactory(unittest.TestCase): + + def test_should_create_tlv_when_parse_method_is_called(self): + # GIVEN + class DummyNetworkDataTlvsFactory: + + def parse(self, data, message_info): + return bytearray(data.read()) + + tlv = any_tlv_data() + + factory = network_layer.NetworkLayerTlvsFactory({tlv[0]: DummyNetworkDataTlvsFactory()}) + + # WHEN + actual_tlvs = factory.parse(io.BytesIO(tlv), None) + + # THEN + self.assertEqual(tlv[2:], actual_tlvs[0]) + +if __name__ == "__main__": + unittest.main()