espressif/openthread
1
0
Fork 0
mirror of https://github.com/espressif/openthread.git synced 2026-08-26 12:11:17 +00:00
Code Issues Packages Projects Releases Wiki Activity

[tests] remove border router MATN tests (#13306)

Browse Source 
Remove MATN border router certification tests as they have been
replaced by equivalent tests in the ot-br-posix repository under:
https://github.com/openthread/ot-br-posix/tree/main/tests/scripts/expect

The new tests run using the docker-in-docker framework.

Deleted tests:
- MATN_02_MLRFirstUse.py
- MATN_03_InvalidCommissionerDeregistration.py
- MATN_04_MulticastListenerTimeout.py
- MATN_05_ReregistrationToSameMulticastGroup.py
- MATN_09_DefaultBRMulticastForwarding.py
- MATN_12_HopLimitProcessing.py
- MATN_15_ChangeOfPrimaryBBRTriggersRegistration.py
- MATN_16_LargeNumberOfMulticastGroupSubscriptionsToBBR.py
This commit is contained in:
Jonathan Hui
2026-07-06 15:16:17 -07:00
committed by GitHub
parent 6cc9359cee
commit 8bb5b30b2b
9 changed files with 0 additions and 1836 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.github/workflows/otbr.yml
-6
View File
@@ -63,12 +63,6 @@ jobs:
packet_verification: 2
nat64: 0
description: ""
- otbr_mdns: "mDNSResponder"
otbr_trel: 0
cert_scripts: ./tests/scripts/thread-cert/border_router/MATN/*.py
packet_verification: 1
nat64: 0
description: "MATN"
- otbr_mdns: "mDNSResponder"
otbr_trel: 0
cert_scripts: ./tests/scripts/thread-cert/border_router/internet/*.py
tests/scripts/thread-cert/border_router/MATN/MATN_02_MLRFirstUse.py
-290
View File
@@ -1,290 +0,0 @@
#!/usr/bin/env python3
#
# Copyright (c) 2021, The OpenThread Authors.
# All rights reserved.
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions are met:
# 1. Redistributions of source code must retain the above copyright
# notice, this list of conditions and the following disclaimer.
# 2. Redistributions in binary form must reproduce the above copyright
# notice, this list of conditions and the following disclaimer in the
# documentation and/or other materials provided with the distribution.
# 3. Neither the name of the copyright holder nor the
# names of its contributors may be used to endorse or promote products
# derived from this software without specific prior written permission.
#
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS 'AS IS'
# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
# ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
# LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
# CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
# SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
# CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
# POSSIBILITY OF SUCH DAMAGE.
#
import logging
import unittest
import pktverify
from pktverify import packet_verifier, packet_filter
from pktverify.consts import MA1, MA1g, MA2, PBBR_ALOC
import config
import thread_cert
# Test description:
# The purpose of this test case is to verify that a Thread device is able to
# register a multicast address with a Primary BBR and that the Primary BBR can
# notify devices on the backbone link of the multicast address. The test also
# verifies that an IPv6 multicast packet originating from the backbone is
# correctly forwarded (in the Thread Network) to the device that registered the
# multicast address.
#
# Topology:
# ----------------(eth)------------------
# | | |
# BR1 (Leader) ----- BR2 HOST
# | |
# | |
# TD --------(when TD is router)
#
BR_1 = 1
BR_2 = 2
TD = 3
HOST = 4
CHANNEL1 = 18
MLR_NOTIFICATION_TIMEOUT = 3600
class MATN_02_MLRFirstUse(thread_cert.TestCase):
USE_MESSAGE_FACTORY = False
TOPOLOGY = {
BR_1: {
'name': 'BR_1',
'is_otbr': True,
'allowlist': [BR_2, TD],
'version': '1.2',
},
BR_2: {
'name': 'BR_2',
'allowlist': [BR_1, TD],
'is_otbr': True,
'version': '1.2',
},
TD: {
'name': 'TD',
'allowlist': [BR_1, BR_2],
'version': '1.2',
},
HOST: {
'name': 'Host',
'is_host': True
},
}
def test(self):
br1 = self.nodes[BR_1]
br2 = self.nodes[BR_2]
td = self.nodes[TD]
host = self.nodes[HOST]
br1.start()
self.simulator.go(config.LEADER_STARTUP_DELAY)
self.assertEqual('leader', br1.get_state())
self.assertTrue(br1.is_primary_backbone_router)
td.start()
self.simulator.go(5)
self.assertEqual('router', td.get_state())
br2.start()
self.simulator.go(5)
self.assertEqual('router', br2.get_state())
host.start(start_radvd=False)
self.simulator.go(10)
# 1. TD registers for multicast address, MA1, at BR_1.
td.add_ipmaddr(MA1)
self.simulator.go(5)
# 7. Host sends a ping packet to the multicast address, MA1. TD should
# respond to the ping request.
self.assertTrue(
host.ping(MA1, backbone=True, ttl=10, interface=host.get_ip6_address(config.ADDRESS_TYPE.ONLINK_ULA)[0]))
self.simulator.go(5)
# 11. Host sends a ping packet to the multicast address, MA2. No one
# should respond.
self.assertFalse(
host.ping(MA2, backbone=True, ttl=10, interface=host.get_ip6_address(config.ADDRESS_TYPE.ONLINK_ULA)[0]))
self.simulator.go(5)
# 14. Host sends a ping packet to the multicast address, MA1g. No one
# should respond.
self.assertFalse(
host.ping(MA1g, backbone=True, ttl=10, interface=host.get_ip6_address(config.ADDRESS_TYPE.ONLINK_ULA)[0]))
self.simulator.go(5)
# 17. Host sends a ping packet to the global unicast address of BR2. BR2
# should respond.
self.assertTrue(host.ping(br2.get_ip6_address(config.ADDRESS_TYPE.BACKBONE_GUA), backbone=True, ttl=10))
self.simulator.go(5)
self.collect_ipaddrs()
self.collect_rloc16s()
self.collect_rlocs()
self.collect_leader_aloc(BR_1)
self.collect_extra_vars()
def verify(self, pv: pktverify.packet_verifier.PacketVerifier):
pkts = pv.pkts
vars = pv.vars
pv.summary.show()
# Ensure the topology is formed correctly
pv.verify_attached('TD', 'BR_1')
pv.verify_attached('BR_2')
# 1. TD Registers for multicast address, MA1, at BR_1.
# TD unicasts an MLR.req CoAP request to BR_1 as
# "coap://[<BR_1 RLOC or PBBR ALOC>]:MM/n/mr".
# The payload contains "IPv6Address TLV: MA1".
pkts.filter_wpan_src64(vars['TD']) \
.filter_ipv6_2dsts(vars['BR_1_RLOC'], PBBR_ALOC) \
.filter_coap_request('/n/mr') \
.filter(lambda p: p.thread_meshcop.tlv.ipv6_addr == [MA1]) \
.must_next()
# 3. BR_1 responds to the multicast registration.
# BR_1 unicasts an MLR.rsp CoAP response to TD as "2.04 changed".
# The payload contains "Status TLV: ST_MLR_SUCCESS".
pkts.copy().filter_wpan_src64(vars['BR_1']) \
.filter_ipv6_dst(vars['TD_RLOC']) \
.filter_coap_ack('/n/mr') \
.filter(lambda p: p.thread_nm.tlv.status == 0) \
.must_next()
# 3a. BR_2 does not respond to the multicast registration.
pkts.filter_wpan_src64(vars['BR_2']) \
.filter_ipv6_dst(vars['TD_RLOC']) \
.filter_coap_ack('/n/mr') \
.must_not_next()
# 4. BR_1 informs other BBRs on the network of multicast registration.
# BR_1 multicasts a BMLR.ntf CoAP request to the Backbone Link including
# to BR_2, as follows
pkts.filter_eth_src(vars['BR_1_ETH']) \
.filter_ipv6_dst(config.ALL_NETWORK_BBRS_ADDRESS) \
.filter_coap_request('/b/bmr') \
.filter(lambda
p: p.thread_meshcop.tlv.ipv6_addr == [MA1] and
p.thread_bl.tlv.timeout == MLR_NOTIFICATION_TIMEOUT) \
.must_next()
# 6. BR_1 multicasts an MLDv2 message to start listening to MA1.
# BR_1 multicasts an MLDv2 message of type “Version 2 Multicast Listener
# Report” (see [RFC 3810] Section 5.2). Where:
# Nr of Mcast Address at least 1 Records (M): Multicast Address Record
# The Multicast Address Record contains the following:
# Record Type: 4 (CHANGE_TO_EXCLUDE_MODE)
# Number of Sources (N): 0
# Multicast Address: MA1
# TODO: Implement this verification
# 7. Host sends a ping packet to the multicast address, MA1.
_pkt = pkts.filter_eth_src(vars['Host_ETH']) \
.filter_ipv6_dst(MA1) \
.filter_ping_request() \
.must_next()
# 8. BR_2 does not forward the ping packet with multicast address MA1 to
# its Thread Network.
pkts.filter_wpan_src64(vars['BR_2']) \
.filter_AMPLFMA(mpl_seed_id=(vars['BR_2_RLOC'])) \
.filter_ping_request(identifier=_pkt.icmpv6.echo.identifier) \
.must_not_next()
# 9. BR_1 forwards the ping packet to its Thread Network.
# BR_1 forwards the ping packet with multicast address, MA1, to its
# Thread Network encapsulated in an MPL packet, where:
# MPL Seed ID: If Source outer IP header = BR_1 RLOC, SeedID length = 0
# Else, SeedID length = 1, and Seed ID = BR_1 RLOC16
pkts.filter_wpan_src64(vars['BR_1']) \
.filter_AMPLFMA(mpl_seed_id=(vars['BR_1_RLOC'])) \
.filter_ping_request(identifier=_pkt.icmpv6.echo.identifier) \
.must_next()
# 10. TD receives the multicast ping packet and sends a ping response
# packet back to Host.
# TD receives the MPL packet containing an encapsulated ping packet to
# MA1, sent by Host, and unicasts a ping response packet back to Host.
pkts.filter_wpan_src64(vars['TD']) \
.filter_ipv6_dst(_pkt.ipv6.src) \
.filter_ping_reply(identifier=_pkt.icmpv6.echo.identifier) \
.must_next()
# 11. Host sends a ping packet to the multicast address, MA2.
_pkt = pkts.filter_eth_src(vars['Host_ETH']) \
.filter_ipv6_dst(MA2) \
.filter_ping_request() \
.must_next()
# 12. BR_2 does not forward the ping packet with multicast address, MA2,
# to the Thread Network in whatever way.
pkts.filter_wpan_src64(vars['BR_2']) \
.filter_AMPLFMA(mpl_seed_id=(vars['BR_2_RLOC'])) \
.filter_ping_request(identifier=_pkt.icmpv6.echo.identifier) \
.must_not_next()
# 13. BR_1 does not forward the ping packet with multicast address, MA2,
# to the Thread Network in whatever way.
pkts.filter_wpan_src64(vars['BR_1']) \
.filter_AMPLFMA(mpl_seed_id=(vars['BR_1_RLOC'])) \
.filter_ping_request(identifier=_pkt.icmpv6.echo.identifier) \
.must_not_next()
# 14. Host sends a ping packet to the multicast address, MA1g.
_pkt = pkts.filter_eth_src(vars['Host_ETH']) \
.filter_ipv6_dst(MA1g) \
.filter_ping_request() \
.must_next()
# 15. BR_2 does not forward the ping packet with multicast address MA1g,
# to the Thread Network in whatever way.
pkts.filter_wpan_src64(vars['BR_2']) \
.filter_AMPLFMA(mpl_seed_id=(vars['BR_2_RLOC'])) \
.filter_ping_request(identifier=_pkt.icmpv6.echo.identifier) \
.must_not_next()
# 16. BR_1 does not forward the ping packet with multicast address MA1g,
# to its Thread Network in whatever way.
pkts.filter_wpan_src64(vars['BR_1']) \
.filter_AMPLFMA(mpl_seed_id=(vars['BR_1_RLOC'])) \
.filter_ping_request(identifier=_pkt.icmpv6.echo.identifier) \
.must_not_next()
# 17. Host sends a ping packet to the BR_2's global unicast address, Gg.
_pkt = pkts.filter_eth_src(vars['Host_ETH']) \
.filter_ipv6_dst(vars['BR_2_BGUA']) \
.filter_ping_request() \
.must_next()
# 18. BR_2 receives and provides the ping response.
# BR_2 Must send back the ping response to the Host.
pkts.filter_eth_src(vars['BR_2_ETH']) \
.filter_ipv6_dst(vars['Host_BGUA']) \
.filter_ping_reply(identifier=_pkt.icmpv6.echo.identifier) \
.must_next()
if __name__ == '__main__':
unittest.main()
tests/scripts/thread-cert/border_router/MATN/MATN_03_InvalidCommissionerDeregistration.py
-184
View File
@@ -1,184 +0,0 @@
#!/usr/bin/env python3
#
# Copyright (c) 2021, The OpenThread Authors.
# All rights reserved.
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions are met:
# 1. Redistributions of source code must retain the above copyright
# notice, this list of conditions and the following disclaimer.
# 2. Redistributions in binary form must reproduce the above copyright
# notice, this list of conditions and the following disclaimer in the
# documentation and/or other materials provided with the distribution.
# 3. Neither the name of the copyright holder nor the
# names of its contributors may be used to endorse or promote products
# derived from this software without specific prior written permission.
#
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS 'AS IS'
# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
# ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
# LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
# CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
# SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
# CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
# POSSIBILITY OF SUCH DAMAGE.
#
import logging
import pktverify
from pktverify import packet_verifier
from pktverify.consts import MA1, PBBR_ALOC
import unittest
import config
import thread_cert
# Test description:
# The purpose of this test case is to verify that a Primary BBR ignores a
# Timeout TLV if included in an MLR.req that does not contain a Commissioner
# Session ID TLV nor is signed by a Commissioner.
#
# Topology:
# ----------------(eth)------------------
# | | |
# BR1 (Leader) ----- BR2 HOST
# | |
# | |
# Router_1------------+
#
BR_1 = 1
BR_2 = 2
ROUTER_1 = 3
HOST = 4
class MATN_03_InvalidCommissionerDeregistration(thread_cert.TestCase):
USE_MESSAGE_FACTORY = False
TOPOLOGY = {
BR_1: {
'name': 'BR_1',
'is_otbr': True,
'allowlist': [BR_2, ROUTER_1],
'version': '1.2',
},
BR_2: {
'name': 'BR_2',
'allowlist': [BR_1, ROUTER_1],
'is_otbr': True,
'version': '1.2',
},
ROUTER_1: {
'name': 'Router_1',
'allowlist': [BR_1, BR_2],
'version': '1.2',
},
HOST: {
'name': 'Host',
'is_host': True
},
}
def test(self):
br1 = self.nodes[BR_1]
br2 = self.nodes[BR_2]
router = self.nodes[ROUTER_1]
host = self.nodes[HOST]
br1.start()
self.simulator.go(config.LEADER_STARTUP_DELAY)
self.assertEqual('leader', br1.get_state())
self.assertTrue(br1.is_primary_backbone_router)
router.start()
self.simulator.go(5)
self.assertEqual('router', router.get_state())
br2.start()
self.simulator.go(5)
self.assertEqual('router', router.get_state())
self.assertFalse(br2.is_primary_backbone_router)
host.start(start_radvd=False)
self.simulator.go(5)
router.add_ipmaddr(MA1)
self.simulator.go(5)
# Ensure that router can receive a multicast ping packet sent to MA1.
self.assertTrue(
host.ping(MA1, backbone=True, ttl=10, interface=host.get_ip6_address(config.ADDRESS_TYPE.ONLINK_ULA)[0]))
self.simulator.go(5)
# 1. Router attempts to deregister for multicast address, MA1, at BR_1.
router.register_multicast_listener(MA1, timeout=0)
# 3. Host sends a ping packet to the multicast address, MA1.
self.assertTrue(
host.ping(MA1, backbone=True, ttl=10, interface=host.get_ip6_address(config.ADDRESS_TYPE.ONLINK_ULA)[0]))
self.simulator.go(5)
self.collect_ipaddrs()
self.collect_rloc16s()
self.collect_rlocs()
self.collect_leader_aloc(BR_1)
self.collect_extra_vars()
def verify(self, pv: pktverify.packet_verifier.PacketVerifier):
pkts = pv.pkts
vars = pv.vars
pv.summary.show()
# Ensure the topology is formed correctly
pv.verify_attached('Router_1', 'BR_1')
pv.verify_attached('BR_2')
# 1. Router attempts to deregister for multicast address, MA1, at BR_1.
# Router unicasts an MLR.req CoAP request to BR_1 as follows:
# coap://[<BR_1 RLOC>]:MM/n/mr
# Where the payload contains:
# IPv6 Addresses TLV: MA1
# Timeout TLV: 0
pkts.filter_wpan_src64(vars['Router_1']) \
.filter_ipv6_2dsts(vars['BR_1_RLOC'], PBBR_ALOC) \
.filter_coap_request('/n/mr') \
.filter(lambda p: p.thread_meshcop.tlv.ipv6_addr == [MA1] and
p.thread_nm.tlv.timeout == 0) \
.must_next()
# 2. BR_1 responds to the multicast registration successfully,
# ignoring the timeout value.
pkts.filter_wpan_src64(vars['BR_1']) \
.filter_ipv6_dst(vars['Router_1_RLOC']) \
.filter_coap_ack('/n/mr') \
.filter(lambda p: p.thread_nm.tlv.status == 0) \
.must_next()
# 3. Host sends a ping packet to the multicast address, MA1.
_pkt = pkts.filter_eth_src(vars['Host_ETH']) \
.filter_ipv6_dst(MA1) \
.filter_ping_request() \
.must_next()
# 4. BR_1 forwards the ping packet with multicast address, MA1, to its
# Thread Network encapsulated in an MPL packet.
pkts.filter_wpan_src64(vars['BR_1']) \
.filter_AMPLFMA(mpl_seed_id=vars['BR_1_RLOC']) \
.filter_ping_request(identifier=_pkt.icmpv6.echo.identifier) \
.must_next()
# 5. Router receives the MPL packet containing an encapsulated ping
# packet to MA1, sent by Host, and unicasts a ping response packet back
# to Host.
pkts.filter_wpan_src64(vars['Router_1']) \
.filter_ipv6_dst(_pkt.ipv6.src) \
.filter_ping_reply(identifier=_pkt.icmpv6.echo.identifier) \
.must_next()
if __name__ == '__main__':
unittest.main()
tests/scripts/thread-cert/border_router/MATN/MATN_04_MulticastListenerTimeout.py
-219
View File
@@ -1,219 +0,0 @@
#!/usr/bin/env python3
#
# Copyright (c) 2021, The OpenThread Authors.
# All rights reserved.
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions are met:
# 1. Redistributions of source code must retain the above copyright
# notice, this list of conditions and the following disclaimer.
# 2. Redistributions in binary form must reproduce the above copyright
# notice, this list of conditions and the following disclaimer in the
# documentation and/or other materials provided with the distribution.
# 3. Neither the name of the copyright holder nor the
# names of its contributors may be used to endorse or promote products
# derived from this software without specific prior written permission.
#
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS 'AS IS'
# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
# ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
# LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
# CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
# SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
# CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
# POSSIBILITY OF SUCH DAMAGE.
#
import logging
import unittest
import pktverify
from pktverify import packet_verifier, packet_filter, consts
from pktverify.consts import MA1
import config
import thread_cert
# Test description:
# The purpose of this test case is to verify that a Primary BBR can remove an
# entry from a Multicast Listeners Table, when the entry expires. The test case
# also verifies that the Primary BBR accepts a new registration to the
# previously expired multicast group.
#
# Topology:
# ----------------(eth)------------------
# | | |
# BR1 (Leader) ----- BR2 HOST
# | |
# | |
# Router_1------------+
#
BR_1 = 1
BR_2 = 2
ROUTER_1 = 3
HOST = 4
REG_DELAY = 10
class MATN_04_MulticastListenerTimeout(thread_cert.TestCase):
USE_MESSAGE_FACTORY = False
TOPOLOGY = {
BR_1: {
'name': 'BR_1',
'is_otbr': True,
'allowlist': [BR_2, ROUTER_1],
'version': '1.2',
},
BR_2: {
'name': 'BR_2',
'allowlist': [BR_1, ROUTER_1],
'is_otbr': True,
'version': '1.2',
},
ROUTER_1: {
'name': 'Router_1',
'allowlist': [BR_1, BR_2],
'version': '1.2',
},
HOST: {
'name': 'Host',
'is_host': True
},
}
def test(self):
br1 = self.nodes[BR_1]
br2 = self.nodes[BR_2]
router = self.nodes[ROUTER_1]
host = self.nodes[HOST]
br1.set_backbone_router(reg_delay=REG_DELAY, mlr_timeout=consts.MLR_TIMEOUT_MIN)
br1.start()
self.simulator.go(config.LEADER_STARTUP_DELAY)
self.assertEqual('leader', br1.get_state())
self.assertTrue(br1.is_primary_backbone_router)
router.start()
self.simulator.go(5)
self.assertEqual('router', router.get_state())
br2.start()
self.simulator.go(5)
self.assertEqual('router', br2.get_state())
self.assertFalse(br2.is_primary_backbone_router)
host.start(start_radvd=False)
self.simulator.go(10)
# Router_1 registers for multicast address, MA1, at BR_1.
router.add_ipmaddr(MA1)
self.simulator.go(5)
# 1. Host sends a ping packet to the multicast address, MA1.
self.assertTrue(
host.ping(MA1, backbone=True, ttl=10, interface=host.get_ip6_address(config.ADDRESS_TYPE.ONLINK_ULA)[0]))
self.simulator.go(5)
# 3a. By internal means, Router_1 stop listening to the multicast
# address, MA1.
router.del_ipmaddr(MA1)
# 4. After (MLR_TIMEOUT_MIN+2) seconds
self.simulator.go(consts.MLR_TIMEOUT_MIN + 2)
# 5. Host sends a ping packet to the multicast address, MA1.
self.assertFalse(
host.ping(MA1, backbone=True, ttl=10, interface=host.get_ip6_address(config.ADDRESS_TYPE.ONLINK_ULA)[0]))
self.simulator.go(5)
# 6a. Router_1 registers for multicast address, MA1, at BR_1.
router.add_ipmaddr(MA1)
self.simulator.go(5)
# 7. Host sends a ping packet to MA1.
self.assertTrue(
host.ping(MA1, backbone=True, ttl=10, interface=host.get_ip6_address(config.ADDRESS_TYPE.ONLINK_ULA)[0]))
self.collect_ipaddrs()
self.collect_rloc16s()
self.collect_rlocs()
self.collect_leader_aloc(BR_1)
self.collect_extra_vars()
def verify(self, pv: pktverify.packet_verifier.PacketVerifier):
pkts = pv.pkts
vars = pv.vars
pv.summary.show()
# Ensure the topology is formed correctly
pv.verify_attached('Router_1', 'BR_1')
pv.verify_attached('BR_2')
# 1. Host sends a ping packet to the multicast address, MA1.
_pkt = pkts.filter_eth_src(vars['Host_ETH']) \
.filter_ipv6_dst(MA1) \
.filter_ping_request() \
.must_next()
# 2. BR_1 forwards the ping packet with multicast address, MA1, to its
# Thread Network encapsulated in an MPL packet.
pkts.filter_wpan_src64(vars['BR_1']) \
.filter_AMPLFMA(vars['BR_1_RLOC']) \
.filter_ping_request(identifier=_pkt.icmpv6.echo.identifier) \
.must_next()
# 3. Router_1 receives the MPL packet containing an encapsulated ping
# packet to MA1, sent by Host, and unicasts a ping response packet back
# to Host.
pkts.filter_wpan_src64(vars['Router_1']) \
.filter_ipv6_dst(_pkt.ipv6.src) \
.filter_ping_reply(identifier=_pkt.icmpv6.echo.identifier) \
.must_next()
# 4. After (MLR_TIMEOUT_MIN+2) seconds, BR_1 multicasts an MLDv2 message
# of type “Version 2 Multicast Listener Report” (see [RFC 3810] Section
# 5.2)
# TODO
# 5. Host sends a ping packet to the multicast address, MA1.
_pkt = pkts.filter_eth_src(vars['Host_ETH']) \
.filter_ipv6_dst(MA1) \
.filter_ping_request() \
.must_next()
# 6. BR_1 does not forward the ping packet with multicast address, MA1,
# to its Thread Network.
pkts.filter_wpan_src64(vars['BR_1']) \
.filter_AMPLFMA(vars['BR_1_RLOC']) \
.filter_ping_request(identifier=_pkt.icmpv6.echo.identifier) \
.must_not_next()
# 7. Host sends a ping packet to the multicast address, MA1.
_pkt = pkts.filter_eth_src(vars['Host_ETH']) \
.filter_ipv6_dst(MA1) \
.filter_ping_request() \
.must_next()
# 8. BR_1 forwards the ping packet with multicast address, MA1, to its
# Thread Network encapsulated in an MPL packet.
pkts.filter_wpan_src64(vars['BR_1']) \
.filter_AMPLFMA(vars['BR_1_RLOC']) \
.filter_ping_request(identifier=_pkt.icmpv6.echo.identifier) \
.must_next()
# 9. Router_1 receives the MPL packet containing an encapsulated ping
# packet to MA1, sent by Host, and unicasts a ping response packet back
# to Host.
pkts.filter_wpan_src64(vars['Router_1']) \
.filter_ipv6_dst(_pkt.ipv6.src) \
.filter_ping_reply(identifier=_pkt.icmpv6.echo.identifier) \
.must_next()
if __name__ == '__main__':
unittest.main()
tests/scripts/thread-cert/border_router/MATN/MATN_05_ReregistrationToSameMulticastGroup.py
-242
View File
@@ -1,242 +0,0 @@
#!/usr/bin/env python3
#
# Copyright (c) 2021, The OpenThread Authors.
# All rights reserved.
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions are met:
# 1. Redistributions of source code must retain the above copyright
# notice, this list of conditions and the following disclaimer.
# 2. Redistributions in binary form must reproduce the above copyright
# notice, this list of conditions and the following disclaimer in the
# documentation and/or other materials provided with the distribution.
# 3. Neither the name of the copyright holder nor the
# names of its contributors may be used to endorse or promote products
# derived from this software without specific prior written permission.
#
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS 'AS IS'
# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
# ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
# LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
# CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
# SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
# CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
# POSSIBILITY OF SUCH DAMAGE.
#
import logging
import unittest
import pktverify
from pktverify import packet_verifier, packet_filter, consts
from pktverify.consts import MA1, PBBR_ALOC
import config
import thread_cert
# Test description:
# The purpose of this test case is to verify that a Primary BBR (DUT) can manage
# a re-registration of a device on its network to remain receiving multicasts.
# The test also verifies the usage of UDP multicast packets across backbone and
# internal Thread network.
#
# Topology:
# ----------------(eth)------------------
# | | |
# BR_1 (Leader) ---- BR_2 HOST
# | |
# | |
# Router_1 -----------+
#
BR_1 = 1
BR_2 = 2
ROUTER_1 = 3
HOST = 4
REG_DELAY = 10
UDP_HEADER_LENGTH = 8
class MATN_05_ReregistrationToSameMulticastGroup(thread_cert.TestCase):
USE_MESSAGE_FACTORY = False
TOPOLOGY = {
BR_1: {
'name': 'BR_1',
'is_otbr': True,
'allowlist': [BR_2, ROUTER_1],
'version': '1.2',
},
BR_2: {
'name': 'BR_2',
'allowlist': [BR_1, ROUTER_1],
'is_otbr': True,
'version': '1.2',
},
ROUTER_1: {
'name': 'Router_1',
'allowlist': [BR_1, BR_2],
'version': '1.2',
},
HOST: {
'name': 'Host',
'is_host': True
},
}
def test(self):
br1 = self.nodes[BR_1]
br2 = self.nodes[BR_2]
router1 = self.nodes[ROUTER_1]
host = self.nodes[HOST]
br1.set_backbone_router(reg_delay=REG_DELAY, mlr_timeout=consts.MLR_TIMEOUT_MIN)
br1.start()
self.simulator.go(config.LEADER_STARTUP_DELAY)
self.assertEqual('leader', br1.get_state())
self.assertTrue(br1.is_primary_backbone_router)
router1.start()
self.simulator.go(10)
self.assertEqual('router', router1.get_state())
br2.start()
self.simulator.go(10)
self.assertEqual('router', br2.get_state())
self.assertFalse(br2.is_primary_backbone_router)
host.start(start_radvd=False)
self.simulator.go(10)
# Router_1 registers for multicast address, MA1, at BR_1.
router1.add_ipmaddr(MA1)
self.simulator.go(5)
# 1. Host sends a ping packet to the multicast address, MA1.
self.assertTrue(
host.ping(MA1, backbone=True, ttl=10, interface=host.get_ip6_address(config.ADDRESS_TYPE.ONLINK_ULA)[0]))
# Ensure Router_1 renews its multicast registration
self.simulator.go(consts.MLR_TIMEOUT_MIN - 10)
# 4. Within MLR_TIMEOUT_MIN seconds, Host sends a ping packet to the
# multicast address, MA1. The destination port 5683 is used for the UDP
# Multicast packet transmission.
host.udp_send_host(data='PING', ipaddr=MA1, port=5683)
self.simulator.go(5)
# 6a. By internal means, Router_1 stops listening to the multicast
# address, MA1.
router1.del_ipmaddr(MA1)
# 7. After (MLR_TIMEOUT_MIN+2) seconds, Host multicasts a ping packet to
# multicast address, MA1, on the backbone link.
self.simulator.go(consts.MLR_TIMEOUT_MIN + 2)
self.assertFalse(
host.ping(MA1, backbone=True, ttl=10, interface=host.get_ip6_address(config.ADDRESS_TYPE.ONLINK_ULA)[0]))
self.collect_ipaddrs()
self.collect_rloc16s()
self.collect_rlocs()
self.collect_leader_aloc(BR_1)
self.collect_extra_vars()
def verify(self, pv: pktverify.packet_verifier.PacketVerifier):
pkts = pv.pkts
vars = pv.vars
pv.summary.show()
logging.info(f'vars = {vars}')
# Ensure the topology is formed correctly
pv.verify_attached('Router_1', 'BR_1')
pv.verify_attached('BR_2')
# Initial registration
# Router_1 registers for multicast address, MA1, at BR_1.
# Router_1 unicasts an MLR.req CoAP request to BR_1 as
# "coap://[<BR_1 RLOC or PBBR ALOC>]:MM/n/mr".
# The payload contains "IPv6Address TLV: MA1".
initial_registration_pkt = pkts.filter_wpan_src64(vars['Router_1']) \
.filter_ipv6_2dsts(vars['BR_1_RLOC'], PBBR_ALOC) \
.filter_coap_request('/n/mr') \
.filter(lambda p: p.thread_meshcop.tlv.ipv6_addr == [MA1]) \
.must_next()
# 1. Host sends a ping packet to the multicast address, MA1.
_pkt = pkts.filter_eth_src(vars['Host_ETH']) \
.filter_ipv6_dst(MA1) \
.filter_ping_request() \
.must_next()
# 2. BR_1 forwards the ping packet with multicast address, MA1, to its
# Thread Network encapsulated in an MPL packet.
pkts.filter_wpan_src64(vars['BR_1']) \
.filter_AMPLFMA(mpl_seed_id=vars['BR_1_RLOC']) \
.filter_ping_request(identifier=_pkt.icmpv6.echo.identifier) \
.must_next()
# 3. Router_1 receives the MPL packet containing an encapsulated ping
# packet to MA1, sent by Host, and unicasts a ping response packet back
# to Host.
pkts.filter_wpan_src64(vars['Router_1']) \
.filter_ipv6_dst(_pkt.ipv6.src) \
.filter_ping_reply(identifier=_pkt.icmpv6.echo.identifier) \
.must_next()
# 3a. Within MLR_TIMEOUT_MIN seconds of initial registration, Router_1
# re-registers for multicast address, MA1, at BR_1.
# Router_1 unicasts an MLR.req CoAP request to BR_1 as
# "coap://[<BR_1 RLOC or PBBR ALOC>]:MM/n/mr".
# The payload contains "IPv6Address TLV: MA1".
pkts.copy() \
.filter_wpan_src64(vars['Router_1']) \
.filter_ipv6_2dsts(vars['BR_1_RLOC'], PBBR_ALOC) \
.filter_coap_request('/n/mr') \
.filter(lambda p: p.thread_meshcop.tlv.ipv6_addr == [MA1] and
p.sniff_timestamp <= initial_registration_pkt.sniff_timestamp + consts.MLR_TIMEOUT_MIN) \
.must_next()
# 4. Within MLR_TIMEOUT_MIN seconds, Host sends a ping packet to the
# multicast address, MA1. The destination port 5683 is used for the UDP
# Multicast packet transmission.
_pkt = pkts.filter_eth_src(vars['Host_ETH']) \
.filter_ipv6_dst(MA1) \
.filter(lambda p: p.udp.length == UDP_HEADER_LENGTH + len('PING')
and p.udp.dstport == 5683) \
.must_next()
# 5. BR_1 forwards the UDP ping packet with multicast address, MA1, to
# its Thread Network encapsulated in an MPL packet.
pkts.filter_wpan_src64(vars['BR_1']) \
.filter_AMPLFMA(mpl_seed_id=vars['BR_1_RLOC']) \
.filter(lambda p: p.udp.length == _pkt.udp.length) \
.must_next()
# 6. Router_1 receives the ping packet.
# Use the port 5683 (CoAP port) to verify that the
# UDP Multicast packet is received.
pkts.filter_wpan_src64(vars['Router_1']) \
.filter(
lambda p: p.udp.length == _pkt.udp.length and p.udp.dstport == 5683) \
.must_next()
# 7. After (MLR_TIMEOUT_MIN+2) seconds, Host multicasts a ping packet to
# multicast address, MA1, on the backbone link.
_pkt = pkts.filter_eth_src(vars['Host_ETH']) \
.filter_ipv6_dst(MA1) \
.filter_ping_request() \
.must_next()
# 8. BR_1 does not forward the ping packet with multicast address, MA1,
# to its Thread Network.
pkts.filter_wpan_src64(vars['BR_1']) \
.filter_AMPLFMA(mpl_seed_id=vars['BR_1_RLOC']) \
.filter_ping_request(identifier=_pkt.icmpv6.echo.identifier) \
.must_not_next()
if __name__ == '__main__':
unittest.main()
tests/scripts/thread-cert/border_router/MATN/MATN_09_DefaultBRMulticastForwarding.py
-191
View File
@@ -1,191 +0,0 @@
#!/usr/bin/env python3
#
# Copyright (c) 2021, The OpenThread Authors.
# All rights reserved.
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions are met:
# 1. Redistributions of source code must retain the above copyright
# notice, this list of conditions and the following disclaimer.
# 2. Redistributions in binary form must reproduce the above copyright
# notice, this list of conditions and the following disclaimer in the
# documentation and/or other materials provided with the distribution.
# 3. Neither the name of the copyright holder nor the
# names of its contributors may be used to endorse or promote products
# derived from this software without specific prior written permission.
#
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS 'AS IS'
# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
# ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
# LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
# CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
# SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
# CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
# POSSIBILITY OF SUCH DAMAGE.
#
import logging
import unittest
import pktverify
from pktverify import packet_verifier, packet_filter, consts
from pktverify.consts import MA1
import config
import thread_cert
# Test description:
# The purpose of this test case is to verify that a Secondary BBR can take over
# forwarding of outbound multicast transmissions from the Thread Network when
# the Primary BBR disconnects. The Secondary in that case becomes Primary.
#
# Topology:
# ----------------(eth)------------------
# | |
# BR_1 (Leader) ---- BR_2
# | |
# | |
# ROUTER_1 -----------+
#
from pktverify.null_field import nullField
BR_1 = 1
BR_2 = 2
ROUTER_1 = 3
class MATN_09_FailureOfPrimaryBBROutboundMulticast(thread_cert.TestCase):
USE_MESSAGE_FACTORY = False
TOPOLOGY = {
BR_1: {
'name': 'BR_1',
'is_otbr': True,
'allowlist': [BR_2, ROUTER_1],
'version': '1.2',
},
BR_2: {
'name': 'BR_2',
'is_otbr': True,
'allowlist': [BR_1, ROUTER_1],
'version': '1.2',
},
ROUTER_1: {
'name': 'Router_1',
'allowlist': [BR_1, BR_2],
'version': '1.2',
'partition_id': 1,
'network_id_timeout': 150,
},
}
def test(self):
br1 = self.nodes[BR_1]
br2 = self.nodes[BR_2]
router1 = self.nodes[ROUTER_1]
br1.start()
self.simulator.go(config.LEADER_STARTUP_DELAY)
self.assertEqual('leader', br1.get_state())
self.assertTrue(br1.is_primary_backbone_router)
router1.start()
self.simulator.go(10)
self.assertEqual('router', router1.get_state())
br2.start()
self.simulator.go(5)
self.assertEqual('router', br2.get_state())
self.assertFalse(br2.is_primary_backbone_router)
# 1. Router sends a ping packet to the multicast address, MA1,
# encapsulated in an MPL packet
self.assertFalse(router1.ping(MA1))
self.simulator.go(5)
# 4a. Switch off BR_1
br1.disable_backbone_router()
br1.thread_stop()
br1.interface_down()
self.simulator.go(180)
# 4b. BR_2 detects the missing Primary BBR and becomes the Leader of
# the Thread Network.
self.assertEqual('disabled', br1.get_state())
self.assertEqual('leader', br2.get_state())
self.assertEqual('router', router1.get_state())
self.assertFalse(br1.is_primary_backbone_router)
self.assertTrue(br2.is_primary_backbone_router)
# 5. Router_1 sends a ping packet to the multicast address, MA1,
# encapsulated in an MPL packet.
self.assertFalse(router1.ping(MA1))
self.collect_ipaddrs()
self.collect_rloc16s()
self.collect_rlocs()
self.collect_leader_aloc(BR_2)
self.collect_extra_vars()
def verify(self, pv: pktverify.packet_verifier.PacketVerifier):
pkts = pv.pkts
vars = pv.vars
pv.summary.show()
logging.info(f'vars = {vars}')
# Ensure the topology is formed correctly
pv.verify_attached('Router_1', 'BR_1')
pv.verify_attached('BR_2')
# 1. Router_1 sends a ping packet to the multicast address, MA1,
# encapsulated in an MPL packet.
_pkt = pkts.filter_wpan_src64(vars['Router_1']) \
.filter_AMPLFMA(mpl_seed_id=vars['Router_1_RLOC']) \
.filter_ping_request() \
.must_next()
initial_identifier = _pkt.icmpv6.echo.identifier
# 2. BR_1 forwards the multicast ping packet with multicast address,
# MA1, to the LAN.
pkts.filter_eth_src(vars['BR_1_ETH']) \
.filter_ipv6_dst(MA1) \
.filter_ping_request(identifier=_pkt.icmpv6.echo.identifier) \
.must_next()
with pkts.save_index():
pv.verify_attached('Router_1', 'BR_2')
# 4b. BR_2 detects the missing Primary BBR and becomes the Leader of the
# Thread Network, distributing its BBR dataset.
# Verify that Router_1 has received the new BBR Dataset from BR_2,
# where:
# • RLOC16 in Server TLV == The RLOC16 of BR_2
# All fields in the Service TLV contain valid values.
pkts.filter_wpan_src64(vars['BR_2']) \
.filter_mle() \
.filter(
lambda p: p.thread_nwd.tlv.server_16 is not nullField and
vars['BR_2_RLOC16'] in p.thread_nwd.tlv.server_16) \
.must_next()
# 5.Router_1 sends a ping packet to the multicast address, MA1,
# encapsulated in an MPL packet.
_pkt = pkts.filter_wpan_src64(vars['Router_1']) \
.filter_AMPLFMA(mpl_seed_id=vars['Router_1_RLOC']) \
.filter_ping_request() \
.filter(lambda p: p.icmpv6.echo.identifier != initial_identifier) \
.must_next()
# 6. BR_2 forwards the multicast ping packet with multicast address,
# MA1, to the LAN.
pkts.filter_eth_src(vars['BR_2_ETH']) \
.filter_ipv6_dst(MA1) \
.filter_ping_request(identifier=_pkt.icmpv6.echo.identifier) \
.must_next()
if __name__ == '__main__':
unittest.main()
tests/scripts/thread-cert/border_router/MATN/MATN_12_HopLimitProcessing.py
-276
View File
@@ -1,276 +0,0 @@
#!/usr/bin/env python3
#
# Copyright (c) 2021, The OpenThread Authors.
# All rights reserved.
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions are met:
# 1. Redistributions of source code must retain the above copyright
# notice, this list of conditions and the following disclaimer.
# 2. Redistributions in binary form must reproduce the above copyright
# notice, this list of conditions and the following disclaimer in the
# documentation and/or other materials provided with the distribution.
# 3. Neither the name of the copyright holder nor the
# names of its contributors may be used to endorse or promote products
# derived from this software without specific prior written permission.
#
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS 'AS IS'
# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
# ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
# LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
# CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
# SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
# CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
# POSSIBILITY OF SUCH DAMAGE.
#
import logging
import unittest
import pktverify
from pktverify import packet_verifier, packet_filter, consts
from pktverify.consts import MA1, MA2
import config
import thread_cert
# Test description:
# The purpose of this test case is to verify that a Primary BBR correctly
# decrements the Hop Limit field by 1 if packet is forwarded with MPL, and if
# forwarded from Thread Network (using MPL) to LAN. This test also verifies that
# the BR drops a packet with Hop Limit 0. It also checks the use of IPv6 packets
# that span multiple 6LoWPAN fragments.
#
# Topology:
# ----------------(eth)------------------
# | |
# BR_1 (Leader) |
# | HOST
# |
# ROUTER_1
#
BR_1 = 1
ROUTER_1 = 2
HOST = 3
ICMP_HEADER_LEN = 8
class MATN_12_HopLimitProcessing(thread_cert.TestCase):
USE_MESSAGE_FACTORY = False
TOPOLOGY = {
BR_1: {
'name': 'BR_1',
'is_otbr': True,
'allowlist': [ROUTER_1],
'version': '1.2',
},
ROUTER_1: {
'name': 'Router_1',
'allowlist': [BR_1],
'version': '1.2',
},
HOST: {
'name': 'Host',
'is_host': True,
},
}
def test(self):
br1 = self.nodes[BR_1]
router1 = self.nodes[ROUTER_1]
host = self.nodes[HOST]
br1.start()
self.simulator.go(config.LEADER_STARTUP_DELAY)
self.assertEqual('leader', br1.get_state())
self.assertTrue(br1.is_primary_backbone_router)
router1.start()
self.simulator.go(5)
self.assertEqual('router', router1.get_state())
host.start(start_radvd=False)
self.simulator.go(10)
# Router_1 registers a multicast address, MA1.
router1.add_ipmaddr(MA1)
self.simulator.go(5)
# 1. Host multicasts a ping packet to the multicast address, MA1, with
# the IPv6 Hop Limit field set to 59. The size of the payload is 130
# bytes.
self.assertTrue(
host.ping(MA1,
backbone=True,
ttl=59,
size=130,
interface=host.get_ip6_address(config.ADDRESS_TYPE.ONLINK_ULA)[0]))
self.simulator.go(5)
# 4. Host multicasts a ping packet to the multicast address, MA1, with
# the IPv6 Hop Limit field set to 1. The size of the payload is 130
# bytes.
host.ping(MA1,
backbone=True,
ttl=1,
size=130,
interface=host.get_ip6_address(config.ADDRESS_TYPE.ONLINK_ULA)[0])
self.simulator.go(5)
# 6. Router_1 sends a ping packet encapsulated in an MPL packet to the
# multicast address, MA2, with the Hop Limit field of the inner packet
# set to 159. The size of the payload is 130 bytes.
self.assertFalse(router1.ping(MA2, hoplimit=159, size=130))
self.simulator.go(5)
# 8. Router_1 sends a ping packet encapsulated in an MPL multicast
# packet to the multicast address, MA2, with the Hop Limit field of
# the inner (encapsulated) packet set to 2. The size of the payload is
# 130 bytes.
self.assertFalse(router1.ping(MA2, hoplimit=2, size=130))
self.simulator.go(5)
# 10. Router_1 sends a ping packet encapsulated in an MPL multicast
# packet to the multicast address, MA2, with the Hop Limit field of the
# inner packet set to 1. The size of the payload is 130 bytes.
self.assertFalse(router1.ping(MA2, hoplimit=1, size=130))
self.simulator.go(5)
# 11. Router_1 sends a ping packet encapsulated in an MPL multicast
# packet to the multicast address, MA2, with the Hop Limit field of the
# inner packet set to 0.
self.assertFalse(router1.ping(MA2, hoplimit=0, size=130))
self.simulator.go(5)
self.collect_ipaddrs()
self.collect_rloc16s()
self.collect_rlocs()
self.collect_extra_vars()
def verify(self, pv: pktverify.packet_verifier.PacketVerifier):
pkts = pv.pkts
vars = pv.vars
pv.summary.show()
logging.info(f'vars = {vars}')
# Ensure the topology is formed correctly
pv.verify_attached('Router_1', 'BR_1')
# 1. Host multicasts a ping packet to the multicast address, MA1, with
# the IPv6 Hop Limit field set to 59. The size of the payload is 130
# bytes.
_pkt = pkts.filter_eth_src(vars['Host_ETH']) \
.filter_ipv6_dst(MA1) \
.filter_ping_request() \
.filter(lambda
p: p.ipv6.hlim == 59 and p.ipv6.plen == 130 + ICMP_HEADER_LEN) \
.must_next()
# 2. BR_1 forwards the ping packet to Router_1 as an MPL packet
# encapsulating the IPv6 packet with the Hop Limit field of the inner
# packet set to 58.
_pkt2 = pkts.filter_wpan_src64(vars['BR_1']) \
.filter_AMPLFMA(mpl_seed_id=vars['BR_1_RLOC']) \
.filter_ping_request(identifier=_pkt.icmpv6.echo.identifier) \
.filter(lambda
p: p.ipv6inner.hlim == 58 and p.ipv6inner.plen == 130 + ICMP_HEADER_LEN) \
.must_next()
# 3. Router_1 receives the multicast ping packet.
pkts.filter_wpan_src64(vars['Router_1']) \
.filter_ipv6_dst(_pkt.ipv6.src) \
.filter_ping_reply(identifier=_pkt.icmpv6.echo.identifier) \
.must_next()
# 4. Host multicasts a ping packet to the multicast address, MA1, with
# the IPv6 Hop Limit field set to 1. The size of the payload is 130
# bytes.
_pkt = pkts.filter_eth_src(vars['Host_ETH']) \
.filter_ipv6_dst(MA1) \
.filter_ping_request() \
.filter(
lambda
p: p.ipv6.hlim == 1 and p.ipv6.plen == 130 + ICMP_HEADER_LEN) \
.must_next()
# 5. BR_1 does not forward the ping packet.
pkts.filter_wpan_src64(vars['BR_1']) \
.filter_AMPLFMA(mpl_seed_id=vars['BR_1']) \
.filter_ping_request(identifier=_pkt.icmpv6.echo.identifier) \
.must_not_next()
# 6. Router_1 sends a ping packet encapsulated in an MPL packet to the
# multicast address, MA2, with the Hop Limit field of the inner
# packet set to 159. The size of the payload is 130 bytes.
_pkt = pkts.filter_wpan_src64(vars['Router_1']) \
.filter_AMPLFMA(mpl_seed_id=vars['Router_1_RLOC']) \
.filter_ping_request() \
.filter(lambda p: p.ipv6inner.dst == MA2 and
p.ipv6inner.hlim == 159 and
p.ipv6inner.plen == 130 + ICMP_HEADER_LEN) \
.must_next()
# 7. BR_1 forwards the multicast ping packet to the LAN with the Hop
# Limit field set to 158.
pkts.filter_eth_src(vars['BR_1_ETH']) \
.filter_ping_request(identifier=_pkt.icmpv6.echo.identifier) \
.filter(lambda p: p.ipv6.hlim == 158) \
.must_next()
# 8. Router_1 sends a ping packet encapsulated in an MPL multicast
# packet to the multicast address, MA2, with the Hop Limit field of the
# inner packet set to 2. The size of the payload is 130 bytes
_pkt = pkts.filter_wpan_src64(vars['Router_1']) \
.filter_AMPLFMA(mpl_seed_id=vars['Router_1_RLOC']) \
.filter_ping_request() \
.filter(lambda p: p.ipv6inner.dst == MA2 and
p.ipv6inner.hlim == 2 and
p.ipv6inner.plen == 130 + ICMP_HEADER_LEN) \
.must_next()
# 9. BR_1 forwards the multicast packet to the LAN with the Hop Limit
# field set to 1.
pkts.filter_eth_src(vars['BR_1_ETH']) \
.filter_ping_request(identifier=_pkt.icmpv6.echo.identifier) \
.filter(lambda p: p.ipv6.hlim == 1) \
.must_next()
# 10. Router_1 sends a ping packet encapsulated in an MPL multicast
# packet to the multicast address, MA2, with the Hop Limit field of the
# inner packet set to 1. The size of the payload is 130 bytes.
_pkt = pkts.filter_wpan_src64(vars['Router_1']) \
.filter_AMPLFMA(mpl_seed_id=vars['Router_1_RLOC']) \
.filter_ping_request() \
.filter(lambda p: p.ipv6inner.dst == MA2 and
p.ipv6inner.hlim == 1 and
p.ipv6inner.plen == 130 + ICMP_HEADER_LEN) \
.must_next()
# 11. BR_1 does not forward the ping packet to the LAN.
pkts.filter_eth_src(vars['BR_1_ETH']) \
.filter_ping_request(identifier=_pkt.icmpv6.echo.identifier) \
.must_not_next()
# 12. Router_1 sends a ping packet encapsulated in an MPL multicast
# packet to the multicast address, MA2, with the Hop Limit field of the
# inner packet set to 0.
_pkt = pkts.filter_wpan_src64(vars['Router_1']) \
.filter_AMPLFMA(mpl_seed_id=vars['Router_1_RLOC']) \
.filter_ping_request() \
.filter(lambda p: p.ipv6inner.dst == MA2 and
p.ipv6inner.hlim == 0 and
p.ipv6inner.plen == 130 + ICMP_HEADER_LEN) \
.must_next()
# 13. BR_1 does not forward the ping packet to the LAN.
pkts.filter_eth_src(vars['BR_1_ETH']) \
.filter_ping_request(identifier=_pkt.icmpv6.echo.identifier) \
.must_not_next()
if __name__ == '__main__':
unittest.main()
tests/scripts/thread-cert/border_router/MATN/MATN_15_ChangeOfPrimaryBBRTriggersRegistration.py
-199
View File
@@ -1,199 +0,0 @@
#!/usr/bin/env python3
#
# Copyright (c) 2021, The OpenThread Authors.
# All rights reserved.
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions are met:
# 1. Redistributions of source code must retain the above copyright
# notice, this list of conditions and the following disclaimer.
# 2. Redistributions in binary form must reproduce the above copyright
# notice, this list of conditions and the following disclaimer in the
# documentation and/or other materials provided with the distribution.
# 3. Neither the name of the copyright holder nor the
# names of its contributors may be used to endorse or promote products
# derived from this software without specific prior written permission.
#
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS 'AS IS'
# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
# ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
# LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
# CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
# SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
# CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
# POSSIBILITY OF SUCH DAMAGE.
#
import logging
import unittest
import pktverify
from pktverify import packet_verifier, packet_filter, consts
from pktverify.consts import MA1, MA2
import config
import thread_cert
# Test description:
# The purpose of this test case is to verify that a Thread End Device detects a
# change of Primary BBR device and triggers a re-registration to its multicast
# groups.
#
# Topology:
#
#
# ----------------(eth)--------------------
# |
# BR_1 (Leader) -----BR_2
# | |
# | |
# | |
# Router_1-------------+
# |
# |
# TD
#
BR_1 = 1
BR_2 = 2
ROUTER_1 = 3
TD = 4
REG_DELAY = 10
NETWORK_ID_TIMEOUT = 120
WAIT_TIME_ALLOWANCE = 60
class MATN_15_ChangeOfPrimaryBBRTriggersRegistration(thread_cert.TestCase):
USE_MESSAGE_FACTORY = False
TOPOLOGY = {
BR_1: {
'name': 'BR_1',
'is_otbr': True,
'allowlist': [BR_2, ROUTER_1],
'version': '1.2',
},
BR_2: {
'name': 'BR_2',
'is_otbr': True,
'allowlist': [BR_1, ROUTER_1],
'version': '1.2',
},
ROUTER_1: {
'name': 'Router_1',
'allowlist': [BR_1, BR_2, TD],
'version': '1.2',
'partition_id': 1,
},
TD: {
'name': 'TD',
'allowlist': [ROUTER_1],
'router_eligible': False,
'version': '1.2',
'partition_id': 1,
},
}
def test(self):
br1 = self.nodes[BR_1]
br2 = self.nodes[BR_2]
router1 = self.nodes[ROUTER_1]
td = self.nodes[TD]
br1.set_backbone_router(reg_delay=REG_DELAY, mlr_timeout=consts.MLR_TIMEOUT_MIN)
br1.start()
self.simulator.go(config.LEADER_STARTUP_DELAY)
self.assertEqual('leader', br1.get_state())
self.assertTrue(br1.is_primary_backbone_router)
br2.set_backbone_router(reg_delay=REG_DELAY, mlr_timeout=consts.MLR_TIMEOUT_MIN)
router1.start()
self.simulator.go(5)
self.assertEqual('router', router1.get_state())
td.start()
self.simulator.go(5)
self.assertEqual('child', td.get_state())
br2.start()
self.simulator.go(5)
self.assertEqual('router', br2.get_state())
self.assertFalse(br2.is_primary_backbone_router)
# TD registers for a multicast address, MA1.
td.add_ipmaddr(MA1)
self.simulator.go(20)
# 1. Disable BR_1 such that BR_2 becomes the Primary BBR for the Thread
# Network.
br1.disable_backbone_router()
br1.thread_stop()
br1.interface_down()
self.simulator.go(NETWORK_ID_TIMEOUT + WAIT_TIME_ALLOWANCE)
# Make sure that BR_2 becomes the primary BBR
self.assertEqual('disabled', br1.get_state())
self.assertFalse(br1.is_primary_backbone_router)
self.assertTrue(br2.is_primary_backbone_router)
self.collect_ipaddrs()
self.collect_rloc16s()
self.collect_rlocs()
self.collect_extra_vars()
def verify(self, pv: pktverify.packet_verifier.PacketVerifier):
pkts = pv.pkts
vars = pv.vars
pv.summary.show()
# Ensure the topology is formed correctly
pv.verify_attached('Router_1', 'BR_1')
# 2. TD automatically detects the Primary BBR change and registers for
# multicast address, MA1, at BR_2.
# TD unicasts an MLR.req CoAP request as follows:
# coap://[<BR_2 RLOC or PBBR ALOC>]:MM/n/mr
# Where the payload contains:
# IPv6 Addresses TLV: MA1
_pkt = pkts.filter_wpan_src64(vars['TD']) \
.filter_ipv6_2dsts(vars['BR_2_RLOC'], consts.PBBR_ALOC) \
.filter_coap_request('/n/mr') \
.filter(lambda p: p.thread_meshcop.tlv.ipv6_addr == [MA1]) \
.must_next()
# 3. Router_1 forwards the registration request to BR_2.
pkts.filter_wpan_src64(vars['Router_1']) \
.filter_ipv6_2dsts(vars['BR_2_RLOC'], consts.PBBR_ALOC) \
.filter_coap_request('/n/mr') \
.filter(lambda
p: p.thread_meshcop.tlv.ipv6_addr == [MA1] and
p.coap.mid == _pkt.coap.mid) \
.must_next()
# 4. BR_2 unicasts an MLR.rsp CoAP response to TD as: 2.04 changed.
# Where the payload contains:
# Status TLV: ST_MLR_SUCCESS
pkts.filter_wpan_src64(vars['BR_2']) \
.filter_ipv6_dst(_pkt.ipv6.src) \
.filter_coap_ack('/n/mr') \
.filter(lambda
p: p.coap.mid == _pkt.coap.mid and
p.thread_nm.tlv.status == 0) \
.must_next()
# 5. Router_1 forwards the response to TD.
pkts.filter_wpan_src64(vars['Router_1']) \
.filter_ipv6_dst(_pkt.ipv6.src) \
.filter_coap_ack('/n/mr') \
.filter(lambda
p: p.coap.mid == _pkt.coap.mid and
p.thread_nm.tlv.status == 0) \
.must_next()
if __name__ == '__main__':
unittest.main()
tests/scripts/thread-cert/border_router/MATN/MATN_16_LargeNumberOfMulticastGroupSubscriptionsToBBR.py
-229
View File
@@ -1,229 +0,0 @@
#!/usr/bin/env python3
#
# Copyright (c) 2021, The OpenThread Authors.
# All rights reserved.
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions are met:
# 1. Redistributions of source code must retain the above copyright
# notice, this list of conditions and the following disclaimer.
# 2. Redistributions in binary form must reproduce the above copyright
# notice, this list of conditions and the following disclaimer in the
# documentation and/or other materials provided with the distribution.
# 3. Neither the name of the copyright holder nor the
# names of its contributors may be used to endorse or promote products
# derived from this software without specific prior written permission.
#
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS 'AS IS'
# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
# ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
# LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
# CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
# SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
# CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
# POSSIBILITY OF SUCH DAMAGE.
#
import logging
import unittest
import pktverify
from pktverify import packet_verifier, packet_filter, consts
from pktverify.consts import MAS, MA2
import config
import thread_cert
# Test description:
# The purpose of this test case is to verify that the Primary BBR can handle a
# large number (75) of subscriptions to different multicast groups. Multicast
# registrations are performed each time with 15 multicast addresses per
# registration request.
#
# Topology:
#
#
# ----------------(eth)--------------------
# | |
# BR_1 (Leader) |
# | Host
# |
# |
# Router_1
#
#
#
BR_1 = 1
ROUTER_1 = 2
HOST = 3
REG_DELAY = 10
class MATN_16_LargeNumberOfMulticastGroupSubscriptionsToBBR(thread_cert.TestCase):
USE_MESSAGE_FACTORY = False
TOPOLOGY = {
BR_1: {
'name': 'BR_1',
'is_otbr': True,
'allowlist': [ROUTER_1],
'version': '1.2',
},
ROUTER_1: {
'name': 'Router_1',
'allowlist': [BR_1],
'version': '1.2',
},
HOST: {
'name': 'Host',
'is_host': True,
},
}
def test(self):
br1 = self.nodes[BR_1]
router1 = self.nodes[ROUTER_1]
host = self.nodes[HOST]
br1.set_backbone_router(reg_delay=REG_DELAY, mlr_timeout=consts.MLR_TIMEOUT_MIN)
br1.start()
self.simulator.go(config.LEADER_STARTUP_DELAY)
self.assertEqual('leader', br1.get_state())
self.assertTrue(br1.is_primary_backbone_router)
router1.start()
self.simulator.go(5)
self.assertEqual('router', router1.get_state())
host.start(start_radvd=False)
self.simulator.go(5)
# Loop steps 1 to 4 with i := 1 to 5.
for i in range(1, 6):
# 1. Router_1 registers for 15 multicast addresses, MASi.
self.assertEqual((0, []), router1.register_multicast_listener(*(MAS[i][1:]),
timeout=consts.MLR_TIMEOUT_MIN))
self.simulator.go(5)
# Loop steps 5 to 6 with i := 1 to 5.
for i in range(1, 6):
# 5. Host sends a ping packet to the multicast address,
# MASi[ 3 * i - 1], on the backbone link.
host.ping(MAS[i][i * 3 - 1],
backbone=True,
ttl=64,
interface=host.get_ip6_address(config.ADDRESS_TYPE.ONLINK_ULA)[0])
self.simulator.go(3)
# 7. Host multicasts a packet to the multicast addresses, MA2.
self.assertFalse(
host.ping(MA2, backbone=True, ttl=64, interface=host.get_ip6_address(config.ADDRESS_TYPE.ONLINK_ULA)[0]))
self.simulator.go(3)
self.collect_ipaddrs()
self.collect_rloc16s()
self.collect_rlocs()
self.collect_extra_vars()
self.collect_leader_aloc(BR_1)
def verify(self, pv: pktverify.packet_verifier.PacketVerifier):
pkts = pv.pkts
vars = pv.vars
pv.summary.show()
# Ensure the topology is formed correctly
pv.verify_attached('Router_1', 'BR_1')
# Loop steps 1 to 4 with i := 1 to 5.
for i in range(1, 6):
# 1. Router_1 registers for 15 multicast addresses, MASi.
# [Automatic result:]
# Unicasts an MLR.req CoAP request to BR_1 as follows:
# coap://[<BR_1 RLOC> or <PBBR ALOC>]:MM/n/mr
# Where the payload contains:
# IPv6 Addresses TLV: MASi (15 addresses)
_pkt = pkts.filter_wpan_src64(vars['Router_1']) \
.filter_ipv6_2dsts(vars['BR_1_RLOC'], consts.PBBR_ALOC) \
.filter_coap_request('/n/mr') \
.filter(lambda p: set(p.thread_meshcop.tlv.ipv6_addr) == set(MAS[i][1:])) \
.must_next()
# 2. BR_1 Responds to the multicast registration.
# BR_1 unicasts an MLR.rsp CoAP response to Router_1 as follows:
# 2.04 changed
# Where the payload contains:
# Status TLV: ST_MLR_SUCCESS
pkts.copy().filter_wpan_src64(vars['BR_1']) \
.filter_ipv6_dst(_pkt.ipv6.src) \
.filter_coap_ack('/n/mr') \
.filter(lambda p: p.coap.mid == _pkt.coap.mid and p.thread_nm.tlv.status == 0) \
.must_next()
# 3. BR_1 informs other BBRs on the network of the multicast
# registrations.
# BR_1 multicasts a BMLR.ntf CoAP request to the Backbone Link,
# as follows:
# coap://[<All network BBRs multicast>]:BB/b/bmr
# Where the payload contains:
# IPv6 Addresses TLV: MASi (15 addresses)
# Timeout TLV: default MLR timeout of BR_1
pkts.filter_eth_src(vars['BR_1_ETH']) \
.filter_ipv6_dst(config.ALL_NETWORK_BBRS_ADDRESS) \
.filter_coap_request('/b/bmr') \
.filter(
lambda p: set(p.thread_meshcop.tlv.ipv6_addr) == set(MAS[i][1:]) and
p.thread_bl.tlv.timeout == consts.MLR_TIMEOUT_MIN) \
.must_next()
# 4. BR_1 multicasts an MLDv2 message of type “Version 2
# Multicast Listener Report” (see [RFC 3810] Section 5.2).
# Where:
# Nr of Mcast Address Records(M) >= 15
# Multicast Address Record[j]:
# Each of the j := 1 ... 15 Multicast Address Record containing an
# address of the set MASi contains the following:
# Record Type: 4 (CHANGE_TO_EXCLUDE_MODE)
# Number of Sources (N): 0
# Multicast Address: MASi[j]
# Alternatively, the DUT may also send multiple of above messages
# each with a portion of the 15 addresses MASi.
# In this case the Nr of Mcast Address Records can be < 15
# but the sum over all messages MUST be >= 15.
# TODO
# Loop steps 5 to 6 with i := 1 to 5.
for i in range(1, 6):
# 5. Host sends a ping packet to the multicast address,
# MASi[ 3 * i - 1], on the backbone link.
_pkt = pkts.filter_eth_src(vars['Host_ETH']) \
.filter_ipv6_dst(MAS[i][3 * i - 1]) \
.filter_ping_request() \
.must_next()
# 6. BR_1 forwards the ping packet of the previous step to its
# Thread Network encapsulated in an MPL packet.
pkts.copy().filter_wpan_src64(vars['BR_1']) \
.filter_AMPLFMA(mpl_seed_id=vars['BR_1_RLOC']) \
.filter_ping_request(identifier=_pkt.icmpv6.echo.identifier) \
.must_next()
# 7. Host multicasts a packet to the multicast addresses, MA2.
_pkt = pkts.filter_eth_src(vars['Host_ETH']) \
.filter_ipv6_dst(MA2) \
.filter_ping_request() \
.must_next()
# 8. BR_1 does not forward the packet to its Thread Network.
pkts.filter_wpan_src64(vars['BR_1']) \
.filter_AMPLFMA() \
.filter_ping_request(identifier=_pkt.icmpv6.echo.identifier) \
.must_not_next()
if __name__ == '__main__':
unittest.main()