espressif/openthread
1
0
Fork 0
mirror of https://github.com/espressif/openthread.git synced 2026-06-05 21:14:49 +00:00
Code Issues Packages Projects Releases Wiki Activity

[thread-cert] properly configure operational dataset (#9189)

Browse Source
This commit is contained in:
Jonathan Hui
2023-06-21 13:44:28 -04:00
committed by GitHub
parent e4aca89f09
commit 330c8088b9
3 changed files with 76 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/scripts/thread-cert/config.py
+7
View File
@@ -123,7 +123,14 @@ RSSI = {
}
SNIFFER_ID = int(os.getenv('SNIFFER_ID', 34))
CHANNEL = 11
CHANNEL_MASK = 0x07fff800
EXTENDED_PANID = 'dead00beef00cafe'
NETWORK_NAME = 'OpenThread'
PANID = 0xface
PSKC = 'c23a76e98f1a6483639b1ac1271e2e27'
SECURITY_POLICY = [672, 'onrc']
LEADER_STARTUP_DELAY = 12
ROUTER_STARTUP_DELAY = 10
tests/scripts/thread-cert/node.py
+40 -20
View File
@@ -762,7 +762,6 @@ class NodeImpl:
super().__init__(nodeid, **kwargs)
self.set_mesh_local_prefix(config.MESH_LOCAL_PREFIX)
self.set_addr64('%016x' % (thread_cert.EXTENDED_ADDRESS_BASE + nodeid))
def _expect(self, pattern, timeout=-1, *args, **kwargs):
@@ -2548,38 +2547,63 @@ class NodeImpl:
def set_active_dataset(
self,
timestamp,
panid=None,
timestamp=None,
channel=None,
channel_mask=None,
extended_panid=None,
mesh_local_prefix=None,
network_key=None,
network_name=None,
panid=None,
pskc=None,
security_policy=[],
):
self.send_command('dataset clear')
self.send_command('dataset clear', go=False)
self._expect_done()
cmd = 'dataset activetimestamp %d' % timestamp
self.send_command(cmd)
self._expect_done()
if panid is not None:
cmd = 'dataset panid %d' % panid
self.send_command(cmd)
if timestamp is not None:
cmd = 'dataset activetimestamp %d' % timestamp
self.send_command(cmd, go=False)
self._expect_done()
if channel is not None:
cmd = 'dataset channel %d' % channel
self.send_command(cmd)
self.send_command(cmd, go=False)
self._expect_done()
if channel_mask is not None:
cmd = 'dataset channelmask %d' % channel_mask
self.send_command(cmd)
self.send_command(cmd, go=False)
self._expect_done()
if extended_panid is not None:
cmd = 'dataset extpanid %s' % extended_panid
self.send_command(cmd, go=False)
self._expect_done()
if mesh_local_prefix is not None:
cmd = 'dataset meshlocalprefix %s' % mesh_local_prefix
self.send_command(cmd, go=False)
self._expect_done()
if network_key is not None:
cmd = 'dataset networkkey %s' % network_key
self.send_command(cmd)
self.send_command(cmd, go=False)
self._expect_done()
if network_name is not None:
cmd = 'dataset networkname %s' % network_name
self.send_command(cmd, go=False)
self._expect_done()
if panid is not None:
cmd = 'dataset panid %d' % panid
self.send_command(cmd, go=False)
self._expect_done()
if pskc is not None:
cmd = 'dataset pskc %s' % pskc
self.send_command(cmd, go=False)
self._expect_done()
if security_policy and len(security_policy) == 2:
@@ -2587,14 +2611,10 @@ class NodeImpl:
str(security_policy[0]),
security_policy[1],
)
self.send_command(cmd)
self.send_command(cmd, go=False)
self._expect_done()
# Set the meshlocal prefix in config.py
self.send_command('dataset meshlocalprefix %s' % config.MESH_LOCAL_PREFIX.split('/')[0])
self._expect_done()
self.send_command('dataset commit active')
self.send_command('dataset commit active', go=False)
self._expect_done()
def set_pending_dataset(self, pendingtimestamp, activetimestamp, panid=None, channel=None, delay=None):
tests/scripts/thread-cert/thread_cert.py
+29 -20
View File
@@ -64,7 +64,6 @@ DEFAULT_PARAMS = {
'mode': 'rdn',
'allowlist': None,
'version': ENV_THREAD_VERSION,
'panid': 0xface,
}
"""Default configurations when creating nodes."""
@@ -179,20 +178,10 @@ class TestCase(NcpSupportMixin, unittest.TestCase):
if node.is_host:
continue
self.nodes[i].set_networkkey(binascii.hexlify(config.DEFAULT_NETWORK_KEY).decode())
self.nodes[i].set_panid(params['panid'])
self.nodes[i].set_mode(params['mode'])
if 'extended_panid' in params:
self.nodes[i].set_extpanid(params['extended_panid'])
if 'partition_id' in params:
self.nodes[i].set_preferred_partition_id(params['partition_id'])
if 'channel' in params:
self.nodes[i].set_channel(params['channel'])
if 'networkkey' in params:
self.nodes[i].set_networkkey(params['networkkey'])
if 'network_name' in params:
self.nodes[i].set_network_name(params['network_name'])
if params['is_ftd']:
self.nodes[i].set_router_selection_jitter(params['router_selection_jitter'])
@@ -209,15 +198,7 @@ class TestCase(NcpSupportMixin, unittest.TestCase):
if 'timeout' in params:
self.nodes[i].set_timeout(params['timeout'])
if 'active_dataset' in params:
if 'network_key' not in params['active_dataset']:
params['active_dataset']['network_key'] = binascii.hexlify(config.DEFAULT_NETWORK_KEY).decode()
self.nodes[i].set_active_dataset(params['active_dataset']['timestamp'],
panid=params['active_dataset'].get('panid'),
channel=params['active_dataset'].get('channel'),
channel_mask=params['active_dataset'].get('channel_mask'),
network_key=params['active_dataset'].get('network_key'),
security_policy=params['active_dataset'].get('security_policy'))
self._set_up_active_dataset(self.nodes[i], params)
if 'pending_dataset' in params:
self.nodes[i].set_pending_dataset(params['pending_dataset']['pendingtimestamp'],
@@ -262,6 +243,34 @@ class TestCase(NcpSupportMixin, unittest.TestCase):
self._inspector = debug.Inspector(self)
self._collect_test_info_after_setup()
def _set_up_active_dataset(self, node, params):
dataset = {
'timestamp': 1,
'channel': config.CHANNEL,
'channel_mask': config.CHANNEL_MASK,
'extended_panid': config.EXTENDED_PANID,
'mesh_local_prefix': config.MESH_LOCAL_PREFIX.split('/')[0],
'network_key': binascii.hexlify(config.DEFAULT_NETWORK_KEY).decode(),
'network_name': config.NETWORK_NAME,
'panid': config.PANID,
'pskc': config.PSKC,
'security_policy': config.SECURITY_POLICY,
}
if 'channel' in params:
dataset['channel'] = params['channel']
if 'networkkey' in params:
dataset['network_key'] = params['networkkey']
if 'network_name' in params:
dataset['network_name'] = params['network_name']
if 'panid' in params:
dataset['panid'] = params['panid']
if 'active_dataset' in params:
dataset.update(params['active_dataset'])
node.set_active_dataset(**dataset)
def inspect(self):
self._inspector.inspect()