espressif/openthread
1
0
Fork 0
mirror of https://github.com/espressif/openthread.git synced 2026-09-02 15:20:07 +00:00
Code Issues Packages Projects Releases Wiki Activity

[link-metrics] add blocking CLI query command and enhance tests (#8108)

Browse Source 
This commit updates CLI `linkmetrics query` command to allow it
to be used in blocking mode (wait for response to query). This
is then used in different tests to validate the `LinkMetrics`
module behavior. In particular, `v1_2_test_single_probe` and
`v1_2_LowPower_7_2_01_ForwardTrackingSeries` are updated to
validate that the received query report message is correctly
parsed.
This commit is contained in:
Abtin Keshavarzian
2022-09-01 10:49:02 -07:00
committed by GitHub
parent c3dd3ba8fb
commit 6dca78e757
5 changed files with 90 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/scripts/thread-cert/node.py
+29 -6
View File
@@ -2861,15 +2861,38 @@ class NodeImpl:
return router_table
def link_metrics_query_single_probe(self, dst_addr: str, linkmetrics_flags: str):
cmd = 'linkmetrics query %s single %s' % (dst_addr, linkmetrics_flags)
def link_metrics_query_single_probe(self, dst_addr: str, linkmetrics_flags: str, block: str = ""):
cmd = 'linkmetrics query %s single %s %s' % (dst_addr, linkmetrics_flags, block)
self.send_command(cmd)
self._expect_done()
self.simulator.go(5)
return self._parse_linkmetrics_query_result(self._expect_command_output())
def link_metrics_query_forward_tracking_series(self, dst_addr: str, series_id: int):
cmd = 'linkmetrics query %s forward %d' % (dst_addr, series_id)
def link_metrics_query_forward_tracking_series(self, dst_addr: str, series_id: int, block: str = ""):
cmd = 'linkmetrics query %s forward %d %s' % (dst_addr, series_id, block)
self.send_command(cmd)
self._expect_done()
self.simulator.go(5)
return self._parse_linkmetrics_query_result(self._expect_command_output())
def _parse_linkmetrics_query_result(self, lines):
"""Parse link metrics query result"""
# Exmaple of command output:
# ['Received Link Metrics Report from: fe80:0:0:0:146e:a00:0:1',
# '- PDU Counter: 1 (Count/Summation)',
# '- LQI: 0 (Exponential Moving Average)',
# '- Margin: 80 (dB) (Exponential Moving Average)',
# '- RSSI: -20 (dBm) (Exponential Moving Average)']
#
# Or 'Link Metrics Report, status: {status}'
result = {}
for line in lines:
if line.startswith('- '):
k, v = line[2:].split(': ')
result[k] = v.split(' ')[0]
elif line.startswith('Link Metrics Report, status: '):
result['Status'] = line[29:]
return result
def link_metrics_mgmt_req_enhanced_ack_based_probing(self,
dst_addr: str,
tests/scripts/thread-cert/v1_2_LowPower_7_2_01_ForwardTrackingSeries.py
+8 -8
View File
@@ -111,9 +111,10 @@ class LowPower_7_2_01_ForwardTrackingSeries(thread_cert.TestCase):
self.simulator.go(30)
# Step 7 - SED_1 sends an MLE Data Request to retrieve aggregated Forward Series Results
self.nodes[SED_1].link_metrics_query_forward_tracking_series(leader_addr, SERIES_ID)
result = self.nodes[SED_1].link_metrics_query_forward_tracking_series(leader_addr, SERIES_ID, 'block')
self.assertIn("PDU Counter", result)
self.simulator.go(5)
#self.simulator.go(5)
# Step 9 - SED_1 clears the Forward Tracking Series
# Forward Series Flags = 0x00:
@@ -144,9 +145,8 @@ class LowPower_7_2_01_ForwardTrackingSeries(thread_cert.TestCase):
self.simulator.go(1)
# Step 19 - SSED_1 sends an MLE Data Request to retrieve aggregated Forward Series Results
self.nodes[SSED_1].link_metrics_query_forward_tracking_series(leader_addr, SERIES_ID_2)
self.simulator.go(5)
result = self.nodes[SSED_1].link_metrics_query_forward_tracking_series(leader_addr, SERIES_ID_2, 'block')
self.assertIn("Margin", result)
# Step 21 - SSED_1 clears the Forward Series Link Metrics
# Forward Series Flags = 0x00:
@@ -157,9 +157,9 @@ class LowPower_7_2_01_ForwardTrackingSeries(thread_cert.TestCase):
self.simulator.go(5)
# Step 23 - SSED_1 sends an MLE Data Request to retrieve aggregated Forward Series Results
self.nodes[SSED_1].link_metrics_query_forward_tracking_series(leader_addr, SERIES_ID_2)
self.simulator.go(5)
result = self.nodes[SSED_1].link_metrics_query_forward_tracking_series(leader_addr, SERIES_ID_2, 'block')
self.assertIn('Status', result)
self.assertEqual(result['Status'], 'Series ID not recognized')
def verify(self, pv):
pkts = pv.pkts
tests/scripts/thread-cert/v1_2_test_single_probe.py
+18 -10
View File
@@ -70,40 +70,48 @@ class SSED_SingleProbe(thread_cert.TestCase):
leader_messages = self.simulator.get_messages_sent_by(LEADER)
# SSED_1 sends a Single Probe Link Metrics for L2 PDU count using MLE Data Request
self.nodes[SSED_1].link_metrics_query_single_probe(leader_addr, 'p')
self.simulator.go(5)
result = self.nodes[SSED_1].link_metrics_query_single_probe(leader_addr, 'p', 'block')
self.assertIn('PDU Counter', result)
self.assertEqual(len(result), 1)
leader_messages = self.simulator.get_messages_sent_by(LEADER)
msg = leader_messages.next_mle_message(mle.CommandType.DATA_RESPONSE)
msg.assertMleMessageContainsTlv(mle.LinkMetricsReport)
# SSED_1 sends a Single Probe Link Metrics for L2 LQI using MLE Data Request
self.nodes[SSED_1].link_metrics_query_single_probe(leader_addr, 'q')
self.simulator.go(5)
result = self.nodes[SSED_1].link_metrics_query_single_probe(leader_addr, 'q', 'block')
self.assertIn('LQI', result)
self.assertEqual(len(result), 1)
leader_messages = self.simulator.get_messages_sent_by(LEADER)
msg = leader_messages.next_mle_message(mle.CommandType.DATA_RESPONSE)
msg.assertMleMessageContainsTlv(mle.LinkMetricsReport)
# SSED_1 sends a Single Probe Link Metrics for Link Margin using MLE Data Request
self.nodes[SSED_1].link_metrics_query_single_probe(leader_addr, 'm')
self.simulator.go(5)
result = self.nodes[SSED_1].link_metrics_query_single_probe(leader_addr, 'm', 'block')
self.assertIn('Margin', result)
self.assertEqual(len(result), 1)
leader_messages = self.simulator.get_messages_sent_by(LEADER)
msg = leader_messages.next_mle_message(mle.CommandType.DATA_RESPONSE)
msg.assertMleMessageContainsTlv(mle.LinkMetricsReport)
# SSED_1 sends a Single Probe Link Metrics for Link Margin using MLE Data Request
self.nodes[SSED_1].link_metrics_query_single_probe(leader_addr, 'r')
self.simulator.go(5)
result = self.nodes[SSED_1].link_metrics_query_single_probe(leader_addr, 'r', 'block')
self.assertIn('RSSI', result)
self.assertEqual(len(result), 1)
leader_messages = self.simulator.get_messages_sent_by(LEADER)
msg = leader_messages.next_mle_message(mle.CommandType.DATA_RESPONSE)
msg.assertMleMessageContainsTlv(mle.LinkMetricsReport)
# SSED_1 sends a Single Probe Link Metrics for all metrics using MLE Data Request
self.nodes[SSED_1].link_metrics_query_single_probe(leader_addr, 'pqmr')
self.simulator.go(5)
result = self.nodes[SSED_1].link_metrics_query_single_probe(leader_addr, 'pqmr', 'block')
self.assertIn('PDU Counter', result)
self.assertIn('LQI', result)
self.assertIn('Margin', result)
self.assertIn('RSSI', result)
self.assertEqual(len(result), 4)
leader_messages = self.simulator.get_messages_sent_by(LEADER)
msg = leader_messages.next_mle_message(mle.CommandType.DATA_RESPONSE)