espressif/openthread
1
0
Fork 0
mirror of https://github.com/espressif/openthread.git synced 2026-08-06 02:37:47 +00:00
Code Issues Packages Projects Releases Wiki Activity

[tests] check whether the unreachable address is included in the DNS-SD response (#9075)

Browse Source 
This commit enhances the DNS-SD server test that it verifies if the
unreachable address is included in the DNS-SD response.
This commit is contained in:
whd
2023-05-24 15:28:01 -07:00
committed by GitHub
parent b82cfddd68
commit c6ea8afc85
2 changed files with 85 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/scripts/thread-cert/border_router/test_dnssd_server_multi_border_routers.py
+61 -5
View File
@@ -155,6 +155,7 @@ class TestDnssdServerOnMultiBr(thread_cert.TestCase):
def is_int(x):
return isinstance(x, int)
# 1. Check hosts & services published by Advertising Proxy.
# check if AAAA query works
dig_result = host.dns_dig(br2_addr, host1_full_name, 'AAAA')
self._assert_dig_result_matches(dig_result, {
@@ -219,7 +220,64 @@ class TestDnssdServerOnMultiBr(thread_cert.TestCase):
],
}])
# check some invalid queries
# 2. Check the host & service published by a WiFi host.
# check if AAAA query works
wifi_host_linklocal_address = 'fe80::1234'
wifi_host_routable_address = '2402::abcd'
wifi_host_full_name = f'wifi-host.{DOMAIN}'
wifi_service_instance_full_name = f'wifi-service._host._tcp.{DOMAIN}'
host.publish_mdns_host('wifi-host', [wifi_host_linklocal_address, wifi_host_routable_address])
host.publish_mdns_service('wifi-service', '_host._tcp', 12345, 'wifi-host', {'k1': 'v1', 'k2': 'v2'})
dig_result = host.dns_dig(br2_addr, wifi_host_full_name, 'AAAA')
self._assert_dig_result_matches(
dig_result, {
'QUESTION': [(wifi_host_full_name, 'IN', 'AAAA')],
'ANSWER': [(wifi_host_full_name, 'IN', 'AAAA', wifi_host_routable_address),],
})
# check if SRV query works
dig_result = host.dns_dig(br2_addr, wifi_service_instance_full_name, 'SRV')
self._assert_dig_result_matches(
dig_result, {
'QUESTION': [(wifi_service_instance_full_name, 'IN', 'SRV')],
'ANSWER': [(wifi_service_instance_full_name, 'IN', 'SRV', is_int, is_int, 12345, wifi_host_full_name),
],
'ADDITIONAL': [(wifi_host_full_name, 'IN', 'AAAA', wifi_host_routable_address),],
})
# check if TXT query works
dig_result = host.dns_dig(br2_addr, wifi_service_instance_full_name, 'TXT')
self._assert_dig_result_matches(
dig_result, {
'QUESTION': [(wifi_service_instance_full_name, 'IN', 'TXT')],
'ANSWER': [(wifi_service_instance_full_name, 'IN', 'TXT', {
'k1': 'v1',
'k2': 'v2'
})],
})
# check if PTR query works
dig_result = host.dns_dig(br2_addr, f'_host._tcp.{DOMAIN}', 'PTR')
self._assert_dig_result_matches_any(dig_result, [{
'QUESTION': [(f'_host._tcp.{DOMAIN}', 'IN', 'PTR')],
'ANSWER': [(f'_host._tcp.{DOMAIN}', 'IN', 'PTR', wifi_service_instance_full_name)],
'ADDITIONAL': [
(wifi_service_instance_full_name, 'IN', 'SRV', is_int, is_int, 12345, wifi_host_full_name),
(wifi_service_instance_full_name, 'IN', 'TXT', {
'k1': 'v1',
'k2': 'v2'
}),
(wifi_host_full_name, 'IN', 'AAAA', wifi_host_routable_address),
],
}])
host.bash('pkill avahi-publish')
# 3. Verify Discovery Proxy works for _meshcop._udp published by BR.
self._verify_discovery_proxy_meshcop(br2_addr, br2.get_network_name(), host)
# 4. Check some invalid queries
for qtype in ['A', 'CNAME']:
dig_result = host.dns_dig(br2_addr, host1_full_name, qtype)
self._assert_dig_result_matches(dig_result, {
@@ -232,9 +290,6 @@ class TestDnssdServerOnMultiBr(thread_cert.TestCase):
'status': 'NXDOMAIN',
})
# verify Discovery Proxy works for _meshcop._udp
self._verify_discovery_proxy_meshcop(br2_addr, br2.get_network_name(), host)
def _verify_discovery_proxy_meshcop(self, server_addr, network_name, digger):
dp_service_name = '_meshcop._udp.default.service.arpa.'
dp_hostname = lambda x: x.endswith('.default.service.arpa.')
@@ -300,7 +355,8 @@ class TestDnssdServerOnMultiBr(thread_cert.TestCase):
client.srp_client_set_host_address(*addrs)
client.srp_client_add_service(instancename, SERVICE, port, priority, weight)
self.simulator.go(5)
self.simulator.go(10)
self.assertEqual(client.srp_client_get_host_state(), 'Registered')
def _assert_have_question(self, dig_result, question):
tests/scripts/thread-cert/node.py
+24
View File
@@ -3644,6 +3644,30 @@ class LinuxHost():
(self.ETH_DEV, self.ETH_DEV))
self.bash(f'ip -6 neigh list dev {self.ETH_DEV}')
def publish_mdns_service(self, instance_name, service_type, port, host_name, txt):
"""Publish an mDNS service on the Ethernet.
:param instance_name: the service instance name.
:param service_type: the service type in format of '<service_type>.<protocol>'.
:param port: the port the service is at.
:param host_name: the host name this service points to. The domain
should not be included.
:param txt: a dictionary containing the key-value pairs of the TXT record.
"""
txt_string = ' '.join([f'{key}={value}' for key, value in txt.items()])
self.bash(f'avahi-publish -s {instance_name} {service_type} {port} -H {host_name}.local {txt_string} &')
def publish_mdns_host(self, hostname, addresses):
"""Publish an mDNS host on the Ethernet
:param host_name: the host name this service points to. The domain
should not be included.
:param addresses: a list of strings representing the addresses to
be registered with the host.
"""
for address in addresses:
self.bash(f'avahi-publish -a {hostname}.local {address} &')
def browse_mdns_services(self, name, timeout=2):
""" Browse mDNS services on the ethernet.