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

[tests] add ipv6 source selection test (#4380)

Browse Source
This commit is contained in:
Rongli Sun
2019-12-06 08:51:23 -08:00
committed by Jonathan Hui
parent bf4dae0c18
commit 76f847babc
4 changed files with 186 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/scripts/thread-cert/Makefile.am
+5 -2
View File
@@ -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 \
tests/scripts/thread-cert/message.py
+6
View File
@@ -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
tests/scripts/thread-cert/node.py
+30
View File
@@ -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()
tests/scripts/thread-cert/test_ipv6_source_selection.py
Executable
+145
View File
@@ -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()