#!/usr/bin/env python3 import unittest import random import socket from ipaddress import IPv4Address, IPv6Address, AddressValueError from framework import VppTestCase, VppTestRunner from util import ppp from scapy.packet import Raw from scapy.layers.l2 import Ether from scapy.layers.inet import IP, UDP from scapy.layers.inet6 import IPv6 from vpp_ip_route import VppIpRoute, VppRoutePath try: text_type = unicode except NameError: text_type = str # # The number of packets to sent. # N_PKTS_IN_STREAM = 300 class TestECMP(VppTestCase): """ Equal-cost multi-path routing Test Case """ @classmethod def setUpClass(cls): """ Perform standard class setup (defined by class method setUpClass in class VppTestCase) before running the test case, set test case related variables and configure VPP. """ super(TestECMP, cls).setUpClass() # create 4 pg interfaces cls.create_pg_interfaces(range(4)) # packet sizes to test cls.pg_if_packet_sizes = [64, 1500, 9018] # setup interfaces for i in cls.pg_interfaces: i.admin_up() i.generate_remote_hosts(5) i.config_ip4() i.resolve_arp() i.configure_ipv4_neighbors() i.config_ip6() i.resolve_ndp() i.configure_ipv6_neighbors() @classmethod def tearDownClass(cls): if not cls.vpp_dead: for i in cls.pg_interfaces: i.unconfig_ip4() i.unconfig_ip6() i.admin_down() super(TestECMP, cls).tearDownClass() def setUp(self): super(TestECMP, self).setUp() self.reset_packet_infos() def tearDown(self): """ Show various debug prints after each test. """ super(TestECMP, self).tearDown() def show_commands_at_teardown(self): self.logger.info(self.vapi.ppcli("show ip4 neighbors")) self.logger.info(self.vapi.ppcli("show ip6 neighbors")) def get_ip_address(self, ip_addr_start, ip_prefix_len): """ :param str ip_addr_start: Starting IPv4 or IPv6 address. :param int ip_prefix_len: IP address prefix length. :return: Random IPv4 or IPv6 address from required range. """ try: ip_addr = IPv4Address(text_type(ip_addr_start)) ip_max_len = 32 except (AttributeError, AddressValueError): ip_addr = IPv6Address(text_type(ip_addr_start)) ip_max_len = 128 return str(ip_addr + random.randint(0, 2 ** (ip_max_len - ip_prefix_len) - 2)) def create_stream(self, src_if, src_ip_start, dst_ip_start, ip_prefix_len, packet_sizes, ip_l=IP): """Create input packet stream for defined interfaces. :param VppInterface src_if: Source Interface for packet stream. :param str src_ip_start: Starting source IPv4 or IPv6 address. :param str dst_ip_start: Starting destination IPv4 or IPv6 address. :param int ip_prefix_len: IP address prefix length. :param list packet_sizes: packet size to test. :param Scapy ip_l: Required IP layer - IP or IPv6. (Default is IP.) """ pkts = [] for i in range(0, N_PKTS_IN_STREAM): info = self.create_packet_info(src_if, src_if) payload = self.info_to_payload(info) src_ip = self.get_ip_address(src_ip_start, ip_prefix_len) dst_ip = self.get_ip_address(dst_ip_start, ip_prefix_len) p = (Ether(dst=src_if.local_mac, src=src_if.remote_mac) / ip_l(src=src_ip, dst=dst_ip) / UDP(sport=1234, dport=1234) / Raw(payload)) info.data = p.copy() size = random.choice(packet_sizes) self.extend_packet(p, size) pkts.append(p) return pkts def verify_capture(self, rx_if, capture, ip_l=IP): """Verify captured input packet stream for defined interface. :param VppInterface rx_if: Interface to verify captured packet stream. :param list capture: Captured packet stream. :param Scapy ip_l: Required IP layer - IP or IPv6. (Default is IP.) """ self.logger.info("Verifying capture on interface %s" % rx_if.name) count = 0 host_counters = {} for host_mac in rx_if._hosts_by_mac: host_counters[host_mac] = 0 for packet in capture: try: ip_received = packet[ip_l] payload_info = self.payload_to_info(packet[Raw]) packet_index = payload_info.index ip_sent = self._packet_infos[packet_index].data[ip_l] self.logger.debug("Got packet on port %s: src=%u (id=%u)" % (rx_if.name, payload_info.src, packet_index)) # Check standard fields self.assertIn(packet.dst, rx_if._hosts_by_mac, "Destination MAC address %s shouldn't be routed " "via interface %s" %