[trel] small updates and adding trel test (#7205)

- Adds platform API `trelDnssdInitialize` to initialize TREL
  DNS-SD module.
- Adds `test_trel_connectivity.py` test.
- Fixed `Border Router` TREL test to really use TREL.
This commit is contained in:
Simon Lin
2022-01-28 12:07:42 -08:00
committed by Jonathan Hui
parent 69ad96675f
commit 40b5fd3749
7 changed files with 223 additions and 11 deletions
+15 -5
View File
@@ -38,7 +38,7 @@ import sys
import time
import traceback
import unittest
from typing import Optional, Callable
from typing import Optional, Callable, Union, Any
import config
import debug
@@ -101,6 +101,7 @@ class TestCase(NcpSupportMixin, unittest.TestCase):
TOPOLOGY = None
CASE_WIRESHARK_PREFS = None
SUPPORT_THREAD_1_1 = True
PACKET_VERIFICATION = config.PACKET_VERIFICATION_DEFAULT
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
@@ -108,9 +109,16 @@ class TestCase(NcpSupportMixin, unittest.TestCase):
logging.basicConfig(level=logging.DEBUG, format='%(asctime)s - %(levelname)s - %(message)s')
self._start_time = None
self._do_packet_verification = PACKET_VERIFICATION and hasattr(self, 'verify')
self._do_packet_verification = PACKET_VERIFICATION and hasattr(self, 'verify') \
and self.PACKET_VERIFICATION == PACKET_VERIFICATION
def skipTest(self, reason: Any) -> None:
self._testSkipped = True
super(TestCase, self).skipTest(reason)
def setUp(self):
self._testSkipped = False
if ENV_THREAD_VERSION == '1.1' and not self.SUPPORT_THREAD_1_1:
self.skipTest('Thread 1.1 not supported.')
@@ -290,7 +298,8 @@ class TestCase(NcpSupportMixin, unittest.TestCase):
self._test_info['pcap'] = pcap_filename
test_info_path = self._output_test_info()
self._verify_packets(test_info_path)
if not self._testSkipped:
self._verify_packets(test_info_path)
def flush_all(self):
"""Flush away all captured messages of all nodes.
@@ -558,8 +567,9 @@ class TestCase(NcpSupportMixin, unittest.TestCase):
if timeout <= 0:
raise RuntimeError(f'wait failed after {timeout} seconds')
def wait_node_state(self, nodeid: int, state: str, timeout: int):
self.wait_until(lambda: self.nodes[nodeid].get_state() == state, timeout)
def wait_node_state(self, node: Union[int, Node], state: str, timeout: int):
node = self.nodes[node] if isinstance(node, int) else node
self.wait_until(lambda: node.get_state() == state, timeout)
def wait_route_established(self, node1: int, node2: int, timeout=10):
node2_addr = self.nodes[node2].get_ip6_address(config.ADDRESS_TYPE.RLOC)