diff --git a/tests/scripts/thread-cert/test_srp_auto_host_address.py b/tests/scripts/thread-cert/test_srp_auto_host_address.py deleted file mode 100755 index b672730c2..000000000 --- a/tests/scripts/thread-cert/test_srp_auto_host_address.py +++ /dev/null @@ -1,341 +0,0 @@ -#!/usr/bin/env python3 -# -# Copyright (c) 2022, 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 unittest - -import command -import config -import thread_cert - -# Test description: -# This test verifies SRP client auto host address mode. -# -# Topology: -# SRP client (leader) -# | -# | -# SRP server (router) -# - -CLIENT = 1 -SERVER = 2 - - -class SrpAutoHostAddress(thread_cert.TestCase): - USE_MESSAGE_FACTORY = False - SUPPORT_NCP = False - - TOPOLOGY = { - CLIENT: { - 'name': 'SRP_CLIENT', - 'mode': 'rdn', - }, - SERVER: { - 'name': 'SRP_SERVER', - 'mode': 'rdn', - }, - } - - def test(self): - client = self.nodes[CLIENT] - server = self.nodes[SERVER] - - # Deprecation interval of an SLAAC address before removal. - deprecate_time = 300 - - #------------------------------------------------------------------- - # Form the network. - - client.srp_server_set_enabled(False) - client.start() - self.simulator.go(15) - self.assertEqual(client.get_state(), 'leader') - client.srp_client_stop() - - server.start() - self.simulator.go(5) - self.assertEqual(server.get_state(), 'router') - - #------------------------------------------------------------------- - # Enable SRP server - - server.srp_server_set_enabled(True) - self.simulator.go(5) - - #------------------------------------------------------------------- - # Check auto start mode on SRP client - - client.srp_client_enable_auto_start_mode() - self.assertEqual(client.srp_client_get_auto_start_mode(), 'Enabled') - self.simulator.go(15) - - self.assertEqual(client.srp_client_get_state(), 'Enabled') - - #------------------------------------------------------------------- - # Set host name and enable auto host address on client - - client.srp_client_set_host_name('host') - client.srp_client_enable_auto_host_address() - - #------------------------------------------------------------------- - # Register a service on client - - client.srp_client_add_service('test_srv', '_test._udo', 12345, 0, 0) - self.simulator.go(2) - self.check_registered_addresses(client, server) - - #------------------------------------------------------------------- - # Add an address and check the SRP client re-registered and updated - # server with new address. - - client.add_ipaddr('fd00:1:2:3:4:5:6:7') - - self.simulator.go(5) - client_addresses = [addr.strip() for addr in client.get_addrs()] - self.assertIn('fd00:1:2:3:4:5:6:7', client_addresses) - self.check_registered_addresses(client, server) - - #------------------------------------------------------------------- - # Remove the address and check the SRP client re-registered and updated - # server. - - client.del_ipaddr('fd00:1:2:3:4:5:6:7') - - self.simulator.go(5) - client_addresses = [addr.strip() for addr in client.get_addrs()] - self.assertNotIn('fd00:1:2:3:4:5:6:7', client_addresses) - self.check_registered_addresses(client, server) - - #------------------------------------------------------------------- - # Add an SLAAC on-mesh prefix (which will trigger an address to be - # added) and check that the SRP client re-registered and updated - # server with the new address. - - client.add_prefix('fd00:abba:cafe:bee::/64', 'paos') - client.register_netdata() - self.simulator.go(15) - - slaac_addr = [addr.strip() for addr in client.get_addrs() if addr.strip().startswith('fd00:abba:cafe:bee:')] - self.assertEqual(len(slaac_addr), 1) - self.check_registered_addresses(client, server) - - #------------------------------------------------------------------- - # Add another SLAAC on-mesh prefix and check that the SRP client - # re-registered and updated server with all address. - - client.add_prefix('fd00:9:8:7::/64', 'paos') - client.register_netdata() - self.simulator.go(15) - - slaac_addr = [addr.strip() for addr in client.get_addrs() if addr.strip().startswith('fd00:9:8:7:')] - self.assertEqual(len(slaac_addr), 1) - self.check_registered_addresses(client, server) - - #------------------------------------------------------------------- - # Add a non-preferred SLAAC on-mesh prefix and check that the - # set of registered addresses remains unchanged and that the - # non-preferred address is not registered by SRP client. - - client.add_prefix('fd00:a:b:c::/64', 'aos') - client.register_netdata() - self.simulator.go(15) - - slaac_addr = [addr.strip() for addr in client.get_addrs() if addr.strip().startswith('fd00:a:b:c:')] - self.assertEqual(len(slaac_addr), 1) - self.check_registered_addresses(client, server) - - #------------------------------------------------------------------- - # Remove the on-mesh prefix. This should trigger the - # associated SLAAC address to be deprecated, but it should - # not yet cause the client to re-register. Verify that the - # registered addresses on server remain unchanged. - - old_registered_addresses = self.get_registered_host_addresses_from_server(server) - - client.remove_prefix('fd00:abba:cafe:bee::/64') - client.register_netdata() - - self.simulator.go(15) - self.assertEqual(old_registered_addresses, self.get_registered_host_addresses_from_server(server)) - - # Wait until the SLAAC address deprecation time has elapsed - # and the address is removed. Verify that the SRP client - # re-registers and updates the server with the remaining - # address. - - self.simulator.go(deprecate_time) - - self.check_registered_addresses(client, server) - - #------------------------------------------------------------------- - # Remove the next on-mesh prefix. Verify that the client does - # not re-register while the address is deprecating. After the - # address is removed, confirm that the SRP client - # re-registers using only the ML-EID. - - old_registered_addresses = self.get_registered_host_addresses_from_server(server) - - client.remove_prefix('fd00:9:8:7::/64') - client.register_netdata() - - self.simulator.go(15) - self.assertEqual(old_registered_addresses, self.get_registered_host_addresses_from_server(server)) - - self.simulator.go(deprecate_time) - - self.check_registered_addresses(client, server) - - #------------------------------------------------------------------- - # Add and remove the on-mesh prefix again. However, before the - # address deprecation time elapses and the address is removed, - # restart the server. This should trigger the client to - # re-register. Verify that the client re-registers with the - # most up-to-date addresses and does not register the deprecating - # address. - - client.add_prefix('fd00:9:8:7::/64', 'paos') - client.register_netdata() - self.simulator.go(15) - - slaac_addr = [addr.strip() for addr in client.get_addrs() if addr.strip().startswith('fd00:9:8:7:')] - self.assertEqual(len(slaac_addr), 1) - self.check_registered_addresses(client, server) - - # Remove the prefix and verify that client does not - # register while the SLAAC address is deprecating. - - old_registered_addresses = self.get_registered_host_addresses_from_server(server) - - client.remove_prefix('fd00:9:8:7::/64') - client.register_netdata() - - self.simulator.go(15) - self.assertEqual(old_registered_addresses, self.get_registered_host_addresses_from_server(server)) - - # Disable and re-enable the server. This should trigger the - # client to re-register. Verify that the ML-EID address is - # now registered. - - server.srp_server_set_enabled(False) - server.srp_server_set_enabled(True) - - self.simulator.go(20) - - self.check_registered_addresses(client, server) - - registered_addresses = self.get_registered_host_addresses_from_server(server) - self.assertEqual(len(registered_addresses), 1) - self.assertEqual(registered_addresses[0], client.get_mleid()) - - # Check that SLAAC address is still deprecating. - - slaac_addr = [addr.strip() for addr in client.get_addrs() if addr.strip().startswith('fd00:9:8:7:')] - self.assertEqual(len(slaac_addr), 1) - - #------------------------------------------------------------------- - # Explicitly set the host addresses (which disables the auto host - # address mode) and check that only the new addresses are registered. - - client.srp_client_set_host_address('fd00:f:e:d:c:b:a:9') - self.simulator.go(5) - - self.assertEqual(client.srp_client_get_host_state(), 'Registered') - server_hosts = server.srp_server_get_hosts() - self.assertEqual(len(server_hosts), 1) - server_host = server_hosts[0] - self.assertEqual(server_host['deleted'], 'false') - self.assertEqual(server_host['fullname'], 'host.default.service.arpa.') - host_addresses = [addr.strip() for addr in server_host['addresses']] - self.assertEqual(len(host_addresses), 1) - self.assertEqual(host_addresses[0], 'fd00:f:e:d:c:b:a:9') - - #------------------------------------------------------------------- - # Re-enable auto host address mode and check that addresses are - # updated and registered properly. - - client.srp_client_enable_auto_host_address() - self.simulator.go(5) - self.check_registered_addresses(client, server) - - def get_registered_host_addresses_from_server(self, server): - # Check the host info on server. - server_hosts = server.srp_server_get_hosts() - self.assertEqual(len(server_hosts), 1) - server_host = server_hosts[0] - self.assertEqual(server_host['deleted'], 'false') - self.assertEqual(server_host['fullname'], 'host.default.service.arpa.') - return [addr.strip() for addr in server_host['addresses']] - - def check_registered_addresses(self, client, server): - # Ensure client has registered successfully. - self.assertEqual(client.srp_client_get_host_state(), 'Registered') - - # Check the host addresses on server to match client. - - host_addresses = self.get_registered_host_addresses_from_server(server) - - client_mleid = client.get_mleid() - client_addresses = [addr.split(' ')[0] for addr in client.get_addrs(verbose=True) if 'preferred:1' in addr] - client_addresses += [client_mleid] - - # All registered addresses must be in client list of addresses. - - for addr in host_addresses: - self.assertIn(addr, client_addresses) - - # All preferred addresses on client excluding link-local and - # mesh-local addresses must be seen on server side. But if there - # was no address, then mesh-local address should be the only - # one registered. - - checked_address = False - - for addr in client_addresses: - if not self.is_address_link_local(addr) and not self.is_address_locator(addr) and addr != client_mleid: - self.assertIn(addr, host_addresses) - checked_address = True - - if not checked_address: - self.assertEqual(len(host_addresses), 1) - self.assertIn(client_mleid, host_addresses) - - def is_address_locator(self, addr): - # Checks if an IPv6 address is a locator (IID should match `0:ff:fe00:xxxx`) - u32s = addr.split(':') - self.assertEqual(len(u32s), 8) - return ':'.join(u32s[4:]).startswith('0:ff:fe00:') - - def is_address_link_local(self, addr): - # Checks if an IPv6 address is link-local - return addr.startswith('fe80:') - - -if __name__ == '__main__': - unittest.main() diff --git a/tests/scripts/thread-cert/test_srp_lease.py b/tests/scripts/thread-cert/test_srp_lease.py deleted file mode 100755 index fa5864f96..000000000 --- a/tests/scripts/thread-cert/test_srp_lease.py +++ /dev/null @@ -1,240 +0,0 @@ -#!/usr/bin/env python3 -# -# Copyright (c) 2021, 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 unittest - -import command -import config -import thread_cert - -# Test description: -# This test verifies the SRP server and client properly handle SRP host -# and service instance lease. -# -# Topology: -# LEADER (SRP server) -# | -# | -# ROUTER (SRP client) -# - -SERVER = 1 -CLIENT = 2 -LEASE = 10 # Seconds -KEY_LEASE = 20 # Seconds - - -class SrpRegisterSingleService(thread_cert.TestCase): - USE_MESSAGE_FACTORY = False - SUPPORT_NCP = False - - TOPOLOGY = { - SERVER: { - 'name': 'SRP_SERVER', - 'network_key': '00112233445566778899aabbccddeeff', - 'mode': 'rdn', - }, - CLIENT: { - 'name': 'SRP_CLIENT', - 'network_key': '00112233445566778899aabbccddeeff', - 'mode': 'rdn', - }, - } - - def test(self): - server = self.nodes[SERVER] - client = self.nodes[CLIENT] - - # - # 0. Start the server and client devices. - # - - server.srp_server_set_enabled(True) - server.srp_server_set_lease_range(LEASE, LEASE, KEY_LEASE, KEY_LEASE) - server.start() - self.simulator.go(config.LEADER_STARTUP_DELAY) - self.assertEqual(server.get_state(), 'leader') - self.simulator.go(5) - - client.start() - self.simulator.go(config.ROUTER_STARTUP_DELAY) - self.assertEqual(client.get_state(), 'router') - - # - # 1. Register a single service and verify that it works. - # - - self.assertEqual(client.srp_client_get_auto_start_mode(), 'Enabled') - - client.srp_client_set_host_name('my-host') - client.srp_client_set_host_address('2001::1') - client.srp_client_add_service('my-service', '_ipps._tcp', 12345) - self.simulator.go(2) - - self.check_host_and_service(server, client) - - # - # 2. Stop the client and wait for the service instance LEASE to expire. - # - - client.srp_client_stop() - self.simulator.go(LEASE + 1) - - # The SRP server should remove the host and service but retain their names - # since the the KEY LEASE hasn't expired yet. - self.assertEqual(server.srp_server_get_host('my-host')['deleted'], 'true') - self.assertEqual(server.srp_server_get_service('my-service', '_ipps._tcp')['deleted'], 'true') - - # Start the client again, the same service should be successfully registered. - client.srp_client_enable_auto_start_mode() - self.simulator.go(15) - - self.check_host_and_service(server, client) - - # - # 3. Stop the client and wait for the KEY LEASE to expire. - # The host and service instance should be fully removed by the SRP server. - # - - client.srp_client_stop() - self.simulator.go(KEY_LEASE + 1) - - # The host and service are expected to be fully removed. - self.assertEqual(len(server.srp_server_get_hosts()), 0) - self.assertEqual(len(server.srp_server_get_services()), 0) - - # Start the client again, the same service should be successfully registered. - client.srp_client_enable_auto_start_mode() - self.simulator.go(15) - - self.check_host_and_service(server, client) - - # - # 4. Clear the first service, lengthen the lease time and register a second service. - # Verify that the lease time of the first service is not affected by new SRP - # registrations. - # - - client.srp_client_clear_service('my-service', '_ipps._tcp') - server.srp_server_set_lease_range(50 * LEASE, 50 * LEASE, 50 * KEY_LEASE, 50 * KEY_LEASE) - client.srp_client_add_service('my-service2', '_ipps._tcp', 12345) - - # Wait for the first service to expire. - self.simulator.go(LEASE + 2) - self.assertEqual(server.srp_server_get_service('my-service', '_ipps._tcp')['deleted'], 'true') - - # Wait for the first service to be fully removed. - self.simulator.go(KEY_LEASE - LEASE + 2) - self.assertEqual(len(server.srp_server_get_services()), 1) - self.assertEqual(len(server.srp_server_get_hosts()), 1) - - # - # 5. Clear the second service, shorten the lease time and register a third service. - # Verify that the lease time of the second service is not affected by new SRP - # registrations. - # - - client.srp_client_clear_service('my-service2', '_ipps._tcp') - server.srp_server_set_lease_range(LEASE, LEASE, KEY_LEASE, KEY_LEASE) - client.srp_client_add_service('my-service3', '_ipps._tcp', 12345) - - # The second service has lease time of 50 * LEASE and should not expire. - self.simulator.go(LEASE + 2) - self.assertEqual(server.srp_server_get_service('my-service2', '_ipps._tcp')['deleted'], 'false') - - # The second service has key-lease time of 50 * KEY_LEASE and should not expire. - self.simulator.go(KEY_LEASE - LEASE + 2) - self.assertEqual(len(server.srp_server_get_services()), 2) - self.assertEqual(len(server.srp_server_get_hosts()), 1) - - # - # 6. Clear the third service. The host and services should expire in lease time. - # Verify that the second service and the third service are removed when their host - # expires. - # - client.srp_client_clear_service('my-service3', '_ipps._tcp') - self.simulator.go(LEASE + 2) - self.assertEqual(len(server.srp_server_get_services()), 2) - self.assertEqual(server.srp_server_get_service('my-service2', '_ipps._tcp')['deleted'], 'true') - self.assertEqual(server.srp_server_get_service('my-service3', '_ipps._tcp')['deleted'], 'true') - self.assertEqual(len(server.srp_server_get_hosts()), 1) - self.assertEqual(server.srp_server_get_host('my-host')['deleted'], 'true') - - def check_host_and_service(self, server, client): - """Check that we have properly registered host and service instance. - """ - - client_services = client.srp_client_get_services() - print(client_services) - self.assertEqual(len(client_services), 1) - client_service = client_services[0] - - # Verify that the client possesses correct service resources. - self.assertEqual(client_service['instance'], 'my-service') - self.assertEqual(client_service['name'], '_ipps._tcp') - self.assertEqual(int(client_service['port']), 12345) - self.assertEqual(int(client_service['priority']), 0) - self.assertEqual(int(client_service['weight']), 0) - - # Verify that the client received a SUCCESS response for the server. - self.assertEqual(client_service['state'], 'Registered') - - # Wait for a KEY LEASE period to make sure that the client has refreshed - # the host and service instance. - self.simulator.go(KEY_LEASE + 1) - - server_services = server.srp_server_get_services() - print(server_services) - self.assertEqual(len(server_services), 1) - server_service = server_services[0] - - # Verify that the server accepted the SRP registration and stored - # the same service resources. - self.assertEqual(server_service['deleted'], 'false') - self.assertEqual(server_service['instance'], client_service['instance']) - self.assertEqual(server_service['name'], client_service['name']) - self.assertEqual(int(server_service['port']), int(client_service['port'])) - self.assertEqual(int(server_service['priority']), int(client_service['priority'])) - self.assertEqual(int(server_service['weight']), int(client_service['weight'])) - self.assertEqual(server_service['host'], 'my-host') - - server_hosts = server.srp_server_get_hosts() - print(server_hosts) - self.assertEqual(len(server_hosts), 1) - server_host = server_hosts[0] - - self.assertEqual(server_host['deleted'], 'false') - self.assertEqual(server_host['fullname'], server_service['host_fullname']) - self.assertEqual(len(server_host['addresses']), 1) - self.assertEqual(ipaddress.ip_address(server_host['addresses'][0]), ipaddress.ip_address('2001::1')) - - -if __name__ == '__main__': - unittest.main() diff --git a/tests/scripts/thread-cert/test_srp_name_conflicts.py b/tests/scripts/thread-cert/test_srp_name_conflicts.py deleted file mode 100755 index 4f41258c1..000000000 --- a/tests/scripts/thread-cert/test_srp_name_conflicts.py +++ /dev/null @@ -1,305 +0,0 @@ -#!/usr/bin/env python3 -# -# Copyright (c) 2021, 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 unittest - -import command -import config -import thread_cert - -# Test description: -# This test verifies if the SRP server can handle name conflicts correctly. -# -# Topology: -# LEADER (SRP server) -# / \ -# / \ -# / \ -# ROUTER1 ROUTER2 -# - -SERVER = 1 -CLIENT1 = 2 -CLIENT2 = 3 - - -class SrpNameConflicts(thread_cert.TestCase): - USE_MESSAGE_FACTORY = False - SUPPORT_NCP = False - - TOPOLOGY = { - SERVER: { - 'name': 'SRP_SERVER', - 'network_key': '00112233445566778899aabbccddeeff', - 'mode': 'rdn', - }, - CLIENT1: { - 'name': 'SRP_CLIENT1', - 'network_key': '00112233445566778899aabbccddeeff', - 'mode': 'rdn', - }, - CLIENT2: { - 'name': 'SRP_CLIENT2', - 'network_key': '00112233445566778899aabbccddeeff', - 'mode': 'rdn', - }, - } - - def test(self): - server = self.nodes[SERVER] - client_1 = self.nodes[CLIENT1] - client_2 = self.nodes[CLIENT2] - - # - # 0. Start the server & client devices. - # - - server.srp_server_set_enabled(True) - server.start() - self.simulator.go(config.LEADER_STARTUP_DELAY) - self.assertEqual(server.get_state(), 'leader') - self.simulator.go(5) - - client_1.srp_server_set_enabled(False) - client_1.start() - self.simulator.go(config.ROUTER_STARTUP_DELAY) - self.assertEqual(client_1.get_state(), 'router') - - client_2.srp_server_set_enabled(False) - client_2.start() - self.simulator.go(config.ROUTER_STARTUP_DELAY) - self.assertEqual(client_2.get_state(), 'router') - - # - # 1. Register a single service and verify that it works. - # - - self.assertEqual(client_1.srp_client_get_auto_start_mode(), 'Enabled') - - client_1.srp_client_set_host_name('my-host-1') - client_1.srp_client_set_host_address('2001::1') - client_1.srp_client_add_service('my-service-1', '_ipps._tcp', 12345) - self.simulator.go(2) - - # Verify that the client possesses correct service resources. - client_1_service = client_1.srp_client_get_services()[0] - self.assertEqual(client_1_service['instance'], 'my-service-1') - self.assertEqual(client_1_service['name'], '_ipps._tcp') - self.assertEqual(int(client_1_service['port']), 12345) - self.assertEqual(int(client_1_service['priority']), 0) - self.assertEqual(int(client_1_service['weight']), 0) - - # Verify that the client receives a SUCCESS response for the server. - self.assertEqual(client_1_service['state'], 'Registered') - - # Verify that the server accepts the SRP registration and stored - # the same service resources. - server_service = server.srp_server_get_services()[0] - self.assertEqual(server_service['deleted'], 'false') - self.assertEqual(server_service['instance'], client_1_service['instance']) - self.assertEqual(server_service['name'], client_1_service['name']) - self.assertEqual(int(server_service['port']), int(client_1_service['port'])) - self.assertEqual(int(server_service['priority']), int(client_1_service['priority'])) - self.assertEqual(int(server_service['weight']), int(client_1_service['weight'])) - self.assertEqual(server_service['host'], 'my-host-1') - - server_host = server.srp_server_get_hosts()[0] - self.assertEqual(server_host['deleted'], 'false') - self.assertEqual(server_host['fullname'], server_service['host_fullname']) - self.assertEqual(len(server_host['addresses']), 1) - self.assertEqual(ipaddress.ip_address(server_host['addresses'][0]), ipaddress.ip_address('2001::1')) - - # - # 2. Register with the same host name from the second client and it should fail. - # - - self.assertEqual(client_2.srp_client_get_auto_start_mode(), 'Enabled') - - client_2.srp_client_set_host_name('my-host-1') - client_2.srp_client_set_host_address('2001::2') - client_2.srp_client_add_service('my-service-2', '_ipps._tcp', 12345) - self.simulator.go(2) - - # It is expected that the registration will be rejected. - client_2_service = client_2.srp_client_get_services()[0] - self.assertNotEqual(client_2_service['state'], 'Registered') - self.assertNotEqual(client_2.srp_client_get_host_state(), 'Registered') - - self.assertEqual(len(server.srp_server_get_services()), 1) - self.assertEqual(len(server.srp_server_get_hosts()), 1) - - client_2.srp_client_clear_host() - client_2.srp_client_stop() - - # - # 3. Register with the same service name from the second client and it should fail. - # - - client_2.srp_client_enable_auto_start_mode() - client_2.srp_client_set_host_name('my-host-2') - client_2.srp_client_set_host_address('2001::2') - client_2.srp_client_add_service('my-service-1', '_ipps._tcp', 12345) - self.simulator.go(2) - - # It is expected that the registration will be rejected. - client_2_service = client_2.srp_client_get_services()[0] - self.assertNotEqual(client_2_service['state'], 'Registered') - self.assertNotEqual(client_2.srp_client_get_host_state(), 'Registered') - - self.assertEqual(len(server.srp_server_get_services()), 1) - self.assertEqual(len(server.srp_server_get_hosts()), 1) - - client_2.srp_client_clear_host() - client_2.srp_client_stop() - - # - # 4. Register the same service instance label with a different service name - # from the second client and it should pass. - # - - client_2.srp_client_enable_auto_start_mode() - client_2.srp_client_set_host_name('my-host-2') - client_2.srp_client_set_host_address('2001::2') - client_2.srp_client_add_service('my-service-1', '_ipps2._tcp', 12345) - self.simulator.go(2) - - # It is expected that the registration will be accepted. - client_2_service = client_2.srp_client_get_services()[0] - self.assertEqual(client_2_service['state'], 'Registered') - self.assertEqual(client_2.srp_client_get_host_state(), 'Registered') - - self.assertEqual(len(server.srp_server_get_services()), 2) - self.assertEqual(len(server.srp_server_get_hosts()), 2) - self.assertEqual(server.srp_server_get_host('my-host-2')['deleted'], 'false') - self.assertEqual(server.srp_server_get_service('my-service-1', '_ipps2._tcp')['deleted'], 'false') - - # Remove the host and all services registered on the SRP server. - client_2.srp_client_remove_host(remove_key=True) - self.simulator.go(2) - - client_2.srp_client_clear_host() - client_2.srp_client_stop() - - # - # 5. Register with different host & service instance name, it should succeed. - # - - client_2.srp_client_enable_auto_start_mode() - client_2.srp_client_set_host_name('my-host-2') - client_2.srp_client_set_host_address('2001::2') - client_2.srp_client_add_service('my-service-2', '_ipps._tcp', 12345) - self.simulator.go(2) - - # It is expected that the registration will be accepted. - client_2_service = client_2.srp_client_get_services()[0] - self.assertEqual(client_2_service['state'], 'Registered') - self.assertEqual(client_2.srp_client_get_host_state(), 'Registered') - - self.assertEqual(len(server.srp_server_get_services()), 2) - self.assertEqual(len(server.srp_server_get_hosts()), 2) - self.assertEqual(server.srp_server_get_host('my-host-2')['deleted'], 'false') - self.assertEqual(server.srp_server_get_service('my-service-2', '_ipps._tcp')['deleted'], 'false') - - # Remove the host and all services registered on the SRP server. - client_2.srp_client_remove_host(remove_key=True) - self.simulator.go(2) - - client_2.srp_client_clear_host() - client_2.srp_client_stop() - - # - # 6. Register with the same service instance full name before its KEY LEASE expires, - # it is expected to fail. - # - - # Remove the service instance from SRP server but retains its name. - client_1.srp_client_remove_service('my-service-1', '_ipps._tcp') - self.simulator.go(2) - - client_2.srp_client_enable_auto_start_mode() - client_2.srp_client_set_host_name('my-host-2') - client_2.srp_client_set_host_address('2001::2') - client_2.srp_client_add_service('my-service-1', '_ipps._tcp', 12345) - self.simulator.go(2) - - # It is expected that the registration will be rejected. - client_2_service = client_2.srp_client_get_services()[0] - self.assertNotEqual(client_2_service['state'], 'Registered') - self.assertNotEqual(client_2.srp_client_get_host_state(), 'Registered') - - # The service 'my-service-1' is removed but its name is retained. - # This is why we can see the service record on the SRP server. - self.assertEqual(len(server.srp_server_get_services()), 1) - self.assertEqual(len(server.srp_server_get_hosts()), 1) - self.assertEqual(server.srp_server_get_host('my-host-1')['deleted'], 'false') - self.assertEqual(server.srp_server_get_service('my-service-1', '_ipps._tcp')['deleted'], 'true') - - client_2.srp_client_clear_host() - client_2.srp_client_stop() - - # - # 7. The service instance name can be re-used by another client when - # the service has been permanently removed (the KEY resource is - # removed) from the host. - # - - # Client 1 adds back the service, it should success. - client_1.srp_client_add_service('my-service-1', '_ipps._tcp', 12345) - self.simulator.go(2) - self.assertEqual(len(server.srp_server_get_services()), 1) - self.assertEqual(len(server.srp_server_get_hosts()), 1) - self.assertEqual(server.srp_server_get_host('my-host-1')['deleted'], 'false') - self.assertEqual(server.srp_server_get_service('my-service-1', '_ipps._tcp')['deleted'], 'false') - - # Permanently removes the service instance. - client_1.srp_client_remove_host(remove_key=True) - self.simulator.go(2) - self.assertEqual(len(server.srp_server_get_services()), 0) - self.assertEqual(len(server.srp_server_get_hosts()), 0) - - # Client 2 registers the same host & service instance name with Client 1. - client_2.srp_client_stop() - client_2.srp_client_enable_auto_start_mode() - client_2.srp_client_clear_host() - client_2.srp_client_set_host_name('my-host-1') - client_2.srp_client_set_host_address('2001::2') - client_2.srp_client_add_service('my-service-1', '_ipps._tcp', 12345) - self.simulator.go(2) - - # It is expected that client 2 will success because those names has been - # released by client 1. - self.assertEqual(len(server.srp_server_get_services()), 1) - self.assertEqual(len(server.srp_server_get_hosts()), 1) - self.assertEqual(server.srp_server_get_host('my-host-1')['deleted'], 'false') - self.assertEqual(server.srp_server_get_service('my-service-1', '_ipps._tcp')['deleted'], 'false') - - -if __name__ == '__main__': - unittest.main() diff --git a/tests/scripts/thread-cert/test_srp_register_single_service.py b/tests/scripts/thread-cert/test_srp_register_single_service.py deleted file mode 100755 index f9a43ee9d..000000000 --- a/tests/scripts/thread-cert/test_srp_register_single_service.py +++ /dev/null @@ -1,221 +0,0 @@ -#!/usr/bin/env python3 -# -# Copyright (c) 2021, 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 unittest - -import command -import config -import thread_cert - -# Test description: -# This test verifies basic SRP function that a service can be registered to -# and unregistered from the SRP server. -# -# Topology: -# LEADER (SRP server) -# | -# | -# ROUTER (SRP client) -# - -SERVER = 1 -CLIENT = 2 - - -class SrpRegisterSingleService(thread_cert.TestCase): - USE_MESSAGE_FACTORY = False - SUPPORT_NCP = False - - TOPOLOGY = { - SERVER: { - 'name': 'SRP_SERVER', - 'network_key': '00112233445566778899aabbccddeeff', - 'mode': 'rdn', - }, - CLIENT: { - 'name': 'SRP_CLIENT', - 'network_key': '00112233445566778899aabbccddeeff', - 'mode': 'rdn', - }, - } - - def test(self): - server = self.nodes[SERVER] - client = self.nodes[CLIENT] - - # - # 0. Start the server & client devices. - # - - server.srp_server_set_enabled(True) - server.start() - self.simulator.go(config.LEADER_STARTUP_DELAY) - self.assertEqual(server.get_state(), 'leader') - self.simulator.go(5) - - client.srp_server_set_enabled(False) - client.start() - self.simulator.go(config.ROUTER_STARTUP_DELAY) - self.assertEqual(client.get_state(), 'router') - - # - # 1. Register a single service and verify that it works. - # - - self.assertEqual(client.srp_client_get_auto_start_mode(), 'Enabled') - - client.srp_client_set_host_name('my-host') - client.srp_client_set_host_address('2001::1') - client.srp_client_add_service('my-service', '_ipps._tcp', 12345, 0, 0, ['abc', 'def=', 'xyz=XYZ']) - self.simulator.go(2) - - self.check_host_and_service(server, client, '2001::1') - - # - # 2. Unregister a service but retain the name. The service name should be - # retained on the server. - # - - client.srp_client_remove_service('my-service', '_ipps._tcp') - self.simulator.go(2) - - client_services = client.srp_client_get_services() - print(client_services) - self.assertEqual(len(client_services), 0) - - server_services = server.srp_server_get_services() - print(server_services) - self.assertEqual(len(server_services), 1) - server_service = server_services[0] - - # Verify that the service has been successfully removed on the server side. - self.assertEqual(server_service['deleted'], 'true') - - server_hosts = server.srp_server_get_hosts() - print(server_hosts) - self.assertEqual(len(server_hosts), 1) - server_host = server_hosts[0] - - # The registered host should not be touched. - self.assertEqual(server_host['deleted'], 'false') - self.assertEqual(server_host['name'], 'my-host') - self.assertEqual(len(server_host['addresses']), 1) - self.assertEqual(ipaddress.ip_address(server_host['addresses'][0]), ipaddress.ip_address('2001::1')) - - # - # 3. Register the same service again. It should succeed and the name should be - # reused. - # - - client.srp_client_add_service('my-service', '_ipps._tcp', 12345, 0, 0, ['abc', 'def=', 'xyz=XYZ']) - self.simulator.go(2) - - self.check_host_and_service(server, client, '2001::1') - - # - # 4. Update the SRP host address and make sure it will succeed. - # - - client.srp_client_set_host_address('2001::2') - self.simulator.go(2) - - self.check_host_and_service(server, client, '2001::2') - - # - # 5. Fully remove the host and all its service instances. - # - - client.srp_client_remove_host(remove_key=True) - self.simulator.go(2) - - client_services = client.srp_client_get_services() - print(client_services) - self.assertEqual(len(client_services), 0) - - print(client.srp_client_get_host_state()) - - server_services = server.srp_server_get_services() - print(server_services) - self.assertEqual(len(server_services), 0) - - server_hosts = server.srp_server_get_hosts() - print(server_hosts) - self.assertEqual(len(server_hosts), 0) - - def check_host_and_service(self, server, client, host_addr): - """Check that we have properly registered host and service instance. - """ - - client_services = client.srp_client_get_services() - print(client_services) - self.assertEqual(len(client_services), 1) - client_service = client_services[0] - - # Verify that the client possesses correct service resources. - self.assertEqual(client_service['instance'], 'my-service') - self.assertEqual(client_service['name'], '_ipps._tcp') - self.assertEqual(int(client_service['port']), 12345) - self.assertEqual(int(client_service['priority']), 0) - self.assertEqual(int(client_service['weight']), 0) - - # Verify that the client received a SUCCESS response for the server. - self.assertEqual(client_service['state'], 'Registered') - - server_services = server.srp_server_get_services() - self.assertEqual(len(server_services), 1) - server_service = server_services[0] - - # Verify that the server accepted the SRP registration and stores - # the same service resources. - self.assertEqual(server_service['deleted'], 'false') - self.assertEqual(server_service['instance'], client_service['instance']) - self.assertEqual(server_service['name'], client_service['name']) - self.assertEqual(server_service['subtypes'], '(null)') - self.assertEqual(int(server_service['port']), int(client_service['port'])) - self.assertEqual(int(server_service['priority']), int(client_service['priority'])) - self.assertEqual(int(server_service['weight']), int(client_service['weight'])) - # We output value of TXT entry as HEX string. - print(server_service['TXT']) - self.assertEqual(server_service['TXT'], ['abc', 'def=', 'xyz=58595a']) - self.assertEqual(server_service['host'], 'my-host') - - server_hosts = server.srp_server_get_hosts() - print(server_hosts) - self.assertEqual(len(server_hosts), 1) - server_host = server_hosts[0] - - self.assertEqual(server_host['deleted'], 'false') - self.assertEqual(server_host['fullname'], server_service['host_fullname']) - self.assertEqual(len(server_host['addresses']), 1) - self.assertEqual(ipaddress.ip_address(server_host['addresses'][0]), ipaddress.ip_address(host_addr)) - - -if __name__ == '__main__': - unittest.main() diff --git a/tests/scripts/thread-cert/test_srp_sub_type.py b/tests/scripts/thread-cert/test_srp_sub_type.py deleted file mode 100755 index bbd8efa17..000000000 --- a/tests/scripts/thread-cert/test_srp_sub_type.py +++ /dev/null @@ -1,142 +0,0 @@ -#!/usr/bin/env python3 -# -# Copyright (c) 2021, 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 unittest - -import command -import config -import thread_cert - -# Test description: -# This test verifies SRP client and server support for registering sub-type services. -# -# Topology: -# -# LEADER (SRP server) -# | -# | -# ROUTER (SRP client) -# - -SERVER = 1 -CLIENT = 2 - - -class SrpSubType(thread_cert.TestCase): - USE_MESSAGE_FACTORY = False - SUPPORT_NCP = False - - TOPOLOGY = { - SERVER: { - 'name': 'SRP_SERVER', - 'mode': 'rdn', - }, - CLIENT: { - 'name': 'SRP_CLIENT', - 'mode': 'rdn', - }, - } - - def test(self): - server = self.nodes[SERVER] - client = self.nodes[CLIENT] - - # Start the server & client devices. - - server.start() - self.simulator.go(config.LEADER_STARTUP_DELAY) - self.assertEqual(server.get_state(), 'leader') - - client.start() - self.simulator.go(config.ROUTER_STARTUP_DELAY) - self.assertEqual(client.get_state(), 'router') - - server.srp_server_set_enabled(True) - client.srp_client_enable_auto_start_mode() - self.simulator.go(15) - - # Register a single service with 3 subtypes and verify that it worked. - - client.srp_client_set_host_name('host1') - client.srp_client_set_host_address('2001::1') - client.srp_client_add_service('ins1', '_srv._udp,_s1,_s2,_s3', 1977) - self.simulator.go(2) - self.check_service_on_client_and_server(server, client) - - # Remove the service on client - - client.srp_client_remove_service('ins1', '_srv._udp') - self.simulator.go(2) - - client_services = client.srp_client_get_services() - self.assertEqual(len(client_services), 0) - server_services = server.srp_server_get_services() - self.assertEqual(len(server_services), 1) - server_service = server_services[0] - self.assertEqual(server_service['fullname'], 'ins1._srv._udp.default.service.arpa.') - self.assertEqual(server_service['instance'], 'ins1') - self.assertEqual(server_service['name'], '_srv._udp') - self.assertEqual(server_service['deleted'], 'true') - - # Register the same service again. - - client.srp_client_add_service('ins1', '_srv._udp,_s1,_s2,_s3', 1977) - self.simulator.go(2) - self.check_service_on_client_and_server(server, client) - - def check_service_on_client_and_server(self, server, client): - # Check the service on client - client_services = client.srp_client_get_services() - self.assertEqual(len(client_services), 1) - client_service = client_services[0] - self.assertEqual(client_service['instance'], 'ins1') - self.assertEqual(client_service['name'], '_srv._udp,_s1,_s2,_s3') - self.assertEqual(int(client_service['port']), 1977) - self.assertEqual(int(client_service['priority']), 0) - self.assertEqual(int(client_service['weight']), 0) - self.assertEqual(client_service['state'], 'Registered') - - # Check the service on server - server_services = server.srp_server_get_services() - self.assertEqual(len(server_services), 1) - server_service = server_services[0] - self.assertEqual(server_service['fullname'], 'ins1._srv._udp.default.service.arpa.') - self.assertEqual(server_service['instance'], 'ins1') - self.assertEqual(server_service['name'], '_srv._udp') - self.assertEqual(server_service['deleted'], 'false') - self.assertEqual(set(server_service['subtypes'].split(',')), {'_s1', '_s2', '_s3'}) - self.assertEqual(int(server_service['port']), 1977) - self.assertEqual(int(server_service['priority']), 0) - self.assertEqual(int(server_service['weight']), 0) - self.assertEqual(server_service['host'], 'host1') - - -if __name__ == '__main__': - unittest.main()