1#!/usr/bin/env python3 2# 3# Copyright (c) 2021, The OpenThread Authors. 4# All rights reserved. 5# 6# Redistribution and use in source and binary forms, with or without 7# modification, are permitted provided that the following conditions are met: 8# 1. Redistributions of source code must retain the above copyright 9# notice, this list of conditions and the following disclaimer. 10# 2. Redistributions in binary form must reproduce the above copyright 11# notice, this list of conditions and the following disclaimer in the 12# documentation and/or other materials provided with the distribution. 13# 3. Neither the name of the copyright holder nor the 14# names of its contributors may be used to endorse or promote products 15# derived from this software without specific prior written permission. 16# 17# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" 18# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE 19# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE 20# ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE 21# LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR 22# CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF 23# SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS 24# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN 25# CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) 26# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE 27# POSSIBILITY OF SUCH DAMAGE. 28# 29 30import ipaddress 31import unittest 32 33import command 34import config 35import thread_cert 36 37# Test description: 38# This test verifies basic SRP function that a service can be registered to 39# and unregistered from the SRP server. 40# 41# Topology: 42# LEADER (SRP server) 43# | 44# | 45# ROUTER (SRP client) 46# 47 48SERVER = 1 49CLIENT = 2 50 51 52class SrpRegisterSingleService(thread_cert.TestCase): 53 USE_MESSAGE_FACTORY = False 54 SUPPORT_NCP = False 55 56 TOPOLOGY = { 57 SERVER: { 58 'name': 'SRP_SERVER', 59 'networkkey': '00112233445566778899aabbccddeeff', 60 'mode': 'rdn', 61 }, 62 CLIENT: { 63 'name': 'SRP_CLIENT', 64 'networkkey': '00112233445566778899aabbccddeeff', 65 'mode': 'rdn', 66 }, 67 } 68 69 def test(self): 70 server = self.nodes[SERVER] 71 client = self.nodes[CLIENT] 72 73 # 74 # 0. Start the server & client devices. 75 # 76 77 server.srp_server_set_enabled(True) 78 server.start() 79 self.simulator.go(config.LEADER_STARTUP_DELAY) 80 self.assertEqual(server.get_state(), 'leader') 81 self.simulator.go(5) 82 83 client.srp_server_set_enabled(False) 84 client.start() 85 self.simulator.go(config.ROUTER_STARTUP_DELAY) 86 self.assertEqual(client.get_state(), 'router') 87 88 # 89 # 1. Register a single service and verify that it works. 90 # 91 92 self.assertEqual(client.srp_client_get_auto_start_mode(), 'Enabled') 93 94 client.srp_client_set_host_name('my-host') 95 client.srp_client_set_host_address('2001::1') 96 client.srp_client_add_service('my-service', '_ipps._tcp', 12345, 0, 0, ['abc', 'def=', 'xyz=XYZ']) 97 self.simulator.go(2) 98 99 self.check_host_and_service(server, client, '2001::1') 100 101 # 102 # 2. Unregister a service but retain the name. The service name should be 103 # retained on the server. 104 # 105 106 client.srp_client_remove_service('my-service', '_ipps._tcp') 107 self.simulator.go(2) 108 109 client_services = client.srp_client_get_services() 110 print(client_services) 111 self.assertEqual(len(client_services), 0) 112 113 server_services = server.srp_server_get_services() 114 print(server_services) 115 self.assertEqual(len(server_services), 1) 116 server_service = server_services[0] 117 118 # Verify that the service has been successfully removed on the server side. 119 self.assertEqual(server_service['deleted'], 'true') 120 121 server_hosts = server.srp_server_get_hosts() 122 print(server_hosts) 123 self.assertEqual(len(server_hosts), 1) 124 server_host = server_hosts[0] 125 126 # The registered host should not be touched. 127 self.assertEqual(server_host['deleted'], 'false') 128 self.assertEqual(server_host['name'], 'my-host') 129 self.assertEqual(len(server_host['addresses']), 1) 130 self.assertEqual(ipaddress.ip_address(server_host['addresses'][0]), ipaddress.ip_address('2001::1')) 131 132 # 133 # 3. Register the same service again. It should succeed and the name should be 134 # reused. 135 # 136 137 client.srp_client_add_service('my-service', '_ipps._tcp', 12345, 0, 0, ['abc', 'def=', 'xyz=XYZ']) 138 self.simulator.go(2) 139 140 self.check_host_and_service(server, client, '2001::1') 141 142 # 143 # 4. Update the SRP host address and make sure it will succeed. 144 # 145 146 client.srp_client_set_host_address('2001::2') 147 self.simulator.go(2) 148 149 self.check_host_and_service(server, client, '2001::2') 150 151 # 152 # 5. Fully remove the host and all its service instances. 153 # 154 155 client.srp_client_remove_host(remove_key=True) 156 self.simulator.go(2) 157 158 client_services = client.srp_client_get_services() 159 print(client_services) 160 self.assertEqual(len(client_services), 0) 161 162 print(client.srp_client_get_host_state()) 163 164 server_services = server.srp_server_get_services() 165 print(server_services) 166 self.assertEqual(len(server_services), 0) 167 168 server_hosts = server.srp_server_get_hosts() 169 print(server_hosts) 170 self.assertEqual(len(server_hosts), 0) 171 172 def check_host_and_service(self, server, client, host_addr): 173 """Check that we have properly registered host and service instance. 174 """ 175 176 client_services = client.srp_client_get_services() 177 print(client_services) 178 self.assertEqual(len(client_services), 1) 179 client_service = client_services[0] 180 181 # Verify that the client possesses correct service resources. 182 self.assertEqual(client_service['instance'], 'my-service') 183 self.assertEqual(client_service['name'], '_ipps._tcp') 184 self.assertEqual(int(client_service['port']), 12345) 185 self.assertEqual(int(client_service['priority']), 0) 186 self.assertEqual(int(client_service['weight']), 0) 187 188 # Verify that the client received a SUCCESS response for the server. 189 self.assertEqual(client_service['state'], 'Registered') 190 191 server_services = server.srp_server_get_services() 192 self.assertEqual(len(server_services), 1) 193 server_service = server_services[0] 194 195 # Verify that the server accepted the SRP registration and stores 196 # the same service resources. 197 self.assertEqual(server_service['deleted'], 'false') 198 self.assertEqual(server_service['instance'], client_service['instance']) 199 self.assertEqual(server_service['name'], client_service['name']) 200 self.assertEqual(server_service['subtypes'], '(null)') 201 self.assertEqual(int(server_service['port']), int(client_service['port'])) 202 self.assertEqual(int(server_service['priority']), int(client_service['priority'])) 203 self.assertEqual(int(server_service['weight']), int(client_service['weight'])) 204 # We output value of TXT entry as HEX string. 205 print(server_service['TXT']) 206 self.assertEqual(server_service['TXT'], ['abc', 'def=', 'xyz=58595a']) 207 self.assertEqual(server_service['host'], 'my-host') 208 209 server_hosts = server.srp_server_get_hosts() 210 print(server_hosts) 211 self.assertEqual(len(server_hosts), 1) 212 server_host = server_hosts[0] 213 214 self.assertEqual(server_host['deleted'], 'false') 215 self.assertEqual(server_host['fullname'], server_service['host_fullname']) 216 self.assertEqual(len(server_host['addresses']), 1) 217 self.assertEqual(ipaddress.ip_address(server_host['addresses'][0]), ipaddress.ip_address(host_addr)) 218 219 220if __name__ == '__main__': 221 unittest.main() 222