diff options
Diffstat (limited to 'scripts/automation')
5 files changed, 122 insertions, 10 deletions
diff --git a/scripts/automation/regression/stateless_tests/stl_ipv6_test.py b/scripts/automation/regression/stateless_tests/stl_ipv6_test.py new file mode 100755 index 00000000..c56c9b5d --- /dev/null +++ b/scripts/automation/regression/stateless_tests/stl_ipv6_test.py @@ -0,0 +1,60 @@ +#!/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.') + + diff --git a/scripts/automation/trex_control_plane/stl/trex_stl_lib/rx_services/trex_stl_rx_service_icmp.py b/scripts/automation/trex_control_plane/stl/trex_stl_lib/rx_services/trex_stl_rx_service_icmp.py index 612b6a2d..8d4e2f57 100644 --- a/scripts/automation/trex_control_plane/stl/trex_stl_lib/rx_services/trex_stl_rx_service_icmp.py +++ b/scripts/automation/trex_control_plane/stl/trex_stl_lib/rx_services/trex_stl_rx_service_icmp.py @@ -14,6 +14,7 @@ class RXServiceICMP(RXServiceAPI): 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" @@ -61,14 +62,21 @@ class RXServiceICMP(RXServiceAPI): # 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 @@ -79,5 +87,7 @@ class RXServiceICMP(RXServiceAPI): # 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) diff --git a/scripts/automation/trex_control_plane/stl/trex_stl_lib/rx_services/trex_stl_rx_service_ipv6.py b/scripts/automation/trex_control_plane/stl/trex_stl_lib/rx_services/trex_stl_rx_service_ipv6.py index c2c8ebc1..b76f523c 100755 --- a/scripts/automation/trex_control_plane/stl/trex_stl_lib/rx_services/trex_stl_rx_service_ipv6.py +++ b/scripts/automation/trex_control_plane/stl/trex_stl_lib/rx_services/trex_stl_rx_service_ipv6.py @@ -126,6 +126,7 @@ class RXServiceICMPv6(RXServiceIPv6): 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' @@ -159,21 +160,38 @@ class RXServiceICMPv6(RXServiceIPv6): 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) diff --git a/scripts/automation/trex_control_plane/stl/trex_stl_lib/trex_stl_client.py b/scripts/automation/trex_control_plane/stl/trex_stl_lib/trex_stl_client.py index a1290a37..e8feebbd 100755 --- a/scripts/automation/trex_control_plane/stl/trex_stl_lib/trex_stl_client.py +++ b/scripts/automation/trex_control_plane/stl/trex_stl_lib/trex_stl_client.py @@ -1916,6 +1916,23 @@ class STLClient(object): 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` @@ -1938,6 +1955,7 @@ class STLClient(object): src_port, pkt_size)) + responses_arr = [] # no async messages with self.logger.supress(level = LoggerApi.VERBOSE_REGULAR_SYNC): self.logger.log('') @@ -1954,10 +1972,13 @@ class STLClient(object): 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 @@ -3148,9 +3169,10 @@ class STLClient(object): 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' diff --git a/scripts/automation/trex_control_plane/stl/trex_stl_lib/trex_stl_streams.py b/scripts/automation/trex_control_plane/stl/trex_stl_lib/trex_stl_streams.py index 8d727d6d..7761be4d 100755 --- a/scripts/automation/trex_control_plane/stl/trex_stl_lib/trex_stl_streams.py +++ b/scripts/automation/trex_control_plane/stl/trex_stl_lib/trex_stl_streams.py @@ -926,6 +926,8 @@ class STLProfile(object): # 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()") |