espressif/openthread
1
0
Fork 0
mirror of https://github.com/espressif/openthread.git synced 2026-07-30 07:37:46 +00:00
Code Issues Packages Projects Releases Wiki Activity

[tests] add test for Advertising Proxy to publish zero or multiple host addrs (#8003)

Browse Source
This commit is contained in:
whd
2022-08-10 21:24:44 -07:00
committed by GitHub
parent c336a1899f
commit df2c11a6cb
3 changed files with 56 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/scripts/thread-cert/border_router/test_advertising_proxy.py
+48 -7
View File
@@ -212,16 +212,54 @@ class SingleHostAndService(thread_cert.TestCase):
self.host_check_mdns_service(host, host_address, 'my-service', service_port)
#
# 9. Check if the expired service is removed by the Advertising Proxy.
# 9. Check if Advertising Proxy filters out Mesh Local and Link Local host addresses
#
client.srp_client_remove_host()
self.simulator.go(2)
client.srp_client_set_host_name('my-host')
client.srp_client_set_host_address('2001::1', '2002::2',
client.get_ip6_address(config.ADDRESS_TYPE.OMR)[0],
client.get_ip6_address(config.ADDRESS_TYPE.LINK_LOCAL),
client.get_ip6_address(config.ADDRESS_TYPE.ML_EID))
client.srp_client_add_service('my-service', '_ipps._tcp', 12345)
client.srp_client_enable_auto_start_mode()
self.simulator.go(10)
self.check_host_and_service(server, client, [
'2001::1', '2002::2',
client.get_ip6_address(config.ADDRESS_TYPE.OMR)[0],
client.get_ip6_address(config.ADDRESS_TYPE.LINK_LOCAL),
client.get_ip6_address(config.ADDRESS_TYPE.ML_EID)
], 'my-service', 12345)
self.host_check_mdns_service(
host, ['2001::1', '2002::2', client.get_ip6_address(config.ADDRESS_TYPE.OMR)[0]], 'my-service', 12345)
client.srp_client_set_host_address(client.get_ip6_address(config.ADDRESS_TYPE.LINK_LOCAL),
client.get_ip6_address(config.ADDRESS_TYPE.ML_EID), client.get_rloc())
self.simulator.go(10)
self.check_host_and_service(server, client, [
client.get_ip6_address(config.ADDRESS_TYPE.LINK_LOCAL),
client.get_ip6_address(config.ADDRESS_TYPE.ML_EID),
client.get_rloc()
], 'my-service', 12345)
self.host_check_mdns_service(host, [], 'my-service', 12345)
client.srp_client_set_host_address('2005::3')
self.simulator.go(10)
self.check_host_and_service(server, client, '2005::3', 'my-service', 12345)
self.host_check_mdns_service(host, '2005::3', 'my-service', 12345)
#
# 10. Check if the expired service is removed by the Advertising Proxy.
#
client.srp_client_stop()
self.simulator.go(LEASE + 2)
self.assertIsNone(host.discover_mdns_service('my-service', '_ipps._tcp', 'my-host'))
self.assertIsNone(host.discover_mdns_service('my-service-1', '_ipps._tcp', 'my-host'))
def host_check_mdns_service(self, host, host_addr, service_instance, service_port=12345):
def host_check_mdns_service(self, host, host_addrs, service_instance, service_port=12345):
if isinstance(host_addrs, str):
host_addrs = [host_addrs]
service = host.discover_mdns_service(service_instance, '_ipps._tcp', 'my-host')
self.assertIsNotNone(service)
self.assertEqual(service['instance'], service_instance)
@@ -230,13 +268,16 @@ class SingleHostAndService(thread_cert.TestCase):
self.assertEqual(service['priority'], 0)
self.assertEqual(service['weight'], 0)
self.assertEqual(service['host'], 'my-host')
self.assertEqual(ipaddress.ip_address(service['addresses'][0]), ipaddress.ip_address(host_addr))
self.assertEqual(len(service['addresses']), 1)
self.assertEqual(len(service['addresses']), len(host_addrs))
self.assertEqual(sorted(map(ipaddress.ip_address, service['addresses'])),
sorted(map(ipaddress.ip_address, host_addrs)))
def check_host_and_service(self, server, client, host_addr, service_instance, service_port=12345):
def check_host_and_service(self, server, client, host_addrs, service_instance, service_port=12345):
"""Check that we have properly registered host and service instance.
"""
if isinstance(host_addrs, str):
host_addrs = [host_addrs]
client_services = client.srp_client_get_services()
print(client_services)
client_services = [service for service in client_services if service['instance'] == service_instance]
@@ -276,8 +317,8 @@ class SingleHostAndService(thread_cert.TestCase):
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))
self.assertEqual(sorted(map(ipaddress.ip_address, server_host['addresses'])),
sorted(map(ipaddress.ip_address, host_addrs)))
class SrpClientRemoveNonExistingHost(thread_cert.TestCase):
tests/scripts/thread-cert/node.py
+4 -2
View File
@@ -987,7 +987,7 @@ class NodeImpl:
addresses = lines.pop(0).strip().split('[')[1].strip(' ]').split(',')
map(str.strip, addresses)
host['addresses'] = [addr for addr in addresses if addr]
host['addresses'] = [addr.strip() for addr in addresses if addr]
host_list.append(host)
@@ -3381,6 +3381,8 @@ class LinuxHost():
fullname = f'{host_name}.local.'
if fullname not in elements:
continue
if 'Add' not in elements:
continue
addresses.append(elements[elements.index(fullname) + 1].split('%')[0])
logging.debug(f'addresses of {host_name}: {addresses}')
@@ -3417,7 +3419,7 @@ class LinuxHost():
assert (service['host_fullname'] == f'{host_name}.local.')
service['host'] = host_name
service['addresses'] = addresses
return service if 'addresses' in service and service['addresses'] else None
return service or None
def start_radvd_service(self, prefix, slaac):
self.bash("""cat >/etc/radvd.conf <<EOF