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

[toranj] test covering multicast traffic over multi-hop (#2983)

Browse Source 
This commit adds a test-case under `toranj` covering multicast
traffic exchange over a multi-hop network to the following multicast
IPv6 addresses: link-local/mesh-local all-nodes address, link-local
/mesh-local all-routers address, link-local/mesh-local all-thread-
nodes address, and user-specified multicast address.
This commit is contained in:
Abtin Keshavarzian
2018-08-27 16:00:11 -07:00
committed by Jonathan Hui
parent 48cbbfd32a
commit 2f605e9953
3 changed files with 272 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/toranj/start.sh
+1
View File
@@ -136,6 +136,7 @@ run test-019-inform-previous-parent.py
run test-020-router-table.py
run test-021-address-cache-table.py
run test-022-multicast-ip6-address.py
run test-023-multicast-traffic.py
run test-100-mcu-power-state.py
run test-600-channel-manager-properties.py
run test-601-channel-manager-channel-change.py
tests/toranj/test-023-multicast-traffic.py
+229
View File
@@ -0,0 +1,229 @@
#!/usr/bin/env python
#
# Copyright (c) 2018, 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 time
import wpan
from wpan import verify
#-----------------------------------------------------------------------------------------------------------------------
# Test description: Multicast traffic
#
# Network topology
#
# r1 ---- r2 ---- r3 ---- r4
# | |
# | |
# fed sed
#
# Test covers the following multicast traffic:
#
# - r2 =>> link-local all-nodes. Expected to receive on [r1, r2, r3, fed].
# - r3 =>> mesh-local all-nodes. Expected to receive on [r1, r2, r3, r4, fed].
# - r3 =>> link-local all-routers. Expected to receive on [r2, r3, r4].
# - r3 =>> mesh-local all-routers. Expected to receive on all routers.
# - r1 =>> link-local all-thread. Expected to receive on [r1, r2].
# - fed =>> mesh-local all-thread. Expected to receive on all nodes.
# - r1 =>> specific address (on r2 and sed). Expected to receive on [r2, sed].
# - Check behavior with different multicast hop limit values (1-hop up to 4-hops).
#
test_name = __file__[:-3] if __file__.endswith('.py') else __file__
print '-' * 120
print 'Starting \'{}\''.format(test_name)
#-----------------------------------------------------------------------------------------------------------------------
# Utility functions
def send_mcast(src_node, src_addr, mcast_addr, recving_nodes, non_recving_nodes=[], msg_len=30, mcast_hops=5):
"""
Send a multicast message with given `len` from `src_node` using `src_addr` to the multicast address `mcast_addr`.
Verify that the message is received on all nodes in `recving_nodes` list and that it is not received on all
nodes in `non_recving_nodes` list.
"""
sender = src_node.prepare_tx(src_addr, mcast_addr, msg_len, mcast_hops=mcast_hops)
recvers = [node.prepare_rx(sender) for node in recving_nodes]
listeners = [node.preapre_listener(sender.dst_port, timeout=0.5) for node in non_recving_nodes]
wpan.Node.perform_async_tx_rx()
verify(sender.was_successful)
for recvr in recvers:
verify(recvr.was_successful)
for lsnr in listeners:
# `all_rx_msg` contains a list of (msg_content, (src_addr, src_port)).
verify(len(lsnr.all_rx_msg) == 0 or
all([msg[1][0] != sender.src_addr and msg[1][1] != sender.src_port for msg in lsnr.all_rx_msg]))
#-----------------------------------------------------------------------------------------------------------------------
# Creating `wpan.Nodes` instances
speedup = 4
wpan.Node.set_time_speedup_factor(speedup)
r1 = wpan.Node()
r2 = wpan.Node()
r3 = wpan.Node()
r4 = wpan.Node()
fed = wpan.Node()
sed = wpan.Node()
all_routers = [r1, r2, r3, r4]
all_nodes = all_routers + [fed, sed]
#-----------------------------------------------------------------------------------------------------------------------
# Init all nodes
wpan.Node.init_all_nodes(disable_logs=False)
#-----------------------------------------------------------------------------------------------------------------------
# Build network topology
#
# Test topology:
#
# r1 ---- r2 ---- r3 ---- r4
# | |
# | |
# fed sed
#
r1.form("mcast-traffic")
r1.whitelist_node(r2)
r2.whitelist_node(r1)
r2.join_node(r1, wpan.JOIN_TYPE_ROUTER)
r2.whitelist_node(fed)
fed.whitelist_node(r2)
fed.join_node(r2, wpan.JOIN_TYPE_END_DEVICE)
r2.whitelist_node(r3)
r3.whitelist_node(r2)
r3.join_node(r2, wpan.JOIN_TYPE_ROUTER)
r3.whitelist_node(r4)
r4.whitelist_node(r3)
r4.join_node(r3, wpan.JOIN_TYPE_ROUTER)
r4.whitelist_node(sed)
sed.whitelist_node(r4)
sed.join_node(r4, wpan.JOIN_TYPE_SLEEPY_END_DEVICE)
sed.set(wpan.WPAN_POLL_INTERVAL, '600')
#-----------------------------------------------------------------------------------------------------------------------
# Test implementation
ml1 = r1.get(wpan.WPAN_IP6_MESH_LOCAL_ADDRESS)[1:-1]
ll1 = r1.get(wpan.WPAN_IP6_LINK_LOCAL_ADDRESS)[1:-1]
ml2 = r2.get(wpan.WPAN_IP6_MESH_LOCAL_ADDRESS)[1:-1]
ll2 = r2.get(wpan.WPAN_IP6_LINK_LOCAL_ADDRESS)[1:-1]
ml3 = r3.get(wpan.WPAN_IP6_MESH_LOCAL_ADDRESS)[1:-1]
ll3 = r3.get(wpan.WPAN_IP6_LINK_LOCAL_ADDRESS)[1:-1]
ml4 = r4.get(wpan.WPAN_IP6_MESH_LOCAL_ADDRESS)[1:-1]
ll4 = r4.get(wpan.WPAN_IP6_LINK_LOCAL_ADDRESS)[1:-1]
ml_fed = fed.get(wpan.WPAN_IP6_MESH_LOCAL_ADDRESS)[1:-1]
ll_fed = fed.get(wpan.WPAN_IP6_LINK_LOCAL_ADDRESS)[1:-1]
# Multicast addresses
ll_all_nodes = "ff02::1"
ml_all_nodes = "ff03::1"
ml_all_mlp_fwder_nodes = "ff03::fc"
ll_all_routers = "ff02::2"
ml_all_routers = "ff03::2"
ml_prefix = r1.get(wpan.WPAN_IP6_MESH_LOCAL_PREFIX)[1:-1].split('/')[0]
ll_all_thread_nodes_addr = 'ff32:40:' + ml_prefix + '1'
ml_all_thread_nodes_addr = 'ff33:40:' + ml_prefix + '1'
#
# r1 ---- r2 ---- r3 ---- r4
# | |
# | |
# fed sed
#
# r2 =>> link-local all-nodes.
send_mcast(r2, ll2, ll_all_nodes, [r1, r2, r3, fed], [r4, sed])
# r3 =>> mesh-local all-nodes.
send_mcast(r3, ml3, ml_all_nodes, [r1, r2, r3, r4, fed])
# r3 =>> link-local all-routers.
send_mcast(r3, ml3, ll_all_routers, [r2, r3, r4], [r1, fed, sed])
# r3 =>> mesh-local all-routers.
send_mcast(r3, ml3, ml_all_routers, all_routers, [sed])
# r1 =>> link-local all-thread.
send_mcast(r1, ll1, ll_all_thread_nodes_addr, [r1, r2], [fed, r3, r4, sed])
# fed =>> mesh-local all-thread.
send_mcast(fed, ml_fed, ml_all_thread_nodes_addr, all_nodes)
#- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
# Send a large multicast message (requiring MAC level fragmentations)
send_mcast(r3, ml3, ml_all_thread_nodes_addr, all_nodes, msg_len=400)
#- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
# Check the hop limit behavior
# r1 =>> mesh-local all-thread (one hop)
send_mcast(r1, ml1, ml_all_thread_nodes_addr, [r1, r2], [fed, r3, r4, sed], mcast_hops=1)
# r1 =>> mesh-local all-thread (two hops)
send_mcast(r1, ml1, ml_all_thread_nodes_addr, [r1, r2, fed, r3], [r4, sed], mcast_hops=2)
# r1 =>> mesh-local all-thread (three hops)
send_mcast(r1, ml1, ml_all_thread_nodes_addr, [r1, r2, fed, r3, r4], [sed], mcast_hops=3)
# r1 =>> mesh-local all-thread (four hops)
send_mcast(r1, ml1, ml_all_thread_nodes_addr, [r1, r2, fed, r3, r4, sed], mcast_hops=4)
#- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
# Subscribe to a specific multicast address on r2 and sed
mcast_addr = "ff03::114"
r2.add(wpan.WPAN_IP6_MULTICAST_ADDRESSES, mcast_addr)
sed.add(wpan.WPAN_IP6_MULTICAST_ADDRESSES, mcast_addr)
time.sleep(1)
# r1 =>> specific address
send_mcast(r1, ml1, mcast_addr, [r2, sed], [r1, r3, r4, fed])
#-----------------------------------------------------------------------------------------------------------------------
# Test finished
wpan.Node.finalize_all_nodes()
print '\'{}\' passed.'.format(test_name)
tests/toranj/wpan.py
+42 -14
View File
@@ -512,7 +512,7 @@ class Node(object):
@classmethod
def init_all_nodes(cls, disable_logs=True, wait_time=15):
"""Issues a `wpanctl.leave` on all `Node` objects and waits for them to be ready"""
random.seed(12345)
random.seed(123456)
time.sleep(0.5)
for node in Node._all_nodes:
start_time = time.time()
@@ -557,7 +557,7 @@ class Node(object):
class _NodeError(Exception):
pass
def prepare_tx(self, src, dst, data=40, count=1):
def prepare_tx(self, src, dst, data=40, count=1, mcast_hops=None):
"""Prepares an IPv6 msg transmission.
- `src` and `dst` can be either a string containing IPv6 address, or a tuple (ipv6 address as string, port),
@@ -565,6 +565,7 @@ class Node(object):
- `data` can be either a string containing the message to be sent, or an int indicating size of the message (a
random message with the given length will be used).
- `count` gives number of times the message will be sent (default is 1).
- `mcast_hops` specifies multicast hop limit (only applicable for multicast tx).
Returns an `AsyncSender` object.
@@ -590,19 +591,15 @@ class Node(object):
else:
msg = data
return AsyncSender(self, src_addr, src_port, dst_addr, dst_port, msg, count)
def prepare_rx(self, sender):
"""Prepare to receive messages from a sender (an `AsyncSender`)"""
local_port = sender.dst_port
return AsyncSender(self, src_addr, src_port, dst_addr, dst_port, msg, count, mcast_hops)
def _get_receiver(self, local_port):
# Gets or creates a receiver (an `AsyncReceiver`) tied to given port number
if local_port in self._recvers:
receiver = self._recvers[local_port]
else:
receiver = AsyncReceiver(self, local_port)
self._recvers[local_port] = receiver
receiver._add_sender(sender.src_addr, sender.src_port, sender.msg, sender.count)
return receiver
def _remove_recver(self, recvr):
@@ -611,9 +608,21 @@ class Node(object):
if local_port in self._recvers:
del self._recvers[local_port]
def prepare_rx(self, sender):
"""Prepare to receive messages from a sender (an `AsyncSender`)"""
receiver = self._get_receiver(sender.dst_port)
receiver._add_sender(sender.src_addr, sender.src_port, sender.msg, sender.count)
return receiver
def preapre_listener(self, local_port, timeout=1):
"""Prepares a listener (an `AsyncReceiver`) listening on the given `local_port` for given `timeout` (sec)"""
receiver = self._get_receiver(local_port)
receiver._set_listen_timeout(timeout)
return receiver
@staticmethod
def perform_async_tx_rx(timeout=20):
"""Called to perform all previously prepared async rx and tx operations"""
"""Called to perform all previously prepared async rx/listen and tx operations"""
try:
start_time = time.time()
while asyncore.socket_map:
@@ -622,7 +631,7 @@ class Node(object):
print 'Performing aysnc tx/tx took too long ({}>{} sec)'.format(elapsed_time, timeout)
raise Node._NodeError('perform_tx_rx timed out ({}>{} sec)'.format(elapsed_time, timeout))
# perform a single asyncore loop
asyncore.loop(timeout=1, count=1)
asyncore.loop(timeout=0.5, count=1)
except:
print 'Failed to perform async rx/tx'
raise
@@ -645,7 +654,7 @@ def _create_socket_address(ip_address, port):
class AsyncSender(asyncore.dispatcher):
""" An IPv6 async message sender - use `Node.prepare_tx()` to create one"""
def __init__(self, node, src_addr, src_port, dst_addr, dst_port, msg, count):
def __init__(self, node, src_addr, src_port, dst_addr, dst_port, msg, count, mcast_hops=None):
self._node = node
self._src_addr = src_addr
self._src_port = src_port
@@ -662,6 +671,10 @@ class AsyncSender(asyncore.dispatcher):
sock.setsockopt(socket.SOL_SOCKET, _SO_BINDTODEVICE, node.interface_name + '\0')
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEPORT, 1)
# Set the IPV6_MULTICAST_HOPS
if mcast_hops is not None:
sock.setsockopt(socket.IPPROTO_IPV6, socket.IPV6_MULTICAST_HOPS, mcast_hops)
# Bind the socket to the given src address
if _is_ipv6_addr_link_local(src_addr):
# If src is a link local address it requires the interface name to be specified.
@@ -742,7 +755,7 @@ class AsyncSender(asyncore.dispatcher):
#- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
class AsyncReceiver(asyncore.dispatcher):
""" An IPv6 async message receiver - use `prepare_tx()` to create one"""
""" An IPv6 async message receiver - use `prepare_rx()` to create one"""
_MAX_RECV_SIZE = 2048
@@ -767,6 +780,9 @@ class AsyncReceiver(asyncore.dispatcher):
self._local_port = local_port
self._senders = [] # list of `_SenderInfo` objects
self._all_rx = [] # contains all received messages as a list of (pkt, (src_addr, src_port))
self._timeout = 0 # listen timeout (zero means forever)
self._started = False
self._start_time = 0
# Create a socket, bind it to the node's interface
sock = socket.socket(socket.AF_INET6, socket.SOCK_DGRAM)
@@ -782,6 +798,9 @@ class AsyncReceiver(asyncore.dispatcher):
def _add_sender(self, sender_addr, sender_port, msg, count):
self._senders.append(AsyncReceiver._SenderInfo(sender_addr, sender_port, msg, count))
def _set_listen_timeout(self, timeout):
self._timeout = timeout
# Property getters
@property
@@ -800,11 +819,20 @@ class AsyncReceiver(asyncore.dispatcher):
@property
def was_successful(self):
"""Indicates if all expected IPv6 messages were received successfully"""
return all([sender._did_recv_all() for sender in self._senders])
return len(self._senders) == 0 or all([sender._did_recv_all() for sender in self._senders])
# asyncore.dispatcher callbacks
def readable(self):
if not self._started:
self._start_time = time.time()
self._started = True
if self._timeout != 0 and time.time() - self._start_time >= self._timeout:
self.handle_close()
if self._node._verbose:
_log('- Node{} finished listening on port {} for {} sec, received {} msg(s)'.format(
self._node._index, self._local_port, self._timeout, len(self._all_rx)))
return False
return True
def writable(self):