diff --git a/tests/scripts/thread-cert/Makefile.am b/tests/scripts/thread-cert/Makefile.am index 35028e6a9..72db53ff3 100644 --- a/tests/scripts/thread-cert/Makefile.am +++ b/tests/scripts/thread-cert/Makefile.am @@ -147,8 +147,9 @@ EXTRA_DIST = \ test_common.py \ test_crypto.py \ test_diag.py \ - test_ipv6_fragmentation.py \ test_ipv6.py \ + test_ipv6_fragmentation.py \ + test_ipv6_source_selection.py \ test_lowpan.py \ test_mac802154.py \ test_mle.py \ @@ -167,8 +168,9 @@ check_SCRIPTS = \ test_common.py \ test_crypto.py \ test_diag.py \ - test_ipv6_fragmentation.py \ test_ipv6.py \ + test_ipv6_fragmentation.py \ + test_ipv6_source_selection.py \ test_lowpan.py \ test_mac802154.py \ test_mle.py \ @@ -284,6 +286,7 @@ XFAIL_NCP_TESTS = \ test_coaps.py \ test_diag.py \ test_ipv6_fragmentation.py \ + test_ipv6_source_selection.py \ test_service.py \ Cert_5_3_10_AddressQuery.py \ Cert_8_1_01_Commissioning.py \ diff --git a/tests/scripts/thread-cert/message.py b/tests/scripts/thread-cert/message.py index 4894efcff..1556acc90 100644 --- a/tests/scripts/thread-cert/message.py +++ b/tests/scripts/thread-cert/message.py @@ -353,6 +353,12 @@ class Message(object): == ipaddress.ip_address(ipv6_address) ) + def assertSentFromSourceAddress(self, ipv6_address): + assert ( + self.ipv6_packet.ipv6_header.source_address + == ipaddress.ip_address(ipv6_address) + ) + def assertSentWithHopLimit(self, hop_limit): assert self.ipv6_packet.ipv6_header.hop_limit == hop_limit diff --git a/tests/scripts/thread-cert/node.py b/tests/scripts/thread-cert/node.py index f5f08f937..65c46c219 100755 --- a/tests/scripts/thread-cert/node.py +++ b/tests/scripts/thread-cert/node.py @@ -533,6 +533,36 @@ class Node: return addrs + def get_mleid(self): + addr = None + cmd = 'ipaddr mleid' + self.send_command(cmd) + i = self._expect(r'(\S+(:\S*)+)\r?\n') + if i == 0: + addr = self.pexpect.match.groups()[0].decode("utf-8") + self._expect('Done') + return addr + + def get_linklocal(self): + addr = None + cmd = 'ipaddr linklocal' + self.send_command(cmd) + i = self._expect(r'(\S+(:\S*)+)\r?\n') + if i == 0: + addr = self.pexpect.match.groups()[0].decode("utf-8") + self._expect('Done') + return addr + + def get_rloc(self): + addr = None + cmd = 'ipaddr rloc' + self.send_command(cmd) + i = self._expect(r'(\S+(:\S*)+)\r?\n') + if i == 0: + addr = self.pexpect.match.groups()[0].decode("utf-8") + self._expect('Done') + return addr + def get_addr(self, prefix): network = ipaddress.ip_network(u'%s' % str(prefix)) addrs = self.get_addrs() diff --git a/tests/scripts/thread-cert/test_ipv6_source_selection.py b/tests/scripts/thread-cert/test_ipv6_source_selection.py new file mode 100755 index 000000000..0c51f0d32 --- /dev/null +++ b/tests/scripts/thread-cert/test_ipv6_source_selection.py @@ -0,0 +1,145 @@ +#!/usr/bin/env python3 +# +# Copyright (c) 2019, The OpenThread Authors. +# All rights reserved. +# +# Redistribution and use in source and binary forms, with or without +# modification, are permitted provided that the following conditions are met: +# 1. Redistributions of source code must retain the above copyright +# notice, this list of conditions and the following disclaimer. +# 2. Redistributions in binary form must reproduce the above copyright +# notice, this list of conditions and the following disclaimer in the +# documentation and/or other materials provided with the distribution. +# 3. Neither the name of the copyright holder nor the +# names of its contributors may be used to endorse or promote products +# derived from this software without specific prior written permission. +# +# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" +# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE +# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE +# ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE +# LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR +# CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF +# SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS +# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN +# CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) +# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE +# POSSIBILITY OF SUCH DAMAGE. +# + +import unittest + +import config +import ipv6 +import node + +LEADER = 1 +ROUTER = 2 + + +class TestIPv6SourceSelection(unittest.TestCase): + + def setUp(self): + self.simulator = config.create_default_simulator() + + self.nodes = {} + for i in range(1, 3): + self.nodes[i] = node.Node(i, simulator=self.simulator) + + self.nodes[LEADER].set_panid(0xcafe) + self.nodes[LEADER].set_mode('rsdn') + self.nodes[LEADER].add_whitelist(self.nodes[ROUTER].get_addr64()) + self.nodes[LEADER].enable_whitelist() + + self.nodes[ROUTER].set_panid(0xcafe) + self.nodes[ROUTER].set_mode('rsdn') + self.nodes[ROUTER].add_whitelist(self.nodes[LEADER].get_addr64()) + self.nodes[ROUTER].enable_whitelist() + self.nodes[ROUTER].set_router_selection_jitter(1) + + def tearDown(self): + for n in list(self.nodes.values()): + n.stop() + n.destroy() + self.simulator.stop() + + def test(self): + self.nodes[LEADER].start() + self.simulator.go(5) + self.assertEqual(self.nodes[LEADER].get_state(), 'leader') + + self.nodes[ROUTER].start() + self.simulator.go(5) + self.assertEqual(self.nodes[ROUTER].get_state(), 'router') + + leader_aloc = self.nodes[LEADER].get_addr_leader_aloc() + leader_mleid = self.nodes[LEADER].get_mleid() + leader_rloc = self.nodes[LEADER].get_rloc() + leader_linklocal = self.nodes[LEADER].get_linklocal() + multicast_addr = 'ff03::1' + external_addr = '2007::1' + + router_rloc = self.nodes[ROUTER].get_rloc() + router_linklocal = self.nodes[ROUTER].get_linklocal() + router_mleid = self.nodes[ROUTER].get_mleid() + + # Source check - RLOC source for RLOC destination + self.assertTrue(self.nodes[ROUTER].ping(leader_rloc)) + router_msgs = self.simulator.get_messages_sent_by(ROUTER) + msg = router_msgs.get_icmp_message(ipv6.ICMP_ECHO_REQUEST) + msg.assertSentFromSourceAddress(router_rloc) + + # Source check - ML-EID source for ALOC destination + self.assertTrue(self.nodes[ROUTER].ping(leader_aloc)) + router_msgs = self.simulator.get_messages_sent_by(ROUTER) + msg = router_msgs.get_icmp_message(ipv6.ICMP_ECHO_REQUEST) + msg.assertSentFromSourceAddress(router_mleid) + + # Source check - ML-EID source for ML-EID destination + self.assertTrue(self.nodes[ROUTER].ping(leader_mleid)) + router_msgs = self.simulator.get_messages_sent_by(ROUTER) + msg = router_msgs.get_icmp_message(ipv6.ICMP_ECHO_REQUEST) + msg.assertSentFromSourceAddress(router_mleid) + + # Source check - link local source source for link local address + self.assertTrue(self.nodes[ROUTER].ping(leader_linklocal)) + router_msgs = self.simulator.get_messages_sent_by(ROUTER) + msg = router_msgs.get_icmp_message(ipv6.ICMP_ECHO_REQUEST) + msg.assertSentFromSourceAddress(router_linklocal) + + # Source check - ML-EID source for realmlocal multicast destination + self.assertTrue(self.nodes[ROUTER].ping(multicast_addr)) + router_msgs = self.simulator.get_messages_sent_by(ROUTER) + msg = router_msgs.get_icmp_message(ipv6.ICMP_ECHO_REQUEST) + msg.assertSentFromSourceAddress(router_mleid) + + # GUA and default gateway + self.nodes[LEADER].add_prefix('2001::/64', 'paros') + self.nodes[LEADER].register_netdata() + self.simulator.go(5) + + # Set lowpan context of sniffer + self.simulator.set_lowpan_context(1, '2001::/64') + + # Flushes message queue before next ping + router_msgs = self.simulator.get_messages_sent_by(ROUTER) + + # Source check - GUA source for GUA destination + leader_gua = self.nodes[LEADER].get_addr("2001::/64") + router_gua = self.nodes[ROUTER].get_addr("2001::/64") + self.assertTrue(leader_gua is not None) + self.assertTrue(router_gua is not None) + self.assertTrue(self.nodes[ROUTER].ping(leader_gua)) + router_msgs = self.simulator.get_messages_sent_by(ROUTER) + msg = router_msgs.get_icmp_message(ipv6.ICMP_ECHO_REQUEST) + msg.assertSentFromSourceAddress(router_gua) + + # Source check - GUA source for external address (default route) + self.nodes[ROUTER].ping(external_addr) + router_msgs = self.simulator.get_messages_sent_by(ROUTER) + msg = router_msgs.get_icmp_message(ipv6.ICMP_ECHO_REQUEST) + msg.assertSentFromSourceAddress(router_gua) + + +if __name__ == '__main__': + unittest.main()