diff --git a/tests/nexus/verify_5_1_10.py b/tests/nexus/verify_5_1_10.py index 14fe52e2d..58771990e 100644 --- a/tests/nexus/verify_5_1_10.py +++ b/tests/nexus/verify_5_1_10.py @@ -67,9 +67,18 @@ def verify(pv): # - Description: Setup the topology without the DUT. Verify all are sending MLE Advertisements. # - Pass Criteria: N/A print("Step 1: Leader, Router_1, Router_2") - pkts.copy().filter_mle_cmd(consts.MLE_ADVERTISEMENT).filter_wpan_src64(LEADER).must_next() - pkts.copy().filter_mle_cmd(consts.MLE_ADVERTISEMENT).filter_wpan_src64(ROUTER_1).must_next() - pkts.copy().filter_mle_cmd(consts.MLE_ADVERTISEMENT).filter_wpan_src64(ROUTER_2).must_next() + pkts.copy().\ + filter_mle_cmd(consts.MLE_ADVERTISEMENT).\ + filter_wpan_src64(LEADER).\ + must_next() + pkts.copy().\ + filter_mle_cmd(consts.MLE_ADVERTISEMENT).\ + filter_wpan_src64(ROUTER_1).\ + must_next() + pkts.copy().\ + filter_mle_cmd(consts.MLE_ADVERTISEMENT).\ + filter_wpan_src64(ROUTER_2).\ + must_next() # Step 2: Test Harness # - Description: Set RSSI of Router_2 at the DUT to -85dBm (Link Quality 2). @@ -107,8 +116,14 @@ def verify(pv): # - Pass Criteria: N/A print("Step 4: Router_1, Router_2") # Verify that we see parent responses from expected nodes. - pkts.copy().filter_mle_cmd(consts.MLE_PARENT_RESPONSE).filter_wpan_src64(ROUTER_1).must_next() - pkts.copy().filter_mle_cmd(consts.MLE_PARENT_RESPONSE).filter_wpan_src64(ROUTER_2).must_next() + pkts.copy().\ + filter_mle_cmd(consts.MLE_PARENT_RESPONSE).\ + filter_wpan_src64(ROUTER_1).\ + must_next() + pkts.copy().\ + filter_mle_cmd(consts.MLE_PARENT_RESPONSE).\ + filter_wpan_src64(ROUTER_2).\ + must_next() # Step 5: Router_3 (DUT) # - Description: DUT sends Child ID Request to Router_1. diff --git a/tests/nexus/verify_5_1_2.py b/tests/nexus/verify_5_1_2.py index 8ee67bc79..3993d6844 100644 --- a/tests/nexus/verify_5_1_2.py +++ b/tests/nexus/verify_5_1_2.py @@ -66,8 +66,12 @@ def verify(pv): # - Description: Verify topology is formed correctly # - Pass Criteria: N/A print("Step 1: Verify topology is formed correctly") - 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() + 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 2: MED_1, SED_1 # - Description: Harness silently powers-off both devices and waits for the keep-alive timeout to expire diff --git a/tests/nexus/verify_5_1_3.py b/tests/nexus/verify_5_1_3.py index f8cc23497..512258cf2 100644 --- a/tests/nexus/verify_5_1_3.py +++ b/tests/nexus/verify_5_1_3.py @@ -73,8 +73,10 @@ def verify(pv): # - Description: Verify topology is formed correctly # - Pass Criteria: N/A print("Step 2: Verifying topology formation.") - pkts.filter_wpan_src64(LEADER).filter_mle_cmd( - consts.MLE_ADVERTISEMENT).filter(lambda p: p.mle.tlv.leader_data.partition_id == P1_ID).must_next() + pkts.filter_wpan_src64(LEADER).\ + filter_mle_cmd(consts.MLE_ADVERTISEMENT).\ + filter(lambda p: p.mle.tlv.leader_data.partition_id == P1_ID).\ + must_next() # Step 3: Leader # - Description: Harness silently powers-off the Leader diff --git a/tests/nexus/verify_5_1_4.py b/tests/nexus/verify_5_1_4.py index fc519bc31..94e8f5beb 100644 --- a/tests/nexus/verify_5_1_4.py +++ b/tests/nexus/verify_5_1_4.py @@ -77,8 +77,10 @@ def verify(pv): # - Description: Verify topology is formed correctly # - Pass Criteria: N/A print("Step 3: Verifying topology formation.") - pkt = pkts.filter_wpan_src64(LEADER).filter_mle_cmd( - consts.MLE_ADVERTISEMENT).filter(lambda p: p.mle.tlv.leader_data.partition_id == P1_ID).must_next() + pkt = pkts.filter_wpan_src64(LEADER).\ + filter_mle_cmd(consts.MLE_ADVERTISEMENT).\ + filter(lambda p: p.mle.tlv.leader_data.partition_id == P1_ID).\ + must_next() original_leader_data = pkt.mle.tlv.leader_data # Step 4: Leader diff --git a/tests/nexus/verify_5_1_5.py b/tests/nexus/verify_5_1_5.py index 679175dd6..9fc9c1a82 100644 --- a/tests/nexus/verify_5_1_5.py +++ b/tests/nexus/verify_5_1_5.py @@ -69,15 +69,15 @@ def verify(pv): print("Step 1: Verify topology is formed correctly") # First attach of Router_1 - _pkt = pkts.filter_wpan_src64(ROUTER_1). \ - filter_wpan_dst16(LEADER_RLOC16). \ - filter_coap_request(consts.ADDR_SOL_URI). \ - must_next() + _pkt = pkts.filter_wpan_src64(ROUTER_1).\ + filter_wpan_dst16(LEADER_RLOC16).\ + filter_coap_request(consts.ADDR_SOL_URI).\ + must_next() - _pkt_res = pkts.filter_wpan_src64(LEADER). \ - filter_wpan_dst16(_pkt.wpan.src16). \ - filter_coap_ack(consts.ADDR_SOL_URI). \ - must_next() + _pkt_res = pkts.filter_wpan_src64(LEADER).\ + filter_wpan_dst16(_pkt.wpan.src16).\ + filter_coap_ack(consts.ADDR_SOL_URI).\ + must_next() first_router_id = _pkt_res.coap.tlv.rloc16 >> ROUTER_ID_OFFSET print(f"Router_1 first Router ID: {first_router_id}") @@ -107,23 +107,23 @@ def verify(pv): # - Router Mask TLV print("Step 4: Leader (DUT) automatically attaches Router_1 and reassigns a different Router ID") - _pkt = pkts.filter_wpan_src64(ROUTER_1). \ - filter_wpan_dst16(LEADER_RLOC16). \ - filter_coap_request(consts.ADDR_SOL_URI). \ - filter(lambda p: p.coap.tlv.rloc16 >> ROUTER_ID_OFFSET == first_router_id). \ - must_next() + _pkt = pkts.filter_wpan_src64(ROUTER_1).\ + filter_wpan_dst16(LEADER_RLOC16).\ + filter_coap_request(consts.ADDR_SOL_URI).\ + filter(lambda p: p.coap.tlv.rloc16 >> ROUTER_ID_OFFSET == first_router_id).\ + must_next() - _pkt_res = pkts.filter_wpan_src64(LEADER). \ - filter_wpan_dst16(_pkt.wpan.src16). \ - filter_coap_ack(consts.ADDR_SOL_URI). \ - filter(lambda p: { - consts.NL_STATUS_TLV, - consts.NL_RLOC16_TLV, - consts.NL_ROUTER_MASK_TLV - } <= set(p.coap.tlv.type) and - p.coap.code == consts.COAP_CODE_ACK and - p.coap.tlv.status == 0). \ - must_next() + _pkt_res = pkts.filter_wpan_src64(LEADER).\ + filter_wpan_dst16(_pkt.wpan.src16).\ + filter_coap_ack(consts.ADDR_SOL_URI).\ + filter(lambda p: { + consts.NL_STATUS_TLV, + consts.NL_RLOC16_TLV, + consts.NL_ROUTER_MASK_TLV + } <= set(p.coap.tlv.type) and + p.coap.code == consts.COAP_CODE_ACK and + p.coap.tlv.status == 0).\ + must_next() second_router_id = _pkt_res.coap.tlv.rloc16 >> ROUTER_ID_OFFSET print(f"Router_1 second Router ID: {second_router_id}") @@ -153,23 +153,23 @@ def verify(pv): # - Router Mask TLV print("Step 7: Leader (DUT) automatically attaches Router_1 and reassigns the requested Router ID") - _pkt = pkts.filter_wpan_src64(ROUTER_1). \ - filter_wpan_dst16(LEADER_RLOC16). \ - filter_coap_request(consts.ADDR_SOL_URI). \ - filter(lambda p: p.coap.tlv.rloc16 >> ROUTER_ID_OFFSET == second_router_id). \ - must_next() + _pkt = pkts.filter_wpan_src64(ROUTER_1).\ + filter_wpan_dst16(LEADER_RLOC16).\ + filter_coap_request(consts.ADDR_SOL_URI).\ + filter(lambda p: p.coap.tlv.rloc16 >> ROUTER_ID_OFFSET == second_router_id).\ + must_next() - _pkt_res = pkts.filter_wpan_src64(LEADER). \ - filter_wpan_dst16(_pkt.wpan.src16). \ - filter_coap_ack(consts.ADDR_SOL_URI). \ - filter(lambda p: { - consts.NL_STATUS_TLV, - consts.NL_RLOC16_TLV, - consts.NL_ROUTER_MASK_TLV - } <= set(p.coap.tlv.type) and - p.coap.code == consts.COAP_CODE_ACK and - p.coap.tlv.status == 0). \ - must_next() + _pkt_res = pkts.filter_wpan_src64(LEADER).\ + filter_wpan_dst16(_pkt.wpan.src16).\ + filter_coap_ack(consts.ADDR_SOL_URI).\ + filter(lambda p: { + consts.NL_STATUS_TLV, + consts.NL_RLOC16_TLV, + consts.NL_ROUTER_MASK_TLV + } <= set(p.coap.tlv.type) and + p.coap.code == consts.COAP_CODE_ACK and + p.coap.tlv.status == 0).\ + must_next() third_router_id = _pkt_res.coap.tlv.rloc16 >> ROUTER_ID_OFFSET print(f"Router_1 third Router ID: {third_router_id}") diff --git a/tests/nexus/verify_5_1_6.py b/tests/nexus/verify_5_1_6.py index e964fd1b9..96962a0b5 100755 --- a/tests/nexus/verify_5_1_6.py +++ b/tests/nexus/verify_5_1_6.py @@ -66,7 +66,8 @@ def verify(pv): # - Pass Criteria: N/A print("Step 0: Verify topology is formed correctly") pv.verify_attached('ROUTER') - pkts.copy().filter_wpan_src64(LEADER).\ + pkts.copy().\ + filter_wpan_src64(LEADER).\ filter_coap_ack(consts.ADDR_SOL_URI).\ filter(lambda p: { consts.NL_STATUS_TLV, diff --git a/tests/nexus/verify_5_2_6.py b/tests/nexus/verify_5_2_6.py index 2f17a1374..47efd4154 100644 --- a/tests/nexus/verify_5_2_6.py +++ b/tests/nexus/verify_5_2_6.py @@ -73,8 +73,8 @@ def verify(pv): # - Pass Criteria: N/A print("Step 2: Router_24 - Harness causes Router_24 to attach to the network") pkts.filter_wpan_src64(ROUTER_24).\ - filter_mle_cmd(consts.MLE_PARENT_REQUEST).\ - must_next() + filter_mle_cmd(consts.MLE_PARENT_REQUEST).\ + must_next() # Step 3: Router_1 (DUT) # - Description: Allow enough time for the DUT to get Network Data Updates and resign its Router ID. @@ -91,41 +91,41 @@ def verify(pv): # Verify DUT sends Parent Request pkts.filter_wpan_src64(ROUTER_1).\ - filter_mle_cmd(consts.MLE_PARENT_REQUEST).\ - must_next() + filter_mle_cmd(consts.MLE_PARENT_REQUEST).\ + must_next() # Verify DUT sends Child ID Request pkts.filter_wpan_src64(ROUTER_1).\ - filter_mle_cmd(consts.MLE_CHILD_ID_REQUEST).\ - must_next() + filter_mle_cmd(consts.MLE_CHILD_ID_REQUEST).\ + must_next() # Verify DUT sends Address Release with required TLVs pkts.filter_wpan_src64(ROUTER_1).\ - filter_coap_request(consts.ADDR_REL_URI).\ - filter(lambda p: { - consts.NL_MAC_EXTENDED_ADDRESS_TLV, - consts.NL_RLOC16_TLV - } <= {int(t, 16) if isinstance(t, str) else t for t in p.coap.tlv.type}).\ - must_next() + filter_coap_request(consts.ADDR_REL_URI).\ + filter(lambda p: { + consts.NL_MAC_EXTENDED_ADDRESS_TLV, + consts.NL_RLOC16_TLV + } <= {int(t, 16) if isinstance(t, str) else t for t in p.coap.tlv.type}).\ + must_next() # Step 4: Leader # - Description: Receives Address Release message and automatically sends a 2.04 Changed CoAP response. # - Pass Criteria: N/A print("Step 4: Leader - Receives Address Release message and automatically sends a 2.04 Changed CoAP response.") pkts.filter_wpan_src64(LEADER).\ - filter_coap_ack(consts.ADDR_REL_URI).\ - filter(lambda p: p.coap.code == consts.COAP_CODE_ACK).\ - must_next() + filter_coap_ack(consts.ADDR_REL_URI).\ + filter(lambda p: p.coap.code == consts.COAP_CODE_ACK).\ + must_next() # Step 5: Leader # - Description: Harness verifies connectivity by instructing the device to send an ICMPv6 Echo Request to the DUT # - Pass Criteria: The DUT MUST respond with an ICMPv6 Echo Reply print("Step 5: Leader - Harness verifies connectivity by instructing the device to send an ICMPv6 Echo Request") _pkt = pkts.filter_ping_request().\ - filter_wpan_src64(LEADER).\ - must_next() + filter_wpan_src64(LEADER).\ + must_next() pkts.filter_ping_reply(identifier=_pkt.icmpv6.echo.identifier).\ - must_next() + must_next() if __name__ == '__main__': diff --git a/tests/nexus/verify_5_3_2.py b/tests/nexus/verify_5_3_2.py index dcdd4a522..3f03e5ebb 100644 --- a/tests/nexus/verify_5_3_2.py +++ b/tests/nexus/verify_5_3_2.py @@ -78,35 +78,39 @@ def verify(pv): # - Description: Build the topology as described and begin the wireless sniffer. # - Pass Criteria: N/A print("Step 1: All") - pkts.filter_wpan_src64(LEADER).filter_mle_cmd(consts.MLE_ADVERTISEMENT).must_next() - pkts.filter_wpan_src64(DUT).filter_mle_cmd(consts.MLE_ADVERTISEMENT).must_next() + pkts.filter_wpan_src64(LEADER).\ + filter_mle_cmd(consts.MLE_ADVERTISEMENT).\ + must_next() + pkts.filter_wpan_src64(DUT).\ + filter_mle_cmd(consts.MLE_ADVERTISEMENT).\ + must_next() # Step 2: Leader # - Description: Harness instructs the device to send an ICMPv6 Echo Request to the DUT ML-EID. # - Pass Criteria: The DUT MUST respond with an ICMPv6 Echo Reply. print("Step 2: Leader") - pkts.filter_wpan_src64(LEADER) \ - .filter_ipv6_dst(DUT_MLEID) \ - .filter_ping_request(identifier=ECHO_ID) \ - .must_next() - pkts.filter_wpan_src64(DUT) \ - .filter_ipv6_dst(LEADER_MLEID) \ - .filter_ping_reply(identifier=ECHO_ID) \ - .must_next() + pkts.filter_wpan_src64(LEADER).\ + filter_ipv6_dst(DUT_MLEID).\ + filter_ping_request(identifier=ECHO_ID).\ + must_next() + pkts.filter_wpan_src64(DUT).\ + filter_ipv6_dst(LEADER_MLEID).\ + filter_ping_reply(identifier=ECHO_ID).\ + must_next() # Step 3: Leader # - Description: Harness instructs the device to send a fragmented ICMPv6 Echo Request to the DUT ML-EID. # - Pass Criteria: The DUT MUST respond with an ICMPv6 Echo Reply. print("Step 3: Leader") - pkts.filter_wpan_src64(LEADER) \ - .filter_ipv6_dst(DUT_MLEID) \ - .filter_ping_request(identifier=ECHO_ID) \ - .filter(lambda p: hasattr(p, 'lowpan') and hasattr(p.lowpan, 'fragment')) \ - .must_next() - pkts.filter_wpan_src64(DUT) \ - .filter_ipv6_dst(LEADER_MLEID) \ - .filter_ping_reply(identifier=ECHO_ID) \ - .must_next() + pkts.filter_wpan_src64(LEADER).\ + filter_ipv6_dst(DUT_MLEID).\ + filter_ping_request(identifier=ECHO_ID).\ + filter(lambda p: hasattr(p, 'lowpan') and hasattr(p.lowpan, 'fragment')).\ + must_next() + pkts.filter_wpan_src64(DUT).\ + filter_ipv6_dst(LEADER_MLEID).\ + filter_ping_reply(identifier=ECHO_ID).\ + must_next() # Step 4: Leader # - Description: Harness instructs the device to send an ICMPv6 Echo Request to the Realm-Local All-Nodes @@ -115,18 +119,19 @@ def verify(pv): # - The DUT MUST respond with an ICMPv6 Echo Reply. # - The DUT MUST NOT forward the ICMPv6 Echo Request to SED_1. print("Step 4: Leader") - pkts.filter_wpan_src64(LEADER) \ - .filter_ipv6_dst(consts.REALM_LOCAL_ALL_NODES_ADDRESS) \ - .filter_ping_request(identifier=ECHO_ID) \ - .must_next() - pkts.filter_wpan_src64(DUT) \ - .filter_ipv6_dst(LEADER_MLEID) \ - .filter_ping_reply(identifier=ECHO_ID) \ - .must_next() - pkts.copy().filter_wpan_src64(DUT) \ - .filter_wpan_dst64(SED_1) \ - .filter_ipv6_dst(consts.REALM_LOCAL_ALL_NODES_ADDRESS) \ - .must_not_next() + pkts.filter_wpan_src64(LEADER).\ + filter_ipv6_dst(consts.REALM_LOCAL_ALL_NODES_ADDRESS).\ + filter_ping_request(identifier=ECHO_ID).\ + must_next() + pkts.filter_wpan_src64(DUT).\ + filter_ipv6_dst(LEADER_MLEID).\ + filter_ping_reply(identifier=ECHO_ID).\ + must_next() + pkts.copy().\ + filter_wpan_src64(DUT).\ + filter_wpan_dst64(SED_1).\ + filter_ipv6_dst(consts.REALM_LOCAL_ALL_NODES_ADDRESS).\ + must_not_next() # Step 5: Leader # - Description: Harness instructs the device to send a fragmented ICMPv6 Echo Request to the Realm-Local @@ -135,19 +140,20 @@ def verify(pv): # - The DUT MUST respond with an ICMPv6 Echo Reply. # - The DUT MUST NOT forward the ICMPv6 Echo Request to SED_1. print("Step 5: Leader") - pkts.filter_wpan_src64(LEADER) \ - .filter_ipv6_dst(consts.REALM_LOCAL_ALL_NODES_ADDRESS) \ - .filter_ping_request(identifier=ECHO_ID) \ - .filter(lambda p: hasattr(p, 'lowpan') and hasattr(p.lowpan, 'fragment')) \ - .must_next() - pkts.filter_wpan_src64(DUT) \ - .filter_ipv6_dst(LEADER_MLEID) \ - .filter_ping_reply(identifier=ECHO_ID) \ - .must_next() - pkts.copy().filter_wpan_src64(DUT) \ - .filter_wpan_dst64(SED_1) \ - .filter_ipv6_dst(consts.REALM_LOCAL_ALL_NODES_ADDRESS) \ - .must_not_next() + pkts.filter_wpan_src64(LEADER).\ + filter_ipv6_dst(consts.REALM_LOCAL_ALL_NODES_ADDRESS).\ + filter_ping_request(identifier=ECHO_ID).\ + filter(lambda p: hasattr(p, 'lowpan') and hasattr(p.lowpan, 'fragment')).\ + must_next() + pkts.filter_wpan_src64(DUT).\ + filter_ipv6_dst(LEADER_MLEID).\ + filter_ping_reply(identifier=ECHO_ID).\ + must_next() + pkts.copy().\ + filter_wpan_src64(DUT).\ + filter_wpan_dst64(SED_1).\ + filter_ipv6_dst(consts.REALM_LOCAL_ALL_NODES_ADDRESS).\ + must_not_next() # Step 6: Leader # - Description: Harness instructs the device to send an ICMPv6 Echo Request to the Realm-Local All-Routers @@ -156,18 +162,19 @@ def verify(pv): # - The DUT MUST respond with an ICMPv6 Echo Reply. # - The DUT MUST NOT forward the ICMPv6 Echo Request to SED_1. print("Step 6: Leader") - pkts.filter_wpan_src64(LEADER) \ - .filter_ipv6_dst(consts.REALM_LOCAL_ALL_ROUTERS_ADDRESS) \ - .filter_ping_request(identifier=ECHO_ID) \ - .must_next() - pkts.filter_wpan_src64(DUT) \ - .filter_ipv6_dst(LEADER_MLEID) \ - .filter_ping_reply(identifier=ECHO_ID) \ - .must_next() - pkts.copy().filter_wpan_src64(DUT) \ - .filter_wpan_dst64(SED_1) \ - .filter_ipv6_dst(consts.REALM_LOCAL_ALL_ROUTERS_ADDRESS) \ - .must_not_next() + pkts.filter_wpan_src64(LEADER).\ + filter_ipv6_dst(consts.REALM_LOCAL_ALL_ROUTERS_ADDRESS).\ + filter_ping_request(identifier=ECHO_ID).\ + must_next() + pkts.filter_wpan_src64(DUT).\ + filter_ipv6_dst(LEADER_MLEID).\ + filter_ping_reply(identifier=ECHO_ID).\ + must_next() + pkts.copy().\ + filter_wpan_src64(DUT).\ + filter_wpan_dst64(SED_1).\ + filter_ipv6_dst(consts.REALM_LOCAL_ALL_ROUTERS_ADDRESS).\ + must_not_next() # Step 7: Leader # - Description: Harness instructs the device to send a fragmented ICMPv6 Echo Request to the Realm-Local @@ -176,19 +183,20 @@ def verify(pv): # - The DUT MUST respond with an ICMPv6 Echo Reply. # - The DUT MUST NOT forward the ICMPv6 Echo Request to SED_1. print("Step 7: Leader") - pkts.filter_wpan_src64(LEADER) \ - .filter_ipv6_dst(consts.REALM_LOCAL_ALL_ROUTERS_ADDRESS) \ - .filter_ping_request(identifier=ECHO_ID) \ - .filter(lambda p: hasattr(p, 'lowpan') and hasattr(p.lowpan, 'fragment')) \ - .must_next() - pkts.filter_wpan_src64(DUT) \ - .filter_ipv6_dst(LEADER_MLEID) \ - .filter_ping_reply(identifier=ECHO_ID) \ - .must_next() - pkts.copy().filter_wpan_src64(DUT) \ - .filter_wpan_dst64(SED_1) \ - .filter_ipv6_dst(consts.REALM_LOCAL_ALL_ROUTERS_ADDRESS) \ - .must_not_next() + pkts.filter_wpan_src64(LEADER).\ + filter_ipv6_dst(consts.REALM_LOCAL_ALL_ROUTERS_ADDRESS).\ + filter_ping_request(identifier=ECHO_ID).\ + filter(lambda p: hasattr(p, 'lowpan') and hasattr(p.lowpan, 'fragment')).\ + must_next() + pkts.filter_wpan_src64(DUT).\ + filter_ipv6_dst(LEADER_MLEID).\ + filter_ping_reply(identifier=ECHO_ID).\ + must_next() + pkts.copy().\ + filter_wpan_src64(DUT).\ + filter_wpan_dst64(SED_1).\ + filter_ipv6_dst(consts.REALM_LOCAL_ALL_ROUTERS_ADDRESS).\ + must_not_next() # Step 8: Leader # - Description: Harness instructs the device to send a Fragmented ICMPv6 Echo Request to the Realm-Local All @@ -203,24 +211,24 @@ def verify(pv): # - group ID set to 1 # - The DUT MUST use IEEE 802.15.4 indirect transmissions to forward packet to SED_1. print("Step 8: Leader") - pkts.filter_wpan_src64(LEADER) \ - .filter_ipv6_dst(realm_local_all_thread_nodes) \ - .filter_ping_request(identifier=ECHO_ID) \ - .filter(lambda p: hasattr(p, 'lowpan') and hasattr(p.lowpan, 'fragment')) \ - .must_next() - pkts.filter(lambda p: p.wpan.src16 == SED_1_RLOC16) \ - .filter_wpan_cmd(consts.WPAN_DATA_REQUEST) \ - .must_next() + pkts.filter_wpan_src64(LEADER).\ + filter_ipv6_dst(realm_local_all_thread_nodes).\ + filter_ping_request(identifier=ECHO_ID).\ + filter(lambda p: hasattr(p, 'lowpan') and hasattr(p.lowpan, 'fragment')).\ + must_next() + pkts.filter(lambda p: p.wpan.src16 == SED_1_RLOC16).\ + filter_wpan_cmd(consts.WPAN_DATA_REQUEST).\ + must_next() # DUT forwards the packet to SED_1 responding to the Data Request. # We use RLOC16 filters because decryption might fail for these packets. - pkts.filter(lambda p: p.wpan.src16 == DUT_RLOC16 and p.wpan.dst16 == SED_1_RLOC16) \ - .must_next() + pkts.filter(lambda p: p.wpan.src16 == DUT_RLOC16 and p.wpan.dst16 == SED_1_RLOC16).\ + must_next() # SED_1 sends Echo Reply (unicast to Leader). # Unicast packets are usually mapped and decrypted correctly if mapping was learned. - pkts.filter(lambda p: p.wpan.src16 == SED_1_RLOC16) \ - .filter_ipv6_dst(LEADER_MLEID) \ - .filter_ping_reply(identifier=ECHO_ID) \ - .must_next() + pkts.filter(lambda p: p.wpan.src16 == SED_1_RLOC16).\ + filter_ipv6_dst(LEADER_MLEID).\ + filter_ping_reply(identifier=ECHO_ID).\ + must_next() if __name__ == '__main__': diff --git a/tests/nexus/verify_5_3_4.py b/tests/nexus/verify_5_3_4.py index 62c77e1e2..21af9e8e8 100644 --- a/tests/nexus/verify_5_3_4.py +++ b/tests/nexus/verify_5_3_4.py @@ -86,13 +86,15 @@ def verify(pv): for mleid in MEDS_MLEID: # For each MED, verify a ping request exists. A fresh copy of `pkts` is used # to ensure the search is independent of the order of packets for other MEDs. - pkts.copy().filter_ping_request().\ + pkts.copy().\ + filter_ping_request().\ filter_wpan_src64(SED_1).\ filter_ipv6_dst(mleid).\ must_next() # Verify a corresponding address query is sent by the DUT. - pkts.copy().filter_wpan_src64(DUT).\ + pkts.copy().\ + filter_wpan_src64(DUT).\ filter_ipv6_dst(consts.REALM_LOCAL_ALL_ROUTERS_ADDRESS).\ filter_coap_request(consts.ADDR_QRY_URI).\ filter(lambda p: p.coap.tlv.target_eid == str(mleid)).\ @@ -146,7 +148,8 @@ def verify(pv): must_next() # Verify NO Address Query was sent by DUT starting from the start of Step 4 - pkts.range(checkpoint_step4).filter_wpan_src64(DUT).\ + pkts.range(checkpoint_step4).\ + filter_wpan_src64(DUT).\ filter_coap_request(consts.ADDR_QRY_URI).\ must_not_next() diff --git a/tests/nexus/verify_5_3_5.py b/tests/nexus/verify_5_3_5.py index 14a2fbe02..1ea01d6cd 100644 --- a/tests/nexus/verify_5_3_5.py +++ b/tests/nexus/verify_5_3_5.py @@ -87,11 +87,16 @@ def verify(pv): # - The hopsLft field of the 6LoWPAN Mesh Header MUST be greater than the route cost to the destination. print("Step 3: Router_3") # Router_3 -> DUT - p1 = pkts.filter_wpan_src16(ROUTER_3_RLOC16).filter_wpan_dst16(DUT_RLOC16).filter_ping_request( - identifier=ECHO_IDENTIFIER).must_next() + p1 = pkts.filter_wpan_src16(ROUTER_3_RLOC16).\ + filter_wpan_dst16(DUT_RLOC16).\ + filter_ping_request(identifier=ECHO_IDENTIFIER).\ + must_next() # DUT -> Leader - p2 = pkts.filter_wpan_src16(DUT_RLOC16).filter_wpan_dst16(LEADER_RLOC16).filter_ping_request( - identifier=ECHO_IDENTIFIER).filter(lambda p: p.lowpan.mesh.hops > 1).must_next() + p2 = pkts.filter_wpan_src16(DUT_RLOC16).\ + filter_wpan_dst16(LEADER_RLOC16).\ + filter_ping_request(identifier=ECHO_IDENTIFIER).\ + filter(lambda p: p.lowpan.mesh.hops > 1).\ + must_next() assert get_hops(p2) == get_hops(p1) - 1, f'Hops left not decremented correctly: {get_hops(p1)} -> {get_hops(p2)}' # Step 4: Harness @@ -106,15 +111,23 @@ def verify(pv): # - The hopsLft field of the 6LoWPAN Mesh Header MUST be greater than the route cost to the destination. print("Step 5: Router_3") # Router_3 -> DUT - p1 = pkts.filter_wpan_src16(ROUTER_3_RLOC16).filter_wpan_dst16(DUT_RLOC16).filter_ping_request( - identifier=ECHO_IDENTIFIER).must_next() + p1 = pkts.filter_wpan_src16(ROUTER_3_RLOC16).\ + filter_wpan_dst16(DUT_RLOC16).\ + filter_ping_request(identifier=ECHO_IDENTIFIER).\ + must_next() # DUT -> Router_2 - p2 = pkts.filter_wpan_src16(DUT_RLOC16).filter_wpan_dst16(ROUTER_2_RLOC16).filter_ping_request( - identifier=ECHO_IDENTIFIER).filter(lambda p: p.lowpan.mesh.hops > 2).must_next() + p2 = pkts.filter_wpan_src16(DUT_RLOC16).\ + filter_wpan_dst16(ROUTER_2_RLOC16).\ + filter_ping_request(identifier=ECHO_IDENTIFIER).\ + filter(lambda p: p.lowpan.mesh.hops > 2).\ + must_next() assert get_hops(p2) == get_hops(p1) - 1, f'Hops left not decremented correctly: {get_hops(p1)} -> {get_hops(p2)}' # Router_2 -> Leader - p3 = pkts.filter_wpan_src16(ROUTER_2_RLOC16).filter_wpan_dst16(LEADER_RLOC16).filter_ping_request( - identifier=ECHO_IDENTIFIER).filter(lambda p: p.lowpan.mesh.hops > 1).must_next() + p3 = pkts.filter_wpan_src16(ROUTER_2_RLOC16).\ + filter_wpan_dst16(LEADER_RLOC16).\ + filter_ping_request(identifier=ECHO_IDENTIFIER).\ + filter(lambda p: p.lowpan.mesh.hops > 1).\ + must_next() assert get_hops(p3) == get_hops(p2) - 1, f'Hops left not decremented correctly: {get_hops(p2)} -> {get_hops(p3)}' # Step 6: Harness @@ -130,11 +143,16 @@ def verify(pv): # - The hopsLft field of the 6LoWPAN Mesh Header MUST be greater than the route cost to the destination. print("Step 7: Router_3") # Router_3 -> DUT - p1 = pkts.filter_wpan_src16(ROUTER_3_RLOC16).filter_wpan_dst16(DUT_RLOC16).filter_ping_request( - identifier=ECHO_IDENTIFIER).must_next() + p1 = pkts.filter_wpan_src16(ROUTER_3_RLOC16).\ + filter_wpan_dst16(DUT_RLOC16).\ + filter_ping_request(identifier=ECHO_IDENTIFIER).\ + must_next() # DUT -> Leader - p2 = pkts.filter_wpan_src16(DUT_RLOC16).filter_wpan_dst16(LEADER_RLOC16).filter_ping_request( - identifier=ECHO_IDENTIFIER).filter(lambda p: p.lowpan.mesh.hops > 2).must_next() + p2 = pkts.filter_wpan_src16(DUT_RLOC16).\ + filter_wpan_dst16(LEADER_RLOC16).\ + filter_ping_request(identifier=ECHO_IDENTIFIER).\ + filter(lambda p: p.lowpan.mesh.hops > 2).\ + must_next() assert get_hops(p2) == get_hops(p1) - 1, f'Hops left not decremented correctly: {get_hops(p1)} -> {get_hops(p2)}' # Step 8: Harness @@ -149,15 +167,23 @@ def verify(pv): # - The hopsLft field of the 6LoWPAN Mesh Header MUST be greater than the route cost to the destination. print("Step 9: Router_3") # Router_3 -> DUT - p1 = pkts.filter_wpan_src16(ROUTER_3_RLOC16).filter_wpan_dst16(DUT_RLOC16).filter_ping_request( - identifier=ECHO_IDENTIFIER).must_next() + p1 = pkts.filter_wpan_src16(ROUTER_3_RLOC16).\ + filter_wpan_dst16(DUT_RLOC16).\ + filter_ping_request(identifier=ECHO_IDENTIFIER).\ + must_next() # DUT -> Router_2 - p2 = pkts.filter_wpan_src16(DUT_RLOC16).filter_wpan_dst16(ROUTER_2_RLOC16).filter_ping_request( - identifier=ECHO_IDENTIFIER).filter(lambda p: p.lowpan.mesh.hops > 2).must_next() + p2 = pkts.filter_wpan_src16(DUT_RLOC16).\ + filter_wpan_dst16(ROUTER_2_RLOC16).\ + filter_ping_request(identifier=ECHO_IDENTIFIER).\ + filter(lambda p: p.lowpan.mesh.hops > 2).\ + must_next() assert get_hops(p2) == get_hops(p1) - 1, f'Hops left not decremented correctly: {get_hops(p1)} -> {get_hops(p2)}' # Router_2 -> Leader - p3 = pkts.filter_wpan_src16(ROUTER_2_RLOC16).filter_wpan_dst16(LEADER_RLOC16).filter_ping_request( - identifier=ECHO_IDENTIFIER).filter(lambda p: p.lowpan.mesh.hops > 1).must_next() + p3 = pkts.filter_wpan_src16(ROUTER_2_RLOC16).\ + filter_wpan_dst16(LEADER_RLOC16).\ + filter_ping_request(identifier=ECHO_IDENTIFIER).\ + filter(lambda p: p.lowpan.mesh.hops > 1).\ + must_next() assert get_hops(p3) == get_hops(p2) - 1, f'Hops left not decremented correctly: {get_hops(p2)} -> {get_hops(p3)}' diff --git a/tests/nexus/verify_5_3_6.py b/tests/nexus/verify_5_3_6.py index 041c0b788..ef9d15cd0 100644 --- a/tests/nexus/verify_5_3_6.py +++ b/tests/nexus/verify_5_3_6.py @@ -66,10 +66,16 @@ def verify(pv): # - Description: Ensure topology is formed correctly. # - Pass Criteria: N/A print("Step 1: All") - pkts.filter_mle_cmd(consts.MLE_ADVERTISEMENT).filter_wpan_src64(LEADER).must_next() - router1_pkt = pkts.filter_mle_cmd(consts.MLE_ADVERTISEMENT).filter_wpan_src64(ROUTER_1).must_next() + pkts.filter_mle_cmd(consts.MLE_ADVERTISEMENT).\ + filter_wpan_src64(LEADER).\ + must_next() + router1_pkt = pkts.filter_mle_cmd(consts.MLE_ADVERTISEMENT).\ + filter_wpan_src64(ROUTER_1).\ + must_next() router1_id = (router1_pkt.mle.tlv.source_addr >> 10) - router2_pkt = pkts.filter_mle_cmd(consts.MLE_ADVERTISEMENT).filter_wpan_src64(ROUTER_2).must_next() + router2_pkt = pkts.filter_mle_cmd(consts.MLE_ADVERTISEMENT).\ + filter_wpan_src64(ROUTER_2).\ + must_next() router2_id = (router2_pkt.mle.tlv.source_addr >> 10) # Step 2: Router_2 @@ -108,7 +114,9 @@ def verify(pv): # - The DUT MUST reset the MLE Advertisement trickle timer and send an Advertisement. print("Step 5: Router_2") # Verify Router 2 becomes router again (sends advertisement) and get its new ID - router2_rejoin_pkt = pkts.filter_mle_cmd(consts.MLE_ADVERTISEMENT).filter_wpan_src64(ROUTER_2).must_next() + router2_rejoin_pkt = pkts.filter_mle_cmd(consts.MLE_ADVERTISEMENT).\ + filter_wpan_src64(ROUTER_2).\ + must_next() router2_id_reattached = (router2_rejoin_pkt.mle.tlv.source_addr >> 10) # Verify Leader sends advertisement after Router 2 re-joins diff --git a/tests/nexus/verify_5_3_7.py b/tests/nexus/verify_5_3_7.py index b2012dbd6..46df38d08 100644 --- a/tests/nexus/verify_5_3_7.py +++ b/tests/nexus/verify_5_3_7.py @@ -79,7 +79,8 @@ def verify(pv): # - Route64 TLV # - Source Address TLV print("Step 2: Leader (DUT)") - pkts.copy().filter_mle_cmd(consts.MLE_ADVERTISEMENT).\ + pkts.copy().\ + filter_mle_cmd(consts.MLE_ADVERTISEMENT).\ filter_wpan_src64(LEADER).\ filter_LLANMA().\ filter(lambda p: { @@ -111,7 +112,8 @@ def verify(pv): # - CoAP Payload: # - Target EID TLV print("Step 5: MED_2") - pkts.copy().filter_wpan_src64(LEADER).\ + pkts.copy().\ + filter_wpan_src64(LEADER).\ filter_RLARMA().\ filter_coap_request(consts.ADDR_QRY_URI).\ filter(lambda p: p.coap.tlv.target_eid == '2001::1').\ @@ -121,8 +123,14 @@ def verify(pv): # - Description: Automatically respond with Address Notification message with matching Target TLVs. # - Pass Criteria: N/A print("Step 6: Router_1, Router_2") - pkts.copy().filter_wpan_src64(ROUTER_1).filter_coap_request(consts.ADDR_NTF_URI).must_next() - pkts.copy().filter_wpan_src64(ROUTER_2).filter_coap_request(consts.ADDR_NTF_URI).must_next() + pkts.copy().\ + filter_wpan_src64(ROUTER_1).\ + filter_coap_request(consts.ADDR_NTF_URI).\ + must_next() + pkts.copy().\ + filter_wpan_src64(ROUTER_2).\ + filter_coap_request(consts.ADDR_NTF_URI).\ + must_next() # Step 7: Leader (DUT) # - Description: Automatically sends a Multicast Address Error Notification. @@ -136,7 +144,8 @@ def verify(pv): # - ML-EID TLV # - The IPv6 Source address MUST be the RLOC of the originator,,, print("Step 7: Leader (DUT)") - pkts.copy().filter_wpan_src64(LEADER).\ + pkts.copy().\ + filter_wpan_src64(LEADER).\ filter_RLARMA().\ filter_coap_request(consts.ADDR_ERR_URI).\ filter(lambda p: p.coap.tlv.target_eid == '2001::1' and p.ipv6.src == pv.vars['LEADER_RLOC']).\ diff --git a/tests/nexus/verify_5_3_8.py b/tests/nexus/verify_5_3_8.py index 011bf919d..8a158246c 100644 --- a/tests/nexus/verify_5_3_8.py +++ b/tests/nexus/verify_5_3_8.py @@ -158,7 +158,8 @@ def _verify_echo(pv, src_ext, dst_ext, target_ip): end_index = pkts.index # Verify no Address Query from DUT (Leader) for the target IP in this range - pkts.range(start_index, end_index).filter_wpan_src64(pv.vars['LEADER']).\ + pkts.range(start_index, end_index).\ + filter_wpan_src64(pv.vars['LEADER']).\ filter(lambda p: p.ipv6.src == DUT_RLOC).\ filter_coap_request(consts.ADDR_QRY_URI).\ filter(lambda p: Ipv6Addr(p.coap.tlv.target_eid) == target_ip_addr or\