#!/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 ipaddress import json import logging import os import subprocess import unittest import otci from otci import OTCI from otci.errors import CommandError from otci import NetifIdentifier logging.basicConfig(level=logging.DEBUG) TEST_CHANNEL = 22 TEST_CHANNEL_MASK = 0x07fff800 TEST_EXTENDED_PANID = '000db80000000000' TEST_MESH_LOCAL_PREFIX = 'fd00:db8::' TEST_NETWORK_KEY = 'ffeeddccbbaa99887766554433221100' TEST_NETWORK_NAME = 'OT CI' TEST_PANID = 0xeeee TEST_PSKC = 'c23a76e98f1a6483639b1ac1271e2e27' TEST_SECURITY_POLICY = (672, 'onrc') REAL_DEVICE = int(os.getenv('REAL_DEVICE', '0')) class TestOTCI(unittest.TestCase): def testCliRealDevice(self): if not REAL_DEVICE: self.skipTest('not for virtual device') if os.getenv('OTBR_SSH'): node = otci.connect_otbr_ssh(os.getenv('OTBR_SSH')) elif os.getenv('OT_CLI_SERIAL'): node = otci.connect_cli_serial(os.getenv('OT_CLI_SERIAL')) else: self.fail("Please set OT_CLI_SERIAL or OTBR_SSH to test the real device.") node.factory_reset() self._test_otci_single_node(node) def testCliSimRealTime(self): if REAL_DEVICE: self.skipTest('not for real device') subprocess.check_call('rm -rf tmp/', shell=True) VIRTUAL_TIME = int(os.getenv('VIRTUAL_TIME', "1")) import simulator if VIRTUAL_TIME: sim = simulator.VirtualTime(use_message_factory=False) else: sim = None if os.getenv('OT_CLI'): executable = os.getenv('OT_CLI') connector = otci.connect_cli_sim elif os.getenv('OT_NCP'): executable = os.getenv('OT_NCP') connector = otci.connect_ncp_sim else: self.fail("Please set OT_CLI to test virtual device") node1 = connector(executable, 1, simulator=sim) self._test_otci_single_node(node1) node1.factory_reset() node2 = connector(executable, 2, simulator=sim) node3 = connector(executable, 3, simulator=sim) node4 = connector(executable, 4, simulator=sim) self._test_otci_example(node1, node2) node1.factory_reset() node2.factory_reset() self._test_otci_multi_nodes(node1, node2, node3, node4) def _test_otci_single_node(self, leader): logging.info('leader version: %r', leader.version) logging.info('leader thread version: %r', leader.thread_version) logging.info('API version: %r', leader.api_version) logging.info('log level: %r', leader.get_log_level()) leader.enable_promiscuous() self.assertTrue(leader.get_promiscuous()) leader.disable_promiscuous() self.assertFalse(leader.get_promiscuous()) try: logging.info("RCP version: %r", leader.get_rcp_version()) except CommandError: pass self.assertTrue(leader.get_router_eligible()) leader.disable_router_eligible() self.assertFalse(leader.get_router_eligible()) leader.enable_router_eligible() self.assertFalse(leader.get_ifconfig_state()) # ifconfig up leader.ifconfig_up() self.assertTrue(leader.get_ifconfig_state()) logging.info('leader eui64 = %r', leader.get_eui64()) logging.info('leader extpanid = %r', leader.get_extpanid()) logging.info('leader networkkey = %r', leader.get_network_key()) extaddr = leader.get_extaddr() self.assertEqual(16, len(extaddr)) int(extaddr, 16) new_extaddr = 'aabbccddeeff0011' leader.set_extaddr(new_extaddr) self.assertEqual(new_extaddr, leader.get_extaddr()) leader.set_network_name(TEST_NETWORK_NAME) leader.set_network_key(TEST_NETWORK_KEY) self.assertEqual(TEST_NETWORK_KEY, leader.get_network_key()) leader.set_panid(TEST_PANID) self.assertEqual(TEST_PANID, leader.get_panid()) leader.set_channel(TEST_CHANNEL) self.assertEqual(TEST_CHANNEL, leader.get_channel()) leader.set_network_name(TEST_NETWORK_NAME) self.assertEqual(TEST_NETWORK_NAME, leader.get_network_name()) self.assertEqual('rdn', leader.get_mode()) leader.set_mode('-') self.assertEqual('-', leader.get_mode()) leader.set_mode('rdn') self.assertEqual('rdn', leader.get_mode()) logging.info('leader weight: %d', leader.get_leader_weight()) leader.set_leader_weight(72) logging.info('domain name: %r', leader.get_domain_name()) leader.set_domain_name("DefaultDomain2") self.assertEqual("DefaultDomain2", leader.get_domain_name()) self.assertEqual(leader.get_preferred_partition_id(), 0) leader.set_preferred_partition_id(0xabcddead) self.assertEqual(leader.get_preferred_partition_id(), 0xabcddead) _setup_default_network(leader) leader.thread_start() leader.wait(10) self.assertEqual('leader', leader.get_state()) self.assertEqual(0xabcddead, leader.get_leader_data()['partition_id']) logging.info('leader key sequence counter = %d', leader.get_key_sequence_counter()) rloc16 = leader.get_rloc16() leader_id = leader.get_router_id() self.assertEqual(rloc16, leader_id << 10) self.assertFalse(leader.get_child_list()) self.assertEqual({}, leader.get_child_table()) leader.enable_allowlist() leader.add_allowlist(leader.get_extaddr()) leader.remove_allowlist(leader.get_extaddr()) leader.set_allowlist([leader.get_extaddr()]) leader.disable_allowlist() self.assertEqual([], leader.backbone_router_get_multicast_listeners()) leader.add_ipmaddr('ff04::1') leader.del_ipmaddr('ff04::1') leader.add_ipmaddr('ff04::2') logging.info('leader ipmaddrs: %r', leader.get_ipmaddrs()) self.assertFalse(leader.has_ipmaddr('ff04::1')) self.assertTrue(leader.has_ipmaddr('ff04::2')) self.assertTrue(leader.get_ipaddr_rloc()) self.assertTrue(leader.get_ipaddr_linklocal()) self.assertTrue(leader.get_ipaddr_mleid()) self.assertTrue(leader.get_ipmaddr_llatn()) self.assertTrue(leader.get_ipmaddr_rlatn()) leader.add_ipaddr('2001::1') leader.del_ipaddr('2001::1') leader.add_ipaddr('2001::2') logging.info('leader ipaddrs: %r', leader.get_ipaddrs()) self.assertFalse(leader.has_ipaddr('2001::1')) self.assertTrue(leader.has_ipaddr('2001::2')) logging.info('leader bbr state: %r', leader.get_backbone_router_state()) bbr_config = leader.get_backbone_router_config() logging.info('leader bbr config: %r', bbr_config) logging.info('leader PBBR: %r', leader.get_primary_backbone_router_info()) new_bbr_seqno = (bbr_config['seqno'] + 1) % 256 leader.set_backbone_router_config(seqno=new_bbr_seqno, delay=10, timeout=301) self.assertEqual({'seqno': new_bbr_seqno, 'delay': 10, 'timeout': 301}, leader.get_backbone_router_config()) leader.enable_backbone_router() leader.wait(3) logging.info('leader bbr state: %r', leader.get_backbone_router_state()) logging.info('leader bbr config: %r', leader.get_backbone_router_config()) logging.info('leader PBBR: %r', leader.get_primary_backbone_router_info()) leader.wait(10) self.assertEqual(1, len(leader.backbone_router_get_multicast_listeners())) self.assertEqual('ff04::2', leader.backbone_router_get_multicast_listeners()[0][0]) logging.info('leader bufferinfo: %r', leader.get_message_buffer_info()) logging.info('child ipaddrs: %r', leader.get_child_ipaddrs()) logging.info('child ipmax: %r', leader.get_child_ip_max()) leader.set_child_ip_max(2) self.assertEqual(2, leader.get_child_ip_max()) logging.info('childmax: %r', leader.get_max_children()) logging.info('counter names: %r', leader.counter_names) for counter_name in leader.counter_names: logging.info('counter %s: %r', counter_name, leader.get_counter(counter_name)) leader.reset_counter(counter_name) self.assertTrue(all(x == 0 for name, x in leader.get_counter(counter_name).items() if "Time" not in name)) logging.info("CSL config: %r", leader.get_csl_config()) leader.config_csl(channel=13, period=16000, timeout=200) logging.info("CSL config: %r", leader.get_csl_config()) logging.info("EID-to-RLOC cache: %r", leader.get_eidcache()) logging.info("leader data: %r", leader.get_leader_data()) logging.info("leader neighbor list: %r", leader.get_neighbor_list()) logging.info("leader neighbor table: %r", leader.get_neighbor_table()) logging.info("Leader external routes: %r", leader.get_local_routes()) leader.add_route('2002::/64') leader.register_network_data() logging.info("Leader external routes: %r", leader.get_local_routes()) self.assertEqual(1, len(leader.get_router_list())) self.assertEqual(1, len(leader.get_router_table())) logging.info("Leader router table: %r", leader.get_router_table()) self.assertFalse(list(leader.get_router_table().values())[0].is_link_established) logging.info('discover: %r', leader.discover()) logging.info('scan: %r', leader.scan()) logging.info('scan energy: %r', leader.scan_energy()) leader.add_service(44970, '112233', 'aabbcc') leader.register_network_data() leader.add_service(44971, b'\x11\x22\x33', b'\xaa\xbb\xcc\xdd') leader.add_prefix("2001::/64") logging.info("network data: %r", leader.get_network_data()) logging.info("network data raw: %r", leader.get_network_data_bytes()) self.assertEqual(leader.get_network_data()['prefixes'], leader.get_prefixes()) self.assertEqual(leader.get_network_data()['routes'], leader.get_routes()) self.assertEqual(leader.get_network_data()['services'], leader.get_services()) logging.info("local prefixes: %r", leader.get_local_prefixes()) logging.info("local routes: %r", leader.get_local_routes()) logging.info('txpower %r', leader.get_txpower()) leader.set_txpower(-10) self.assertEqual(-10, leader.get_txpower()) self.assertTrue(leader.is_singleton()) leader.coap_start() leader.coap_set_test_resource_path('test') leader.coap_test_set_resource_content('helloworld') leader.coap_get(leader.get_ipaddr_rloc(), 'test') leader.coap_put(leader.get_ipaddr_rloc(), 'test', 'con', 'xxx') leader.coap_post(leader.get_ipaddr_rloc(), 'test', 'con', 'xxx') leader.coap_delete(leader.get_ipaddr_rloc(), 'test', 'con', 'xxx') leader.wait(1) leader.coap_stop() for netif in (NetifIdentifier.THERAD, NetifIdentifier.UNSPECIFIED, NetifIdentifier.BACKBONE): leader.udp_open() leader.udp_bind("::", 1234, netif=netif) leader.udp_send(leader.get_ipaddr_rloc(), 1234, text='hello') leader.udp_send(leader.get_ipaddr_rloc(), 1234, random_bytes=3) leader.udp_send(leader.get_ipaddr_rloc(), 1234, hex='112233') leader.wait(1) leader.udp_close() logging.info('dataset: %r', leader.get_dataset()) logging.info('dataset active: %r', leader.get_dataset('active')) leader.dataset_init_buffer() leader.dataset_commit_buffer('pending') leader.dataset_init_buffer(get_active_dataset=True) leader.dataset_init_buffer(get_pending_dataset=True) logging.info('dataset: %r', leader.get_dataset()) logging.info('dataset active: %r', leader.get_dataset('active')) logging.info('dataset pending: %r', leader.get_dataset('pending')) logging.info('dataset active -x: %r', leader.get_dataset_bytes('active')) logging.info('dataset pending -x: %r', leader.get_dataset_bytes('pending')) # Test SRP server & client self._test_otci_srp(leader, leader) # Test DNS client and server self._test_otci_dns(leader, leader) self._test_otci_srp_remove(leader, leader) def _test_otci_dns(self, client: OTCI, server: OTCI): dns_cfg = client.dns_get_config() self.assertTrue(dns_cfg['server']) self.assertIn('response_timeout', dns_cfg) self.assertIn('max_tx_attempts', dns_cfg) self.assertIn('recursion_desired', dns_cfg) client.dns_set_config(server=(server.get_ipaddr_rloc(), 53), response_timeout=10000, max_tx_attempts=4, recursion_desired=False) self.assertEqual( { 'server': (server.get_ipaddr_rloc(), 53), 'response_timeout': 10000, 'max_tx_attempts': 4, 'recursion_desired': False }, client.dns_get_config()) self.assertTrue(client.dns_get_compression()) client.dns_disable_compression() self.assertFalse(client.dns_get_compression()) client.dns_enable_compression() self.assertTrue(client.dns_get_compression()) logging.info('dns browse: %r', client.dns_browse('_ipps._tcp.default.service.arpa.')) logging.info('dns browse: %r', client.dns_browse('_meshcop._udp.default.service.arpa.')) logging.info('dns resolve: %r', client.dns_resolve_service('ins1', '_ipps._tcp.default.service.arpa.')) logging.info('dns resolve: %r', client.dns_resolve('host1.default.service.arpa.')) def _test_otci_srp(self, client: OTCI, server: OTCI): self.assertEqual('disabled', server.srp_server_get_state()) self.assertEqual('default.service.arpa.', server.srp_server_get_domain()) server.srp_server_set_domain('example1.com') self.assertEqual('example1.com.', server.srp_server_get_domain()) server.srp_server_set_domain('example2.com.') self.assertEqual('example2.com.', server.srp_server_get_domain()) server.srp_server_set_domain('default.service.arpa.') self.assertEqual('default.service.arpa.', server.srp_server_get_domain()) default_leases = server.srp_server_get_lease() self.assertEqual(default_leases, (30, 97200, 30, 680400)) server.srp_server_set_lease(1801, 7201, 86401, 1209601) leases = server.srp_server_get_lease() self.assertEqual(leases, (1801, 7201, 86401, 1209601)) self.assertFalse(client.srp_client_get_state()) self.assertEqual('Removed', client.srp_client_get_host_state()) self.assertEqual(('::', 0), client.srp_client_get_server()) self.assertFalse(client.srp_client_get_service_key()) client.srp_client_enable_service_key() self.assertTrue(client.srp_client_get_service_key()) client.srp_client_disable_service_key() self.assertFalse(client.srp_client_get_service_key()) server.srp_server_disable() client.wait(3) server.srp_server_enable() client.wait(10) self.assertEqual([], server.srp_server_get_hosts()) self.assertEqual('running', server.srp_server_get_state()) self.assertTrue(client.srp_client_get_autostart()) client.wait(3) self.assertTrue(client.srp_client_get_state()) self.assertNotEqual(('::', 0), client.srp_client_get_server()) self.assertEqual('', client.srp_client_get_host_name()) client.srp_client_set_host_name('host1') self.assertEqual('host1', client.srp_client_get_host_name()) self.assertEqual([], client.srp_client_get_host_addresses()) client.srp_client_set_host_addresses('2001::1') self.assertEqual(['2001::1'], client.srp_client_get_host_addresses()) client.srp_client_set_host_addresses('2001::1', '2001::2') self.assertEqual(['2001::1', '2001::2'], client.srp_client_get_host_addresses()) srp_client_host = client.srp_client_get_host() self.assertEqual('host1', srp_client_host['host']) self.assertEqual('ToAdd', srp_client_host['state']) self.assertEqual( {ipaddress.IPv6Address('2001::1'), ipaddress.IPv6Address('2001::2')}, set(srp_client_host['addresses'])) self.assertEqual([], client.srp_client_get_services()) client.srp_client_add_service('ins1', '_ipps._tcp', 1000, 1, 1, txt={ 'txt11': 'val11', 'txt12': b'val12', 'txt13': True }) client.srp_client_add_service('ins2', '_meshcop._udp', 2000, 2, 2, txt={ 'txt21': 'val21', 'txt22': b'val22', 'txt23': True }) self.assertEqual(2, len(client.srp_client_get_services())) self.assertIn( { 'instance': 'ins1', 'service': '_ipps._tcp', 'state': 'ToAdd', 'port': 1000, 'priority': 1, 'weight': 1, }, client.srp_client_get_services()) self.assertIn( { 'instance': 'ins2', 'service': '_meshcop._udp', 'state': 'ToAdd', 'port': 2000, 'priority': 2, 'weight': 2, }, client.srp_client_get_services()) client.wait(3) self.assertEqual('Registered', client.srp_client_get_host()['state']) srp_server_hosts = server.srp_server_get_hosts() logging.info('srp_server_hosts %r', srp_server_hosts) self.assertEqual(1, len(srp_server_hosts)) self.assertEqual('host1.default.service.arpa.', srp_server_hosts[0]['host']) self.assertEqual(False, srp_server_hosts[0]['deleted']) self.assertEqual( {ipaddress.IPv6Address('2001::1'), ipaddress.IPv6Address('2001::2')}, set(srp_server_hosts[0]['addresses'])) srp_server_services = server.srp_server_get_services() logging.info('srp_server_services %r', srp_server_services) self.assertEqual(2, len(srp_server_services)) for service in srp_server_services: if service['instance'] == 'ins1._ipps._tcp.default.service.arpa.': self.assertEqual(False, service['deleted']) self.assertEqual(1000, service['port']) self.assertEqual(1, service['priority']) self.assertEqual(1, service['weight']) self.assertEqual('host1.default.service.arpa.', service['host']) self.assertEqual({ipaddress.IPv6Address('2001::1'), ipaddress.IPv6Address('2001::2')}, set(service['addresses'])) self.assertEqual({'txt11': b'val11', 'txt12': b'val12', 'txt13': True}, service['txt']) elif service['instance'] == 'ins2._meshcop._udp.default.service.arpa.': self.assertEqual(False, service['deleted']) self.assertEqual(2000, service['port']) self.assertEqual(2, service['priority']) self.assertEqual(2, service['weight']) self.assertEqual('host1.default.service.arpa.', service['host']) self.assertEqual({ipaddress.IPv6Address('2001::1'), ipaddress.IPv6Address('2001::2')}, set(service['addresses'])) self.assertEqual({'txt21': b'val21', 'txt22': b'val22', 'txt23': True}, service['txt']) else: self.fail(service) def _test_otci_srp_remove(self, client: OTCI, server: OTCI): client.srp_client_remove_host(remove_key_lease=True) client.wait(3) self.assertEqual([], client.srp_client_get_services()) self.assertEqual('Removed', client.srp_client_get_host()['state']) self.assertEqual([], server.srp_server_get_hosts()) self.assertEqual([], server.srp_server_get_services()) def _test_otci_example(self, node1, node2): node1.dataset_init_buffer() node1.dataset_set_buffer(network_name='test', network_key='00112233445566778899aabbccddeeff', panid=0xface, channel=11) node1.dataset_commit_buffer('active') node1.ifconfig_up() node1.thread_start() node1.wait(10) assert node1.get_state() == "leader" node1.commissioner_start() node1.wait(3) node1.commissioner_add_joiner("TEST123", eui64='*') node2.ifconfig_up() node2.set_router_selection_jitter(1) node2.joiner_start("TEST123") node2.wait(10, expect_line="Join success") node2.thread_start() node2.wait(10) assert node2.get_state() == "router" def _test_otci_multi_nodes(self, leader, commissioner, child1, child2): self.assertFalse(leader.get_ifconfig_state()) # ifconfig up leader.ifconfig_up() self.assertTrue(leader.get_ifconfig_state()) logging.info('leader eui64 = %r', leader.get_eui64()) logging.info('leader extpanid = %r', leader.get_extpanid()) logging.info('leader networkkey = %r', leader.get_network_key()) extaddr = leader.get_extaddr() self.assertEqual(16, len(extaddr)) int(extaddr, 16) new_extaddr = 'aabbccddeeff0011' leader.set_extaddr(new_extaddr) self.assertEqual(new_extaddr, leader.get_extaddr()) _setup_default_network(leader) self.assertEqual(TEST_CHANNEL, leader.get_channel()) self.assertEqual(TEST_NETWORK_KEY, leader.get_network_key()) self.assertEqual(TEST_NETWORK_NAME, leader.get_network_name()) self.assertEqual(TEST_PANID, leader.get_panid()) self.assertEqual('rdn', leader.get_mode()) leader.thread_start() leader.wait(10) self.assertEqual('leader', leader.get_state()) logging.info('leader key sequence counter = %d', leader.get_key_sequence_counter()) rloc16 = leader.get_rloc16() leader_id = leader.get_router_id() self.assertEqual(rloc16, leader_id << 10) commissioner.dataset_clear_buffer() commissioner.dataset_set_buffer( channel=TEST_CHANNEL, network_key=TEST_NETWORK_KEY, panid=TEST_PANID, ) commissioner.dataset_commit_buffer('active') commissioner.set_router_selection_jitter(1) commissioner.ifconfig_up() commissioner.thread_start() commissioner.wait(10) self.assertEqual('router', commissioner.get_state()) for dst_ip in leader.get_ipaddrs(): statistics = commissioner.ping(dst_ip, size=10, count=10, interval=2, hoplimit=3) self.assertEqual(statistics['transmitted_packets'], 10) self.assertEqual(statistics['received_packets'], 10) self.assertAlmostEqual(statistics['packet_loss'], 0.0, delta=1e-9) rtt = statistics['round_trip_time'] self.assertTrue(rtt['min'] - 1e-9 <= rtt['avg'] <= rtt['max'] + 1e-9) commissioner.wait(1) self.assertEqual('disabled', commissioner.get_commissioiner_state()) commissioner.commissioner_start() commissioner.wait(5) self.assertEqual('active', commissioner.get_commissioiner_state()) logging.info('commissioner.get_network_id_timeout() = %d', commissioner.get_network_id_timeout()) commissioner.set_network_id_timeout(60) self.assertEqual(60, commissioner.get_network_id_timeout()) commissioner.commissioner_add_joiner('TEST123', eui64='*') commissioner.wait(3) child1.ifconfig_up() logging.info("child1 discover: %r", child1.discover()) logging.info("child1 scan: %r", child1.scan()) logging.info("child1 scan energy: %r", child1.scan_energy()) child1.set_mode('rn') child1.set_router_selection_jitter(1) child1.joiner_start('TEST123') logging.info('joiner id = %r', child1.get_joiner_id()) child1.wait(10, expect_line="Join success") child1.enable_allowlist() child1.disable_allowlist() child1.add_allowlist(commissioner.get_extaddr()) child1.remove_allowlist(commissioner.get_extaddr()) child1.set_allowlist([commissioner.get_extaddr()]) child1.thread_start() child1.wait(3) self.assertEqual('child', child1.get_state()) child1.thread_stop() child1.set_mode('n') child1.set_poll_period(1000) self.assertEqual(1000, child1.get_poll_period()) child1.thread_start() child1.wait(3) self.assertEqual('child', child1.get_state()) child2.ifconfig_up() child2.set_mode('rn') child2.set_router_selection_jitter(1) child2.joiner_start('TEST123') logging.info('joiner id = %r', child2.get_joiner_id()) child2.wait(10, expect_line="Join success") child2.enable_allowlist() child2.disable_allowlist() child2.add_allowlist(commissioner.get_extaddr()) child2.remove_allowlist(commissioner.get_extaddr()) child2.set_allowlist([commissioner.get_extaddr()]) child2.thread_start() child2.wait(3) self.assertEqual('child', child2.get_state()) child_table = commissioner.get_child_table() logging.info('commissioiner child table: \n%s\n', json.dumps(child_table, indent=True)) child_list = commissioner.get_child_list() logging.info('commissioiner child list: %r', child_list) for child_id in child_list: logging.info('child %s info: %r', child_id, commissioner.get_child_info(child_id)) logging.info('child1 info: %r', commissioner.get_child_info(child1.get_rloc16())) logging.info('child2 info: %r', commissioner.get_child_info(child2.get_rloc16())) self.assertEqual(set(commissioner.get_child_list()), set(commissioner.get_child_table().keys())) child1.add_ipmaddr('ff04::1') child1.del_ipmaddr('ff04::1') child1.add_ipmaddr('ff04::2') logging.info('child1 ipmaddrs: %r', child1.get_ipmaddrs()) self.assertFalse(child1.has_ipmaddr('ff04::1')) self.assertTrue(child1.has_ipmaddr('ff04::2')) child1.add_ipaddr('2001::1') child1.del_ipaddr('2001::1') child1.add_ipaddr('2001::2') logging.info('child1 ipaddrs: %r', child1.get_ipaddrs()) self.assertFalse(child1.has_ipaddr('2001::1')) self.assertTrue(child1.has_ipaddr('2001::2')) logging.info('child ipaddrs: %r', commissioner.get_child_ipaddrs()) logging.info("EID-to-RLOC cache: %r", leader.get_eidcache()) logging.info("leader neighbor list: %r", leader.get_neighbor_list()) logging.info("leader neighbor table: %r", leader.get_neighbor_table()) logging.info("prefixes: %r", commissioner.get_local_prefixes()) commissioner.add_prefix('2001::/64') commissioner.register_network_data() commissioner.wait(1) logging.info("prefixes: %r", commissioner.get_local_prefixes()) self.assertEqual(2, len(leader.get_router_list())) self.assertEqual(2, len(leader.get_router_table())) logging.info('leader router table: %r', leader.get_router_table()) self.assertEqual({False, True}, set(router.is_link_established for router in leader.get_router_table().values())) self.assertFalse(leader.is_singleton()) statistics = commissioner.ping("ff02::1", size=1, count=10, interval=1, hoplimit=255) self.assertEqual(statistics['transmitted_packets'], 10) self.assertEqual(statistics['received_packets'], 20) rtt = statistics['round_trip_time'] self.assertTrue(rtt['min'] - 1e-9 <= rtt['avg'] <= rtt['max'] + 1e-9) # Shutdown leader.thread_stop() logging.info("node state: %s", leader.get_state()) leader.ifconfig_down() self.assertFalse(leader.get_ifconfig_state()) leader.close() def _setup_default_network(node): node.dataset_clear_buffer() node.dataset_set_buffer( active_timestamp=1, channel=TEST_CHANNEL, channel_mask=TEST_CHANNEL_MASK, extpanid=TEST_EXTENDED_PANID, mesh_local_prefix=TEST_MESH_LOCAL_PREFIX, network_key=TEST_NETWORK_KEY, network_name=TEST_NETWORK_NAME, panid=TEST_PANID, pskc=TEST_PSKC, security_policy=TEST_SECURITY_POLICY, ) node.dataset_commit_buffer('active') if __name__ == '__main__': unittest.main()