From 80285206d23d5646d18bacb59eaf7d43c15b710a Mon Sep 17 00:00:00 2001 From: kangping Date: Wed, 29 Apr 2020 04:37:21 +0800 Subject: [PATCH] [tests] add 5_7_01 network diagnostic test cases (#4663) This commit adds Network Diagnostic TLV parsing and 5.7.01 test case. --- .../Cert_5_7_01_CoapDiagCommands_A.py | 229 ++++++++++++ tests/scripts/thread-cert/Makefile.am | 4 + tests/scripts/thread-cert/config.py | 67 +++- tests/scripts/thread-cert/network_diag.py | 336 ++++++++++++++++++ tests/scripts/thread-cert/node.py | 24 ++ 5 files changed, 651 insertions(+), 9 deletions(-) create mode 100755 tests/scripts/thread-cert/Cert_5_7_01_CoapDiagCommands_A.py create mode 100644 tests/scripts/thread-cert/network_diag.py diff --git a/tests/scripts/thread-cert/Cert_5_7_01_CoapDiagCommands_A.py b/tests/scripts/thread-cert/Cert_5_7_01_CoapDiagCommands_A.py new file mode 100755 index 000000000..8fde1efce --- /dev/null +++ b/tests/scripts/thread-cert/Cert_5_7_01_CoapDiagCommands_A.py @@ -0,0 +1,229 @@ +#!/usr/bin/env python3 +# +# Copyright (c) 2020, 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 unittest + +import config +import mle +import network_diag +import network_layer +import node + +from network_diag import TlvType + +LEADER = 1 +ROUTER1 = 2 +REED1 = 3 +SED1 = 4 +MED1 = 5 +FED1 = 6 + +DUT = ROUTER1 +MTDS = [MED1, SED1] + + +class Cert_5_7_01_CoapDiagCommands_A(unittest.TestCase): + + def setUp(self): + self.simulator = config.create_default_simulator() + + self.nodes = {} + for i in range(1, 7): + self.nodes[i] = node.Node(i, (i in MTDS), simulator=self.simulator) + + self.nodes[LEADER].set_panid(0xface) + self.nodes[LEADER].set_mode('rsdn') + self.nodes[LEADER].add_whitelist(self.nodes[ROUTER1].get_addr64()) + self.nodes[LEADER].enable_whitelist() + + self.nodes[ROUTER1].set_panid(0xface) + self.nodes[ROUTER1].set_mode('rsdn') + self.nodes[ROUTER1].add_whitelist(self.nodes[LEADER].get_addr64()) + self.nodes[ROUTER1].add_whitelist(self.nodes[REED1].get_addr64()) + self.nodes[ROUTER1].add_whitelist(self.nodes[SED1].get_addr64()) + self.nodes[ROUTER1].add_whitelist(self.nodes[MED1].get_addr64()) + self.nodes[ROUTER1].add_whitelist(self.nodes[FED1].get_addr64()) + self.nodes[ROUTER1].enable_whitelist() + self.nodes[ROUTER1].set_router_selection_jitter(1) + + self.nodes[REED1].set_panid(0xface) + self.nodes[REED1].set_mode('rsdn') + self.nodes[REED1].add_whitelist(self.nodes[ROUTER1].get_addr64()) + self.nodes[REED1].set_router_upgrade_threshold(0) + self.nodes[REED1].enable_whitelist() + + self.nodes[SED1].set_panid(0xface) + self.nodes[SED1].set_mode('s') + self.nodes[SED1].add_whitelist(self.nodes[ROUTER1].get_addr64()) + self.nodes[SED1].enable_whitelist() + self.nodes[SED1].set_timeout(config.DEFAULT_CHILD_TIMEOUT) + + self.nodes[MED1].set_panid(0xface) + self.nodes[MED1].set_mode('rsn') + self.nodes[MED1].add_whitelist(self.nodes[ROUTER1].get_addr64()) + self.nodes[MED1].enable_whitelist() + + self.nodes[FED1].set_panid(0xface) + self.nodes[FED1].set_mode('rsdn') + self.nodes[FED1].add_whitelist(self.nodes[ROUTER1].get_addr64()) + self.nodes[FED1].set_router_upgrade_threshold(0) + self.nodes[FED1].enable_whitelist() + + def tearDown(self): + for n in list(self.nodes.values()): + n.stop() + n.destroy() + self.simulator.stop() + + def test(self): + # 1 - Form topology + self.nodes[LEADER].start() + self.simulator.go(5) + self.assertEqual(self.nodes[LEADER].get_state(), 'leader') + + self.nodes[ROUTER1].start() + self.simulator.go(5) + self.assertEqual(self.nodes[ROUTER1].get_state(), 'router') + + self.nodes[REED1].start() + self.simulator.go(5) + self.assertEqual(self.nodes[REED1].get_state(), 'child') + + self.nodes[SED1].start() + self.simulator.go(10) + self.assertEqual(self.nodes[SED1].get_state(), 'child') + + self.nodes[MED1].start() + self.simulator.go(5) + self.assertEqual(self.nodes[MED1].get_state(), 'child') + + self.nodes[FED1].start() + self.simulator.go(5) + self.assertEqual(self.nodes[FED1].get_state(), 'child') + + dut_rloc = self.nodes[DUT].get_ip6_address(config.ADDRESS_TYPE.RLOC) + + # 2 - Leader sends DIAG_GET.req + tlv_types = [ + TlvType.EXT_ADDRESS, TlvType.ADDRESS16, TlvType.MODE, + TlvType.CONNECTIVITY, TlvType.ROUTE64, TlvType.LEADER_DATA, + TlvType.NETWORK_DATA, TlvType.IPV6_ADDRESS_LIST, + TlvType.CHANNEL_PAGES + ] + self.nodes[LEADER].send_network_diag_get(dut_rloc, tlv_types) + self.simulator.go(2) + + dut_messages = self.simulator.get_messages_sent_by(DUT) + + diag_get_rsp = dut_messages.next_coap_message(code='2.04') + diag_get_rsp.assertCoapMessageContainsTlv( + network_layer.MacExtendedAddress) + diag_get_rsp.assertCoapMessageContainsTlv(mle.Address16) + diag_get_rsp.assertCoapMessageContainsTlv(mle.Mode) + diag_get_rsp.assertCoapMessageContainsTlv(mle.Connectivity) + diag_get_rsp.assertCoapMessageContainsTlv(mle.Route64) + diag_get_rsp.assertCoapMessageContainsTlv(mle.LeaderData) + diag_get_rsp.assertCoapMessageContainsTlv(mle.NetworkData) + diag_get_rsp.assertCoapMessageContainsTlv(network_diag.Ipv6AddressList) + diag_get_rsp.assertCoapMessageContainsTlv(network_diag.ChannelPages) + + # 3 - Leader sends DIAG_GET.req (MAC Counters TLV type included) + self.nodes[LEADER].send_network_diag_get(dut_rloc, + [TlvType.MAC_COUNTERS]) + self.simulator.go(2) + + dut_messages = self.simulator.get_messages_sent_by(DUT) + diag_get_rsp = dut_messages.next_coap_message(code='2.04') + diag_get_rsp.assertCoapMessageContainsTlv(network_diag.MacCounters) + mac_counters = diag_get_rsp.get_coap_message_tlv( + network_diag.MacCounters) + + # 4 - Leader sends DIAG_GET.req (Timeout/Polling Period TLV type included) + self.nodes[LEADER].send_network_diag_get(dut_rloc, + [TlvType.POLLING_PERIOD]) + self.simulator.go(2) + + dut_messages = self.simulator.get_messages_sent_by(DUT) + diag_get_rsp = dut_messages.next_coap_message(code='2.04') + diag_get_rsp.assertCoapMessageDoesNotContainTlv(mle.Timeout) + + # 5 - Leader sends DIAG_GET.req (Battery Level and Supply Voltage TLV types included) + self.nodes[LEADER].send_network_diag_get( + dut_rloc, [TlvType.BATTERY_LEVEL, TlvType.SUPPLY_VOLTAGE]) + self.simulator.go(2) + + dut_messages = self.simulator.get_messages_sent_by(DUT) + diag_get_rsp = dut_messages.next_coap_message(code='2.04') + diag_get_rsp.assertCoapMessageContainsOptionalTlv( + network_diag.BatteryLevel) + diag_get_rsp.assertCoapMessageContainsOptionalTlv( + network_diag.SupplyVoltage) + + # 6 - Leader sends DIAG_GET.req (Child Table TLV types included) + self.nodes[LEADER].send_network_diag_get(dut_rloc, + [TlvType.CHILD_TABLE]) + self.simulator.go(2) + + dut_messages = self.simulator.get_messages_sent_by(DUT) + diag_get_rsp = dut_messages.next_coap_message(code='2.04') + diag_get_rsp.assertCoapMessageContainsTlv(network_diag.ChildTable) + child_table = diag_get_rsp.get_coap_message_tlv(network_diag.ChildTable) + self.assertEqual(len(child_table.children), 4) + # TODO(wgtdkp): more validations + + # 7 - Leader sends DIAG_RST.ntf (MAC Counters TLV type included) + self.nodes[LEADER].send_network_diag_reset(dut_rloc, + [TlvType.MAC_COUNTERS]) + self.simulator.go(2) + + dut_messages = self.simulator.get_messages_sent_by(DUT) + + # Make sure the response is there. + dut_messages.next_coap_message(code='2.04') + + # 8 - Leader Sends DIAG_GET.req (MAC Counters TLV type included) + self.nodes[LEADER].send_network_diag_get(dut_rloc, + [TlvType.MAC_COUNTERS]) + self.simulator.go(2) + + dut_messages = self.simulator.get_messages_sent_by(DUT) + diag_get_rsp = dut_messages.next_coap_message(code='2.04') + diag_get_rsp.assertCoapMessageContainsTlv(network_diag.MacCounters) + reset_mac_counters = diag_get_rsp.get_coap_message_tlv( + network_diag.MacCounters) + + self.assertEqual(len(mac_counters.counters), + len(reset_mac_counters.counters)) + for old_counter, new_counter in zip(mac_counters.counters, + reset_mac_counters.counters): + self.assertTrue(new_counter == 0 or new_counter < old_counter) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/scripts/thread-cert/Makefile.am b/tests/scripts/thread-cert/Makefile.am index d59d60ce0..94c69b23f 100644 --- a/tests/scripts/thread-cert/Makefile.am +++ b/tests/scripts/thread-cert/Makefile.am @@ -78,6 +78,7 @@ EXTRA_DIST = \ Cert_5_6_07_NetworkDataRequestREED.py \ Cert_5_6_08_ContextManagement.py \ Cert_5_6_09_NetworkDataForwarding.py \ + Cert_5_7_01_CoapDiagCommands_A.py \ Cert_5_8_01_KeySynchronization.py \ Cert_5_8_02_KeyIncrement.py \ Cert_5_8_03_KeyIncrementRollOver.py \ @@ -136,6 +137,7 @@ EXTRA_DIST = \ mle.py \ net_crypto.py \ network_data.py \ + network_diag.py \ network_layer.py \ node.py \ pcap.py \ @@ -227,6 +229,7 @@ check_SCRIPTS = \ Cert_5_6_07_NetworkDataRequestREED.py \ Cert_5_6_08_ContextManagement.py \ Cert_5_6_09_NetworkDataForwarding.py \ + Cert_5_7_01_CoapDiagCommands_A.py \ Cert_5_8_01_KeySynchronization.py \ Cert_5_8_02_KeyIncrement.py \ Cert_5_8_03_KeyIncrementRollOver.py \ @@ -295,6 +298,7 @@ XFAIL_NCP_TESTS = \ test_reed_address_solicit_rejected.py \ test_service.py \ Cert_5_3_10_AddressQuery.py \ + Cert_5_7_01_CoapDiagCommands_A.py \ Cert_8_1_01_Commissioning.py \ Cert_8_1_02_Commissioning.py \ Cert_8_2_01_JoinerRouter.py \ diff --git a/tests/scripts/thread-cert/config.py b/tests/scripts/thread-cert/config.py index 38ca752f8..dcac3bd67 100644 --- a/tests/scripts/thread-cert/config.py +++ b/tests/scripts/thread-cert/config.py @@ -39,6 +39,7 @@ import message import mle import net_crypto import network_data +import network_diag import network_layer import simulator import sniffer @@ -423,18 +424,66 @@ def create_default_mesh_cop_tlvs_factory(): sub_tlvs_factories=create_default_mesh_cop_tlvs_factories()) +def create_default_network_diag_tlv_factories(): + return { + network_diag.TlvType.EXT_ADDRESS: + network_layer.MacExtendedAddressFactory(), + network_diag.TlvType.ADDRESS16: + mle.Address16Factory(), + network_diag.TlvType.MODE: + mle.ModeFactory(), + network_diag.TlvType.POLLING_PERIOD: + mle.TimeoutFactory(), + network_diag.TlvType.CONNECTIVITY: + mle.ConnectivityFactory(), + network_diag.TlvType.ROUTE64: + create_default_mle_tlv_route64_factory(), + network_diag.TlvType.LEADER_DATA: + mle.LeaderDataFactory(), + network_diag.TlvType.NETWORK_DATA: + create_default_mle_tlv_network_data_factory(), + network_diag.TlvType.IPV6_ADDRESS_LIST: + network_diag.Ipv6AddressListFactory(), + network_diag.TlvType.MAC_COUNTERS: + network_diag.MacCountersFactory(), + network_diag.TlvType.BATTERY_LEVEL: + network_diag.BatteryLevelFactory(), + network_diag.TlvType.SUPPLY_VOLTAGE: + network_diag.SupplyVoltageFactory(), + network_diag.TlvType.CHILD_TABLE: + network_diag.ChildTableFactory(), + network_diag.TlvType.CHANNEL_PAGES: + network_diag.ChannelPagesFactory(), + network_diag.TlvType.TYPE_LIST: + network_diag.TypeListFactory(), + network_diag.TlvType.MAX_CHILD_TIMEOUT: + network_diag.MaxChildTimeoutFactory() + } + + +def create_default_network_diag_tlvs_factory(): + return SubTlvsFactory( + sub_tlvs_factories=create_default_network_diag_tlv_factories()) + + def create_default_uri_path_based_payload_factories(): network_layer_tlvs_factory = create_default_network_tlvs_factory() mesh_cop_tlvs_factory = create_default_mesh_cop_tlvs_factory() + network_diag_tlvs_factory = create_default_network_diag_tlvs_factory() + return { - "/a/as": network_layer_tlvs_factory, - "/a/aq": network_layer_tlvs_factory, - "/a/ar": network_layer_tlvs_factory, - "/a/ae": network_layer_tlvs_factory, - "/a/an": network_layer_tlvs_factory, - "/a/sd": network_layer_tlvs_factory, - "/c/lp": mesh_cop_tlvs_factory, - "/c/cs": mesh_cop_tlvs_factory, + '/a/as': network_layer_tlvs_factory, + '/a/aq': network_layer_tlvs_factory, + '/a/ar': network_layer_tlvs_factory, + '/a/ae': network_layer_tlvs_factory, + '/a/an': network_layer_tlvs_factory, + '/a/sd': network_layer_tlvs_factory, + '/c/lp': mesh_cop_tlvs_factory, + '/c/cs': mesh_cop_tlvs_factory, + '/d/da': network_diag_tlvs_factory, + '/d/dg': network_diag_tlvs_factory, + '/d/dq': network_diag_tlvs_factory, + '/d/dr': network_diag_tlvs_factory, } @@ -483,7 +532,7 @@ def create_default_ipv6_icmp_body_factories(): ipv6.ICMPv6EchoBodyFactory(), ipv6.ICMP_ECHO_RESPONSE: ipv6.ICMPv6EchoBodyFactory(), - "default": + 'default': ipv6.BytesPayloadFactory(), } diff --git a/tests/scripts/thread-cert/network_diag.py b/tests/scripts/thread-cert/network_diag.py new file mode 100644 index 000000000..5cdf1e628 --- /dev/null +++ b/tests/scripts/thread-cert/network_diag.py @@ -0,0 +1,336 @@ +#!/usr/bin/env python3 +# +# Copyright (c) 2020, 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 struct + +from enum import IntEnum +from typing import List + +import common +import ipaddress +import mle + + +class TlvType(IntEnum): + EXT_ADDRESS = 0 + ADDRESS16 = 1 + MODE = 2 + POLLING_PERIOD = 3 + CONNECTIVITY = 4 + ROUTE64 = 5 + LEADER_DATA = 6 + NETWORK_DATA = 7 + IPV6_ADDRESS_LIST = 8 + MAC_COUNTERS = 9 + BATTERY_LEVEL = 14 + SUPPLY_VOLTAGE = 15 + CHILD_TABLE = 16 + CHANNEL_PAGES = 17 + TYPE_LIST = 18 + MAX_CHILD_TIMEOUT = 19 + + +class Ipv6AddressList: + + def __init__(self, addresses: List[ipaddress.IPv6Address]): + self._addresses = addresses + + @property + def addresses(self): + return self._addresses + + def __eq__(self, other): + common.expect_the_same_class(self, other) + return self.addresses == other.addresses + + def __repr__(self): + return f'Ipv6AddressList({self.addresses})' + + +class Ipv6AddressListFactory: + + def parse(self, data, message_info): + addresses = [] + while data.tell() < message_info.length: + addresses.append(ipaddress.IPv6Address(data.read(16))) + return Ipv6AddressList(addresses) + + +class MacCounters: + + def __init__(self, counters: List[int]): + self._counters = counters + + @property + def if_in_unknown_protos(self): + return self._counters[0] + + @property + def if_in_errors(self): + return self._counters[1] + + @property + def if_out_errors(self): + return self._counters[2] + + @property + def if_in_ucast_pkts(self): + return self._counters[3] + + @property + def if_in_broadcast_pkts(self): + return self._counters[4] + + @property + def if_in_discards(self): + return self._counters[5] + + @property + def if_out_ucast_pkts(self): + return self._counters[6] + + @property + def if_out_broadcast_pkts(self): + return self._counters[7] + + @property + def if_out_discards(self): + return self._counters[8] + + @property + def counters(self): + return self._counters + + def __eq__(self, other): + common.expect_the_same_class(self, other) + + return self.counters == other.counters + + def __repr__(self): + return ('MacCounters(' + + f'if_in_unknown_protos={self.if_in_unknown_protos}, ' + + f'if_in_errors={self.if_in_errors}, ' + + f'if_out_errors={self.if_out_errors}, ' + + f'if_in_ucast_pkts={self.if_in_ucast_pkts}, ' + + f'if_in_broadcast_pkts={self.if_in_broadcast_pkts}, ' + + f'if_in_discards={self.if_in_discards}, ' + + f'if_out_ucast_pkts={self.if_out_ucast_pkts}, ' + + f'if_out_broadcast_pkts={self.if_out_broadcast_pkts}, ' + + f'if_out_discards={self.if_out_discards})') + + +class MacCountersFactory: + + def parse(self, data, message_info): + return MacCounters(struct.unpack('>9I', data.read(4 * 9))) + + +class BatteryLevel: + + def __init__(self, battery_level: int): + self._battery_level = battery_level + + @property + def battery_level(self): + return self._battery_level + + def __eq__(self, other): + common.expect_the_same_class(self, other) + + return self.battery_level == other.battery_level + + def __repr__(self): + return f'BatteryLevel(battery_level={self.battery_level})' + + +class BatteryLevelFactory: + + def parse(self, data, message_info): + return BatteryLevel(struct.unpack('>B', data.read(1))[0]) + + +class SupplyVoltage: + + def __init__(self, supply_voltage: int): + self._supply_voltage = supply_voltage + + @property + def supply_voltage(self): + return self._supply_voltage + + def __eq__(self, other): + common.expect_the_same_class(self, other) + + return self.supply_voltage == other.supply_voltage + + def __repr__(self): + return f'SupplyVoltage(supply_voltage={self.supply_voltage})' + + +class SupplyVoltageFactory: + + def parse(self, data, message_info): + return SupplyVoltage(struct.unpack('>H', data.read(2))[0]) + + +class ChildTableEntry: + + def __init__(self, timeout: int, child_id: int, mode: mle.Mode): + self._timeout = timeout + self._child_id = child_id + self._mode = mode + + @property + def timeout(self): + return self._timeout + + @property + def child_id(self): + return self._child_id + + @property + def mode(self): + return self._mode + + def __eq__(self, other): + common.expect_the_same_class(self, other) + + return (self.timeout == other.timeout and + self.child_id == other.child_id and self.mode == other.mode) + + def __repr__(self): + return f'ChildTableEntry(timeout={self.timeout}, child_id={self.child_id}, mode={self.mode})' + + +class ChildTable: + + def __init__(self, children: List[ChildTableEntry]): + self._children = sorted(children, key=lambda child: child.child_id) + + @property + def children(self): + return self._children + + def __eq__(self, other): + common.expect_the_same_class(self, other) + + return self.children == other.children + + def __repr__(self): + return f'ChildTable({self.children})' + + +class ChildTableFactory: + + def parse(self, data, message_info): + children = [] + while message_info.length > 0: + timeout_and_id = struct.unpack('>H', data.read(2))[0] + message_info.length -= 2 + + timeout = (timeout_and_id & 0xf800) >> 11 + child_id = timeout_and_id & 0x1fff + + mode = mle.ModeFactory().parse(data, message_info) + message_info.length -= 1 + + children.append(ChildTableEntry(timeout, child_id, mode)) + return ChildTable(children) + + +class ChannelPages: + + def __init__(self, channel_pages: bytes): + self._channel_pages = channel_pages + + @property + def channel_pages(self): + return self._channel_pages + + def __eq__(self, other): + common.expect_the_same_class(self, other) + + return self.channel_pages == other.channel_pages + + def __repr__(self): + return f'ChannelPages(channel_pages={self.channel_pages})' + + +class ChannelPagesFactory: + + def parse(self, data, message_info): + return ChannelPages(data.getvalue()) + + +class TypeList: + + def __init__(self, tlv_types: List[int]): + self._tlv_types = tlv_types + + @property + def tlv_types(self): + return self._tlv_types + + def __eq__(self, other): + common.expect_the_same_class(self, other) + + return self.tlv_types == other.tlv_types + + def __repr__(self): + return f'TypeList(tlv_types={self.tlv_types})' + + +class TypeListFactory: + + def parse(self, data, message_info): + return TypeList([ord(t) for t in data.getvalue()]) + + +class MaxChildTimeout: + + def __init__(self, max_child_timeout: int): + self._max_child_timeout = max_child_timeout + + @property + def max_child_timeout(self): + return self._max_child_timeout + + def __eq__(self, other): + common.expect_the_same_class(self, other) + + return self.max_child_timeout == other.max_child_timeout + + def __repr__(self): + return f'MaxChildTimeout(max_child_timeout={self.max_child_timeout})' + + +class MaxChildTimeoutFactory: + + def parse(self, data, message_info): + return MaxChildTimeout(struct.unpack('>I', data.read(4))[0]) diff --git a/tests/scripts/thread-cert/node.py b/tests/scripts/thread-cert/node.py index 28672c01a..c7c1fec3f 100755 --- a/tests/scripts/thread-cert/node.py +++ b/tests/scripts/thread-cert/node.py @@ -901,6 +901,30 @@ class Node: self.send_command('netdataregister') self._expect('Done') + def send_network_diag_get(self, addr, tlv_types): + self.send_command('networkdiagnostic get %s %s' % + (addr, ' '.join([str(t.value) for t in tlv_types]))) + + if isinstance(self.simulator, simulator.VirtualTime): + self.simulator.go(8) + timeout = 1 + else: + timeout = 8 + + self._expect('Done', timeout=timeout) + + def send_network_diag_reset(self, addr, tlv_types): + self.send_command('networkdiagnostic reset %s %s' % + (addr, ' '.join([str(t.value) for t in tlv_types]))) + + if isinstance(self.simulator, simulator.VirtualTime): + self.simulator.go(8) + timeout = 1 + else: + timeout = 8 + + self._expect('Done', timeout=timeout) + def energy_scan(self, mask, count, period, scan_duration, ipaddr): cmd = 'commissioner energy %d %d %d %d %s' % ( mask,