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

[tests] migrate low power and 1.2 tests to nexus (#12641)

Browse Source 
This commit removes several legacy Python-based certification tests for
Low Power (CSL) and Thread 1.2 features, as they have been migrated to
the Nexus test framework.

Specifically, the following tests and their associated CI workflows
(including packet verification for low power) are removed:
- CSL Transmission and Timeout
- Enhanced Frame Pending and Keep-Alive
- Single Probe and Forward Tracking Series Link Metrics
- SSED Attachment and Parent Selection

The removal of these scripts from tests/scripts/thread-cert/ and
the corresponding GitHub Action workflows reduces CI overhead while
maintaining coverage through the more scalable Nexus tests.
This commit is contained in:
Jonathan Hui
2026-03-06 14:35:51 -06:00
committed by GitHub
parent 9080e827d8
commit 0740bff909
16 changed files with 0 additions and 2734 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.github/workflows/otbr.yml
-6
View File
@@ -137,12 +137,6 @@ jobs:
packet_verification: 1
nat64: 0
description: "MATN"
- otbr_mdns: "mDNSResponder"
otbr_trel: 0
cert_scripts: ./tests/scripts/thread-cert/border_router/LowPower/*.py
packet_verification: 1
nat64: 0
description: "LowPower"
- otbr_mdns: "mDNSResponder"
otbr_trel: 0
cert_scripts: ./tests/scripts/thread-cert/border_router/internet/*.py
.github/workflows/simulation-1.4.yml
-150
View File
@@ -121,85 +121,6 @@ jobs:
path: tmp/coverage.info
retention-days: 1
packet-verification-low-power:
runs-on: ubuntu-24.04
env:
REFERENCE_DEVICE: 1
VIRTUAL_TIME: 1
TEST_TIMEOUT: 300
OT_VT_USE_UNIX_SOCKET: 1
COVERAGE: 1
PACKET_VERIFICATION: 1
THREAD_VERSION: 1.4
MAC_FILTER: 1
INTER_OP: 1
INTER_OP_BBR: 0
steps:
- name: Harden Runner
uses: step-security/harden-runner@5ef0c079ce82195b2a36a210272d6b661572d83e # v2.14.2
with:
egress-policy: audit # TODO: change to 'egress-policy: block' after couple of runs
- uses: actions/checkout@08c6903cd8c0fde910a37f88322edcfb5dd907a8 # v5.0.0
with:
submodules: recursive
- uses: actions/setup-python@a309ff8b426b58ec0e2a45f0f869d46889d02405 # v6.2.0
with:
python-version: '3.12'
cache: pip
- name: Bootstrap
run: |
sudo apt-get update
sudo apt-get --no-install-recommends install -y ninja-build lcov
python3 -m pip install -r tests/scripts/thread-cert/requirements.txt
- name: Build
run: |
./script/test build
- name: Build with OT_CSL_RECEIVER_LOCAL_TIME_SYNC
run: |
OT_BUILDDIR="${PWD}/build_csl_receiver_local_time_sync" OT_OPTIONS="-DOT_CSL_RECEIVER_LOCAL_TIME_SYNC=ON" ./script/test build
- name: Get Thread-Wireshark
run: |
./script/test get_thread_wireshark
- name: Run
run: |
ulimit -c unlimited
./script/test prepare_coredump_upload
for i in {1..10}
do
./script/test cert_suite ./tests/scripts/thread-cert/v1_2_LowPower*.py
done
- name: Run with OT_CSL_RECEIVER_LOCAL_TIME_SYNC
run: |
OT_BUILDDIR="${PWD}/build_csl_receiver_local_time_sync" ./script/test cert_suite ./tests/scripts/thread-cert/v1_2_LowPower*.py
- name: Check Crash
if: ${{ failure() }}
run: |
CRASHED=$(./script/test check_crash | tail -1)
[[ $CRASHED -eq "1" ]] && echo "Crashed!" || echo "Not crashed."
echo "CRASHED=$CRASHED" >> $GITHUB_ENV
- uses: actions/upload-artifact@b7c566a772e6b6bfb58ed0dc250532a479d7789f # v6.0.0
if: ${{ failure() }}
with:
name: packet-verification-low-power-pcaps
path: |
*.pcap
*.json
- uses: actions/upload-artifact@b7c566a772e6b6bfb58ed0dc250532a479d7789f # v6.0.0
if: ${{ failure() && env.CRASHED == '1' }}
with:
name: core-packet-verification-low-power
path: |
./ot-core-dump/*
- name: Generate Coverage
run: |
./script/test generate_coverage gcc
- uses: actions/upload-artifact@b7c566a772e6b6bfb58ed0dc250532a479d7789f # v6.0.0
with:
name: cov-packet-verification-low-power
path: tmp/coverage.info
retention-days: 1
packet-verification-1-1-on-1-4:
runs-on: ubuntu-24.04
env:
@@ -352,82 +273,11 @@ jobs:
path: tmp/coverage.info
retention-days: 1
thread-1-4-posix:
runs-on: ubuntu-24.04
env:
COVERAGE: 1
PYTHONUNBUFFERED: 1
READLINE: readline
THREAD_VERSION: 1.4
OT_NODE_TYPE: rcp
USE_MTD: 1
VIRTUAL_TIME: 1
OT_VT_USE_UNIX_SOCKET: 1
INTER_OP: 1
steps:
- name: Harden Runner
uses: step-security/harden-runner@5ef0c079ce82195b2a36a210272d6b661572d83e # v2.14.2
with:
egress-policy: audit # TODO: change to 'egress-policy: block' after couple of runs
- uses: actions/checkout@08c6903cd8c0fde910a37f88322edcfb5dd907a8 # v5.0.0
with:
submodules: recursive
- uses: actions/setup-python@a309ff8b426b58ec0e2a45f0f869d46889d02405 # v6.2.0
with:
python-version: '3.12'
cache: pip
- name: Bootstrap
run: |
sudo apt-get update
sudo apt-get --no-install-recommends install -y libreadline6-dev ninja-build llvm lcov
python3 -m pip install -r tests/scripts/thread-cert/requirements.txt
- name: Build
run: |
./script/test build
- name: Run
run: |
ulimit -c unlimited
./script/test prepare_coredump_upload
./script/test cert tests/scripts/thread-cert/v1_2_LowPower_5_3_01_SSEDAttachment.py
./script/test cert tests/scripts/thread-cert/v1_2_LowPower_6_1_07_PreferringARouterOverAReed.py
./script/test cert tests/scripts/thread-cert/v1_2_router_5_1_1.py
./script/test cert tests/scripts/thread-cert/v1_2_test_csl_transmission.py
./script/test cert tests/scripts/thread-cert/v1_2_test_enhanced_frame_pending.py
./script/test cert tests/scripts/thread-cert/v1_2_test_parent_selection.py
- name: Check Crash
if: ${{ failure() }}
run: |
CRASHED=$(./script/test check_crash | tail -1)
[[ $CRASHED -eq "1" ]] && echo "Crashed!" || echo "Not crashed."
echo "CRASHED=$CRASHED" >> $GITHUB_ENV
- uses: actions/upload-artifact@b7c566a772e6b6bfb58ed0dc250532a479d7789f # v6.0.0
if: ${{ failure() }}
with:
name: thread-1-4-posix-pcaps
path: "*.pcap"
- uses: actions/upload-artifact@b7c566a772e6b6bfb58ed0dc250532a479d7789f # v6.0.0
if: ${{ failure() && env.CRASHED == '1' }}
with:
name: core-thread-1-4-posix
path: |
./ot-core-dump/*
- name: Generate Coverage
run: |
./script/test generate_coverage gcc
- uses: actions/upload-artifact@b7c566a772e6b6bfb58ed0dc250532a479d7789f # v6.0.0
with:
name: cov-thread-1-4-posix
path: tmp/coverage.info
retention-days: 1
upload-coverage:
needs:
- thread-1-4
- packet-verification-low-power
- packet-verification-1-1-on-1-4
- expects
- thread-1-4-posix
runs-on: ubuntu-22.04
steps:
- name: Harden Runner
tests/scripts/thread-cert/border_router/LowPower/v1_2_LowPower_5_3_01_SSEDAttachment_BR.py
-38
View File
@@ -1,38 +0,0 @@
#!/usr/bin/env python3
#
# Copyright (c) 2022, 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 unittest
from v1_2_LowPower_5_3_01_SSEDAttachment import LowPower_5_3_01_SSEDAttachment
from v1_2_LowPower_5_3_01_SSEDAttachment import LEADER
LowPower_5_3_01_SSEDAttachment.TOPOLOGY[LEADER]['is_otbr'] = True
if __name__ == '__main__':
unittest.main()
tests/scripts/thread-cert/border_router/LowPower/v1_2_LowPower_7_1_01_SingleProbeLinkMetricsWithEnhancedAcks_BR.py
-38
View File
@@ -1,38 +0,0 @@
#!/usr/bin/env python3
#
# Copyright (c) 2022, 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 unittest
from v1_2_LowPower_7_1_01_SingleProbeLinkMetricsWithEnhancedAcks import LowPower_7_1_01
from v1_2_LowPower_7_1_01_SingleProbeLinkMetricsWithEnhancedAcks import LEADER
LowPower_7_1_01.TOPOLOGY[LEADER]['is_otbr'] = True
if __name__ == '__main__':
unittest.main()
tests/scripts/thread-cert/border_router/LowPower/v1_2_LowPower_7_2_01_ForwardTrackingSeries_BR.py
-37
View File
@@ -1,37 +0,0 @@
#!/usr/bin/env python3
#
# Copyright (c) 2022, 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 unittest
from v1_2_LowPower_7_2_01_ForwardTrackingSeries import LEADER
from v1_2_LowPower_7_2_01_ForwardTrackingSeries import LowPower_7_2_01_ForwardTrackingSeries
LowPower_7_2_01_ForwardTrackingSeries.TOPOLOGY[LEADER]['is_otbr'] = True
if __name__ == '__main__':
unittest.main()
tests/scripts/thread-cert/v1_2_LowPower_5_3_01_SSEDAttachment.py
-145
View File
@@ -1,145 +0,0 @@
#!/usr/bin/env python3
#
# Copyright (c) 2020, 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 unittest
from config import ADDRESS_TYPE
from mac802154 import MacHeader
from pktverify import consts
from pktverify.packet_verifier import PacketVerifier
import config
import thread_cert
LEADER = 1
ROUTER = 2
SSED_1 = 3
class LowPower_5_3_01_SSEDAttachment(thread_cert.TestCase):
USE_MESSAGE_FACTORY = False
TOPOLOGY = {
LEADER: {
'version': '1.2',
'name': 'LEADER',
'mode': 'rdn',
'allowlist': [ROUTER, SSED_1]
},
ROUTER: {
'version': '1.2',
'name': 'ROUTER',
'mode': 'rdn',
'allowlist': [LEADER],
},
SSED_1: {
'version': '1.2',
'name': 'SSED_1',
'is_mtd': True,
'mode': '-',
'allowlist': [LEADER],
},
}
def test(self):
self.nodes[LEADER].start()
self.simulator.go(config.LEADER_STARTUP_DELAY)
self.assertEqual(self.nodes[LEADER].get_state(), 'leader')
self.nodes[ROUTER].start()
self.simulator.go(config.ROUTER_STARTUP_DELAY)
self.assertEqual(self.nodes[ROUTER].get_state(), 'router')
self.nodes[SSED_1].set_csl_period(consts.CSL_DEFAULT_PERIOD)
self.nodes[SSED_1].start()
self.simulator.go(5)
self.assertEqual(self.nodes[SSED_1].get_state(), 'child')
self.simulator.go(5)
self.collect_rloc16s()
self.collect_rlocs()
# Verify connectivity by sending an ICMP Echo Request from the Router to SSED_1 mesh-local address via the DUT
self.assertTrue(self.nodes[ROUTER].ping(self.nodes[SSED_1].get_ip6_address(ADDRESS_TYPE.RLOC)))
# Verify fragmented CSL transmission
self.assertTrue(self.nodes[ROUTER].ping(self.nodes[SSED_1].get_ip6_address(ADDRESS_TYPE.RLOC), size=128))
self.simulator.go(5)
def verify(self, pv):
pkts = pv.pkts
pv.summary.show()
LEADER = pv.vars['LEADER']
ROUTER = pv.vars['ROUTER']
SSED_1 = pv.vars['SSED_1']
# Step 1: ensure the Leader is sending MLE Advertisements
pkts.filter_wpan_src64(LEADER) \
.filter_mle_cmd(consts.MLE_ADVERTISEMENT) \
.must_next()
# Step 2: Verify that the frame version is IEEE Std 802.15.4-2015 of the MLE Child Update Response from the
# DUT to SSED_1
pkts.filter_wpan_src64(LEADER) \
.filter_wpan_dst64(SSED_1) \
.filter_mle_cmd(consts.MLE_CHILD_UPDATE_RESPONSE) \
.filter_wpan_version(consts.MAC_FRAME_VERSION_2015) \
.must_next()
# Step 3:
# Verify that SSED_1 does not send a MAC Data Request prior to receiving the ICMP Echo Request from the Leader.
# Verify that the frame version is IEEE Std 802.15.4-2015 of the ICMP Echo Request sent to SSED_1 from the DUT
# Verify that the Router successfully receives an Echo Response from SSED_1
ssed_rloc16 = pv.vars['SSED_1_RLOC16']
ssed_rloc = pv.vars['SSED_1_RLOC']
leader_rloc16 = pv.vars['LEADER_RLOC16']
router_rloc16 = pv.vars['ROUTER_RLOC16']
router_rloc = pv.vars['ROUTER_RLOC']
pkts.filter_wpan_src64(LEADER) \
.filter('wpan.dst16 == {rloc16}', rloc16 = ssed_rloc16) \
.filter_ipv6_dst(ssed_rloc) \
.filter_ping_request() \
.filter_wpan_version(consts.MAC_FRAME_VERSION_2015) \
.must_next()
idx1 = pkts.index
pkts.filter_wpan_src64(SSED_1) \
.filter('wpan.dst16 == {rloc16}', rloc16 = leader_rloc16) \
.filter_ipv6_dst(router_rloc) \
.filter_ping_reply() \
.filter_wpan_version(consts.MAC_FRAME_VERSION_2015) \
.must_next() \
.must_verify(lambda p: (p.wpan.ie_present == 1) and (consts.CSL_IE_ID in p.wpan.header_ie.id))
idx2 = pkts.index
pkts.range(idx1, idx2).filter(lambda p: p.wpan.cmd == MacHeader.CommandIdentifier.DATA_REQUEST) \
.must_not_next()
if __name__ == '__main__':
unittest.main()
tests/scripts/thread-cert/v1_2_LowPower_6_1_07_PreferringARouterOverAReed.py
-172
View File
@@ -1,172 +0,0 @@
#!/usr/bin/env python3
#
# Copyright (c) 2020, 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 copy
import unittest
from mle import TlvType
from pktverify import consts
from pktverify.packet_verifier import PacketVerifier
import config
import thread_cert
LEADER = 1
ROUTER_1 = 2
REED_1 = 3
DUT = 4
class LowPower_6_1_07_PreferringARouterOverAReed_Base(thread_cert.TestCase):
USE_MESSAGE_FACTORY = False
TOPOLOGY = {
LEADER: {
'version': '1.2',
'name': 'LEADER',
'mode': 'rdn',
'allowlist': [ROUTER_1, REED_1]
},
ROUTER_1: {
'version': '1.1',
'name': 'ROUTER_1',
'mode': 'rdn',
'allowlist': [LEADER, DUT],
},
REED_1: {
'version': '1.2',
'name': 'REED_1',
'mode': 'rdn',
'router_upgrade_threshold': 0,
'allowlist': [LEADER, DUT],
},
DUT: {
'version': '1.2',
'name': 'DUT',
'is_mtd': True,
'allowlist': [ROUTER_1, REED_1],
},
}
def test(self):
self.nodes[LEADER].start()
self.simulator.go(config.LEADER_STARTUP_DELAY)
self.assertEqual(self.nodes[LEADER].get_state(), 'leader')
self.nodes[ROUTER_1].start()
self.simulator.go(config.ROUTER_STARTUP_DELAY)
self.assertEqual(self.nodes[ROUTER_1].get_state(), 'router')
self.nodes[REED_1].start()
self.simulator.go(5)
self.assertEqual(self.nodes[REED_1].get_state(), 'child')
self.simulator.go(5) # wait for advertisements
# Step 2 - set outbound link quality of ROUTER_1 and REED_1 to 2 (medium)
self.nodes[ROUTER_1].set_outbound_link_quality(2)
self.nodes[REED_1].set_outbound_link_quality(2)
# Step 3 - DUT Automatically begins attach process by sending a multicast MLE Parent Request
self.nodes[DUT].start()
self.simulator.go(5)
self.assertEqual(self.nodes[DUT].get_state(), 'child')
def verify(self, pv):
pkts = pv.pkts
pv.summary.show()
LEADER = pv.vars['LEADER']
ROUTER_1 = pv.vars['ROUTER_1']
REED_1 = pv.vars['REED_1']
DUT = pv.vars['DUT']
# Step 1 - Ensure all routers and leader are sending MLE advertisements
pkts.filter_wpan_src64(LEADER) \
.filter_mle_cmd(consts.MLE_ADVERTISEMENT) \
.must_next()
pkts.filter_wpan_src64(ROUTER_1) \
.filter_mle_cmd(consts.MLE_ADVERTISEMENT) \
.must_next()
# Step 3 - DUT begins attach process by sending a multicast MLE Parent Request
pkts.filter_wpan_src64(DUT) \
.filter_mle_cmd(consts.MLE_PARENT_REQUEST) \
.filter_mle_has_tlv(TlvType.CHALLENGE, TlvType.MODE) \
.filter(lambda p: p.mle.tlv.scan_mask.r == 1 and p.mle.tlv.scan_mask.e == 0) \
.filter(lambda p: p.mle.tlv.version >= config.THREAD_VERSION_1_2) \
.filter(lambda p: p.ipv6.hlim == 255) \
.filter_LLARMA() \
.must_next()
# Step 4 - Router_1 automatically Respond with MLE Parent Response
pkts.filter_wpan_src64(ROUTER_1) \
.filter_wpan_dst64(DUT) \
.filter_mle_cmd(consts.MLE_PARENT_RESPONSE) \
.filter(lambda p: p.mle.tlv.version == config.THREAD_VERSION_1_1) \
.must_next()
# Step 5 - DUT sends another multicast MLE Parent Request to the all-routers multicast with the Scan Mask TLV
# set for all routers and all REEDs
pkts.filter_wpan_src64(DUT) \
.filter_mle_cmd(consts.MLE_PARENT_REQUEST) \
.filter_mle_has_tlv(TlvType.CHALLENGE, TlvType.MODE) \
.filter(lambda p: p.mle.tlv.scan_mask.r == 1 and p.mle.tlv.scan_mask.e == 1) \
.filter(lambda p: p.mle.tlv.version >= consts.THREAD_VERSION_1_2) \
.filter(lambda p: p.ipv6.hlim == 255) \
.filter_LLARMA() \
.must_next()
# Step 6 - REED_1 sends MLE Parent Response to the DUT
pkts.filter_wpan_src64(REED_1) \
.filter_wpan_dst64(DUT) \
.filter_mle_cmd(consts.MLE_PARENT_RESPONSE) \
.must_next()
# # Step 7 - DUT sends a Child ID Request to REED_1 due to its better link quality
pkts.filter_wpan_src64(DUT) \
.filter_mle_cmd(consts.MLE_CHILD_ID_REQUEST) \
.filter_wpan_dst64(ROUTER_1) \
.filter_mle_has_tlv(TlvType.ADDRESS_REGISTRATION, TlvType.LINK_LAYER_FRAME_COUNTER, TlvType.MODE, TlvType.RESPONSE, TlvType.TIMEOUT, TlvType.TLV_REQUEST) \
.filter(lambda p: p.mle.tlv.version >= consts.THREAD_VERSION_1_2) \
.must_next()
class LowPower_6_1_07_PreferringARouterOverAReed_MED(LowPower_6_1_07_PreferringARouterOverAReed_Base):
TOPOLOGY = copy.deepcopy(LowPower_6_1_07_PreferringARouterOverAReed_Base.TOPOLOGY)
TOPOLOGY[DUT]['mode'] = 'rn'
class LowPower_6_1_07_PreferringARouterOverAReed_SED(LowPower_6_1_07_PreferringARouterOverAReed_Base):
TOPOLOGY = copy.deepcopy(LowPower_6_1_07_PreferringARouterOverAReed_Base.TOPOLOGY)
TOPOLOGY[DUT]['mode'] = '-'
del (LowPower_6_1_07_PreferringARouterOverAReed_Base)
if __name__ == '__main__':
unittest.main()
tests/scripts/thread-cert/v1_2_LowPower_7_1_01_SingleProbeLinkMetricsWithEnhancedAcks.py
-393
View File
@@ -1,393 +0,0 @@
#!/usr/bin/env python3
#
# Copyright (c) 2020, 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 unittest
from config import ADDRESS_TYPE
from pktverify import consts
import config
import thread_cert
LEADER = 1
SED_1 = 2
SSED_1 = 3
POLL_PERIOD = 3000 # 3s
class LowPower_7_1_01(thread_cert.TestCase):
USE_MESSAGE_FACTORY = False
TOPOLOGY = {
LEADER: {
'version': '1.2',
'name': 'LEADER',
'mode': 'rdn',
'allowlist': [SED_1, SSED_1],
},
SED_1: {
'version': '1.2',
'name': 'SED_1',
'mode': '-',
'allowlist': [LEADER],
},
SSED_1: {
'version': '1.2',
'name': 'SSED_1',
'mode': '-',
'allowlist': [LEADER],
}
}
"""All nodes are created with default configurations"""
def test(self):
self.nodes[LEADER].start()
self.simulator.go(config.LEADER_STARTUP_DELAY)
self.assertEqual(self.nodes[LEADER].get_state(), 'leader')
self.nodes[SED_1].set_pollperiod(POLL_PERIOD)
self.nodes[SED_1].start()
self.simulator.go(5)
self.assertEqual(self.nodes[SED_1].get_state(), 'child')
self.nodes[SSED_1].set_csl_period(consts.CSL_DEFAULT_PERIOD)
self.nodes[SSED_1].start()
self.simulator.go(5)
self.assertEqual(self.nodes[SSED_1].get_state(), 'child')
leader_addr = self.nodes[LEADER].get_ip6_address(ADDRESS_TYPE.LINK_LOCAL)
# Step 3 - Verify connectivity by instructing each device to send an ICMPv6 Echo Request to the DUT
self.assertTrue(self.nodes[SED_1].ping(leader_addr, timeout=POLL_PERIOD * 2 / 1000))
self.assertTrue(self.nodes[SSED_1].ping(leader_addr))
self.simulator.go(5)
# Step 4 - SED_1 enables IEEE 802.15.4-2015 Enhanced ACK based Probing by sending a Link Metrics Management
# Request to the DUT
# MLE Link Metrics Management TLV Payload:
# - Enhanced ACK Configuration Sub-TLV
# -- Enh-ACK Flags = 1 (register a configuration)
# -- Concatenation of Link Metric Type ID Flags = 0x00:
# --- Item1: (0)(0)(001)(010) = 0x0a
# ---- E = 0
# ---- L = 0
# ---- Type/Average Enum = 1 (Exponential Moving Avg)
# ---- Metrics Enum = 2 (Link Margin)
self.nodes[SED_1].link_metrics_config_req_enhanced_ack_based_probing(leader_addr, True, 'm', mode='async')
self.simulator.go(5)
# Step 6 - SSED_1 enables IEEE 802.15.4-2015 Enhanced ACK based Probing by sending a Link Metrics Management
# Request to the DUT
# MLE Link Metrics Management TLV Payload:
# - Enhanced ACK Configuration Sub-TLV
# -- Enh-ACK Flags = 1 (register a configuration)
# -- Concatenation of Link Metric Type ID Flags = 0x00:
# --- Item1: (0)(0)(001)(010) = 0x0a
# --- Item2: (0)(0)(001)(011) = 0x0b
# ---- E = 0
# ---- L = 0
# ---- Type/Average Enum = 1 (Exponential Moving Avg)
# ---- Metrics Enum = 2 (Link Margin)
# ---- Metrics Enum = 3 (RSSI)
self.nodes[SSED_1].link_metrics_config_req_enhanced_ack_based_probing(leader_addr, True, 'mr', mode='async')
self.simulator.go(5)
# Step 8 - SSED_1 sends an MLE Data Message with setting Frame Version subfield within the MAC header of the
# message to 0b10
self.nodes[SSED_1].send_mac_emptydata()
self.simulator.go(5)
# Step 10 - SED_1 sends an MLE Data Message with setting Frame Version subfield within the MAC header of the
# message # to 0b10
self.nodes[SED_1].send_mac_emptydata()
self.simulator.go(5)
# Step 12 - SSED_1 clears its Enhanced ACK link metrics configuration by # sending a Link Metrics Management
# Request to the DUT
# Enh-ACK Flags = 0 (clear enhanced ACK link metric config)
self.nodes[SSED_1].link_metrics_config_req_enhanced_ack_based_probing(leader_addr, False, '', mode='async')
self.simulator.go(5)
# Step 14 - SSED_1 Sends a MLE Data Message with setting Frame Version subfield within the MAC header of the
# message to 0b10
self.nodes[SSED_1].send_mac_emptydata()
self.simulator.go(5)
# Step 16 - This step verifies that Enhanced ACKs cannot be enabled while requesting 3 metric types by
# instructing the device to send the following Link Metrics Management Request to the DUT
# MLE Link Metrics Management TLV Payload:
# - Enhanced ACK Configuration Sub-TLV
# -- Enh-ACK Flags = 1 (register a configuration)
# -- Concatenation of Link Metric Type ID Flags = 0x00:
# --- Item1: (0)(0)(001)(001) = 0x09
# --- Item2: (0)(0)(001)(010) = 0x0a
# --- Item3: (0)(0)(001)(011) = 0x0b
# ---- E = 0
# ---- L = 0
# ---- Type/Average Enum = 1 (Exponential Moving Avg)
# ---- Metrics Enum = 1 (Layer 2 LQI)
# ---- Metrics Enum = 2 (Link Margin)
# ---- Metrics Enum = 3 (RSSI)
self.nodes[SSED_1].link_metrics_config_req_enhanced_ack_based_probing(leader_addr, True, 'qmr', mode='async')
self.simulator.go(5)
# Step 18 - This step verifies that Enhanced ACKs cannot be enabled while requesting a reserved Type/Average
# enum of value 2 by instructing the device to send the following Link Metrics Management Request to the DUT
# MLE Link Metrics Management TLV Payload:
# - Enhanced ACK Configuration Sub-TLV
# -- Enh-ACK Flags = 1 (register a configuration)
# -- Concatenation of Link Metric Type ID Flags = 0x00:
# --- Item1: (0)(0)(010)(010) = 0x12
# ---- E = 0
# ---- L = 0
# ---- Type/Average Enum = 2 (Reserved)
# ---- Metrics Enum = 2 (Link Margin)
self.nodes[SSED_1].link_metrics_config_req_enhanced_ack_based_probing(leader_addr,
True,
'm',
'r',
mode='async')
self.simulator.go(5)
def verify(self, pv):
pkts = pv.pkts
pv.summary.show()
LEADER = pv.vars['LEADER']
SED_1 = pv.vars['SED_1']
SSED_1 = pv.vars['SSED_1']
# Step 3 - The DUT MUST send ICMPv6 Echo Responses to both SED1 & SSED1
pkts.filter_wpan_src64(LEADER) \
.filter_wpan_dst64(SED_1) \
.filter_ping_reply() \
.must_next()
pkts.filter_wpan_src64(LEADER) \
.filter_wpan_dst64(SSED_1) \
.filter_ping_reply() \
.must_next()
# Step 4 - SED_1 enables IEEE 802.15.4-2015 Enhanced ACK based Probing by sending a Link Metrics Management
# Request to the DUT
# MLE Link Metrics Management TLV Payload:
# - Enhanced ACK Configuration Sub-TLV
# -- Enh-ACK Flags = 1 (register a configuration)
# -- Concatenation of Link Metric Type ID Flags = 0x00:
# --- Item1: (0)(0)(001)(010) = 0x0a
# ---- E = 0
# ---- L = 0
# ---- Type/Average Enum = 1 (Exponential Moving Avg)
# ---- Metrics Enum = 2 (Link Margin)
pkts.filter_wpan_src64(SED_1) \
.filter_wpan_dst64(LEADER) \
.filter_mle_cmd(consts.MLE_LINK_METRICS_MANAGEMENT_REQUEST) \
.filter(lambda p: consts.LM_ENHANCED_ACK_CONFIGURATION_SUB_TLV in p.mle.tlv.link_sub_tlv) \
.filter(lambda p: p.mle.tlv.link_enh_ack_flags == consts.LINK_METRICS_ENH_ACK_PROBING_REGISTER) \
.filter(lambda p: p.mle.tlv.link_requested_type_id_flags == '0a') \
.must_next()
# Step 5 - The DUT MUST send a Link Metrics Management Response to SED_1 containing the following TLVs:
# - MLE LInk Metrics Management TLV
# -- Link Metrics Status Sub-TLV = 0 (Success)
pkts.filter_wpan_src64(LEADER) \
.filter_wpan_dst64(SED_1) \
.filter_mle_cmd(consts.MLE_LINK_METRICS_MANAGEMENT_RESPONSE) \
.filter(lambda p: p.mle.tlv.link_status_sub_tlv == consts.LINK_METRICS_STATUS_SUCCESS) \
.must_next()
# Step 6 - SSED_1 enables IEEE 802.15.4-2015 Enhanced ACK based Probing by sending a Link Metrics Management
# Request to the DUT
# MLE Link Metrics Management TLV Payload:
# - Enhanced ACK Configuration Sub-TLV
# -- Enh-ACK Flags = 1 (register a configuration)
# -- Concatenation of Link Metric Type ID Flags = 0x00:
# --- Item1: (0)(0)(001)(010) = 0x0a
# --- Item2: (0)(0)(001)(011) = 0x0b
# ---- E = 0
# ---- L = 0
# ---- Type/Average Enum = 1 (Exponential Moving Avg)
# ---- Metrics Enum = 2 (Link Margin)
# ---- Metrics Enum = 3 (RSSI)
pkts.filter_wpan_src64(SSED_1) \
.filter_mle_cmd(consts.MLE_LINK_METRICS_MANAGEMENT_REQUEST) \
.filter(lambda p: consts.LM_ENHANCED_ACK_CONFIGURATION_SUB_TLV in p.mle.tlv.link_sub_tlv) \
.filter(lambda p: p.mle.tlv.link_enh_ack_flags == consts.LINK_METRICS_ENH_ACK_PROBING_REGISTER) \
.filter(lambda p: p.mle.tlv.link_requested_type_id_flags == '0a0b') \
.must_next()
# Step 7 - The DUT MUST send a Link Metrics Management Response to SSED_1
pkts.filter_wpan_src64(LEADER) \
.filter_mle_cmd(consts.MLE_LINK_METRICS_MANAGEMENT_RESPONSE) \
.filter(lambda p: p.mle.tlv.link_status_sub_tlv == consts.LINK_METRICS_STATUS_SUCCESS) \
.must_next()
# Step 8 - SSED_1 sends an MLE Data Message with setting Frame Version subfield within the MAC header of the
# message to 0b10
pkt = pkts.filter_wpan_src64(SSED_1) \
.filter_wpan_data() \
.filter_wpan_version(consts.MAC_FRAME_VERSION_2015) \
.must_next()
ack_seq_no = pkt.wpan.seq_no
# Step 9 - The DUT MUST reply to SSED_1 with an Enhanced ACK containing the following:
# - Frame Control Field
# -- Security Enabled = True
# - Header IE
# -- Element ID = 0x00
# --- Vendor CID = 0xEAB89B (Thread Group)
# --- Vendor Specific Information
# ---- 1st byte = 0 (Enhanced ACK Link Metrics)
# ---- 2nd byte ... Link Margin data
# ---- 3rd byte ... RSSI data
pkts.filter_wpan_ack() \
.filter_wpan_seq(ack_seq_no) \
.filter(lambda p: p.wpan.payload_ie.vendor.oui == consts.THREAD_IEEE_802154_COMPANY_ID) \
.must_next()
# Step 10 - SED_1 sends an MLE Data Message with setting Frame Version subfield within the MAC header of the
# message to 0b10
pkt = pkts.filter_wpan_src64(SED_1) \
.filter_wpan_data() \
.filter_wpan_version(consts.MAC_FRAME_VERSION_2015) \
.must_next()
ack_seq_no = pkt.wpan.seq_no
# Step 11 - The DUT MUST reply to SED_1 with an Enhanced ACK containing the following:
# - Frame Control Field
# -- Security Enabled = True
# - Header IE
# -- Element ID = 0x00
# --- Vendor CID = 0xEAB89B (Thread Group)
# --- Vendor Specific Information
# ---- 1st byte = 0 (Enhanced ACK Link Metrics)
# ---- 2nd byte ... Link Margin data
pkts.filter_wpan_ack() \
.filter_wpan_seq(ack_seq_no) \
.filter(lambda p: p.wpan.payload_ie.vendor.oui == consts.THREAD_IEEE_802154_COMPANY_ID) \
.must_next()
# Step 12 - SSED_1 clears enhanced ACK link metrics configuration by instructing it to send a Link Metrics
# Management Request to the DUT
# - MLE Link Metrics Management TLV Payload
# -- Enhanced ACK Configuration Sub-TLV
# -- Enh-ACK Flags = 0 (clear enhanced ACK link metric config)
pkts.filter_wpan_src64(SSED_1) \
.filter_wpan_dst64(LEADER) \
.filter_mle_cmd(consts.MLE_LINK_METRICS_MANAGEMENT_REQUEST) \
.filter(lambda p: consts.LM_ENHANCED_ACK_CONFIGURATION_SUB_TLV in p.mle.tlv.link_sub_tlv) \
.filter(lambda p: p.mle.tlv.link_enh_ack_flags == consts.LINK_METRICS_ENH_ACK_PROBING_CLEAR) \
.must_next()
# Step 13 - The DUT MUST send Link Metrics Management Response to SSED_1 containing the following:
# - MLE Link Metrics Management TLV
# -- Link Metrics Status Sub-TLV = 0 (Success)
pkts.filter_wpan_src64(LEADER) \
.filter_wpan_dst64(SSED_1) \
.filter_mle_cmd(consts.MLE_LINK_METRICS_MANAGEMENT_RESPONSE) \
.filter(lambda p: p.mle.tlv.link_status_sub_tlv == consts.LINK_METRICS_STATUS_SUCCESS) \
.must_next()
# Step 14 - SSED_1 sends an MLE Data Message with setting Frame Version subfield within the MAC header of the
# message to 0b10
pkt = pkts.filter_wpan_src64(SSED_1) \
.filter_wpan_data() \
.filter_wpan_version(consts.MAC_FRAME_VERSION_2015) \
.must_next()
ack_seq_no = pkt.wpan.seq_no
# Step 15 - The DUT MUST NOT include a Link Metrics Report in the ACK
pkts.filter_wpan_ack() \
.filter_wpan_seq(ack_seq_no) \
.filter_wpan_ie_not_present() \
.must_next()
# Step 16 - This step verifies that Enhanced ACKS cannot be enabled while requesting 3 metric types by
# instructing the device to send the following Link Metrics Management Request to the DUT:
# MLE Link Metrics Management TLV Payload:
# - Enhanced ACK Configuration Sub-TLV
# -- Enh-ACK Flags = 1 (register a configuration)
# -- Concatenation of Link Metric Type ID Flags = 0x00:
# --- Item1: (0)(0)(001)(001) = 0x09
# --- Item2: (0)(0)(001)(010) = 0x0a
# --- Item3: (0)(0)(001)(011) = 0x0b
# ---- E = 0
# ---- L = 0
# ---- Type/Average Enum = 1 (Exponential Moving Avg)
# ---- Metrics Enum = 1 (Layer 2 LQI)
# ---- Metrics Enum = 2 (Link Margin)
# ---- Metrics Enum = 3 (RSSI)
pkts.filter_wpan_src64(SSED_1) \
.filter_wpan_dst64(LEADER) \
.filter_mle_cmd(consts.MLE_LINK_METRICS_MANAGEMENT_REQUEST) \
.filter(lambda p: consts.LM_ENHANCED_ACK_CONFIGURATION_SUB_TLV in p.mle.tlv.link_sub_tlv) \
.filter(lambda p: p.mle.tlv.link_enh_ack_flags == consts.LINK_METRICS_ENH_ACK_PROBING_REGISTER) \
.filter(lambda p: p.mle.tlv.link_requested_type_id_flags == '090a0b') \
.must_next()
# Step 17 - Leader automatically responds to the invalid query from SSED_1 with a failure
# The DUT MUST send Link Metrics Management Response to SSED_1containing the following:
# - MLE Link Metrics Management TLV
# -- Link Metrics Status Sub-TLV = 254 (Failure)
pkts.filter_wpan_src64(LEADER) \
.filter_wpan_dst64(SSED_1) \
.filter_mle_cmd(consts.MLE_LINK_METRICS_MANAGEMENT_RESPONSE) \
.filter(lambda p: p.mle.tlv.link_status_sub_tlv == consts.LINK_METRICS_STATUS_OTHER_ERROR) \
.must_next()
# Step 18 - This step verifies that Enhanced ACKs cannot be enabled while requesting a reserved Type/Average
# enum of value 2 by instructing the device to send the following Link Metrics Management Request to the DUT
# MLE Link Metrics Management TLV Payload:
# - Enhanced ACK Configuration Sub-TLV
# -- Enh-ACK Flags = 1 (register a configuration)
# -- Concatenation of Link Metric Type ID Flags = 0x00:
# --- Item1: (0)(0)(010)(010) = 0x12
# ---- E = 0
# ---- L = 0
# ---- Type/Average Enum = 2 (Reserved)
# ---- Metrics Enum = 2 (Link Margin)
pkts.filter_wpan_src64(SSED_1) \
.filter_wpan_dst64(LEADER) \
.filter_mle_cmd(consts.MLE_LINK_METRICS_MANAGEMENT_REQUEST) \
.filter(lambda p: consts.LM_ENHANCED_ACK_CONFIGURATION_SUB_TLV in p.mle.tlv.link_sub_tlv) \
.filter(lambda p: p.mle.tlv.link_enh_ack_flags == consts.LINK_METRICS_ENH_ACK_PROBING_REGISTER) \
.filter(lambda p: p.mle.tlv.link_requested_type_id_flags == '12') \
.must_next()
# Step 19 - Leader automatically responds to the invalid query from SSED_1 with a failure
# The DUT MUST send Link Metrics Management Response to SSED_1containing the following:
# - MLE Link Metrics Management TLV
# -- Link Metrics Status Sub-TLV = 254 (Failure)
pkts.filter_wpan_src64(LEADER) \
.filter_wpan_dst64(SSED_1) \
.filter_mle_cmd(consts.MLE_LINK_METRICS_MANAGEMENT_RESPONSE) \
.filter(lambda p: p.mle.tlv.link_status_sub_tlv == consts.LINK_METRICS_STATUS_OTHER_ERROR) \
.must_next()
if __name__ == '__main__':
unittest.main()
tests/scripts/thread-cert/v1_2_LowPower_7_1_02_SingleProbeLinkMetricsWithoutEnhancedAck.py
-388
View File
@@ -1,388 +0,0 @@
#!/usr/bin/env python3
#
# Copyright (c) 2023, 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 unittest
from config import ADDRESS_TYPE
from mle import LinkMetricsSubTlvType
from pktverify import consts
from pktverify.null_field import nullField
from pktverify.packet_verifier import PacketVerifier
import config
import thread_cert
LEADER = 1
SED_1 = 2
SSED_1 = 3
SERIES_ID = 1
SERIES_ID_2 = 2
POLL_PERIOD = 2000 # 2s
class LowPower_7_1_02_SingleProbeLinkMetricsWithoutEnhancedAck(thread_cert.TestCase):
USE_MESSAGE_FACTORY = False
TOPOLOGY = {
LEADER: {
'version': '1.2',
'name': 'LEADER',
'mode': 'rdn',
'allowlist': [SED_1, SSED_1],
},
SED_1: {
'version': '1.2',
'name': 'SED_1',
'mode': '-',
'allowlist': [LEADER],
},
SSED_1: {
'version': '1.2',
'name': 'SSED_1',
'mode': '-',
'allowlist': [LEADER],
}
}
"""All nodes are created with default configurations"""
def test(self):
self.nodes[LEADER].start()
self.simulator.go(config.LEADER_STARTUP_DELAY)
self.assertEqual(self.nodes[LEADER].get_state(), 'leader')
self.nodes[SED_1].set_pollperiod(POLL_PERIOD)
self.nodes[SED_1].start()
self.simulator.go(5)
self.assertEqual(self.nodes[SED_1].get_state(), 'child')
self.nodes[SSED_1].set_csl_period(consts.CSL_DEFAULT_PERIOD)
self.nodes[SSED_1].start()
self.simulator.go(5)
self.assertEqual(self.nodes[SSED_1].get_state(), 'child')
leader_addr = self.nodes[LEADER].get_ip6_address(ADDRESS_TYPE.LINK_LOCAL)
sed_extaddr = self.nodes[SED_1].get_addr64()
# Step 3 - Verify connectivity by instructing each device to send an ICMPv6 Echo Request to the DUT
self.assertTrue(self.nodes[SED_1].ping(leader_addr, timeout=POLL_PERIOD / 1000))
self.assertTrue(self.nodes[SSED_1].ping(leader_addr, timeout=(2 * consts.CSL_DEFAULT_PERIOD_IN_SECOND)))
self.simulator.go(5)
# Step 4 - SED_1 sends a Single Probe Link Metric for RSSI using MLE Data Request
# MLE Data Request Payload:
# - TLV Request TLV (Link Metrics Report TLV specified)
# - Link Metrics Query TLV
# -- Link Metrics Query ID Sub-TLV
# --- Query ID = 0 (Single Probe Query)
# -- Link Metrics Query Options Sub-TLV
# --- Metric Type ID Flags
# ---- Type / Average Enum = 1 (Exponential Moving Avg)
# ---- Metric Enum = 3 (RSSI)
#
# In this step, SED_1 should set its TxPower to 'High'. In simulation, this will be implemented by
# setting Macfilter on the Rx side (Leader).
self.nodes[LEADER].add_allowlist(sed_extaddr, -30)
res = self.nodes[SED_1].link_metrics_request_single_probe(leader_addr, 'r')
rss_1 = int(res['RSSI'])
self.simulator.go(5)
# Step 6 - SED_1 sends a Single Probe Link Metric for RSSI using MLE Data Request
# MLE Data Request Payload:
# - TLV Request TLV (Link Metrics Report TLV specified)
# - Link Metrics Query TLV
# -- Link Metrics Query ID Sub-TLV
# --- Query ID = 0 (Single Probe Query)
# -- Link Metrics Query Options Sub-TLV
# --- Metric Type ID Flags
# ---- Type / Average Enum = 1 (Exponential Moving Avg)
# ---- Metric Enum = 3 (RSSI)
#
# In this step, SED_1 should set its TxPower to 'Low'.
self.nodes[LEADER].add_allowlist(sed_extaddr, -95)
res = self.nodes[SED_1].link_metrics_request_single_probe(leader_addr, 'r')
rss_2 = int(res['RSSI'])
self.simulator.go(5)
# Step 8 - Compare the rssi value in step 5 & 7, RSSI in 5 should be larger than RSSI in 7
self.assertTrue(rss_1 > rss_2)
# Step 9 - SSED_1 sends a Single Probe Link Metric for Layer 2 LQI using MLE Data Request
# MLE Data Request Payload:
# - TLV Request TLV (Link Metrics Report TLV specified)
# - Link Metrics Query TLV
# -- Link Metrics Query ID Sub-TLV
# --- Query ID = 0 (Single Probe Query)
# -- Link Metrics Query Options Sub-TLV
# --- Metric Type ID Flags
# ---- Type / Average Enum = 1 (Exponential Moving Avg)
# ---- Metric Enum = 1 (Layer 2 LQI)
self.nodes[SSED_1].link_metrics_request_single_probe(leader_addr, 'q')
self.simulator.go(5)
# Step 11 - SSED_1 sends a Single Probe Link Metric for Link Margin using MLE Data Request
# MLE Data Request Payload:
# - TLV Request TLV (Link Metrics Report TLV specified)
# - Link Metrics Query TLV
# -- Link Metrics Query ID Sub-TLV
# --- Query ID = 0 (Single Probe Query)
# -- Link Metrics Query Options Sub-TLV
# --- Metric Type ID Flags
# ---- Type / Average Enum = 1 (Exponential Moving Avg)
# ---- Metric Enum = 2 (Link Margin)
self.nodes[SSED_1].link_metrics_request_single_probe(leader_addr, 'm')
self.simulator.go(5)
# Step 13 - SSED_1 sends a Single Probe Link Metric using MLE Data Request
# MLE Data Request Payload:
# - TLV Request TLV (Link Metrics Report TLV specified)
# - Link Metrics Query TLV
# -- Link Metrics Query ID Sub-TLV
# --- Query ID = 0 (Single Probe Query)
# -- Link Metrics Query Options Sub-TLV
# --- Metric Type ID Flags
# ---- Type / Average Enum = 1 (Exponential Moving Avg)
# ---- Metric Enum = 1 (Layer 2 LQI)
# ---- Metric Enum = 2 (Link Margin)
# ---- Metric Enum = 3 (RSSI)
self.nodes[SSED_1].link_metrics_request_single_probe(leader_addr, 'qmr')
self.simulator.go(5)
def verify(self, pv):
pkts = pv.pkts
pv.summary.show()
LEADER = pv.vars['LEADER']
SED_1 = pv.vars['SED_1']
SSED_1 = pv.vars['SSED_1']
# Step 3 - The DUT MUST send ICMPv6 Echo Responses to both SED1 & SSED1
pkts.filter_wpan_src64(LEADER) \
.filter_wpan_dst64(SED_1) \
.filter_ping_reply() \
.must_next()
pkts.filter_wpan_src64(LEADER) \
.filter_wpan_dst64(SSED_1) \
.filter_ping_reply() \
.must_next()
# Step 4 - SED_1 sends a Single Probe Link Metric for RSSI using MLE Data Request
# MLE Data Request Payload:
# - TLV Request TLV (Link Metrics Report TLV specified)
# - Link Metrics Query TLV
# -- Link Metrics Query ID Sub-TLV
# --- Query ID = 0 (Single Probe Query)
# -- Link Metrics Query Options Sub-TLV
# --- Metric Type ID Flags
# ---- Type / Average Enum = 1 (Exponential Moving Avg)
# ---- Metric Enum = 3 (RSSI)
# TODO: Currently the ot-pktverify version of wireshark cannot parse Link Metrics Query Options Sub-TLV correctly. Will add check for the fields after fixing this.
pkts.filter_wpan_src64(SED_1) \
.filter_wpan_dst64(LEADER) \
.filter_mle_cmd(consts.MLE_DATA_REQUEST) \
.filter_mle_has_tlv(consts.TLV_REQUEST_TLV) \
.filter_mle_has_tlv(consts.LINK_METRICS_QUERY_TLV) \
.filter(lambda p: LinkMetricsSubTlvType.LINK_METRICS_QUERY_ID in p.mle.tlv.link_sub_tlv) \
.filter(lambda p: p.mle.tlv.query_id == 0) \
.filter(lambda p: LinkMetricsSubTlvType.LINK_METRICS_QUERY_OPTIONS in p.mle.tlv.link_sub_tlv) \
.must_next()
# Step 5 The DUT MUST reply to SED_1 with MLE Data Response with the following:
# - Link Metrics Report TLV
# -- Link Metrics Report Sub-TLV
# --- Metric Type ID Flags
# ---- Type / Average Enum = 1 (Exponential Moving Avg)
# ---- Metric Enum = 3 (RSSI)
# --- RSSI Value (1-byte)
pkts.filter_wpan_src64(LEADER) \
.filter_wpan_dst64(SED_1) \
.filter_mle_cmd(consts.MLE_DATA_RESPONSE) \
.filter_mle_has_tlv(consts.LINK_METRICS_REPORT_TLV) \
.filter(lambda p: LinkMetricsSubTlvType.LINK_METRICS_REPORT in p.mle.tlv.link_sub_tlv) \
.filter(lambda p: consts.LINK_METRICS_TYPE_AVERAGE_ENUM_EXPONENTIAL in p.mle.tlv.metric_type_id_flags.type) \
.filter(lambda p: consts.LINK_METRICS_METRIC_TYPE_ENUM_RSSI in p.mle.tlv.metric_type_id_flags.metric) \
.must_next()
# Step 6 - SED_1 sends a Single Probe Link Metric for RSSI using MLE Data Request
# MLE Data Request Payload:
# - TLV Request TLV (Link Metrics Report TLV specified)
# - Link Metrics Query TLV
# -- Link Metrics Query ID Sub-TLV
# --- Query ID = 0 (Single Probe Query)
# -- Link Metrics Query Options Sub-TLV
# --- Metric Type ID Flags
# ---- Type / Average Enum = 1 (Exponential Moving Avg)
# ---- Metric Enum = 3 (RSSI)
pkts.filter_wpan_src64(SED_1) \
.filter_wpan_dst64(LEADER) \
.filter_mle_cmd(consts.MLE_DATA_REQUEST) \
.filter_mle_has_tlv(consts.TLV_REQUEST_TLV) \
.filter_mle_has_tlv(consts.LINK_METRICS_QUERY_TLV) \
.filter(lambda p: LinkMetricsSubTlvType.LINK_METRICS_QUERY_ID in p.mle.tlv.link_sub_tlv) \
.filter(lambda p: p.mle.tlv.query_id == 0) \
.filter(lambda p: LinkMetricsSubTlvType.LINK_METRICS_QUERY_OPTIONS in p.mle.tlv.link_sub_tlv) \
.must_next()
# Step 7 The DUT MUST reply to SED_1 with MLE Data Response with the following:
# - Link Metrics Report TLV
# -- Link Metrics Report Sub-TLV
# --- Metric Type ID Flags
# ---- Type / Average Enum = 1 (Exponential Moving Avg)
# ---- Metric Enum = 3 (RSSI)
# --- RSSI Value (1-byte)
pkts.filter_wpan_src64(LEADER) \
.filter_wpan_dst64(SED_1) \
.filter_mle_cmd(consts.MLE_DATA_RESPONSE) \
.filter_mle_has_tlv(consts.LINK_METRICS_REPORT_TLV) \
.filter(lambda p: LinkMetricsSubTlvType.LINK_METRICS_REPORT in p.mle.tlv.link_sub_tlv) \
.filter(lambda p: consts.LINK_METRICS_TYPE_AVERAGE_ENUM_EXPONENTIAL in p.mle.tlv.metric_type_id_flags.type) \
.filter(lambda p: consts.LINK_METRICS_METRIC_TYPE_ENUM_RSSI in p.mle.tlv.metric_type_id_flags.metric) \
.must_next()
# Step 9 - SSED_1 sends a Single Probe Link Metric for Layer 2 LQI using MLE Data Request
# MLE Data Request Payload:
# - TLV Request TLV (Link Metrics Report TLV specified)
# - Link Metrics Query TLV
# -- Link Metrics Query ID Sub-TLV
# --- Query ID = 0 (Single Probe Query)
# -- Link Metrics Query Options Sub-TLV
# --- Metric Type ID Flags
# ---- Type / Average Enum = 1 (Exponential Moving Avg)
# ---- Metric Enum = 1 (Layer 2 LQI)
pkts.filter_wpan_src64(SSED_1) \
.filter_wpan_dst64(LEADER) \
.filter_mle_cmd(consts.MLE_DATA_REQUEST) \
.filter_mle_has_tlv(consts.TLV_REQUEST_TLV) \
.filter_mle_has_tlv(consts.LINK_METRICS_QUERY_TLV) \
.filter(lambda p: LinkMetricsSubTlvType.LINK_METRICS_QUERY_ID in p.mle.tlv.link_sub_tlv) \
.filter(lambda p: p.mle.tlv.query_id == 0) \
.filter(lambda p: LinkMetricsSubTlvType.LINK_METRICS_QUERY_OPTIONS in p.mle.tlv.link_sub_tlv) \
.must_next()
# Step 10 The DUT MUST reply to SSED_1 with MLE Data Response with the following:
# - Link Metrics Report TLV
# -- Link Metrics Report Sub-TLV
# --- Metric Type ID Flags
# ---- Type / Average Enum = 1 (Exponential Moving Avg)
# ---- Metric Enum = 1 (Layer 2 LQI)
# --- Layer 2 LQI value (1-byte)
pkts.filter_wpan_src64(LEADER) \
.filter_wpan_dst64(SSED_1) \
.filter_mle_cmd(consts.MLE_DATA_RESPONSE) \
.filter_mle_has_tlv(consts.LINK_METRICS_REPORT_TLV) \
.filter(lambda p: LinkMetricsSubTlvType.LINK_METRICS_REPORT in p.mle.tlv.link_sub_tlv) \
.filter(lambda p: consts.LINK_METRICS_TYPE_AVERAGE_ENUM_EXPONENTIAL in p.mle.tlv.metric_type_id_flags.type) \
.filter(lambda p: consts.LINK_METRICS_METRIC_TYPE_ENUM_LQI in p.mle.tlv.metric_type_id_flags.metric) \
.must_next()
# Step 11 - SSED_1 sends a Single Probe Link Metric for Link Margin using MLE Data Request
# MLE Data Request Payload:
# - TLV Request TLV (Link Metrics Report TLV specified)
# - Link Metrics Query TLV
# -- Link Metrics Query ID Sub-TLV
# --- Query ID = 0 (Single Probe Query)
# -- Link Metrics Query Options Sub-TLV
# --- Metric Type ID Flags
# ---- Type / Average Enum = 1 (Exponential Moving Avg)
# ---- Metric Enum = 2 (Link Margin)
pkts.filter_wpan_src64(SSED_1) \
.filter_wpan_dst64(LEADER) \
.filter_mle_cmd(consts.MLE_DATA_REQUEST) \
.filter_mle_has_tlv(consts.TLV_REQUEST_TLV) \
.filter_mle_has_tlv(consts.LINK_METRICS_QUERY_TLV) \
.filter(lambda p: LinkMetricsSubTlvType.LINK_METRICS_QUERY_ID in p.mle.tlv.link_sub_tlv) \
.filter(lambda p: p.mle.tlv.query_id == 0) \
.filter(lambda p: LinkMetricsSubTlvType.LINK_METRICS_QUERY_OPTIONS in p.mle.tlv.link_sub_tlv) \
.must_next()
# Step 12 The DUT MUST reply to SSED_1 with MLE Data Response with the following:
# - Link Metrics Report TLV
# -- Link Metrics Report Sub-TLV
# --- Metric Type ID Flags
# ---- Type / Average Enum = 1 (Exponential Moving Avg)
# ---- Metric Enum = 2 (Link Margin)
# --- Link Margin value (1-byte)
pkts.filter_wpan_src64(LEADER) \
.filter_wpan_dst64(SSED_1) \
.filter_mle_cmd(consts.MLE_DATA_RESPONSE) \
.filter_mle_has_tlv(consts.LINK_METRICS_REPORT_TLV) \
.filter(lambda p: LinkMetricsSubTlvType.LINK_METRICS_REPORT in p.mle.tlv.link_sub_tlv) \
.filter(lambda p: consts.LINK_METRICS_TYPE_AVERAGE_ENUM_EXPONENTIAL in p.mle.tlv.metric_type_id_flags.type) \
.filter(lambda p: consts.LINK_METRICS_METRIC_TYPE_ENUM_LINK_MARGIN in p.mle.tlv.metric_type_id_flags.metric) \
.must_next()
# Step 13 - SSED_1 sends a Single Probe Link Metric using MLE Data Request
# MLE Data Request Payload:
# - TLV Request TLV (Link Metrics Report TLV specified)
# - Link Metrics Query TLV
# -- Link Metrics Query ID Sub-TLV
# --- Query ID = 0 (Single Probe Query)
# -- Link Metrics Query Options Sub-TLV
# --- Metric Type ID Flags
# ---- Type / Average Enum = 1 (Exponential Moving Avg)
# ---- Metric Enum = 1 (Layer 2 LQI)
# ---- Type / Average Enum = 1 (Exponential Moving Avg)
# ---- Metric Enum = 2 (Link Margin)
# ---- Type / Average Enum = 1 (Exponential Moving Avg)
# ---- Metric Enum = 3 (RSSI)
pkts.filter_wpan_src64(SSED_1) \
.filter_wpan_dst64(LEADER) \
.filter_mle_cmd(consts.MLE_DATA_REQUEST) \
.filter_mle_has_tlv(consts.TLV_REQUEST_TLV) \
.filter_mle_has_tlv(consts.LINK_METRICS_QUERY_TLV) \
.filter(lambda p: LinkMetricsSubTlvType.LINK_METRICS_QUERY_ID in p.mle.tlv.link_sub_tlv) \
.filter(lambda p: p.mle.tlv.query_id == 0) \
.filter(lambda p: LinkMetricsSubTlvType.LINK_METRICS_QUERY_OPTIONS in p.mle.tlv.link_sub_tlv) \
.must_next()
# Step 14 The DUT MUST reply to SSED_1 with MLE Data Response with the following:
# - Link Metrics Report TLV
# -- Link Metrics Report Sub-TLV
# --- Metric Type ID Flags
# ---- Type / Average Enum = 1 (Exponential Moving Avg)
# ---- Metric Enum = 1 (Layer 2 LQI)
# ---- Layer 2 LQI value (1-byte)
# ---- Type / Average Enum = 1 (Exponential Moving Avg)
# ---- Metric Enum = 2 (Link Margin)
# ---- Link Margin value (1-byte)
# ---- Type / Average Enum = 1 (Exponential Moving Avg)
# ---- Metric Enum = 3 (RSSI)
# ---- RSSI value (1-byte)
pkts.filter_wpan_src64(LEADER) \
.filter_wpan_dst64(SSED_1) \
.filter_mle_cmd(consts.MLE_DATA_RESPONSE) \
.filter_mle_has_tlv(consts.LINK_METRICS_REPORT_TLV) \
.filter(lambda p: LinkMetricsSubTlvType.LINK_METRICS_REPORT in p.mle.tlv.link_sub_tlv) \
.filter(lambda p: consts.LINK_METRICS_TYPE_AVERAGE_ENUM_EXPONENTIAL in p.mle.tlv.metric_type_id_flags.type) \
.filter(lambda p: consts.LINK_METRICS_METRIC_TYPE_ENUM_LQI in p.mle.tlv.metric_type_id_flags.metric) \
.filter(lambda p: consts.LINK_METRICS_METRIC_TYPE_ENUM_LINK_MARGIN in p.mle.tlv.metric_type_id_flags.metric) \
.filter(lambda p: consts.LINK_METRICS_METRIC_TYPE_ENUM_RSSI in p.mle.tlv.metric_type_id_flags.metric) \
.must_next()
if __name__ == '__main__':
unittest.main()
tests/scripts/thread-cert/v1_2_LowPower_7_2_01_ForwardTrackingSeries.py
-339
View File
@@ -1,339 +0,0 @@
#!/usr/bin/env python3
#
# Copyright (c) 2020, 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 unittest
from config import ADDRESS_TYPE
from mle import LinkMetricsSubTlvType, TlvType
from pktverify import consts
from pktverify.null_field import nullField
from pktverify.packet_verifier import PacketVerifier
import config
import thread_cert
LEADER = 1
SED_1 = 2
SSED_1 = 3
SERIES_ID = 1
SERIES_ID_2 = 2
POLL_PERIOD = 2000 # 2s
class LowPower_7_2_01_ForwardTrackingSeries(thread_cert.TestCase):
USE_MESSAGE_FACTORY = False
TOPOLOGY = {
LEADER: {
'version': '1.2',
'name': 'LEADER',
'mode': 'rdn',
'allowlist': [SED_1, SSED_1],
},
SED_1: {
'version': '1.2',
'name': 'SED_1',
'mode': '-',
'allowlist': [LEADER],
},
SSED_1: {
'version': '1.2',
'name': 'SSED_1',
'mode': '-',
'allowlist': [LEADER],
}
}
"""All nodes are created with default configurations"""
def test(self):
self.nodes[LEADER].start()
self.simulator.go(config.LEADER_STARTUP_DELAY)
self.assertEqual(self.nodes[LEADER].get_state(), 'leader')
self.nodes[SED_1].set_pollperiod(POLL_PERIOD)
self.nodes[SED_1].start()
self.simulator.go(5)
self.assertEqual(self.nodes[SED_1].get_state(), 'child')
self.nodes[SSED_1].set_csl_period(consts.CSL_DEFAULT_PERIOD)
self.nodes[SSED_1].start()
self.simulator.go(5)
self.assertEqual(self.nodes[SSED_1].get_state(), 'child')
leader_addr = self.nodes[LEADER].get_ip6_address(ADDRESS_TYPE.LINK_LOCAL)
# Step 3 - Verify connectivity by instructing each device to send an ICMPv6 Echo Request to the DUT
self.assertTrue(self.nodes[SED_1].ping(leader_addr, timeout=POLL_PERIOD / 1000))
self.assertTrue(self.nodes[SSED_1].ping(leader_addr))
self.simulator.go(5)
# Step 4 - SED_1 requests a Forward Tracking Series by sending a Link Metrics Management Request
# Forward Series Flags = 0x04:
# - Bit 2 = 1, MAC Data Request
# - Bits 0,1 & 3-7 = 0
# Concatenation of Link Metric Type ID Flags = 0x00:
# - Item1: (0)(1)(000)(000) = 0x40
# -- E = 0
# -- L = 1
# -- Type/Average Enum = 0 (count)
# -- Metrics Enum = 0 (count)
self.nodes[SED_1].link_metrics_config_req_forward_tracking_series(leader_addr, SERIES_ID, 'r', 'p', 'async')
self.simulator.go(5)
# Step 6 - Configures SED_1 to send MAC Data Request every 2 seconds and wait for 30 seconds
self.simulator.go(30)
# Step 7 - SED_1 sends an MLE Data Request to retrieve aggregated Forward Series Results
result = self.nodes[SED_1].link_metrics_request_forward_tracking_series(leader_addr, SERIES_ID)
self.assertIn("PDU Counter", result)
#self.simulator.go(5)
# Step 9 - SED_1 clears the Forward Tracking Series
# Forward Series Flags = 0x00:
# - Bits 0-7 = 0
# Concatenation of Link Metric Type ID Flags = NULL:
self.nodes[SED_1].link_metrics_config_req_forward_tracking_series(leader_addr, SERIES_ID, 'X', '', 'async')
self.simulator.go(5)
# Step 16 - SSED_1 requests a Forward Tracking Series by sending a Link Metrics Management Request
# Forward Series Flags = 0x02:
# - Bit 1 = 1, (all Link Layer Data Frames)
# - Bits 0, 2-7 = 0
# Concatenation of Link Metric Type ID Flags = 0x0a:
# - Item1: (0)(0)(001)(010) = 0x0a
# -- E = 0
# -- L = 0
# -- Type/Average Enum = 1 (Exponential Moving Average)
# -- Metrics Enum = 2 (Link Margin)
self.nodes[SSED_1].link_metrics_config_req_forward_tracking_series(leader_addr, SERIES_ID_2, 'd', 'm', 'async')
self.simulator.go(5)
# Step 18 - SSED_1 sends MAC Data packet every 1 second for 30 seconds
# Note: As the api to send MAC Data frame hasn't been implemented, we send MLE Link Probe message here
for i in range(30):
self.nodes[SSED_1].link_metrics_send_link_probe(leader_addr, SERIES_ID_2, 1)
self.simulator.go(1)
# Step 19 - SSED_1 sends an MLE Data Request to retrieve aggregated Forward Series Results
result = self.nodes[SSED_1].link_metrics_request_forward_tracking_series(leader_addr, SERIES_ID_2)
self.assertIn("Margin", result)
# Step 21 - SSED_1 clears the Forward Series Link Metrics
# Forward Series Flags = 0x00:
# - Bits 0-7 = 0
# Concatenation of Link Metric Type ID Flags = NULL
self.nodes[SSED_1].link_metrics_config_req_forward_tracking_series(leader_addr, SERIES_ID_2, 'X', '', 'async')
self.simulator.go(5)
# Step 23 - SSED_1 sends an MLE Data Request to retrieve aggregated Forward Series Results
result = self.nodes[SSED_1].link_metrics_request_forward_tracking_series(leader_addr, SERIES_ID_2)
self.assertIn('Status', result)
self.assertEqual(result['Status'], 'Series ID not recognized')
def verify(self, pv):
pkts = pv.pkts
pv.summary.show()
LEADER = pv.vars['LEADER']
SED_1 = pv.vars['SED_1']
SSED_1 = pv.vars['SSED_1']
# Step 3 - The DUT MUST send ICMPv6 Echo Responses to both SED1 & SSED1
pkts.filter_wpan_src64(LEADER) \
.filter_wpan_dst64(SED_1) \
.filter_ping_reply() \
.must_next()
pkts.filter_wpan_src64(LEADER) \
.filter_wpan_dst64(SSED_1) \
.filter_ping_reply() \
.must_next()
# Step 4 - SED_1 requests a Forward Series Link Metrics by sending a Link Metrics Management Request
# Forward Series Flags = 0x04:
# - Bit 2 = 1, MAC Data Request
# - Bits 0,1 & 3-7 = 0
# Concatenation of Link Metric Type ID Flags = 0x40:
# - Item1: (0)(1)(000)(000) = 0x40
# -- E = 0
# -- L = 1
# -- Type/Average Enum = 0 (count)
# -- Metrics Enum = 0 (count)
pkts.filter_wpan_src64(SED_1) \
.filter_wpan_dst64(LEADER) \
.filter_mle_cmd(consts.MLE_LINK_METRICS_MANAGEMENT_REQUEST) \
.filter(lambda p: LinkMetricsSubTlvType.FORWARD_PROBING_REGISTRATION in p.mle.tlv.link_sub_tlv) \
.filter(lambda p: 4 in p.mle.tlv.link_forward_series) \
.filter(lambda p: consts.LINK_METRICS_TYPE_AVERAGE_ENUM_COUNT in p.mle.tlv.metric_type_id_flags.type) \
.filter(lambda p: consts.LINK_METRICS_METRIC_TYPE_ENUM_PDU_COUNT in p.mle.tlv.metric_type_id_flags.metric) \
.filter(lambda p: 1 in p.mle.tlv.metric_type_id_flags.l) \
.must_next()
# Step 5 - Leader responds with a Link Metrics Management Response
pkts.filter_wpan_src64(LEADER) \
.filter_wpan_dst64(SED_1) \
.filter_mle_cmd(consts.MLE_LINK_METRICS_MANAGEMENT_RESPONSE) \
.filter(lambda p: p.mle.tlv.link_status_sub_tlv == consts.LINK_METRICS_STATUS_SUCCESS) \
.must_next()
# Step 8 - Leader responds with an MLE Data Response
pkts.filter_wpan_src64(LEADER) \
.filter_wpan_dst64(SED_1) \
.filter_mle_cmd(consts.MLE_DATA_RESPONSE) \
.filter(lambda p: TlvType.LINK_METRICS_REPORT in p.mle.tlv.type) \
.filter(lambda p: consts.LINK_METRICS_TYPE_AVERAGE_ENUM_COUNT in p.mle.tlv.metric_type_id_flags.type) \
.filter(lambda p: consts.LINK_METRICS_METRIC_TYPE_ENUM_PDU_COUNT in p.mle.tlv.metric_type_id_flags.metric) \
.must_next()
# Step 9 - SED_1 clears the Forward Series Link Metrics
# Forward Series Flags = 0x00:
# - Bits 0-7 = 0
# Concatenation of Link Metric Type ID Flags = NULL:
pkts.filter_wpan_src64(SED_1) \
.filter_wpan_dst64(LEADER) \
.filter_mle_cmd(consts.MLE_LINK_METRICS_MANAGEMENT_REQUEST) \
.filter(lambda p: LinkMetricsSubTlvType.FORWARD_PROBING_REGISTRATION in p.mle.tlv.link_sub_tlv) \
.filter(lambda p: 0 in p.mle.tlv.link_forward_series) \
.filter(lambda p: p.mle.tlv.metric_type_id_flags.type is nullField) \
.filter(lambda p: p.mle.tlv.metric_type_id_flags.metric is nullField) \
.must_next()
# Step 10 - Leader responds with a Link Metrics Management Response
pkts.filter_wpan_src64(LEADER) \
.filter_wpan_dst64(SED_1) \
.filter_mle_cmd(consts.MLE_LINK_METRICS_MANAGEMENT_RESPONSE) \
.filter(lambda p: p.mle.tlv.link_status_sub_tlv == consts.LINK_METRICS_STATUS_SUCCESS) \
.must_next()
# Step 16 - SSED_1 requests a Forward Series Link Metrics by sending a Link Metrics Management Request
# Forward Series Flags = 0x02:
# - Bit 1 = 1, (all Link Layer Data Frames)
# - Bits 0, 2-7 = 0
# Concatenation of Link Metric Type ID Flags = 0x0a:
# - Item1: (0)(0)(001)(010) = 0x0a
# -- E = 0
# -- L = 0
# -- Type/Average Enum = 1 (Exponential Moving Average)
# -- Metrics Enum = 2 (Link Margin)
pkts.filter_wpan_src64(SSED_1) \
.filter_wpan_dst64(LEADER) \
.filter_mle_cmd(consts.MLE_LINK_METRICS_MANAGEMENT_REQUEST) \
.filter(lambda p: LinkMetricsSubTlvType.FORWARD_PROBING_REGISTRATION in p.mle.tlv.link_sub_tlv) \
.filter(lambda p: 2 in p.mle.tlv.link_forward_series) \
.filter(lambda p: consts.LINK_METRICS_TYPE_AVERAGE_ENUM_EXPONENTIAL in p.mle.tlv.metric_type_id_flags.type) \
.filter(lambda p: consts.LINK_METRICS_METRIC_TYPE_ENUM_LINK_MARGIN in p.mle.tlv.metric_type_id_flags.metric) \
.must_next()
# Step 17 - Leader responds with a Link Metrics Management Response
pkts.filter_wpan_src64(LEADER) \
.filter_wpan_dst64(SSED_1) \
.filter_mle_cmd(consts.MLE_LINK_METRICS_MANAGEMENT_RESPONSE) \
.filter(lambda p: p.mle.tlv.link_status_sub_tlv == consts.LINK_METRICS_STATUS_SUCCESS) \
.must_next()
# Step 19 - SSED_1 sends an MLE Data Request to retrieve aggregated Forward Series results
# - TLV Request TLV(Link Metrics Report TLV specified
# - Link Metrics Query TLV
# -- Link Metrics Query ID Sub-TLV
# --- Query ID: (ID set in step 16)
# - Verify that the Link Metrics Query Options Sub-TLV is NOT present
pkts.filter_wpan_src64(SSED_1) \
.filter_wpan_dst64(LEADER) \
.filter_mle_cmd(consts.MLE_DATA_REQUEST) \
.filter(lambda p: TlvType.LINK_METRICS_QUERY in p.mle.tlv.type) \
.filter(lambda p: p.mle.tlv.query_id == SERIES_ID_2) \
.must_next()
# Step 20 - Leader MUST reply to SSED_1 with an MLE Data Response with the following:
# - Link Metrics Report TLV
# -- Link Metrics Report Sub-TLV
# --- Metric Type ID Flags
# ---- Type / Average Enum = 1 (Exponential Moving Average)
# ---- Metric Enum = 2 (Link Margin)
# --- Value (1 bytes)
pkts.filter_wpan_src64(LEADER) \
.filter_wpan_dst64(SSED_1) \
.filter_mle_cmd(consts.MLE_DATA_RESPONSE) \
.filter_mle_has_tlv(TlvType.LINK_METRICS_REPORT) \
.filter(lambda p: LinkMetricsSubTlvType.LINK_METRICS_REPORT in p.mle.tlv.link_sub_tlv) \
.filter(lambda p: consts.LINK_METRICS_TYPE_AVERAGE_ENUM_EXPONENTIAL in p.mle.tlv.metric_type_id_flags.type) \
.filter(lambda p: consts.LINK_METRICS_METRIC_TYPE_ENUM_LINK_MARGIN in p.mle.tlv.metric_type_id_flags.metric) \
.must_next()
# Step 21 - SSED_1 clears the Forward Series Link Metrics
# Forward Series Flags = 0x00:
# - Bits 0-7 = 0
# Concatenation of Link Metric Type ID Flags = NULL
pkts.filter_wpan_src64(SSED_1) \
.filter_wpan_dst64(LEADER) \
.filter_mle_cmd(consts.MLE_LINK_METRICS_MANAGEMENT_REQUEST) \
.filter(lambda p: LinkMetricsSubTlvType.FORWARD_PROBING_REGISTRATION in p.mle.tlv.link_sub_tlv) \
.filter(lambda p: 0 in p.mle.tlv.link_forward_series) \
.filter(lambda p: p.mle.tlv.metric_type_id_flags.type is nullField) \
.filter(lambda p: p.mle.tlv.metric_type_id_flags.metric is nullField) \
.must_next()
# Step 22 - Leader responds with a Link Metrics Management Response
pkts.filter_wpan_src64(LEADER) \
.filter_wpan_dst64(SSED_1) \
.filter_mle_cmd(consts.MLE_LINK_METRICS_MANAGEMENT_RESPONSE) \
.filter(lambda p: p.mle.tlv.link_status_sub_tlv == consts.LINK_METRICS_STATUS_SUCCESS) \
.must_next()
# Step 23 - SSED_1 sends an MLE Data Request to retrieve aggregated Forward Series results
# - TLV Request TLV(Link Metrics Report TLV specified
# - Link Metrics Query TLV
# -- Link Metrics Query ID Sub-TLV
# --- Query ID: (ID set in step 16)
# This is a negative step
pkts.filter_wpan_src64(SSED_1) \
.filter_wpan_dst64(LEADER) \
.filter_mle_cmd(consts.MLE_DATA_REQUEST) \
.filter(lambda p: TlvType.LINK_METRICS_QUERY in p.mle.tlv.type) \
.filter(lambda p: p.mle.tlv.query_id == SERIES_ID_2) \
.must_next()
# Step 24 - Leader MUST reply to SSED_1 with an MLE Data Response with the following:
# - Link Metrics Report TLV
# -- Link Metrics Status Sub-TLV
# --- Status = 3 (Series ID not recognized)
pkts.filter_wpan_src64(LEADER) \
.filter_wpan_dst64(SSED_1) \
.filter_mle_cmd(consts.MLE_DATA_RESPONSE) \
.filter_mle_has_tlv(TlvType.LINK_METRICS_REPORT) \
.filter(lambda p: LinkMetricsSubTlvType.LINK_METRICS_STATUS in p.mle.tlv.link_sub_tlv) \
.filter(lambda p: p.mle.tlv.link_status_sub_tlv == consts.LINK_METRICS_STATUS_SERIES_ID_NOT_RECOGNIZED) \
.must_next()
if __name__ == '__main__':
unittest.main()
tests/scripts/thread-cert/v1_2_LowPower_test_forward_tracking_series.py
-194
View File
@@ -1,194 +0,0 @@
#!/usr/bin/env python3
#
# Copyright (c) 2020, 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 unittest
from config import ADDRESS_TYPE
from mle import LinkMetricsSubTlvType, TlvType
from pktverify import consts
from pktverify.null_field import nullField
from pktverify.packet_verifier import PacketVerifier
import config
import thread_cert
LEADER = 1
CHILD = 2
SERIES_ID_1 = 1
SERIES_ID_2 = 2
class LowPower_test_ForwardTrackingSeries(thread_cert.TestCase):
TOPOLOGY = {
LEADER: {
'version': '1.2',
'name': 'LEADER',
'mode': 'rdn',
'allowlist': [CHILD],
},
CHILD: {
'version': '1.2',
'name': 'CHILD',
'mode': 'rn',
'allowlist': [LEADER],
}
}
"""All nodes are created with default configurations"""
def test(self):
self.nodes[LEADER].start()
self.simulator.go(config.LEADER_STARTUP_DELAY)
self.assertEqual(self.nodes[LEADER].get_state(), 'leader')
self.nodes[CHILD].start()
self.simulator.go(5)
self.assertEqual(self.nodes[CHILD].get_state(), 'child')
leader_addr = self.nodes[LEADER].get_ip6_address(ADDRESS_TYPE.LINK_LOCAL)
# 1. Child configures a Forward Tracking Series successfully.
# The Series tracks the count of MAC Data Request.
# Child should get a response with status 0 (SUCCESS).
self.nodes[CHILD].link_metrics_config_req_forward_tracking_series(leader_addr, SERIES_ID_1, 'r', 'p', 'async')
self.simulator.go(1)
# 2. Child configures the same Forward Tracking Series again.
# Child should get a response with status 2 (SERIES_ID_ALREADY_REGISTERED).
self.nodes[CHILD].link_metrics_config_req_forward_tracking_series(leader_addr, SERIES_ID_1, 'r', 'p', 'async')
self.simulator.go(1)
# 3. Child queries a Series that doesn't exist (using a wrong Series ID).
# Child should get a report with status 3 (SERIES_ID_NOT_RECOGNIZED).
self.nodes[CHILD].link_metrics_request_forward_tracking_series(leader_addr, SERIES_ID_2, 'async')
self.simulator.go(1)
# 4. Child queries a Series that don't have matched frames yet.
# Child should get a report with status 4 (NO_MATCHING_FRAMES_RECEIVED).
self.nodes[CHILD].link_metrics_request_forward_tracking_series(leader_addr, SERIES_ID_1, 'async')
self.simulator.go(1)
# 5. Child clears a Forward Tracking Series that doesn't exist.
# Child should get a response with status 3 (SERIES_ID_NOT_RECOGNIZED).
self.nodes[CHILD].link_metrics_config_req_forward_tracking_series(leader_addr, SERIES_ID_2, 'X', '', 'async')
self.simulator.go(1)
# 6. Child clears a Forward Tracking Series successfully.
# Child should get a response with status 0 (SUCCESS).
self.nodes[CHILD].link_metrics_config_req_forward_tracking_series(leader_addr, SERIES_ID_1, 'X', '', 'async')
self.simulator.go(1)
# 7. Child configures a new Forward Tracking Series successfully.
# The Series tracks the count of all MAC Data frames.
# Child should get a response with status 0 (SUCCESS).
self.nodes[CHILD].link_metrics_config_req_forward_tracking_series(leader_addr, SERIES_ID_2, 'd', 'pqmr',
'async')
self.simulator.go(1)
# 8. Child sends an MLE Link Probe message to the Subject for the newly configured Series.
self.nodes[CHILD].link_metrics_send_link_probe(leader_addr, SERIES_ID_2, 1)
# 9. Child queries the newly configured Series successfully.
# Child should get a report with valid values.
self.nodes[CHILD].link_metrics_request_forward_tracking_series(leader_addr, SERIES_ID_2, 'async')
self.simulator.go(1)
def verify(self, pv):
pkts = pv.pkts
pv.summary.show()
LEADER = pv.vars['LEADER']
CHILD = pv.vars['CHILD']
# 1. Child should get a response with status 0 (SUCCESS).
pkts.filter_wpan_src64(LEADER) \
.filter_wpan_dst64(CHILD) \
.filter_mle_cmd(consts.MLE_LINK_METRICS_MANAGEMENT_RESPONSE) \
.filter(lambda p: p.mle.tlv.link_status_sub_tlv == consts.LINK_METRICS_STATUS_SUCCESS) \
.must_next()
# 2. Child should get a response with status 2 (SERIES_ID_ALREADY_REGISTERED).
pkts.filter_wpan_src64(LEADER) \
.filter_wpan_dst64(CHILD) \
.filter_mle_cmd(consts.MLE_LINK_METRICS_MANAGEMENT_RESPONSE) \
.filter(lambda p: p.mle.tlv.link_status_sub_tlv == consts.LINK_METRICS_STATUS_SERIES_ID_ALREADY_REGISTERED) \
.must_next()
# 3. Child should get a report with status 3 (SERIES_ID_NOT_RECOGNIZED).
pkts.filter_wpan_src64(LEADER) \
.filter_wpan_dst64(CHILD) \
.filter_mle_cmd(consts.MLE_DATA_RESPONSE) \
.filter_mle_has_tlv(TlvType.LINK_METRICS_REPORT) \
.filter(lambda p: p.mle.tlv.link_status_sub_tlv == consts.LINK_METRICS_STATUS_SERIES_ID_NOT_RECOGNIZED) \
.must_next()
# 4. Child should get a report with status 4 (NO_MATCHING_FRAMES_RECEIVED).
pkts.filter_wpan_src64(LEADER) \
.filter_wpan_dst64(CHILD) \
.filter_mle_cmd(consts.MLE_DATA_RESPONSE) \
.filter_mle_has_tlv(TlvType.LINK_METRICS_REPORT) \
.filter(lambda p: p.mle.tlv.link_status_sub_tlv == consts.LINK_METRICS_STATUS_NO_MATCHING_FRAMES_RECEIVED) \
.must_next()
# 5. Child should get a response with status 3 (SERIES_ID_NOT_RECOGNIZED).
pkts.filter_wpan_src64(LEADER) \
.filter_wpan_dst64(CHILD) \
.filter_mle_cmd(consts.MLE_LINK_METRICS_MANAGEMENT_RESPONSE) \
.filter(lambda p: p.mle.tlv.link_status_sub_tlv == consts.LINK_METRICS_STATUS_SERIES_ID_NOT_RECOGNIZED) \
.must_next()
# 6. Child should get a response with status 0 (SUCCESS).
pkts.filter_wpan_src64(LEADER) \
.filter_wpan_dst64(CHILD) \
.filter_mle_cmd(consts.MLE_LINK_METRICS_MANAGEMENT_RESPONSE) \
.filter(lambda p: p.mle.tlv.link_status_sub_tlv == consts.LINK_METRICS_STATUS_SUCCESS) \
.must_next()
# 7. Child should get a response with status 0 (SUCCESS).
pkts.filter_wpan_src64(LEADER) \
.filter_wpan_dst64(CHILD) \
.filter_mle_cmd(consts.MLE_LINK_METRICS_MANAGEMENT_RESPONSE) \
.filter(lambda p: p.mle.tlv.link_status_sub_tlv == consts.LINK_METRICS_STATUS_SUCCESS) \
.must_next()
# 8. Child sends an MLE Link Probe message to the Subject for the newly configured Series.
pkts.filter_wpan_src64(CHILD) \
.filter_wpan_dst64(LEADER) \
.filter_mle_cmd(consts.MLE_LINK_PROBE) \
.must_next()
# 9. Child should get a report with valid values.
pkts.filter_wpan_src64(LEADER) \
.filter_wpan_dst64(CHILD) \
.filter_mle_cmd(consts.MLE_DATA_RESPONSE) \
.filter(lambda p: TlvType.LINK_METRICS_REPORT in p.mle.tlv.type) \
.must_next()
if __name__ == '__main__':
unittest.main()
tests/scripts/thread-cert/v1_2_LowPower_test_link_metrics_manager.py
-271
View File
@@ -1,271 +0,0 @@
#!/usr/bin/env python3
#
# Copyright (c) 2023, 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 unittest
from config import ADDRESS_TYPE
from mle import LinkMetricsSubTlvType, TlvType
from pktverify import consts
from pktverify.null_field import nullField
from pktverify.packet_verifier import PacketVerifier
import config
import thread_cert
LEADER = 1
ROUTER = 2
CHILD = 3
"""
Acronyms:
- EAP - Enhanced-ACK Based Probing
Test Process:
1. Initiate a leader and a router at the beginning.
2. Enable Link Metrics Manager on leader.
3. Wait a few seconds.
- At this moment, leader should have configured EAP successfully at router.
4. Instruct the leader ping the router.
- Enhanced ACK sent by router should have Thread IE containing Link Metrics data.
5. Add a child into the network. The child should attach to the leader.
6. Wait 150 seconds.
- At this moment, leader should have configured EAP successfully at the child.
7. Instruct both the router and the child ping the leader.
- Enhanced ACK sent by router and child should have Thread IE containing Link Metrics data.
8. Shutdown the child.
9. Wait 300 seconds.
- The leader should not send a Link Management Request to the child.
- The leader should have sent a Link Management Request to the router and get a response.
10. Disable Link Metrics Manager on leader.
- The leader shonld send a Link Management Request to the router to unregister EAP.
"""
class LowPower_test_LinkMetricsManager(thread_cert.TestCase):
TOPOLOGY = {
LEADER: {
'version': '1.2',
'name': 'LEADER',
'mode': 'rdn',
'allowlist': [ROUTER, CHILD],
},
ROUTER: {
'version': '1.2',
'name': 'ROUTER',
'mode': 'rdn',
'allowlist': [LEADER],
},
CHILD: {
'version': '1.2',
'name': 'CHILD',
'mode': 'r',
'allowlist': [LEADER],
}
}
"""All nodes are created with default configurations"""
def test(self):
self.nodes[LEADER].start()
self.simulator.go(config.LEADER_STARTUP_DELAY)
self.assertEqual(self.nodes[LEADER].get_state(), 'leader')
self.nodes[ROUTER].start()
self.simulator.go(config.ROUTER_STARTUP_DELAY)
self.assertEqual(self.nodes[ROUTER].get_state(), 'router')
leader_addr = self.nodes[LEADER].get_ip6_address(ADDRESS_TYPE.LINK_LOCAL)
router_addr = self.nodes[ROUTER].get_ip6_address(ADDRESS_TYPE.LINK_LOCAL)
# Step 2 - Enable Link Metrics Manager on leader.
self.nodes[LEADER].link_metrics_mgr_set_enabled(True)
self.simulator.go(10)
# Step 4 - Instruct the leader ping the router.
self.nodes[LEADER].ping(router_addr)
# Step 5 - Add a child into the network.
self.nodes[CHILD].start()
self.simulator.go(10)
self.assertEqual(self.nodes[CHILD].get_state(), 'child')
child_addr = self.nodes[CHILD].get_ip6_address(ADDRESS_TYPE.LINK_LOCAL)
# Step 6 - Wait the leader to configure EAP at the child.
self.simulator.go(150)
# Step 7 - Instruct both the router and the child ping the leader.
self.nodes[LEADER].ping(router_addr)
self.nodes[LEADER].ping(child_addr)
# Step 8 - Shutdown the child.
self.nodes[CHILD].stop()
# Step 9 - Wait the leader to refresh EAP.
self.simulator.go(300)
# Step 10 - Disable Link Metrics Manager on leader.
self.nodes[LEADER].link_metrics_mgr_set_enabled(False)
def verify(self, pv):
pkts = pv.pkts
pv.summary.show()
LEADER = pv.vars['LEADER']
ROUTER = pv.vars['ROUTER']
CHILD = pv.vars['CHILD']
# Step 3 - Leader enables IEEE 802.15.4-2015 Enhanced ACK based Probing by sending a Link Metrics Management
# Request to the Router
# MLE Link Metrics Management TLV Payload:
# - Enhanced ACK Configuration Sub-TLV
# -- Enh-ACK Flags = 1 (register a configuration)
# -- Concatenation of Link Metric Type ID Flags = 0x00:
# --- Item1: (0)(0)(001)(010) = 0x0a
# ---- E = 0
# ---- L = 0
# ---- Type/Average Enum = 1 (Exponential Moving Avg)
# ---- Metrics Enum = 2 (Link Margin)
pkts.filter_wpan_src64(LEADER) \
.filter_wpan_dst64(ROUTER) \
.filter_mle_cmd(consts.MLE_LINK_METRICS_MANAGEMENT_REQUEST) \
.filter(lambda p: consts.LM_ENHANCED_ACK_CONFIGURATION_SUB_TLV in p.mle.tlv.link_sub_tlv) \
.filter(lambda p: p.mle.tlv.link_enh_ack_flags == consts.LINK_METRICS_ENH_ACK_PROBING_REGISTER) \
.filter(lambda p: p.mle.tlv.link_requested_type_id_flags == '0a0b') \
.must_next()
# Step 3 - The Router MUST send a Link Metrics Management Response to Leader containing the following TLVs:
# - MLE LInk Metrics Management TLV
# -- Link Metrics Status Sub-TLV = 0 (Success)
pkts.filter_wpan_src64(ROUTER) \
.filter_wpan_dst64(LEADER) \
.filter_mle_cmd(consts.MLE_LINK_METRICS_MANAGEMENT_RESPONSE) \
.filter(lambda p: p.mle.tlv.link_status_sub_tlv == consts.LINK_METRICS_STATUS_SUCCESS) \
.must_next()
# Step 4 - The Enhanced Ack sent by both Router should contain Thread specific IE
pkt = pkts.filter_wpan_src64(LEADER) \
.filter_wpan_dst64(ROUTER) \
.filter_ping_request() \
.must_next()
ack_seq_no = pkt.wpan.seq_no
pkts.filter_wpan_ack() \
.filter_wpan_seq(ack_seq_no) \
.filter(lambda p: p.wpan.payload_ie.vendor.oui == consts.THREAD_IEEE_802154_COMPANY_ID) \
.must_next()
# Step 6 - Leader enables IEEE 802.15.4-2015 Enhanced ACK based Probing by sending a Link Metrics Management
# Request to the Child
# MLE Link Metrics Management TLV Payload:
# - Enhanced ACK Configuration Sub-TLV
# -- Enh-ACK Flags = 1 (register a configuration)
# -- Concatenation of Link Metric Type ID Flags = 0x00:
# --- Item1: (0)(0)(001)(010) = 0x0a
# ---- E = 0
# ---- L = 0
# ---- Type/Average Enum = 1 (Exponential Moving Avg)
# ---- Metrics Enum = 2 (Link Margin)
pkts.filter_wpan_src64(LEADER) \
.filter_wpan_dst64(CHILD) \
.filter_mle_cmd(consts.MLE_LINK_METRICS_MANAGEMENT_REQUEST) \
.filter(lambda p: consts.LM_ENHANCED_ACK_CONFIGURATION_SUB_TLV in p.mle.tlv.link_sub_tlv) \
.filter(lambda p: p.mle.tlv.link_enh_ack_flags == consts.LINK_METRICS_ENH_ACK_PROBING_REGISTER) \
.filter(lambda p: p.mle.tlv.link_requested_type_id_flags == '0a0b') \
.must_next()
# Step 6 - The Child MUST send a Link Metrics Management Response to Leader containing the following TLVs:
# - MLE LInk Metrics Management TLV
# -- Link Metrics Status Sub-TLV = 0 (Success)
pkts.filter_wpan_src64(CHILD) \
.filter_wpan_dst64(LEADER) \
.filter_mle_cmd(consts.MLE_LINK_METRICS_MANAGEMENT_RESPONSE) \
.filter(lambda p: p.mle.tlv.link_status_sub_tlv == consts.LINK_METRICS_STATUS_SUCCESS) \
.must_next()
# Step 7 - The Enhanced Ack sent by both Router and Child should contain Thread specific IE
pkt1 = pkts.filter_wpan_src64(LEADER) \
.filter_wpan_dst64(ROUTER) \
.filter_ping_request() \
.must_next()
ack_seq_no = pkt1.wpan.seq_no
pkts.filter_wpan_ack() \
.filter_wpan_seq(ack_seq_no) \
.filter(lambda p: p.wpan.payload_ie.vendor.oui == consts.THREAD_IEEE_802154_COMPANY_ID) \
.must_next()
pkt2 = pkts.filter_wpan_src64(LEADER) \
.filter_wpan_dst64(CHILD) \
.filter_ping_request() \
.must_next()
ack_seq_no = pkt2.wpan.seq_no
pkts.filter_wpan_ack() \
.filter_wpan_seq(ack_seq_no) \
.filter(lambda p: p.wpan.payload_ie.vendor.oui == consts.THREAD_IEEE_802154_COMPANY_ID) \
.must_next()
# Step 9 - Leader refreshes IEEE 802.15.4-2015 Enhanced ACK based Probing by sending a Link Metrics Management
# Request to the Router
# MLE Link Metrics Management TLV Payload:
# - Enhanced ACK Configuration Sub-TLV
# -- Enh-ACK Flags = 1 (register a configuration)
# -- Concatenation of Link Metric Type ID Flags = 0x00:
# --- Item1: (0)(0)(001)(010) = 0x0a
# ---- E = 0
# ---- L = 0
# ---- Type/Average Enum = 1 (Exponential Moving Avg)
# ---- Metrics Enum = 2 (Link Margin)
pkts.filter_wpan_src64(LEADER) \
.filter_wpan_dst64(ROUTER) \
.filter_mle_cmd(consts.MLE_LINK_METRICS_MANAGEMENT_REQUEST) \
.filter(lambda p: consts.LM_ENHANCED_ACK_CONFIGURATION_SUB_TLV in p.mle.tlv.link_sub_tlv) \
.filter(lambda p: p.mle.tlv.link_enh_ack_flags == consts.LINK_METRICS_ENH_ACK_PROBING_REGISTER) \
.filter(lambda p: p.mle.tlv.link_requested_type_id_flags == '0a0b') \
.must_next()
# Step 9 - The Router MUST send a Link Metrics Management Response to Leader containing the following TLVs:
# - MLE LInk Metrics Management TLV
# -- Link Metrics Status Sub-TLV = 0 (Success)
pkts.filter_wpan_src64(ROUTER) \
.filter_wpan_dst64(LEADER) \
.filter_mle_cmd(consts.MLE_LINK_METRICS_MANAGEMENT_RESPONSE) \
.filter(lambda p: p.mle.tlv.link_status_sub_tlv == consts.LINK_METRICS_STATUS_SUCCESS) \
.must_next()
# Step 10 - Leader unregisters IEEE 802.15.4-2015 Enhanced ACK based Probing by sending a Link Metrics Management
# Request to the Router
# MLE Link Metrics Management TLV Payload:
# - Enhanced ACK Configuration Sub-TLV
# -- Enh-ACK Flags = 0 (unregister a configuration)
pkts.filter_wpan_src64(LEADER) \
.filter_wpan_dst64(ROUTER) \
.filter_mle_cmd(consts.MLE_LINK_METRICS_MANAGEMENT_REQUEST) \
.filter(lambda p: consts.LM_ENHANCED_ACK_CONFIGURATION_SUB_TLV in p.mle.tlv.link_sub_tlv) \
.filter(lambda p: p.mle.tlv.link_enh_ack_flags == consts.LINK_METRICS_ENH_ACK_PROBING_CLEAR) \
.must_next()
if __name__ == '__main__':
unittest.main()
tests/scripts/thread-cert/v1_2_test_csl_transmission.py
-145
View File
@@ -1,145 +0,0 @@
#!/usr/bin/env python3
#
# Copyright (c) 2020, 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 unittest
import config
import mle
import thread_cert
from pktverify import consts
LEADER = 1
SSED_1 = 2
class SSED_CslTransmission(thread_cert.TestCase):
TOPOLOGY = {
LEADER: {
'version': '1.2',
},
SSED_1: {
'version': '1.2',
'is_mtd': True,
'mode': '-',
},
}
"""All nodes are created with default configurations"""
def test(self):
self.nodes[SSED_1].set_csl_period(consts.CSL_DEFAULT_PERIOD)
self.nodes[SSED_1].set_csl_timeout(consts.CSL_DEFAULT_TIMEOUT)
self.nodes[SSED_1].get_csl_info()
self.nodes[LEADER].start()
self.simulator.go(config.LEADER_STARTUP_DELAY)
self.assertEqual(self.nodes[LEADER].get_state(), 'leader')
self.nodes[SSED_1].start()
self.simulator.go(7)
self.assertEqual(self.nodes[SSED_1].get_state(), 'child')
print('SSED rloc:%s' % self.nodes[SSED_1].get_rloc())
self.assertTrue(self.nodes[LEADER].ping(self.nodes[SSED_1].get_rloc()))
self.simulator.go(5)
ssed_messages = self.simulator.get_messages_sent_by(SSED_1)
msg = ssed_messages.next_mle_message(mle.CommandType.CHILD_UPDATE_REQUEST)
self.nodes[SSED_1].set_csl_channel(consts.CSL_DEFAULT_CHANNEL)
self.simulator.go(1)
ssed_messages = self.simulator.get_messages_sent_by(SSED_1)
msg = ssed_messages.next_mle_message(mle.CommandType.CHILD_UPDATE_REQUEST)
msg.assertMleMessageContainsTlv(mle.CslChannel)
self.assertTrue(self.nodes[LEADER].ping(self.nodes[SSED_1].get_rloc()))
self.simulator.go(5)
self.nodes[SSED_1].set_csl_channel(0)
self.simulator.go(1)
self.assertTrue(self.nodes[LEADER].ping(self.nodes[SSED_1].get_rloc()))
self.simulator.go(5)
ssed_messages = self.simulator.get_messages_sent_by(SSED_1)
msg = ssed_messages.next_mle_message(mle.CommandType.CHILD_UPDATE_REQUEST)
self.nodes[SSED_1].set_csl_period(0)
self.assertFalse(self.nodes[LEADER].ping(self.nodes[SSED_1].get_rloc()))
self.simulator.go(2)
self.nodes[SSED_1].set_pollperiod(1000)
self.simulator.go(2)
self.nodes[SSED_1].set_pollperiod(0)
self.simulator.go(5)
# Check if data poll is not sent if data packet with CSL IE is sent to parent
self.nodes[SSED_1].set_csl_period(consts.CSL_DEFAULT_PERIOD)
self.simulator.go(consts.CSL_DEFAULT_TIMEOUT / 2)
self.assertTrue(self.nodes[LEADER].ping(self.nodes[SSED_1].get_rloc()))
self.flush_all()
self.simulator.go(consts.CSL_DEFAULT_TIMEOUT / 2)
ssed_messages = self.simulator.get_messages_sent_by(SSED_1)
self.assertIsNone(ssed_messages.next_data_poll())
# Check if data poll is used to sync SSED's parent before CSL timeout
self.assertTrue(self.nodes[LEADER].ping(self.nodes[SSED_1].get_rloc()))
self.flush_all()
self.simulator.go(consts.CSL_DEFAULT_TIMEOUT)
ssed_messages = self.simulator.get_messages_sent_by(SSED_1)
self.assertIsNotNone(ssed_messages.next_data_poll())
# Check if data poll is used to resync SSED's parent after retransmission
leader_rloc = self.nodes[LEADER].get_rloc()
self.nodes[LEADER].stop()
self.flush_all()
self.assertFalse(self.nodes[SSED_1].ping(leader_rloc))
self.simulator.go(2)
ssed_messages = self.simulator.get_messages_sent_by(SSED_1)
self.assertIsNotNone(ssed_messages.next_data_poll())
# Check if SSED is able to resynchronize with the parent after it is gone longer than the timeout
self.nodes[LEADER].start()
self.simulator.go(config.LEADER_RESET_DELAY)
self.nodes[SSED_1].set_csl_timeout(8)
self.nodes[SSED_1].set_timeout(10)
self.simulator.go(2)
self.nodes[LEADER].stop()
self.simulator.go(25)
self.flush_all()
self.nodes[LEADER].start()
self.simulator.go(config.LEADER_RESET_DELAY)
self.assertEqual(self.nodes[LEADER].get_state(), 'leader')
self.simulator.go(5)
self.assertEqual(self.nodes[SSED_1].get_state(), 'child')
ssed_messages = self.simulator.get_messages_sent_by(SSED_1)
self.assertIsNotNone(ssed_messages.next_mle_message(mle.CommandType.CHILD_UPDATE_REQUEST))
if __name__ == '__main__':
unittest.main()
tests/scripts/thread-cert/v1_2_test_enhanced_frame_pending.py
-111
View File
@@ -1,111 +0,0 @@
#!/usr/bin/env python3
#
# Copyright (c) 2020, 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.
#
"""This test case verifies Thread 1.2 enhanced frame pending feature.
Topology
Leader ----- SED_1
1. Set up topology a Leader and a SED.
2. SED pings leader so a data poll is just sent.
3. Leader sends a UDP packet to SED to put a pending frame in queue.
4. Wait for half polling period, verify no packet received by SED.
5. SED sends a UDP packet to Leader to solicit an ACK with frame pending set.
* Verify Leader receives a UDP packet
* Verify SED sends a data poll
* Verify SED receives a UDP packet
"""
import unittest
import pexpect
import thread_cert
import config
import common
LEADER = 1
SED_1 = 2
CHILD_TIMEOUT = 30
UDP_BYTES_COUNT = 73
DEFAULT_POLL_PERIOD = CHILD_TIMEOUT - 4
"""The default poll period calculated from child timeout."""
class SED_EnhancedFramePending(thread_cert.TestCase):
TOPOLOGY = {
LEADER: {
'version': '1.2'
},
SED_1: {
'mode': '-',
'version': '1.2',
},
}
def test(self):
self.nodes[SED_1].set_timeout(CHILD_TIMEOUT)
# 1 - Set up topology
self.nodes[LEADER].start()
self.simulator.go(config.LEADER_STARTUP_DELAY)
self.assertEqual(self.nodes[LEADER].get_state(), 'leader')
self.nodes[SED_1].start()
self.simulator.go(7)
self.assertEqual(self.nodes[SED_1].get_state(), 'child')
self.nodes[LEADER].udp_start('::', common.UDP_TEST_PORT)
self.nodes[SED_1].udp_start('::', common.UDP_TEST_PORT)
# 2 - Ping Leader
self.assertTrue(self.nodes[SED_1].ping(self.nodes[LEADER].get_rloc(), timeout=CHILD_TIMEOUT))
self.flush_all()
# 3 - Send to SED
self.nodes[LEADER].udp_send(UDP_BYTES_COUNT, self.nodes[SED_1].get_rloc(), common.UDP_TEST_PORT)
# 4 - Wait for half polling period
self.simulator.go(DEFAULT_POLL_PERIOD // 2)
with self.assertRaises(pexpect.TIMEOUT):
self.nodes[SED_1].udp_check_rx(UDP_BYTES_COUNT)
# 5 - Send to Leader
self.nodes[SED_1].udp_send(UDP_BYTES_COUNT, self.nodes[LEADER].get_rloc(), common.UDP_TEST_PORT)
self.simulator.go(1)
self.nodes[LEADER].udp_check_rx(UDP_BYTES_COUNT)
sed_messages = self.simulator.get_messages_sent_by(SED_1)
self.assertNotEqual(sed_messages.next_data_poll(), None)
self.nodes[SED_1].udp_check_rx(UDP_BYTES_COUNT)
if __name__ == '__main__':
unittest.main()
tests/scripts/thread-cert/v1_2_test_enhanced_keep_alive.py
-183
View File
@@ -1,183 +0,0 @@
#!/usr/bin/env python3
#
# Copyright (c) 2019, 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 unittest
import thread_cert
import config
import mle
LEADER = 1
SED_1 = 2
CHILD_TIMEOUT = 30
DEFAULT_POLL_PERIOD = CHILD_TIMEOUT - 4
"""The default poll period calculated by ot::Mac::DataPollSender::GetDefaultPollPeriod()."""
USER_POLL_PERIOD = CHILD_TIMEOUT // 3
"""The poll period explicitly set by this test for verifying enhanced keep-alive."""
class SED_EnhancedKeepAlive(thread_cert.TestCase):
TOPOLOGY = {
LEADER: {
'version': '1.2'
},
SED_1: {
'mode': '-',
'version': '1.2',
},
}
"""All nodes are created with default configurations"""
def test(self):
self.nodes[SED_1].set_timeout(CHILD_TIMEOUT)
self.nodes[SED_1].set_pollperiod(USER_POLL_PERIOD * 1000)
self.nodes[LEADER].start()
self.simulator.go(config.LEADER_STARTUP_DELAY)
self.assertEqual(self.nodes[LEADER].get_state(), 'leader')
self.nodes[SED_1].start()
self.simulator.go(7)
self.assertEqual(self.nodes[SED_1].get_state(), 'child')
leader_messages = self.simulator.get_messages_sent_by(LEADER)
sed_messages = self.simulator.get_messages_sent_by(SED_1)
# 1 - Leader transmits MLE advertisements
msg = leader_messages.next_mle_message(mle.CommandType.ADVERTISEMENT)
msg.assertSentWithHopLimit(255)
msg.assertSentToDestinationAddress('ff02::1')
msg.assertMleMessageContainsTlv(mle.SourceAddress)
msg.assertMleMessageContainsTlv(mle.LeaderData)
msg.assertMleMessageContainsTlv(mle.Route64)
# 2 - SED_1 begins attach process by sending a multicast MLE Parent Request
msg = sed_messages.next_mle_message(mle.CommandType.PARENT_REQUEST)
msg.assertSentWithHopLimit(255)
msg.assertSentToDestinationAddress('ff02::2')
msg.assertMleMessageContainsTlv(mle.Mode)
msg.assertMleMessageContainsTlv(mle.Challenge)
msg.assertMleMessageContainsTlv(mle.ScanMask)
msg.assertMleMessageContainsTlv(mle.Version)
self.assertGreaterEqual(msg.get_mle_message_tlv(mle.Version).version, config.THREAD_VERSION_1_2)
scan_mask_tlv = msg.get_mle_message_tlv(mle.ScanMask)
self.assertEqual(1, scan_mask_tlv.router)
self.assertEqual(0, scan_mask_tlv.end_device)
# 3 - Leader sends a MLE Parent Response
msg = leader_messages.next_mle_message(mle.CommandType.PARENT_RESPONSE)
msg.assertSentToNode(self.nodes[SED_1])
msg.assertMleMessageContainsTlv(mle.SourceAddress)
msg.assertMleMessageContainsTlv(mle.LeaderData)
msg.assertMleMessageContainsTlv(mle.LinkLayerFrameCounter)
msg.assertMleMessageContainsOptionalTlv(mle.MleFrameCounter)
msg.assertMleMessageContainsTlv(mle.Response)
msg.assertMleMessageContainsTlv(mle.Challenge)
msg.assertMleMessageContainsTlv(mle.LinkMargin)
msg.assertMleMessageContainsTlv(mle.Connectivity)
msg.assertMleMessageContainsTlv(mle.Version)
self.assertGreaterEqual(msg.get_mle_message_tlv(mle.Version).version, config.THREAD_VERSION_1_2)
# 4 - SED_1 receives the MLE Parent Response and sends a Child ID Request
msg = sed_messages.next_mle_message(mle.CommandType.CHILD_ID_REQUEST)
msg.assertSentToNode(self.nodes[LEADER])
msg.assertMleMessageContainsTlv(mle.Response)
msg.assertMleMessageContainsTlv(mle.LinkLayerFrameCounter)
msg.assertMleMessageContainsOptionalTlv(mle.MleFrameCounter)
msg.assertMleMessageContainsTlv(mle.Mode)
msg.assertMleMessageContainsTlv(mle.Timeout)
msg.assertMleMessageContainsTlv(mle.Version)
msg.assertMleMessageContainsTlv(mle.TlvRequest)
self.assertGreaterEqual(msg.get_mle_message_tlv(mle.Version).version, config.THREAD_VERSION_1_2)
# 5 - Leader responds with a Child ID Response
msg = leader_messages.next_mle_message(mle.CommandType.CHILD_ID_RESPONSE)
msg.assertSentToNode(self.nodes[SED_1])
msg.assertMleMessageContainsTlv(mle.SourceAddress)
msg.assertMleMessageContainsTlv(mle.LeaderData)
msg.assertMleMessageContainsTlv(mle.Address16)
msg.assertMleMessageContainsOptionalTlv(mle.NetworkData)
msg.assertMleMessageContainsOptionalTlv(mle.Route64)
msg.assertMleMessageContainsOptionalTlv(mle.AddressRegistration)
leader_aloc = self.nodes[LEADER].get_addr_leader_aloc()
self.assertTrue(self.nodes[SED_1].ping(leader_aloc, timeout=USER_POLL_PERIOD * 2))
# 6 - Timeout Child
self.nodes[SED_1].set_pollperiod(CHILD_TIMEOUT * 1000 * 4)
self.simulator.go(CHILD_TIMEOUT + 1)
self.assertEqual(self.nodes[SED_1].get_state(), 'child')
self.assertFalse(self.nodes[SED_1].ping(leader_aloc, timeout=USER_POLL_PERIOD * 2))
self.flush_all()
self.nodes[SED_1].stop()
self.nodes[SED_1].set_pollperiod(USER_POLL_PERIOD * 1000)
self.nodes[SED_1].start()
# 7 - Wait SED_1 to re-attach
self.simulator.go(240)
self.assertEqual(self.nodes[SED_1].get_state(), 'child')
leader_messages = self.simulator.get_messages_sent_by(LEADER)
msg = leader_messages.next_mle_message(mle.CommandType.CHILD_ID_RESPONSE)
msg.assertSentToNode(self.nodes[SED_1])
msg.assertMleMessageContainsTlv(mle.SourceAddress)
msg.assertMleMessageContainsTlv(mle.LeaderData)
msg.assertMleMessageContainsTlv(mle.Address16)
msg.assertMleMessageContainsOptionalTlv(mle.NetworkData)
msg.assertMleMessageContainsOptionalTlv(mle.Route64)
msg.assertMleMessageContainsOptionalTlv(mle.AddressRegistration)
self.assertTrue(self.nodes[SED_1].ping(leader_aloc, timeout=USER_POLL_PERIOD * 2))
self.flush_all()
# 8 - Verify enhanced keep-alive works
self.nodes[SED_1].set_pollperiod(CHILD_TIMEOUT * 1000 * 4)
self.simulator.go(CHILD_TIMEOUT // 2)
self.assertEqual(self.nodes[SED_1].get_state(), 'child')
non_exist_addr = leader_aloc.replace('fc00', 'fc12')
self.assertFalse(self.nodes[SED_1].ping(non_exist_addr))
self.simulator.go(CHILD_TIMEOUT // 2)
self.nodes[SED_1].set_pollperiod(USER_POLL_PERIOD * 1000)
self.assertTrue(self.nodes[SED_1].ping(leader_aloc, timeout=USER_POLL_PERIOD * 2))
# 9 - Verify child resets keep-alive timer
self.nodes[SED_1].set_pollperiod(DEFAULT_POLL_PERIOD * 1000)
self.simulator.go(DEFAULT_POLL_PERIOD // 3 * 2)
self.flush_all()
self.nodes[SED_1].ping(leader_aloc, timeout=1)
self.simulator.go(DEFAULT_POLL_PERIOD // 3 * 2)
sed_messages = self.simulator.get_messages_sent_by(SED_1)
self.assertEqual(sed_messages.next_data_poll(), None)
if __name__ == '__main__':
unittest.main()
tests/scripts/thread-cert/v1_2_test_single_probe.py
-124
View File
@@ -1,124 +0,0 @@
#!/usr/bin/env python3
#
# Copyright (c) 2020, 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 unittest
import mle
import thread_cert
import config
LEADER = 1
SSED_1 = 2
CSL_PERIOD = 500 * 1000 # 0.5 in usec
CSL_TIMEOUT = 30 # 30s
class SSED_SingleProbe(thread_cert.TestCase):
TOPOLOGY = {
LEADER: {
'version': '1.2',
},
SSED_1: {
'version': '1.2',
'mode': '-',
},
}
"""All nodes are created with default configurations"""
def test(self):
self.nodes[SSED_1].set_csl_period(CSL_PERIOD)
self.nodes[SSED_1].set_csl_timeout(CSL_TIMEOUT)
self.nodes[LEADER].start()
self.simulator.go(config.LEADER_STARTUP_DELAY)
self.assertEqual(self.nodes[LEADER].get_state(), 'leader')
self.nodes[SSED_1].start()
self.simulator.go(7)
self.assertEqual(self.nodes[SSED_1].get_state(), 'child')
leader_addr = self.nodes[LEADER].get_linklocal()
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
result = self.nodes[SSED_1].link_metrics_request_single_probe(leader_addr, 'p')
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
result = self.nodes[SSED_1].link_metrics_request_single_probe(leader_addr, 'q')
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
result = self.nodes[SSED_1].link_metrics_request_single_probe(leader_addr, 'm')
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
result = self.nodes[SSED_1].link_metrics_request_single_probe(leader_addr, 'r')
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
result = self.nodes[SSED_1].link_metrics_request_single_probe(leader_addr, 'pqmr')
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)
msg.assertMleMessageContainsTlv(mle.LinkMetricsReport)
self.assertTrue(self.nodes[LEADER].ping(self.nodes[SSED_1].get_rloc()))
if __name__ == '__main__':
unittest.main()