[travis] test parent selection rules (#4423)

This commit has five scenarios to test the four Mesh Impacting
Criteria and one Child Impacting Criterion (version).
This commit is contained in:
Rongli Sun
2020-02-25 14:34:53 -08:00
committed by Jonathan Hui
parent 835d9e2505
commit b339a8946b
8 changed files with 448 additions and 46 deletions
+8
View File
@@ -34,6 +34,14 @@ from enum import IntEnum
import ipaddress
# Map of 2 bits of parent priority.
pp_map = {1: 1, 0: 0, 3: -1, 2: -2}
# Get the signed parent priority from the byte that parent priority is in.
def map_pp(self, pp_byte):
return pp_map[((pp_byte & 0xC0) >> 6)]
def expect_the_same_class(self, other):
if not isinstance(other, self.__class__):
+9 -5
View File
@@ -573,7 +573,7 @@ class Connectivity(object):
def __init__(
self,
pp,
pp_byte,
link_quality_3,
link_quality_2,
link_quality_1,
@@ -583,7 +583,7 @@ class Connectivity(object):
sed_buffer_size=None,
sed_datagram_count=None,
):
self._pp = pp
self._pp_byte = pp_byte
self._link_quality_3 = link_quality_3
self._link_quality_2 = link_quality_2
self._link_quality_1 = link_quality_1
@@ -593,9 +593,13 @@ class Connectivity(object):
self._sed_buffer_size = sed_buffer_size
self._sed_datagram_count = sed_datagram_count
@property
def pp_byte(self):
return self._pp_byte
@property
def pp(self):
return self._pp
return common.map_pp(self, self._pp_byte)
@property
def link_quality_3(self):
@@ -667,7 +671,7 @@ class Connectivity(object):
class ConnectivityFactory:
def parse(self, data, message_info):
pp = ord(data.read(1)) & 0x03
pp_byte = ord(data.read(1))
link_quality_3 = ord(data.read(1))
link_quality_2 = ord(data.read(1))
link_quality_1 = ord(data.read(1))
@@ -685,7 +689,7 @@ class ConnectivityFactory:
sed_datagram_count = None
return Connectivity(
pp,
pp_byte,
link_quality_3,
link_quality_2,
link_quality_1,
+51 -9
View File
@@ -42,10 +42,17 @@ import unittest
class Node:
def __init__(self, nodeid, is_mtd=False, simulator=None):
def __init__(self, nodeid, is_mtd=False, simulator=None, version=None):
self.nodeid = nodeid
self.verbose = int(float(os.getenv('VERBOSE', 0)))
self.node_type = os.getenv('NODE_TYPE', 'sim')
self.env_version = os.getenv('THREAD_VERSION', '1.1')
if version is not None:
self.version = version
else:
self.version = self.env_version
self.simulator = simulator
if self.simulator:
self.simulator.add_node(self)
@@ -69,11 +76,18 @@ class Node:
""" Initialize a simulation node. """
if 'OT_CLI_PATH' in os.environ:
cmd = os.environ['OT_CLI_PATH']
elif 'top_builddir' in os.environ:
elif (self.version == '1.1' and self.version != self.env_version):
# Posix app
if 'OT_CLI_PATH_1_1' in os.environ:
cmd = os.environ['OT_CLI_PATH_1_1']
elif ('top_builddir_1_1') in os.environ:
srcdir = os.environ['top_builddir_1_1']
cmd = '%s/examples/apps/cli/ot-cli-%s' % (srcdir, mode)
elif ('top_builddir') in os.environ:
srcdir = os.environ['top_builddir']
cmd = '%s/examples/apps/cli/ot-cli-%s' % (srcdir, mode)
else:
cmd = './ot-cli-%s' % mode
cmd = '%s/ot-cli-%s' % (self.version, mode)
if 'RADIO_DEVICE' in os.environ:
cmd += ' -v %s' % os.environ['RADIO_DEVICE']
@@ -107,15 +121,33 @@ class Node:
os.environ['OT_NCP_PATH'],
args,
)
elif "top_builddir" in os.environ:
builddir = os.environ['top_builddir']
cmd = 'spinel-cli.py -p "%s/examples/apps/ncp/ot-ncp-%s%s" -n' % (
builddir,
mode,
elif (self.version == '1.1' and self.version != self.env_version):
if 'OT_NCP_PATH_1_1' in os.environ:
cmd = 'spinel-cli.py -p "%s%s" -n' % (
os.environ['OT_NCP_PATH_1_1'],
args,
)
elif ('top_builddir_1_1') in os.environ:
srcdir = os.environ['top_builddir_1_1']
cmd = '%s/examples/apps/ncp/ot-ncp-%s' % (srcdir, mode)
cmd = 'spinel-cli.py -p "%s%s" -n' % (
cmd,
args,
)
elif ('top_builddir') in os.environ:
srcdir = os.environ['top_builddir']
cmd = '%s/examples/apps/ncp/ot-ncp-%s' % (srcdir, mode)
cmd = 'spinel-cli.py -p "%s%s" -n' % (
cmd,
args,
)
else:
cmd = 'spinel-cli.py -p "./ot-ncp-%s%s" -n' % (mode, args)
cmd = 'spinel-cli.py -p "%s/ot-ncp-%s%s" -n' % (
self.version,
mode,
args,
)
cmd += ' %d' % nodeid
print("%s" % cmd)
@@ -312,6 +344,11 @@ class Node:
self.send_command(cmd)
self._expect('Done')
def set_link_quality(self, addr, lqi):
cmd = 'macfilter rss add-lqi %s %s' % (addr, lqi)
self.send_command(cmd)
self._expect('Done')
def remove_whitelist(self, addr):
cmd = 'macfilter addr remove %s' % addr
self.send_command(cmd)
@@ -436,6 +473,11 @@ class Node:
self.send_command(cmd)
self._expect('Done')
def set_parent_priority(self, priority):
cmd = 'parentpriority %d' % priority
self.send_command(cmd)
self._expect('Done')
def get_partition_id(self):
self.send_command('leaderpartitionid')
i = self._expect(r'(\d+)\r?\n')
+12 -12
View File
@@ -294,7 +294,7 @@ def any_u():
def any_pp():
return random.getrandbits(2)
return (random.getrandbits(2) << 6)
def any_link_quality_3():
@@ -1115,9 +1115,9 @@ class TestConnectivity(unittest.TestCase):
def test_should_return_pp_value_when_pp_property_is_called(self):
# GIVEN
pp = any_pp()
pp_byte = any_pp()
connectivity_obj = mle.Connectivity(pp, any_link_quality_3(),
connectivity_obj = mle.Connectivity(pp_byte, any_link_quality_3(),
any_link_quality_2(),
any_link_quality_1(),
any_leader_cost(),
@@ -1130,7 +1130,7 @@ class TestConnectivity(unittest.TestCase):
actual_pp = connectivity_obj.pp
# THEN
self.assertEqual(pp, actual_pp)
self.assertEqual(common.map_pp(self, pp_byte), actual_pp)
def test_should_return_link_quality_3_value_when_link_quality_3_property_is_called(
self):
@@ -1294,7 +1294,7 @@ class TestConnectivityFactory(unittest.TestCase):
def test_should_create_Connectivity_from_bytearray_when_parse_method_is_called(
self):
# GIVEN
pp = any_pp()
pp_byte = any_pp()
link_quality_3 = any_link_quality_3()
link_quality_2 = any_link_quality_2()
link_quality_1 = any_link_quality_1()
@@ -1307,8 +1307,8 @@ class TestConnectivityFactory(unittest.TestCase):
factory = mle.ConnectivityFactory()
data = bytearray([
pp, link_quality_3, link_quality_2, link_quality_1, leader_cost,
id_sequence, active_routers
pp_byte, link_quality_3, link_quality_2, link_quality_1,
leader_cost, id_sequence, active_routers
]) + struct.pack(">H", sed_buffer_size) + bytearray(
[sed_datagram_count])
@@ -1317,7 +1317,7 @@ class TestConnectivityFactory(unittest.TestCase):
# THEN
self.assertTrue(isinstance(actual_connectivity, mle.Connectivity))
self.assertEqual(pp, actual_connectivity.pp)
self.assertEqual(common.map_pp(self, pp_byte), actual_connectivity.pp)
self.assertEqual(link_quality_3, actual_connectivity.link_quality_3)
self.assertEqual(link_quality_2, actual_connectivity.link_quality_2)
self.assertEqual(link_quality_1, actual_connectivity.link_quality_1)
@@ -1331,7 +1331,7 @@ class TestConnectivityFactory(unittest.TestCase):
def test_should_create_Connectivity_without_sed_data_when_parse_method_is_called(
self):
# GIVEN
pp = any_pp()
pp_byte = any_pp()
link_quality_3 = any_link_quality_3()
link_quality_2 = any_link_quality_2()
link_quality_1 = any_link_quality_1()
@@ -1344,8 +1344,8 @@ class TestConnectivityFactory(unittest.TestCase):
factory = mle.ConnectivityFactory()
data = bytearray([
pp, link_quality_3, link_quality_2, link_quality_1, leader_cost,
id_sequence, active_routers
pp_byte, link_quality_3, link_quality_2, link_quality_1,
leader_cost, id_sequence, active_routers
])
# WHEN
@@ -1353,7 +1353,7 @@ class TestConnectivityFactory(unittest.TestCase):
# THEN
self.assertTrue(isinstance(actual_connectivity, mle.Connectivity))
self.assertEqual(pp, actual_connectivity.pp)
self.assertEqual(common.map_pp(self, pp_byte), actual_connectivity.pp)
self.assertEqual(link_quality_3, actual_connectivity.link_quality_3)
self.assertEqual(link_quality_2, actual_connectivity.link_quality_2)
self.assertEqual(link_quality_1, actual_connectivity.link_quality_1)
+18 -2
View File
@@ -38,7 +38,7 @@ DEFAULT_PARAMS = {
'mode': 'rsdn',
'panid': 0xface,
'whitelist': None,
'version': '1.2'
'version': '1.2',
}
"""Default configurations when creating nodes."""
@@ -68,7 +68,12 @@ class TestCase(unittest.TestCase):
params = DEFAULT_PARAMS.copy()
initial_topology[i] = params
self.nodes[i] = Node(i, params['is_mtd'], simulator=self.simulator)
self.nodes[i] = Node(
i,
params['is_mtd'],
simulator=self.simulator,
version=params['version'],
)
self.nodes[i].set_panid(params['panid'])
self.nodes[i].set_mode(params['mode'])
self.nodes[i].set_addr64(format(EXTENDED_ADDRESS_BASE + i, '016x'))
@@ -101,3 +106,14 @@ class TestCase(unittest.TestCase):
"""
for i in list(self.nodes.keys()):
self.simulator.get_messages_sent_by(i)
def flush_nodes(self, nodes):
"""Flush away all captured messages of specified nodes.
Args:
nodes (list): nodes whose messages to flush.
"""
for i in nodes:
if i in list(self.nodes.keys()):
self.simulator.get_messages_sent_by(i)
+307
View File
@@ -0,0 +1,307 @@
#!/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 thread_cert
import mle
LEADER_1_2 = 1
ROUTER_1_1 = 2
REED_1_2 = 3
ROUTER_1_2 = 4
REED_1_1 = 5
MED_1_1 = 6
MED_1_2 = 7
# Topology
# (lq:2) (pp:1)
# REED_1_2 ----- ROUTER_1_2
# | \ / | \
# | \/ REED_1_1 \
# (lq:2) | / \ / `router` \
# | (lq:2) \ \
# | / / \ \
# LEADER_1_2 --- ROUTER_1_1 -- MED_1_2
# \ |
# \ |
# \ |
# MED_1_1
#
# 1) Bring up LEADER_1_2 and ROUTER_1_1,
# 2) Config link quality (LEADER_1_2->REED_1_2) as 2, bring up REED_1_2 which would attach to ROUTER_1_1
# due to higher two-way link quality,
# 3) Config link quality(LEADER_1_2->ROUTER_1_2) and link quality(REED_1_2->ROUTER_1_2) as 2, bring up
# ROUTER_1_2 which would attach to LEADER_1_2 due to active router is preferred,
# 4) Config parent priority as 1 on ROUTER_1_2, bring up REED_1_1 which would attach to ROUTER_1_2 due to
# higher parent priority,
# 5) Upgrade REED_1_1 to `router` role, bring up MED_1_1 which would attach to LEADER_1_2 which has higher
# link quality of 3,
# 6) Config parent priority as 1 on ROUTER_1_1, bring up MED_1_2 which would attach to ROUTER_1_2 due to
# higher version
#
class TestParentSelection(thread_cert.TestCase):
topology = {
LEADER_1_2: {
'version': '1.2',
'whitelist': [REED_1_2, ROUTER_1_2, REED_1_1, ROUTER_1_1, MED_1_1],
},
ROUTER_1_1: {
'version': '1.1',
'whitelist': [LEADER_1_2, REED_1_2, MED_1_2, MED_1_1],
},
REED_1_2: {
'version': '1.2',
'whitelist': [ROUTER_1_2, ROUTER_1_1, LEADER_1_2],
},
ROUTER_1_2: {
'version': '1.2',
'whitelist': [REED_1_2, MED_1_2, REED_1_1, LEADER_1_2],
},
REED_1_1: {
'version': '1.1',
'whitelist': [ROUTER_1_2, LEADER_1_2]
},
MED_1_1: {
'mode': 'rs',
'version': '1.1',
'whitelist': [LEADER_1_2, ROUTER_1_1],
},
MED_1_2: {
'mode': 'rs',
'version': '1.2',
'whitelist': [ROUTER_1_1, ROUTER_1_2],
},
}
"""All nodes are created with default configurations"""
def test(self):
self.nodes[LEADER_1_2].start()
self.simulator.go(5)
self.assertEqual(self.nodes[LEADER_1_2].get_state(), 'leader')
self.nodes[ROUTER_1_1].set_router_selection_jitter(1)
self.nodes[ROUTER_1_1].start()
self.simulator.go(5)
self.assertEqual(self.nodes[ROUTER_1_1].get_state(), 'router')
# Mesh Impacting Criteria - Highest Two-way link quality
# REED_1_2 would attach to ROUTER_1_1
# Attach to ROUTER_1_1 which has highest two-way link quality
# Flush relative message queues
self.flush_nodes([LEADER_1_2, ROUTER_1_1])
self.nodes[LEADER_1_2].set_link_quality(
self.nodes[REED_1_2].get_addr64(), 2)
self.nodes[REED_1_2].set_router_selection_jitter(1)
self.nodes[REED_1_2].set_router_upgrade_threshold(1)
self.nodes[REED_1_2].start()
self.simulator.go(5)
self.assertEqual(self.nodes[REED_1_2].get_state(), 'child')
# Check Parent Response
messages = self.simulator.get_messages_sent_by(ROUTER_1_1)
parent_prefer = messages.next_mle_message(
mle.CommandType.PARENT_RESPONSE)
assert (parent_prefer), "Error: Expected parent response not found"
messages = self.simulator.get_messages_sent_by(LEADER_1_2)
parent_cmp = messages.next_mle_message(mle.CommandType.PARENT_RESPONSE)
assert (parent_cmp), "Error: Expected parent response not found"
# Known that link margin for link quality 3 is 80 and link quality 2 is 15
assert ((parent_prefer.get_mle_message_tlv(mle.LinkMargin).link_margin -
parent_cmp.get_mle_message_tlv(mle.LinkMargin).link_margin) >
20)
# Check Child Id Request
messages = self.simulator.get_messages_sent_by(REED_1_2)
msg = messages.next_mle_message(mle.CommandType.CHILD_ID_REQUEST)
msg.assertSentToNode(self.nodes[ROUTER_1_1])
# Mesh Impacting Criteria - Active Routers over REEDs
# ROUTER_1_2 would attach to LEADER_1_2
# Link quality configuration, so that REED_1_2 has the chance to respond
# Flush relative message queues
self.flush_nodes([LEADER_1_2, REED_1_2])
self.nodes[LEADER_1_2].set_link_quality(
self.nodes[ROUTER_1_2].get_addr64(), 2)
self.nodes[REED_1_2].set_link_quality(
self.nodes[ROUTER_1_2].get_addr64(), 2)
self.nodes[ROUTER_1_2].set_router_selection_jitter(1)
self.nodes[ROUTER_1_2].start()
self.simulator.go(5)
self.assertEqual(self.nodes[ROUTER_1_2].get_state(), 'router')
# Check Parent Response
messages = self.simulator.get_messages_sent_by(LEADER_1_2)
# Skip first response for first parent request
assert messages.next_mle_message(mle.CommandType.PARENT_RESPONSE)
parent_prefer = messages.next_mle_message(
mle.CommandType.PARENT_RESPONSE)
assert (parent_prefer), "Error: Expected parent response not found"
messages = self.simulator.get_messages_sent_by(REED_1_2)
parent_cmp = messages.next_mle_message(mle.CommandType.PARENT_RESPONSE)
assert (parent_cmp), "Error: Expected parent response not found"
assert (parent_prefer.get_mle_message_tlv(
mle.LinkMargin).link_margin == parent_cmp.get_mle_message_tlv(
mle.LinkMargin).link_margin)
# Check Child Id Request
messages = self.simulator.get_messages_sent_by(ROUTER_1_2)
msg = messages.next_mle_message(mle.CommandType.CHILD_ID_REQUEST)
msg.assertSentToNode(self.nodes[LEADER_1_2])
# Mesh Impacting Criteria - Highest Parent Priority value in the Connectivity TLV
# REED_1_1 would attach to ROUTER_1_2
# Flush relative message queues
self.flush_nodes([LEADER_1_2, ROUTER_1_2])
self.nodes[ROUTER_1_2].set_parent_priority(1)
self.nodes[REED_1_1].set_router_selection_jitter(1)
self.nodes[REED_1_1].set_router_upgrade_threshold(1)
self.nodes[REED_1_1].start()
self.simulator.go(5)
self.assertEqual(self.nodes[REED_1_1].get_state(), 'child')
# Check Parent Response
messages = self.simulator.get_messages_sent_by(ROUTER_1_2)
parent_prefer = messages.next_mle_message(
mle.CommandType.PARENT_RESPONSE)
assert (parent_prefer), "Error: Expected parent response not found"
messages = self.simulator.get_messages_sent_by(LEADER_1_2)
parent_cmp = messages.next_mle_message(mle.CommandType.PARENT_RESPONSE)
assert (parent_cmp), "Error: Expected parent response not found"
assert (parent_prefer.get_mle_message_tlv(
mle.LinkMargin).link_margin == parent_cmp.get_mle_message_tlv(
mle.LinkMargin).link_margin)
assert (parent_prefer.get_mle_message_tlv(mle.Connectivity).pp >
parent_cmp.get_mle_message_tlv(mle.Connectivity).pp)
# Check Child Id Request
messages = self.simulator.get_messages_sent_by(REED_1_1)
msg = messages.next_mle_message(mle.CommandType.CHILD_ID_REQUEST)
msg.assertSentToNode(self.nodes[ROUTER_1_2])
# Mesh Impacting Criteria - Router with the most high-quality neighbors
# (Link Quality 3 field in the Connectivity TLV)
# MED_1_1 would attach to LEADER_1_2
self.nodes[REED_1_1].set_state('router')
self.simulator.go(5)
self.assertEqual(self.nodes[REED_1_1].get_state(), 'router')
# Flush relative message queues
self.flush_nodes([LEADER_1_2, ROUTER_1_1])
self.nodes[MED_1_1].start()
self.simulator.go(5)
self.assertEqual(self.nodes[MED_1_1].get_state(), 'child')
# Check Parent Response
messages = self.simulator.get_messages_sent_by(LEADER_1_2)
parent_prefer = messages.next_mle_message(
mle.CommandType.PARENT_RESPONSE)
assert (parent_prefer), "Error: Expected parent response not found"
messages = self.simulator.get_messages_sent_by(ROUTER_1_1)
parent_cmp = messages.next_mle_message(mle.CommandType.PARENT_RESPONSE)
assert (parent_cmp), "Error: Expected parent response not found"
assert (parent_prefer.get_mle_message_tlv(
mle.LinkMargin).link_margin == parent_cmp.get_mle_message_tlv(
mle.LinkMargin).link_margin)
assert (parent_prefer.get_mle_message_tlv(
mle.Connectivity).pp == parent_cmp.get_mle_message_tlv(
mle.Connectivity).pp)
assert (parent_prefer.get_mle_message_tlv(
mle.Connectivity).link_quality_3 > parent_cmp.get_mle_message_tlv(
mle.Connectivity).link_quality_3)
# Check Child Id Request
messages = self.simulator.get_messages_sent_by(MED_1_1)
msg = messages.next_mle_message(mle.CommandType.CHILD_ID_REQUEST)
msg.assertSentToNode(self.nodes[LEADER_1_2])
# Child Impacting Criteria - A Version number in the Version TLV
# equal to or higher than the version that implements features
# desirable to the Child MED_1_2 would attach to ROUTER_1_2
# Flush relative message queues
self.flush_nodes([ROUTER_1_2, ROUTER_1_1])
self.nodes[ROUTER_1_1].set_parent_priority(1)
self.nodes[MED_1_2].start()
self.simulator.go(5)
self.assertEqual(self.nodes[MED_1_2].get_state(), 'child')
# Check Parent Response
messages = self.simulator.get_messages_sent_by(ROUTER_1_2)
parent_prefer = messages.next_mle_message(
mle.CommandType.PARENT_RESPONSE)
assert (parent_prefer), "Error: Expected parent response not found"
messages = self.simulator.get_messages_sent_by(ROUTER_1_1)
parent_cmp = messages.next_mle_message(mle.CommandType.PARENT_RESPONSE)
assert (parent_cmp), "Error: Expected parent response not found"
assert (parent_prefer.get_mle_message_tlv(
mle.LinkMargin).link_margin == parent_cmp.get_mle_message_tlv(
mle.LinkMargin).link_margin)
assert (parent_prefer.get_mle_message_tlv(
mle.Connectivity).pp == parent_cmp.get_mle_message_tlv(
mle.Connectivity).pp)
assert (parent_prefer.get_mle_message_tlv(
mle.Connectivity).link_quality_3 == parent_cmp.get_mle_message_tlv(
mle.Connectivity).link_quality_3)
assert (parent_prefer.get_mle_message_tlv(mle.Version).version >
parent_cmp.get_mle_message_tlv(mle.Version).version)
# Check Child Id Request
messages = self.simulator.get_messages_sent_by(MED_1_2)
msg = messages.next_mle_message(mle.CommandType.CHILD_ID_REQUEST)
msg.assertSentToNode(self.nodes[ROUTER_1_2])
if __name__ == '__main__':
unittest.main()