endif::backend-docbook[]
+== Release 2.16 ==
+
+* IPv6 console basic support:
+** scan6 function to search for IPv6-enabled neighbors
+** ping can be used with IPv6 addresses to ping nearby devices
+
== Release 2.15 ==
* XL710/X710 VF (SR-IOV) mode works better
--- /dev/null
+#!/usr/bin/python
+from .stl_general_test import CStlGeneral_Test
+from trex_stl_lib.api import *
+
+class STLIPv6_Test(CStlGeneral_Test):
+ """Tests for IPv6 scan6/ping_ip """
+
+ def setUp(self):
+ CStlGeneral_Test.setUp(self)
+ print('')
+ self.stl_trex.set_service_mode(ports = [0])
+
+ def tearDown(self):
+ CStlGeneral_Test.tearDown(self)
+ self.stl_trex.set_service_mode(ports = [0], enabled = False)
+
+ def test_ipv6_ping(self):
+ ping_count = 5
+ expected_replies = 4 # allow one loss
+ results = self.stl_trex.ping_ip(src_port = 0, dst_ip = 'ff02::1', count = ping_count)
+ good_replies = len(filter(lambda result: result['status'] == 'success', results))
+ if self.is_loopback:
+ # negative test, loopback
+ if good_replies > 0:
+ self.fail('We should not respond to IPv6 in loopback at this stage, bug!\nOutput: %s' % results)
+ else:
+ print('No IPv6 replies in loopback as expected.')
+ else:
+ # positive test, DUT
+ if good_replies < expected_replies:
+ self.fail('Got only %s good replies out of %s.\nOutput: %s' % (good_replies, ping_count, results))
+ else:
+ print('Got replies from DUT as expected.')
+
+ # negative test, unknown IP
+ results = self.stl_trex.ping_ip(src_port = 0, dst_ip = '1234::1234', count = ping_count)
+ good_replies = len(filter(lambda result: result['status'] == 'success', results))
+ if good_replies > 0:
+ self.fail('We have answers from unknown IPv6, bug!\nOutput: %s' % results)
+ else:
+ print('Got no replies from unknown IPv6 as expected.')
+
+ def test_ipv6_scan6(self):
+ results = self.stl_trex.scan6(ports = 0)
+ if self.is_loopback:
+ # negative test, loopback
+ if results[0]:
+ self.fail("Scan6 found devices in loopback, we don't answer to IPv6 now, bug!\nOutput: %s" % results)
+ else:
+ print('No devices found in loopback as expected.')
+ else:
+ # positive test, DUT
+ if len(results[0]) > 1:
+ self.fail('Found more than one device at port 0: %s' % results)
+ elif len(results[0]) == 1:
+ print('Found one device as expected:\n{type:10} - {mac:20} - {ipv6}'.format(**results[0][0]))
+ else:
+ self.fail('Did not find IPv6 devices.')
+
+
super(RXServiceICMP, self).__init__(port, layer_mode = RXServiceAPI.LAYER_MODE_L3, *a, **k)
self.ping_ip = ping_ip
self.pkt_size = pkt_size
+ self.result = {}
def get_name (self):
return "PING"
# check seq
if icmp.seq != self.base_pkt['ICMP'].seq:
return None
- return self.port.ok('Reply from {0}: bytes={1}, time={2:.2f}ms, TTL={3}'.format(ip.src, len(pkt['binary']), dt * 1000, ip.ttl))
+ self.result['formatted_string'] = 'Reply from {0}: bytes={1}, time={2:.2f}ms, TTL={3}'.format(ip.src, len(pkt['binary']), dt * 1000, ip.ttl)
+ self.result['src_ip'] = ip.src
+ self.result['rtt'] = dt * 1000
+ self.result['ttl'] = ip.ttl
+ self.result['status'] = 'success'
+ return self.port.ok(self.result)
# unreachable
elif icmp.type == 3:
# check seq
if icmp.payload.seq != self.base_pkt['ICMP'].seq:
return None
- return self.port.ok('Reply from {0}: Destination host unreachable'.format(icmp.src))
+ self.result['formatted_string'] = 'Reply from {0}: Destination host unreachable'.format(icmp.src)
+ self.result['status'] = 'unreachable'
+ return self.port.ok(self.result)
else:
# skip any other types
# return the str of a timeout err
def on_timeout(self):
- return self.port.ok('Request timed out.')
+ self.result['formatted_string'] = 'Request timed out.'
+ self.result['status'] = 'timeout'
+ return self.port.ok(self.result)
super(RXServiceICMPv6, self).__init__(port, *a, **k)
self.pkt_size = pkt_size
self.dst_mac = dst_mac
+ self.result = {}
def get_name(self):
return 'PING6'
node_ip = scapy_pkt.getlayer(IPv6).src
hlim = scapy_pkt.getlayer(IPv6).hlim
dst_ip = scapy_pkt.getlayer(IPv6).dst
-
if dst_ip != self.src_ip: # not our ping
return
dt = pkt['ts'] - start_ts
- return self.port.ok('Reply from {0}: bytes={1}, time={2:.2f}ms, hlim={3}'.format(node_ip, len(pkt['binary']), dt * 1000, hlim))
+ self.result['formatted_string'] = 'Reply from {0}: bytes={1}, time={2:.2f}ms, hlim={3}'.format(node_ip, len(pkt['binary']), dt * 1000, hlim)
+ self.result['src_ip'] = node_ip
+ self.result['rtt'] = dt * 1000
+ self.result['ttl'] = hlim
+ self.result['status'] = 'success'
+ return self.port.ok(self.result)
if scapy_pkt.haslayer('ICMPv6ND_NS') and scapy_pkt.haslayer('ICMPv6NDOptSrcLLAddr'):
node_mac = scapy_pkt.getlayer(ICMPv6NDOptSrcLLAddr).lladdr
node_ip = scapy_pkt.getlayer(IPv6).src
+ dst_ip = scapy_pkt.getlayer(IPv6).dst
+ if dst_ip != self.src_ip: # not our ping
+ return
self.send_intermediate(self.generate_ns_na(node_mac, node_ip))
+ if scapy_pkt.haslayer('ICMPv6DestUnreach'):
+ node_ip = scapy_pkt.getlayer(IPv6).src
+ dst_ip = scapy_pkt.getlayer(IPv6).dst
+ if dst_ip != self.src_ip: # not our ping
+ return
+ self.result['formatted_string'] = 'Reply from {0}: Destination host unreachable'.format(node_ip)
+ self.result['status'] = 'unreachable'
+ return self.port.ok(self.result)
# return the str of a timeout err
def on_timeout(self):
- return self.port.ok('Request timed out.')
+ self.result['formatted_string'] = 'Request timed out.'
+ self.result['status'] = 'timeout'
+ return self.port.ok(self.result)
pkt_size - packet size to use
count - how many times to ping
interval_sec - how much time to wait between pings
+
+ :returns:
+ List of replies per 'count'
+
+ Each element is dictionary with following items:
+
+ Always available keys:
+
+ * formatted_string - string, human readable output, for example: 'Request timed out.'
+ * status - string, one of options: 'success', 'unreachable', 'timeout'
+
+ Available only if status is 'success':
+
+ * src_ip - string, IP replying to request
+ * rtt - float, latency of the ping (round trip time)
+ * ttl - int, time to live in IPv4 or hop limit in IPv6
+
:raises:
+ :exc:`STLError`
src_port,
pkt_size))
+ responses_arr = []
# no async messages
with self.logger.supress(level = LoggerApi.VERBOSE_REGULAR_SYNC):
self.logger.log('')
if not rc:
raise STLError(rc)
- self.logger.log(rc.data())
+ responses_arr.append(rc.data())
+ self.logger.log(rc.data()['formatted_string'])
if i != (count - 1):
time.sleep(interval_sec)
+
+ return responses_arr
verbose - log for each request the response
:return:
list of dictionaries per neighbor:
- type - type of device: 'Router' or 'Host'
- mac - MAC address of device
- ipv6 - IPv6 address of device
+
+ * type - type of device: 'Router' or 'Host'
+ * mac - MAC address of device
+ * ipv6 - IPv6 address of device
:raises:
+ :exe:'STLError'
# fetch defaults
defaults = func.__defaults__
+ if defaults is None:
+ return {}
if len(defaults) != (argc - 1):
raise STLError("Module should provide default values for all arguments on get_streams()")
Code Review / trex.git / commitdiff