From 5bc2b8113ba62e75642fffdf6d15ec9ef68354a7 Mon Sep 17 00:00:00 2001 From: hjian2017 <30794429+hjian2017@users.noreply.github.com> Date: Wed, 18 Oct 2017 11:25:22 -0500 Subject: [PATCH] [tests] add traffic to 4 test cases: from Cert_5_2_07 to Cert_5_3_03 (#2214) - Add a method in message.py to verify if receive a ICMPv6 message. - Add two methods in command.py to verify link request and link accept. --- .../Cert_5_2_07_REEDSynchronization.py | 107 ++++++------ .../thread-cert/Cert_5_3_01_LinkLocal.py | 33 ++-- .../thread-cert/Cert_5_3_02_RealmLocal.py | 66 ++++--- .../thread-cert/Cert_5_3_03_AddressQuery.py | 161 ++++++++++-------- tests/scripts/thread-cert/command.py | 37 +++- tests/scripts/thread-cert/config.py | 3 + tests/scripts/thread-cert/message.py | 8 + tests/scripts/thread-cert/node_cli.py | 23 ++- 8 files changed, 265 insertions(+), 173 deletions(-) diff --git a/tests/scripts/thread-cert/Cert_5_2_07_REEDSynchronization.py b/tests/scripts/thread-cert/Cert_5_2_07_REEDSynchronization.py index 9eda71f66..3504693ae 100755 --- a/tests/scripts/thread-cert/Cert_5_2_07_REEDSynchronization.py +++ b/tests/scripts/thread-cert/Cert_5_2_07_REEDSynchronization.py @@ -29,86 +29,85 @@ import time import unittest +import ipv6 import node +import mle +import config +import command LEADER = 1 -ROUTER1 = 2 -REED = 3 -ROUTER2 = 4 -ROUTER3 = 5 +DUT_ROUTER1 = 2 +DUT_REED = 17 class Cert_5_2_7_REEDSynchronization(unittest.TestCase): def setUp(self): self.nodes = {} - for i in range(1,6): + for i in range(1, 18): self.nodes[i] = node.Node(i) + self.nodes[i].set_panid(0xface) + self.nodes[i].set_mode('rsdn') + self.nodes[i].set_router_selection_jitter(1) - self.nodes[LEADER].set_panid(0xface) - self.nodes[LEADER].set_mode('rsdn') - self.nodes[LEADER].add_whitelist(self.nodes[ROUTER1].get_addr64()) - self.nodes[LEADER].add_whitelist(self.nodes[REED].get_addr64()) - self.nodes[LEADER].add_whitelist(self.nodes[ROUTER2].get_addr64()) - self.nodes[LEADER].enable_whitelist() - - self.nodes[ROUTER1].set_panid(0xface) - self.nodes[ROUTER1].set_mode('rsdn') - self.nodes[ROUTER1].add_whitelist(self.nodes[LEADER].get_addr64()) - self.nodes[ROUTER1].add_whitelist(self.nodes[REED].get_addr64()) - self.nodes[ROUTER1].add_whitelist(self.nodes[ROUTER3].get_addr64()) - self.nodes[ROUTER1].enable_whitelist() - self.nodes[ROUTER1].set_router_selection_jitter(1) - - self.nodes[REED].set_panid(0xface) - self.nodes[REED].set_mode('rsdn') - self.nodes[REED].add_whitelist(self.nodes[LEADER].get_addr64()) - self.nodes[REED].add_whitelist(self.nodes[ROUTER1].get_addr64()) - self.nodes[REED].add_whitelist(self.nodes[ROUTER2].get_addr64()) - self.nodes[REED].add_whitelist(self.nodes[ROUTER3].get_addr64()) - self.nodes[REED].enable_whitelist() - self.nodes[REED].set_router_upgrade_threshold(0) - - self.nodes[ROUTER2].set_panid(0xface) - self.nodes[ROUTER2].set_mode('rsdn') - self.nodes[ROUTER2].add_whitelist(self.nodes[LEADER].get_addr64()) - self.nodes[ROUTER2].add_whitelist(self.nodes[REED].get_addr64()) - self.nodes[ROUTER2].add_whitelist(self.nodes[ROUTER3].get_addr64()) - self.nodes[ROUTER2].enable_whitelist() - self.nodes[ROUTER2].set_router_selection_jitter(1) - - self.nodes[ROUTER3].set_panid(0xface) - self.nodes[ROUTER3].set_mode('rsdn') - self.nodes[ROUTER3].add_whitelist(self.nodes[REED].get_addr64()) - self.nodes[ROUTER3].add_whitelist(self.nodes[ROUTER1].get_addr64()) - self.nodes[ROUTER3].add_whitelist(self.nodes[ROUTER2].get_addr64()) - self.nodes[ROUTER3].enable_whitelist() - self.nodes[ROUTER3].set_router_selection_jitter(1) + self.sniffer = config.create_default_thread_sniffer() + self.sniffer.start() def tearDown(self): + self.sniffer.stop() + del self.sniffer + for node in list(self.nodes.values()): node.stop() del self.nodes def test(self): + # 1. Ensure topology is formed correctly without DUT_ROUTER1. self.nodes[LEADER].start() self.nodes[LEADER].set_state('leader') self.assertEqual(self.nodes[LEADER].get_state(), 'leader') - self.nodes[REED].start() + for i in range(2, 17): + self.nodes[i].start() time.sleep(5) - self.assertEqual(self.nodes[REED].get_state(), 'child') - self.nodes[ROUTER1].start() - time.sleep(5) - self.assertEqual(self.nodes[ROUTER1].get_state(), 'router') + for i in range(2, 17): + self.assertEqual(self.nodes[i].get_state(), 'router') - self.nodes[ROUTER2].start() - time.sleep(5) - self.assertEqual(self.nodes[ROUTER2].get_state(), 'router') + # 2. DUT_REED: Attach to network. Verify it didn't send an Address Solicit Request. + # Avoid DUT_REED attach to DUT_ROUTER1. + self.nodes[DUT_REED].add_whitelist(self.nodes[DUT_ROUTER1].get_addr64(), config.RSSI['LINK_QULITY_1']) - self.nodes[ROUTER3].start() + self.nodes[DUT_REED].start() time.sleep(5) - self.assertEqual(self.nodes[ROUTER3].get_state(), 'router') + self.assertEqual(self.nodes[DUT_REED].get_state(), 'child') + + # The DUT_REED must not send a coap message here. + reed_messages = self.sniffer.get_messages_sent_by(DUT_REED) + msg = reed_messages.does_not_contain_coap_message() + assert msg is True, "Error: The DUT_REED sent an Address Solicit Request" + + # 3. DUT_REED: Verify sent a Link Request to at least 3 neighboring Routers. + for i in range(0, 3): + msg = reed_messages.next_mle_message(mle.CommandType.LINK_REQUEST) + command.check_link_request(msg) + + # 4. DUT_ROUTER1: Verify sent a Link Accept to DUT_REED. + time.sleep(30) + dut_messages = self.sniffer.get_messages_sent_by(DUT_ROUTER1) + flag_link_accept = False + while True: + msg = dut_messages.next_mle_message(mle.CommandType.LINK_ACCEPT, False) + if msg == None : + break + + destination_link_local = self.nodes[DUT_REED].get_ip6_address(config.ADDRESS_TYPE.LINK_LOCAL) + if ipv6.ip_address(destination_link_local) == msg.ipv6_packet.ipv6_header.destination_address: + flag_link_accept = True + break + + assert flag_link_accept is True, "Error: DUT_ROUTER1 didn't send a Link Accept to DUT_REED" + + command.check_link_accept(msg, self.nodes[DUT_REED]) if __name__ == '__main__': unittest.main() diff --git a/tests/scripts/thread-cert/Cert_5_3_01_LinkLocal.py b/tests/scripts/thread-cert/Cert_5_3_01_LinkLocal.py index 6b26b3c73..6168fad46 100755 --- a/tests/scripts/thread-cert/Cert_5_3_01_LinkLocal.py +++ b/tests/scripts/thread-cert/Cert_5_3_01_LinkLocal.py @@ -31,9 +31,11 @@ import time import unittest import node +import config +import command LEADER = 1 -ROUTER1 = 2 +DUT_ROUTER1 = 2 class Cert_5_3_1_LinkLocal(unittest.TestCase): def setUp(self): @@ -43,14 +45,10 @@ class Cert_5_3_1_LinkLocal(unittest.TestCase): self.nodes[LEADER].set_panid(0xface) self.nodes[LEADER].set_mode('rsdn') - self.nodes[LEADER].add_whitelist(self.nodes[ROUTER1].get_addr64()) - self.nodes[LEADER].enable_whitelist() - self.nodes[ROUTER1].set_panid(0xface) - self.nodes[ROUTER1].set_mode('rsdn') - self.nodes[ROUTER1].add_whitelist(self.nodes[LEADER].get_addr64()) - self.nodes[ROUTER1].enable_whitelist() - self.nodes[ROUTER1].set_router_selection_jitter(1) + self.nodes[DUT_ROUTER1].set_panid(0xface) + self.nodes[DUT_ROUTER1].set_mode('rsdn') + self.nodes[DUT_ROUTER1].set_router_selection_jitter(1) def tearDown(self): for node in list(self.nodes.values()): @@ -58,25 +56,30 @@ class Cert_5_3_1_LinkLocal(unittest.TestCase): del self.nodes def test(self): + # 1 self.nodes[LEADER].start() self.nodes[LEADER].set_state('leader') self.assertEqual(self.nodes[LEADER].get_state(), 'leader') - self.nodes[ROUTER1].start() + self.nodes[DUT_ROUTER1].start() time.sleep(5) - self.assertEqual(self.nodes[ROUTER1].get_state(), 'router') + self.assertEqual(self.nodes[DUT_ROUTER1].get_state(), 'router') - addrs = self.nodes[ROUTER1].get_addrs() - for addr in addrs: - if addr[0:4] == 'fe80': - self.assertTrue(self.nodes[LEADER].ping(addr, size=256)) - self.assertTrue(self.nodes[LEADER].ping(addr)) + # 2 & 3 + link_local = self.nodes[DUT_ROUTER1].get_ip6_address(config.ADDRESS_TYPE.LINK_LOCAL) + self.assertTrue(self.nodes[LEADER].ping(link_local, size=256)) + self.assertTrue(self.nodes[LEADER].ping(link_local)) + # 4 & 5 self.assertTrue(self.nodes[LEADER].ping('ff02::1', size=256)) self.assertTrue(self.nodes[LEADER].ping('ff02::1')) + # 6 & 7 self.assertTrue(self.nodes[LEADER].ping('ff02::2', size=256)) self.assertTrue(self.nodes[LEADER].ping('ff02::2')) + # 8 + self.assertTrue(self.nodes[LEADER].ping(config.LINK_LOCAL_All_THREAD_NODES_MULTICAST_ADDRESS)) + if __name__ == '__main__': unittest.main() diff --git a/tests/scripts/thread-cert/Cert_5_3_02_RealmLocal.py b/tests/scripts/thread-cert/Cert_5_3_02_RealmLocal.py index cc4cb5323..49886ae7e 100755 --- a/tests/scripts/thread-cert/Cert_5_3_02_RealmLocal.py +++ b/tests/scripts/thread-cert/Cert_5_3_02_RealmLocal.py @@ -31,10 +31,12 @@ import time import unittest import node +import config +import command LEADER = 1 ROUTER1 = 2 -ROUTER2 = 3 +DUT_ROUTER2 = 3 SED1 = 4 class Cert_5_3_2_RealmLocal(unittest.TestCase): @@ -51,29 +53,36 @@ class Cert_5_3_2_RealmLocal(unittest.TestCase): self.nodes[ROUTER1].set_panid(0xface) self.nodes[ROUTER1].set_mode('rsdn') self.nodes[ROUTER1].add_whitelist(self.nodes[LEADER].get_addr64()) - self.nodes[ROUTER1].add_whitelist(self.nodes[ROUTER2].get_addr64()) + self.nodes[ROUTER1].add_whitelist(self.nodes[DUT_ROUTER2].get_addr64()) self.nodes[ROUTER1].enable_whitelist() self.nodes[ROUTER1].set_router_selection_jitter(1) - self.nodes[ROUTER2].set_panid(0xface) - self.nodes[ROUTER2].set_mode('rsdn') - self.nodes[ROUTER2].add_whitelist(self.nodes[ROUTER1].get_addr64()) - self.nodes[ROUTER2].add_whitelist(self.nodes[SED1].get_addr64()) - self.nodes[ROUTER2].enable_whitelist() - self.nodes[ROUTER2].set_router_selection_jitter(1) + self.nodes[DUT_ROUTER2].set_panid(0xface) + self.nodes[DUT_ROUTER2].set_mode('rsdn') + self.nodes[DUT_ROUTER2].add_whitelist(self.nodes[ROUTER1].get_addr64()) + self.nodes[DUT_ROUTER2].add_whitelist(self.nodes[SED1].get_addr64()) + self.nodes[DUT_ROUTER2].enable_whitelist() + self.nodes[DUT_ROUTER2].set_router_selection_jitter(1) self.nodes[SED1].set_panid(0xface) self.nodes[SED1].set_mode('sn') - self.nodes[SED1].add_whitelist(self.nodes[ROUTER2].get_addr64()) + self.nodes[SED1].add_whitelist(self.nodes[DUT_ROUTER2].get_addr64()) self.nodes[SED1].enable_whitelist() self.nodes[SED1].set_timeout(3) + self.sniffer = config.create_default_thread_sniffer() + self.sniffer.start() + def tearDown(self): + self.sniffer.stop() + del self.sniffer + for node in list(self.nodes.values()): node.stop() del self.nodes def test(self): + # 1 self.nodes[LEADER].start() self.nodes[LEADER].set_state('leader') self.assertEqual(self.nodes[LEADER].get_state(), 'leader') @@ -82,28 +91,47 @@ class Cert_5_3_2_RealmLocal(unittest.TestCase): time.sleep(5) self.assertEqual(self.nodes[ROUTER1].get_state(), 'router') - self.nodes[ROUTER2].start() + self.nodes[DUT_ROUTER2].start() time.sleep(5) - self.assertEqual(self.nodes[ROUTER2].get_state(), 'router') + self.assertEqual(self.nodes[DUT_ROUTER2].get_state(), 'router') self.nodes[SED1].start() time.sleep(5) self.assertEqual(self.nodes[SED1].get_state(), 'child') - addrs = self.nodes[ROUTER2].get_addrs() - for addr in addrs: - if addr[0:4] != 'fe80': - self.assertTrue(self.nodes[LEADER].ping(addr, size=256)) - self.assertTrue(self.nodes[LEADER].ping(addr)) + # 2 & 3 + mleid = self.nodes[DUT_ROUTER2].get_ip6_address(config.ADDRESS_TYPE.ML_EID) + self.assertTrue(self.nodes[LEADER].ping(mleid, size=256)) + self.assertTrue(self.nodes[LEADER].ping(mleid)) + # 4 & 5 self.assertTrue(self.nodes[LEADER].ping('ff03::1', num_responses=2, size=256)) + sed_messages = self.sniffer.get_messages_sent_by(SED1) + self.assertFalse(sed_messages.contains_icmp_message()) + self.assertTrue(self.nodes[LEADER].ping('ff03::1', num_responses=2)) + sed_messages = self.sniffer.get_messages_sent_by(SED1) + self.assertFalse(sed_messages.contains_icmp_message()) + # 6 & 7 self.assertTrue(self.nodes[LEADER].ping('ff03::2', num_responses=2, size=256)) - self.assertTrue(self.nodes[LEADER].ping('ff03::2', num_responses=2)) + sed_messages = self.sniffer.get_messages_sent_by(SED1) + self.assertFalse(sed_messages.contains_icmp_message()) - self.assertTrue(self.nodes[LEADER].ping('ff33:0040:fdde:ad00:beef:0:0:1', num_responses=3, size=256)) - self.assertTrue(self.nodes[LEADER].ping('ff33:0040:fdde:ad00:beef:0:0:1', num_responses=3)) + self.assertTrue(self.nodes[LEADER].ping('ff03::2', num_responses=2)) + sed_messages = self.sniffer.get_messages_sent_by(SED1) + self.assertFalse(sed_messages.contains_icmp_message()) + + # 8 + self.assertTrue(self.nodes[LEADER].ping(config.REALM_LOCAL_All_THREAD_NODES_MULTICAST_ADDRESS, num_responses=3, size=256)) + time.sleep(2) + sed_messages = self.sniffer.get_messages_sent_by(SED1) + self.assertTrue(sed_messages.contains_icmp_message()) + + self.assertTrue(self.nodes[LEADER].ping(config.REALM_LOCAL_All_THREAD_NODES_MULTICAST_ADDRESS, num_responses=3)) + time.sleep(2) + sed_messages = self.sniffer.get_messages_sent_by(SED1) + self.assertTrue(sed_messages.contains_icmp_message()) if __name__ == '__main__': unittest.main() diff --git a/tests/scripts/thread-cert/Cert_5_3_03_AddressQuery.py b/tests/scripts/thread-cert/Cert_5_3_03_AddressQuery.py index 75c1eaf3f..61e1db371 100755 --- a/tests/scripts/thread-cert/Cert_5_3_03_AddressQuery.py +++ b/tests/scripts/thread-cert/Cert_5_3_03_AddressQuery.py @@ -31,121 +31,136 @@ import time import unittest import node +import config +import command -BR = 1 -LEADER = 2 -ROUTER2 = 3 +LEADER = 1 +ROUTER1 = 2 +DUT_ROUTER2 = 3 ROUTER3 = 4 -ED2 = 5 +MED1 = 5 +MED1_TIMEOUT = 3 class Cert_5_3_3_AddressQuery(unittest.TestCase): def setUp(self): self.nodes = {} for i in range(1,6): - self.nodes[i] = node.Node(i, (i == ED2)) + self.nodes[i] = node.Node(i, (i == MED1)) - self.nodes[LEADER].set_panid(0xface) + self.nodes[LEADER].set_panid() self.nodes[LEADER].set_mode('rsdn') - self.nodes[LEADER].add_whitelist(self.nodes[BR].get_addr64()) - self.nodes[LEADER].add_whitelist(self.nodes[ROUTER2].get_addr64()) + self.nodes[LEADER].add_whitelist(self.nodes[ROUTER1].get_addr64()) + self.nodes[LEADER].add_whitelist(self.nodes[DUT_ROUTER2].get_addr64()) self.nodes[LEADER].add_whitelist(self.nodes[ROUTER3].get_addr64()) self.nodes[LEADER].enable_whitelist() - self.nodes[BR].set_panid(0xface) - self.nodes[BR].set_mode('rsdn') - self.nodes[BR].add_whitelist(self.nodes[LEADER].get_addr64()) - self.nodes[BR].enable_whitelist() - self.nodes[BR].set_router_selection_jitter(1) + self.nodes[ROUTER1].set_panid() + self.nodes[ROUTER1].set_mode('rsdn') + self.nodes[ROUTER1].add_whitelist(self.nodes[LEADER].get_addr64()) + self.nodes[ROUTER1].enable_whitelist() + self.nodes[ROUTER1].set_router_selection_jitter(1) - self.nodes[ROUTER2].set_panid(0xface) - self.nodes[ROUTER2].set_mode('rsdn') - self.nodes[ROUTER2].add_whitelist(self.nodes[LEADER].get_addr64()) - self.nodes[ROUTER2].add_whitelist(self.nodes[ROUTER3].get_addr64()) - self.nodes[ROUTER2].add_whitelist(self.nodes[ED2].get_addr64()) - self.nodes[ROUTER2].enable_whitelist() - self.nodes[ROUTER2].set_router_selection_jitter(1) + self.nodes[DUT_ROUTER2].set_panid() + self.nodes[DUT_ROUTER2].set_mode('rsdn') + self.nodes[DUT_ROUTER2].add_whitelist(self.nodes[LEADER].get_addr64()) + self.nodes[DUT_ROUTER2].add_whitelist(self.nodes[ROUTER3].get_addr64()) + self.nodes[DUT_ROUTER2].add_whitelist(self.nodes[MED1].get_addr64()) + self.nodes[DUT_ROUTER2].enable_whitelist() + self.nodes[DUT_ROUTER2].set_router_selection_jitter(1) - self.nodes[ROUTER3].set_panid(0xface) + self.nodes[ROUTER3].set_panid() self.nodes[ROUTER3].set_mode('rsdn') self.nodes[ROUTER3].add_whitelist(self.nodes[LEADER].get_addr64()) - self.nodes[ROUTER3].add_whitelist(self.nodes[ROUTER2].get_addr64()) + self.nodes[ROUTER3].add_whitelist(self.nodes[DUT_ROUTER2].get_addr64()) self.nodes[ROUTER3].enable_whitelist() self.nodes[ROUTER3].set_router_selection_jitter(1) - self.nodes[ED2].set_panid(0xface) - self.nodes[ED2].set_mode('rsn') - self.nodes[ED2].add_whitelist(self.nodes[ROUTER2].get_addr64()) - self.nodes[ED2].set_timeout(3) - self.nodes[ED2].enable_whitelist() + self.nodes[MED1].set_panid() + self.nodes[MED1].set_mode('rsn') + self.nodes[MED1].add_whitelist(self.nodes[DUT_ROUTER2].get_addr64()) + self.nodes[MED1].set_timeout(MED1_TIMEOUT) + self.nodes[MED1].enable_whitelist() + + self.sniffer = config.create_default_thread_sniffer() + self.sniffer.start() def tearDown(self): + self.sniffer.stop() + del self.sniffer + for node in list(self.nodes.values()): node.stop() del self.nodes def test(self): + # 1 self.nodes[LEADER].start() self.nodes[LEADER].set_state('leader') self.assertEqual(self.nodes[LEADER].get_state(), 'leader') - self.nodes[BR].start() - time.sleep(5) - self.assertEqual(self.nodes[BR].get_state(), 'router') - - self.nodes[BR].add_prefix('2001:2:0:3::/64', 'paros') - self.nodes[BR].add_prefix('2001:2:0:4::/64', 'paros') - self.nodes[BR].register_netdata() - - self.nodes[ROUTER2].start() - time.sleep(5) - self.assertEqual(self.nodes[ROUTER2].get_state(), 'router') - + self.nodes[ROUTER1].start() + self.nodes[DUT_ROUTER2].start() self.nodes[ROUTER3].start() + self.nodes[MED1].start() time.sleep(5) + + self.assertEqual(self.nodes[ROUTER1].get_state(), 'router') + self.assertEqual(self.nodes[DUT_ROUTER2].get_state(), 'router') self.assertEqual(self.nodes[ROUTER3].get_state(), 'router') + self.assertEqual(self.nodes[MED1].get_state(), 'child') + dut_messages = self.sniffer.get_messages_sent_by(DUT_ROUTER2) - self.nodes[ED2].start() - time.sleep(5) - self.assertEqual(self.nodes[ED2].get_state(), 'child') + # 2 + router3_mleid = self.nodes[ROUTER3].get_ip6_address(config.ADDRESS_TYPE.ML_EID) + self.assertTrue(self.nodes[MED1].ping(router3_mleid)) - addrs = self.nodes[ROUTER3].get_addrs() - for addr in addrs: - if addr[0:4] != 'fe80': - self.assertTrue(self.nodes[ED2].ping(addr)) - time.sleep(1) + # Verify DUT_ROUTER2 sent an Address Query Request to the Realm local address. + dut_messages = self.sniffer.get_messages_sent_by(DUT_ROUTER2) + msg = dut_messages.next_coap_message('0.02', '/a/aq') + command.check_address_query(msg, self.nodes[DUT_ROUTER2], config.REALM_LOCAL_ALL_ROUTERS_ADDRESS) - addrs = self.nodes[ED2].get_addrs() - for addr in addrs: - if addr[0:4] != 'fe80': - self.assertTrue(self.nodes[LEADER].ping(addr)) - time.sleep(1) + # 3 + med1_mleid = self.nodes[MED1].get_ip6_address(config.ADDRESS_TYPE.ML_EID) + self.assertTrue(self.nodes[ROUTER1].ping(med1_mleid)) - addrs = self.nodes[BR].get_addrs() - for addr in addrs: - if addr[0:4] != 'fe80': - self.assertTrue(self.nodes[ED2].ping(addr)) - time.sleep(1) + # Verify DUT_ROUTER2 responded with an Address Notification. + dut_messages = self.sniffer.get_messages_sent_by(DUT_ROUTER2) + msg = dut_messages.next_coap_message('0.02', '/a/an') + command.check_address_notification(msg, self.nodes[DUT_ROUTER2], self.nodes[ROUTER1]) - addrs = self.nodes[ROUTER3].get_addrs() - for addr in addrs: - if addr[0:4] != 'fe80': - self.assertTrue(self.nodes[ED2].ping(addr)) - time.sleep(1) + # 4 + dut_messages = self.sniffer.get_messages_sent_by(DUT_ROUTER2) + self.assertTrue(self.nodes[MED1].ping(router3_mleid)) - addrs = self.nodes[ROUTER3].get_addrs() + # Verify DUT_ROUTER2 didn't send an Address Query Request. + dut_messages = self.sniffer.get_messages_sent_by(DUT_ROUTER2) + msg = dut_messages.next_coap_message('0.02', '/a/aq', False) + assert msg is None, "The Address Query Request is not expected." + + # 5 + dut_messages = self.sniffer.get_messages_sent_by(DUT_ROUTER2) self.nodes[ROUTER3].stop() - time.sleep(140) - for addr in addrs: - if addr[0:4] != 'fe80': - self.assertFalse(self.nodes[ED2].ping(addr)) + # Wait for the Leader to expire its Router ID. + # MAX_NEIGHBOR_AGE + INFINITE_COST_TIMEOUT + ID_REUSE_DELAY + propagation time + transmission time ~ 550s. + time.sleep(550) + self.assertFalse(self.nodes[MED1].ping(router3_mleid)) - addrs = self.nodes[ED2].get_addrs() - self.nodes[ED2].stop() - time.sleep(10) - for addr in addrs: - if addr[0:4] != 'fe80': - self.assertFalse(self.nodes[BR].ping(addr)) + # Verify DUT_ROUTER2 sent an Address Query Request to the Realm local address. + dut_messages = self.sniffer.get_messages_sent_by(DUT_ROUTER2) + msg = dut_messages.next_coap_message('0.02', '/a/aq') + command.check_address_query(msg, self.nodes[DUT_ROUTER2], config.REALM_LOCAL_ALL_ROUTERS_ADDRESS) + + # 6 + self.nodes[MED1].stop() + time.sleep(MED1_TIMEOUT) + self.assertFalse(self.nodes[ROUTER1].ping(med1_mleid)) + self.assertFalse(self.nodes[ROUTER1].ping(med1_mleid)) + + # Verify DUT_ROUTER2 didn't respond with an Address Notification. + dut_messages = self.sniffer.get_messages_sent_by(DUT_ROUTER2) + msg = dut_messages.next_coap_message('0.02', '/a/an', False) + assert msg is None, "The Address Notification is not expected." if __name__ == '__main__': unittest.main() diff --git a/tests/scripts/thread-cert/command.py b/tests/scripts/thread-cert/command.py index 459fd28f7..c683f716f 100644 --- a/tests/scripts/thread-cert/command.py +++ b/tests/scripts/thread-cert/command.py @@ -2,6 +2,17 @@ import ipv6 import network_layer import config +import mle + +def check_address_query(command_msg, source_node, destination_address): + """Verify source_node sent a properly formatted Address Query Request message to the destination_address. + """ + command_msg.assertCoapMessageContainsTlv(network_layer.TargetEid) + + source_rloc = source_node.get_ip6_address(config.ADDRESS_TYPE.RLOC) + assert ipv6.ip_address(source_rloc) == command_msg.ipv6_packet.ipv6_header.source_address, "Error: The IPv6 source address is not the RLOC of the originator." + + assert ipv6.ip_address(destination_address.decode('utf-8')) == command_msg.ipv6_packet.ipv6_header.destination_address, "Error: The IPv6 destination address is not expected." def check_address_notification(command_msg, source_node, destination_node): """Verify source_node sent a properly formatted Address Notification command message to destination_node. @@ -12,10 +23,10 @@ def check_address_notification(command_msg, source_node, destination_node): command_msg.assertCoapMessageContainsTlv(network_layer.MlEid) source_rloc = source_node.get_ip6_address(config.ADDRESS_TYPE.RLOC) - assert ipv6.ip_address(source_rloc) == command_msg.ipv6_packet.ipv6_header.source_address, "Error: The IPv6 Source address is not the RLOC of the originator." + assert ipv6.ip_address(source_rloc) == command_msg.ipv6_packet.ipv6_header.source_address, "Error: The IPv6 source address is not the RLOC of the originator." destination_rloc = destination_node.get_ip6_address(config.ADDRESS_TYPE.RLOC) - assert ipv6.ip_address(destination_rloc) == command_msg.ipv6_packet.ipv6_header.destination_address, "Error: The IPv6 Destination address is not the RLOC of the destination." + assert ipv6.ip_address(destination_rloc) == command_msg.ipv6_packet.ipv6_header.destination_address, "Error: The IPv6 destination address is not the RLOC of the destination." def check_address_release(command_msg, destination_node): """Verify the message is a properly formatted address release destined to the given node. @@ -25,4 +36,24 @@ def check_address_release(command_msg, destination_node): command_msg.assertCoapMessageContainsTlv(network_layer.MacExtendedAddress) destination_rloc = destination_node.get_ip6_address(config.ADDRESS_TYPE.RLOC) - assert ipv6.ip_address(destination_rloc) == command_msg.ipv6_packet.ipv6_header.destination_address, "Error: The Destination is not RLOC address" + assert ipv6.ip_address(destination_rloc) == command_msg.ipv6_packet.ipv6_header.destination_address, "Error: The destination is not RLOC address" + +def check_link_request(command_msg): + """Verify a properly formatted Link Request command message. + """ + command_msg.assertMleMessageContainsTlv(mle.Challenge) + command_msg.assertMleMessageContainsTlv(mle.LeaderData) + command_msg.assertMleMessageContainsTlv(mle.SourceAddress) + command_msg.assertMleMessageContainsTlv(mle.Version) + +def check_link_accept(command_msg, destination_node): + """Verify a properly formatted Link Accept command message. + """ + command_msg.assertMleMessageContainsTlv(mle.LinkLayerFrameCounter) + command_msg.assertMleMessageContainsTlv(mle.SourceAddress) + command_msg.assertMleMessageContainsTlv(mle.Response) + command_msg.assertMleMessageContainsTlv(mle.Version) + command_msg.assertMleMessageContainsOptionalTlv(mle.MleFrameCounter) + + destination_link_local = destination_node.get_ip6_address(config.ADDRESS_TYPE.LINK_LOCAL) + assert ipv6.ip_address(destination_link_local) == command_msg.ipv6_packet.ipv6_header.destination_address, "Error: The destination is unexpected" diff --git a/tests/scripts/thread-cert/config.py b/tests/scripts/thread-cert/config.py index bdbfe3ad9..7a33068c9 100644 --- a/tests/scripts/thread-cert/config.py +++ b/tests/scripts/thread-cert/config.py @@ -46,6 +46,9 @@ ROUTING_LOCATOR_REGEX_PATTERN = '.*:(0)?:0{0,2}ff:fe00:\w{1,4}$' LINK_LOCAL = 'fe80:/112' LINK_LOCAL_REGEX_PATTERN = '^fe80:.*' ALOC_FLAG_REGEX_PATTERN = '.*:fc..$' +LINK_LOCAL_All_THREAD_NODES_MULTICAST_ADDRESS = 'ff32:40:fdde:ad00:beef:0:0:1' +REALM_LOCAL_All_THREAD_NODES_MULTICAST_ADDRESS = 'ff33:40:fdde:ad00:beef:0:0:1' +REALM_LOCAL_ALL_ROUTERS_ADDRESS = 'ff03::2' DEFAULT_MASTER_KEY = bytearray([0x00, 0x11, 0x22, 0x33, 0x44, 0x55, 0x66, 0x77, 0x88, 0x99, 0xaa, 0xbb, 0xcc, 0xdd, 0xee, 0xff]) diff --git a/tests/scripts/thread-cert/message.py b/tests/scripts/thread-cert/message.py index 1e177f8df..99657c36a 100644 --- a/tests/scripts/thread-cert/message.py +++ b/tests/scripts/thread-cert/message.py @@ -377,6 +377,14 @@ class MessagesSet(object): return message + def contains_icmp_message(self): + for m in self.messages: + if m.type == MessageType.ICMP: + return True + + return False + + def contains_mle_message(self, command_type): for m in self.messages: if m.type != MessageType.MLE: diff --git a/tests/scripts/thread-cert/node_cli.py b/tests/scripts/thread-cert/node_cli.py index 80691cc6b..b0b36b80f 100644 --- a/tests/scripts/thread-cert/node_cli.py +++ b/tests/scripts/thread-cert/node_cli.py @@ -427,14 +427,20 @@ class otCli: Returns: IPv6 address string. """ - switcher = { - config.ADDRESS_TYPE.LINK_LOCAL: self.__getLinkLocalAddress(), - config.ADDRESS_TYPE.GLOBAL: self.__getGlobalAddress(), - config.ADDRESS_TYPE.RLOC: self.__getRloc(), - config.ADDRESS_TYPE.ALOC: self.__getAloc(), - config.ADDRESS_TYPE.ML_EID: self.__getMleid() - } - return switcher.get(address_type, None) + if address_type == config.ADDRESS_TYPE.LINK_LOCAL: + return self.__getLinkLocalAddress() + elif address_type == config.ADDRESS_TYPE.GLOBAL: + return self.__getGlobalAddress() + elif address_type == config.ADDRESS_TYPE.RLOC: + return self.__getRloc() + elif address_type == config.ADDRESS_TYPE.ALOC: + return self.__getAloc() + elif address_type == config.ADDRESS_TYPE.ML_EID: + return self.__getMleid() + else: + return None + + return None def get_context_reuse_delay(self): self.send_command('contextreusedelay') @@ -623,7 +629,6 @@ class otCli: def send_mgmt_pending_set(self, pending_timestamp=None, active_timestamp=None, delay_timer=None, channel=None, panid=None, master_key=None, mesh_local=None, network_name=None): cmd = 'dataset mgmtsetcommand pending ' - if pending_timestamp != None: cmd += 'pendingtimestamp %d ' % pending_timestamp