[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.

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__':

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

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: