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 SRP client `RemoveHostAndServices()` behavior when host info is not yet registered.
39#
40# Topology:
41#
42#     LEADER (SRP server)
43#       |
44#       |
45#     ROUTER (SRP client)
46#
47
48SERVER = 1
49CLIENT = 2
50
51
52class SrpRemoveHost(thread_cert.TestCase):
53    USE_MESSAGE_FACTORY = False
54    SUPPORT_NCP = False
55
56    TOPOLOGY = {
57        SERVER: {
58            'name': 'SRP_SERVER',
59            'mode': 'rdn',
60        },
61        CLIENT: {
62            'name': 'SRP_CLIENT',
63            'mode': 'rdn',
64        },
65    }
66
67    def test(self):
68        server = self.nodes[SERVER]
69        client = self.nodes[CLIENT]
70
71        #-------------------------------------------------------------------------------------
72        # Start the server & client devices.
73
74        server.start()
75        self.simulator.go(config.LEADER_STARTUP_DELAY)
76        self.assertEqual(server.get_state(), 'leader')
77
78        client.start()
79        self.simulator.go(config.ROUTER_STARTUP_DELAY)
80        self.assertEqual(client.get_state(), 'router')
81
82        server.srp_server_set_enabled(True)
83        client.srp_client_enable_auto_start_mode()
84        self.simulator.go(15)
85
86        #-------------------------------------------------------------------------------------
87        # Register a single service and verify that it worked.
88
89        client.srp_client_set_host_name('host')
90        client.srp_client_set_host_address('2001::1')
91        client.srp_client_add_service('ins', '_srv._udp', 1977)
92        self.simulator.go(2)
93
94        client_services = client.srp_client_get_services()
95        self.assertEqual(len(client_services), 1)
96        client_service = client_services[0]
97        self.assertEqual(client_service['instance'], 'ins')
98        self.assertEqual(client_service['name'], '_srv._udp')
99        self.assertEqual(int(client_service['port']), 1977)
100        self.assertEqual(int(client_service['priority']), 0)
101        self.assertEqual(int(client_service['weight']), 0)
102        self.assertEqual(client_service['state'], 'Registered')
103
104        server_hosts = server.srp_server_get_hosts()
105        self.assertEqual(len(server_hosts), 1)
106        server_host = server_hosts[0]
107        self.assertEqual(server_host['fullname'], 'host.default.service.arpa.')
108        self.assertEqual(server_host['deleted'], 'false')
109        self.assertEqual(len(server_host['addresses']), 1)
110        self.assertIn('2001:0:0:0:0:0:0:1', server_host['addresses'])
111
112        server_services = server.srp_server_get_services()
113        self.assertEqual(len(server_services), 1)
114        server_service = server_services[0]
115        self.assertEqual(server_service['fullname'], 'ins._srv._udp.default.service.arpa.')
116        self.assertEqual(server_service['instance'], 'ins')
117        self.assertEqual(server_service['name'], '_srv._udp')
118        self.assertEqual(server_service['deleted'], 'false')
119        self.assertEqual(int(server_service['port']), 1977)
120        self.assertEqual(int(server_service['priority']), 0)
121        self.assertEqual(int(server_service['weight']), 0)
122        self.assertEqual(server_service['host'], 'host')
123
124        #-------------------------------------------------------------------------------------
125        # Clear the info on client, and verify that it is still present on server.
126
127        client.srp_client_clear_host()
128        self.simulator.go(2)
129
130        client_services = client.srp_client_get_services()
131        self.assertEqual(len(client_services), 0)
132        self.assertIsNone(client.srp_client_get_host_name())
133        self.assertIsNone(client.srp_client_get_host_address())
134
135        server_services = server.srp_server_get_services()
136        self.assertEqual(len(server_services), 1)
137
138        #-------------------------------------------------------------------------------------
139        # Now set the host name and request "host remove" requiring that client updates server
140        # even if host is not yet registered. Verify that client updates the server by checking
141        # that the host and service entries are marked as "deleted" on the server (i.e. deleted
142        # by the name (and associated key) are retained).
143
144        client.srp_client_set_host_name('host')
145        client.srp_client_remove_host(send_unreg_to_server=True)
146        self.simulator.go(2)
147
148        self.assertIsNone(client.srp_client_get_host_name())
149
150        server_hosts = server.srp_server_get_hosts()
151        self.assertEqual(len(server_hosts), 1)
152        server_host = server_hosts[0]
153        self.assertEqual(server_host['fullname'], 'host.default.service.arpa.')
154        self.assertEqual(server_host['deleted'], 'true')
155
156        server_services = server.srp_server_get_services()
157        self.assertEqual(len(server_services), 1)
158        server_service = server_services[0]
159        self.assertEqual(server_service['fullname'], 'ins._srv._udp.default.service.arpa.')
160        self.assertEqual(server_service['deleted'], 'true')
161
162        #-------------------------------------------------------------------------------------
163        # Again request "host remove" but this time request `remove_key`. Verify that entries
164        # on the server are fully removed .
165
166        client.srp_client_set_host_name('host')
167        client.srp_client_remove_host(remove_key=True, send_unreg_to_server=True)
168        self.simulator.go(2)
169
170        self.assertIsNone(client.srp_client_get_host_name())
171        self.assertEqual(len(server.srp_server_get_services()), 0)
172        self.assertEqual(len(server.srp_server_get_hosts()), 0)
173
174
175if __name__ == '__main__':
176    unittest.main()
177