From 01f1489ec825bf83b0ea65aa4879184029ce12c3 Mon Sep 17 00:00:00 2001 From: whd <7058128+superwhd@users.noreply.github.com> Date: Wed, 27 Oct 2021 13:48:28 +0800 Subject: [PATCH] [tests] update test_publish_meshcop_service.py (#7067) - Relax the assumptions on the names of meshcop services so that it's OK for services to change names to resolve conflicts. - Adjust timeouts. --- .../test_publish_meshcop_service.py | 50 +++++++++++++------ tests/scripts/thread-cert/node.py | 16 +++--- tests/scripts/thread-cert/thread_cert.py | 2 +- 3 files changed, 44 insertions(+), 24 deletions(-) diff --git a/tests/scripts/thread-cert/border_router/test_publish_meshcop_service.py b/tests/scripts/thread-cert/border_router/test_publish_meshcop_service.py index 7f2186f8b..395f8c7b1 100755 --- a/tests/scripts/thread-cert/border_router/test_publish_meshcop_service.py +++ b/tests/scripts/thread-cert/border_router/test_publish_meshcop_service.py @@ -58,12 +58,14 @@ class PublishMeshCopService(thread_cert.TestCase): 'allowlist': [], 'is_otbr': True, 'version': '1.2', + 'network_name': 'ot-br1', }, BR2: { 'name': 'BR_2', 'allowlist': [], 'is_otbr': True, 'version': '1.2', + 'network_name': 'ot-br2', }, HOST: { 'name': 'Host', @@ -77,37 +79,47 @@ class PublishMeshCopService(thread_cert.TestCase): br2 = self.nodes[BR2] br2.disable_br() - # TODO: verify the behavior when thread is disabled host.bash('service otbr-agent stop') host.start(start_radvd=False) - self.simulator.go(10) + self.simulator.go(20) + self.assertEqual(br1.get_state(), 'disabled') br1.start() - self.simulator.go(10) + self.simulator.go(20) self.assertEqual('leader', br1.get_state()) - self.check_meshcop_service(br1, host) br1.disable_backbone_router() self.simulator.go(10) self.check_meshcop_service(br1, host) + br1.stop() + br1.set_network_name('ot-br1-1') + br1.start() + self.simulator.go(10) + self.check_meshcop_service(br1, host) + # verify that there are two meshcop services br2.start() br2.disable_backbone_router() br2.enable_br() - self.simulator.go(10) + self.simulator.go(25) service_instances = host.browse_mdns_services('_meshcop._udp') self.assertEqual(len(service_instances), 2) - br1_service = self.check_meshcop_service(br1, host, 'OpenThread_BorderRouter') - for instance in service_instances: - if instance != 'OpenThread_BorderRouter': - br2_service = self.check_meshcop_service(br2, host, instance) + br1_service = self.check_meshcop_service(br1, host) + br2_service = self.check_meshcop_service(br2, host) self.assertNotEqual(br1_service['host'], br2_service['host']) - def check_meshcop_service(self, br, host, instance_name='OpenThread_BorderRouter'): - data = host.discover_mdns_service(instance_name, '_meshcop._udp', None) - sb_data = data['txt']['sb'].encode('raw_unicode_escape') + def check_meshcop_service(self, br, host): + services = self.discover_all_meshcop_services(host) + for service in services: + if service['txt']['nn'] == br.get_network_name(): + self.check_meshcop_service_by_data(br, service) + return service + self.fail('MeshCoP service not found') + + def check_meshcop_service_by_data(self, br, service_data): + sb_data = service_data['txt']['sb'].encode('raw_unicode_escape') state_bitmap = int.from_bytes(sb_data, byteorder='big') logging.info(bin(state_bitmap)) self.assertEqual((state_bitmap & 7), 1) # connection mode = PskC @@ -121,10 +133,16 @@ class PublishMeshCopService(thread_cert.TestCase): self.assertEqual((state_bitmap >> 7 & 1), br.get_backbone_router_state() != 'Disabled') # BBR is enabled or not self.assertEqual((state_bitmap >> 8 & 1), br.get_backbone_router_state() == 'Primary') # BBR is primary or not - self.assertEqual(data['txt']['nn'], br.get_network_name()) - self.assertEqual(data['txt']['rv'], '1') - self.assertIn(data['txt']['tv'], ['1.1.0', '1.1.1', '1.2.0']) - return data + self.assertEqual(service_data['txt']['nn'], br.get_network_name()) + self.assertEqual(service_data['txt']['rv'], '1') + self.assertIn(service_data['txt']['tv'], ['1.1.0', '1.1.1', '1.2.0']) + + def discover_all_meshcop_services(self, host): + instance_names = host.browse_mdns_services('_meshcop._udp') + services = [] + for instance_name in instance_names: + services.append(host.discover_mdns_service(instance_name, '_meshcop._udp', None)) + return services if __name__ == '__main__': diff --git a/tests/scripts/thread-cert/node.py b/tests/scripts/thread-cert/node.py index ed2470e24..c57f48fa6 100755 --- a/tests/scripts/thread-cert/node.py +++ b/tests/scripts/thread-cert/node.py @@ -3091,35 +3091,37 @@ class LinuxHost(): The return value is a dict with the same key/values of srp_server_get_service except that we don't have a `deleted` field here. """ + host_name_file = self.bash('mktemp')[0].strip() + service_data_file = self.bash('mktemp')[0].strip() - self.bash(f'dns-sd -Z {name} local. > /tmp/{name} 2>&1 &') + self.bash(f'dns-sd -Z {name} local. > {service_data_file} 2>&1 &') time.sleep(timeout) full_service_name = f'{instance}.{name}' # When hostname is unspecified, extract hostname from browse result if host_name is None: - for line in self.bash(f'cat /tmp/{name}', encoding='raw_unicode_escape'): + for line in self.bash(f'cat {service_data_file}', encoding='raw_unicode_escape'): elements = line.split() if len(elements) >= 6 and elements[0] == full_service_name and elements[1] == 'SRV': host_name = elements[5].split('.')[0] break assert (host_name is not None) - self.bash(f'dns-sd -G v6 {host_name}.local. > /tmp/{host_name} 2>&1 &') + self.bash(f'dns-sd -G v6 {host_name}.local. > {host_name_file} 2>&1 &') time.sleep(timeout) self.bash('pkill dns-sd') addresses = [] service = {} - logging.debug(self.bash(f'cat /tmp/{host_name}', encoding='raw_unicode_escape')) - logging.debug(self.bash(f'cat /tmp/{name}', encoding='raw_unicode_escape')) + logging.debug(self.bash(f'cat {host_name_file}', encoding='raw_unicode_escape')) + logging.debug(self.bash(f'cat {service_data_file}', encoding='raw_unicode_escape')) # example output in the host file: # Timestamp A/R Flags if Hostname Address TTL # 9:38:09.274 Add 23 48 my-host.local. 2001:0000:0000:0000:0000:0000:0000:0002%<0> 120 # - for line in self.bash(f'cat /tmp/{host_name}', encoding='raw_unicode_escape'): + for line in self.bash(f'cat {host_name_file}', encoding='raw_unicode_escape'): elements = line.split() fullname = f'{host_name}.local.' if fullname not in elements: @@ -3135,7 +3137,7 @@ class LinuxHost(): # is_txt = False txt = '' - for line in self.bash(f'cat /tmp/{name}', encoding='raw_unicode_escape'): + for line in self.bash(f'cat {service_data_file}', encoding='raw_unicode_escape'): elements = line.split() if len(elements) >= 2 and elements[0] == full_service_name and elements[1] == 'TXT': is_txt = True diff --git a/tests/scripts/thread-cert/thread_cert.py b/tests/scripts/thread-cert/thread_cert.py index dbbddc731..e6236ce0f 100755 --- a/tests/scripts/thread-cert/thread_cert.py +++ b/tests/scripts/thread-cert/thread_cert.py @@ -233,7 +233,7 @@ class TestCase(NcpSupportMixin, unittest.TestCase): # we have to add allowlist after nodes are all created for i, params in initial_topology.items(): allowlist = params['allowlist'] - if not allowlist: + if allowlist is None: continue for j in allowlist: