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

[border-agent] update meshcop test cases (#6865)

Browse Source 
Add a new test case to verify the meshcop service is published with
proper fields.
This commit is contained in:
whd
2021-08-02 18:25:13 -07:00
committed by GitHub
parent 744db375a7
commit c33df6f0ee
3 changed files with 170 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/scripts/thread-cert/border_router/test_dnssd_server.py
+4 -1
View File
@@ -222,13 +222,16 @@ class TestDnssdServerOnBr(thread_cert.TestCase):
def _verify_discovery_proxy_meshcop(self, server_addr, network_name, digger):
dp_service_name = '_meshcop._udp.default.service.arpa.'
dp_instance_name = f'{network_name}._meshcop._udp.default.service.arpa.'
dp_hostname = lambda x: x.endswith('.default.service.arpa.')
def check_border_agent_port(port):
return 0 < port <= 65535
dig_result = digger.dns_dig(server_addr, dp_service_name, 'PTR')
for answer in dig_result['ANSWER']:
if len(answer) >= 2 and answer[-2] == 'PTR':
dp_instance_name = answer[-1]
break
self._assert_dig_result_matches(
dig_result, {
'QUESTION': [(dp_service_name, 'IN', 'PTR'),],
tests/scripts/thread-cert/border_router/test_publish_meshcop_service.py
Executable
+120
View File
@@ -0,0 +1,120 @@
#!/usr/bin/env python3
#
# Copyright (c) 2021, The OpenThread Authors.
# All rights reserved.
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions are met:
# 1. Redistributions of source code must retain the above copyright
# notice, this list of conditions and the following disclaimer.
# 2. Redistributions in binary form must reproduce the above copyright
# notice, this list of conditions and the following disclaimer in the
# documentation and/or other materials provided with the distribution.
# 3. Neither the name of the copyright holder nor the
# names of its contributors may be used to endorse or promote products
# derived from this software without specific prior written permission.
#
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS 'AS IS'
# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
# ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
# LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
# CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
# SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
# CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
# POSSIBILITY OF SUCH DAMAGE.
#
import ipaddress
import logging
import unittest
import config
import thread_cert
# Test description:
# This test verifies that OTBR publishes the meshcop service using a proper
# configuration.
#
# Topology:
# ----------------(eth)--------------------
# | |
# BR (Leader) HOST (mDNS Browser)
# |
# ROUTER
#
BR = 1
ROUTER = 2
HOST = 3
class PublishMeshCopService(thread_cert.TestCase):
USE_MESSAGE_FACTORY = False
TOPOLOGY = {
BR: {
'name': 'BR',
'allowlist': [ROUTER],
'is_otbr': True,
'version': '1.2',
},
ROUTER: {
'name': 'Router',
'allowlist': [BR],
'version': '1.2',
},
HOST: {
'name': 'Host',
'is_host': True
},
}
def test(self):
host = self.nodes[HOST]
br = self.nodes[BR]
client = self.nodes[ROUTER]
# TODO: verify the behavior when thread is disabled
host.bash('service otbr-agent stop')
host.start(start_radvd=False)
self.simulator.go(5)
br.start()
self.simulator.go(5)
self.assertEqual('leader', br.get_state())
client.start()
self.simulator.go(5)
self.assertEqual('router', client.get_state())
self.check_meshcop_service(br, host)
br.disable_backbone_router()
self.check_meshcop_service(br, host)
def check_meshcop_service(self, br, host):
instance_name = r'OpenThread'
data = host.discover_mdns_service(instance_name, '_meshcop._udp', None)
sb_data = 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
if br.get_state() == 'disabled':
self.assertEqual((state_bitmap >> 3 & 3), 0) # Thread is disabled
elif br.get_state() == 'detached':
self.assertEqual((state_bitmap >> 3 & 3), 1) # Thread is detached
else:
self.assertEqual((state_bitmap >> 3 & 3), 2) # Thread is attached
self.assertEqual((state_bitmap >> 5 & 3), 1) # high availability
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'])
if __name__ == '__main__':
unittest.main()
tests/scripts/thread-cert/node.py
+46 -10
View File
@@ -191,13 +191,13 @@ class OtbrDocker:
self._socat_proc.wait()
del self._socat_proc
def bash(self, cmd: str) -> List[str]:
def bash(self, cmd: str, encoding='ascii') -> List[str]:
logging.info("%s $ %s", self, cmd)
proc = subprocess.Popen(['docker', 'exec', '-i', self._docker_name, 'bash', '-c', cmd],
stdin=subprocess.DEVNULL,
stdout=subprocess.PIPE,
stderr=sys.stderr,
encoding='ascii')
encoding=encoding)
with proc:
@@ -247,7 +247,7 @@ class OtbrDocker:
],
}
"""
output = self.bash(f'dig -6 @{server} {name} {qtype}')
output = self.bash(f'dig -6 @{server} \'{name}\' {qtype}', encoding='raw_unicode_escape')
section = None
dig_result = {
@@ -286,7 +286,6 @@ class OtbrDocker:
if section == 'QUESTION':
assert line.startswith(';')
line = line[1:]
record = list(line.split())
if section != 'QUESTION':
@@ -2803,6 +2802,18 @@ class LinuxHost():
"""
self.bash(f'dns-sd -Z {name} local. > /tmp/{name} 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'):
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 &')
time.sleep(timeout)
@@ -2810,14 +2821,14 @@ class LinuxHost():
addresses = []
service = {}
logging.debug(self.bash(f'cat /tmp/{host_name}'))
logging.debug(self.bash(f'cat /tmp/{name}'))
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'))
# 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}'):
for line in self.bash(f'cat /tmp/{host_name}', encoding='raw_unicode_escape'):
elements = line.split()
fullname = f'{host_name}.local.'
if fullname not in elements:
@@ -2831,9 +2842,21 @@ class LinuxHost():
# my-service._ipps._tcp SRV 0 0 12345 my-host.local. ; Replace with unicast FQDN of target host
# my-service._ipps._tcp TXT ""
#
for line in self.bash(f'cat /tmp/{name}'):
is_txt = False
txt = ''
for line in self.bash(f'cat /tmp/{name}', encoding='raw_unicode_escape'):
elements = line.split()
if not elements or elements[0] != f'{instance}.{name}':
if len(elements) >= 2 and elements[1] == 'TXT':
is_txt = True
if is_txt:
txt += line.strip()
if line.strip().endswith('"'):
is_txt = False
txt_dict = self.__parse_dns_sd_txt(txt)
logging.info(f'txt = {txt_dict}')
service['txt'] = txt_dict
if not elements or elements[0] != full_service_name:
continue
if elements[1] == 'SRV':
service['fullname'] = elements[0]
@@ -2846,7 +2869,7 @@ class LinuxHost():
assert (service['host_fullname'] == f'{host_name}.local.')
service['host'] = host_name
service['addresses'] = addresses
return service if service['addresses'] else None
return service if 'addresses' in service and service['addresses'] else None
def start_radvd_service(self, prefix, slaac):
self.bash("""cat >/etc/radvd.conf <<EOF
@@ -2879,6 +2902,19 @@ EOF
def kill_radvd_service(self):
self.bash('pkill radvd')
def __parse_dns_sd_txt(self, line: str):
# Example TXT entry:
# "xp=\\000\\013\\184\\000\\000\\000\\000\\000"
txt = {}
for entry in re.findall(r'"((?:[^\\]|\\.)*?)"', line):
if '=' not in entry:
continue
k, v = entry.split('=', 1)
txt[k] = v
return txt
class OtbrNode(LinuxHost, NodeImpl, OtbrDocker):
is_otbr = True