diff --git a/tests/toranj/start.sh b/tests/toranj/start.sh index 510f76a23..0437bc2e5 100755 --- a/tests/toranj/start.sh +++ b/tests/toranj/start.sh @@ -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 diff --git a/tests/toranj/test-023-multicast-traffic.py b/tests/toranj/test-023-multicast-traffic.py new file mode 100644 index 000000000..c3ac9c430 --- /dev/null +++ b/tests/toranj/test-023-multicast-traffic.py @@ -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) diff --git a/tests/toranj/wpan.py b/tests/toranj/wpan.py index 0610b0e2e..1ddcbc112 100644 --- a/tests/toranj/wpan.py +++ b/tests/toranj/wpan.py @@ -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):