#!/usr/bin/env python3 import binascii import random import socket import unittest import scapy.compat from scapy.contrib.mpls import MPLS from scapy.layers.inet import IP, UDP, TCP, ICMP, icmptypes, icmpcodes from scapy.layers.l2 import Ether, Dot1Q, ARP from scapy.packet import Raw from six import moves from framework import VppTestCase, VppTestRunner from util import ppp from vpp_ip_route import VppIpRoute, VppRoutePath, VppIpMRoute, \ VppMRoutePath, MRouteItfFlags, MRouteEntryFlags, VppMplsIpBind, \ VppMplsTable, VppIpTable, FibPathType, find_route, \ VppIpInterfaceAddress, find_route_in_dump, find_mroute_in_dump from vpp_sub_interface import VppSubInterface, VppDot1QSubint, VppDot1ADSubint from vpp_papi import VppEnum from vpp_neighbor import VppNeighbor from vpp_lo_interface import VppLoInterface NUM_PKTS = 67 class TestIPv4(VppTestCase): """ IPv4 Test Case """ @classmethod def setUpClass(cls): super(TestIPv4, cls).setUpClass() @classmethod def tearDownClass(cls): super(TestIPv4, cls).tearDownClass() def setUp(self): """ Perform test setup before test case. **Config:** - create 3 pg interfaces - untagged pg0 interface - Dot1Q subinterface on pg1 - Dot1AD subinterface on pg2 - setup interfaces: - put it into UP state - set IPv4 addresses - resolve neighbor address using ARP - configure 200 fib entries :ivar list interfaces: pg interfaces and subinterfaces. :ivar dict flows: IPv4 packet flows in test. """ super(TestIPv4, self).setUp() # create 3 pg interfaces self.create_pg_interfaces(range(3)) # create 2 subinterfaces for pg1 and pg2 self.sub_interfaces = [ VppDot1QSubint(self, self.pg1, 100), VppDot1ADSubint(self, self.pg2, 200, 300, 400)] # packet flows mapping pg0 -> pg1.sub, pg2.sub, etc. self.flows = dict() self.flows[self.pg0] = [self.pg1.sub_if, self.pg2.sub_if] self.flows[self.pg1.sub_if] = [self.pg0, self.pg2.sub_if] self.flows[self.pg2.sub_if] = [self.pg0, self.pg1.sub_if] # packet sizes self.pg_if_packet_sizes = [64, 1500, 9020] self.interfaces = list(self.pg_interfaces) self.interfaces.extend(self.sub_interfaces) # setup all interfaces for i in self.interfaces: i.admin_up() i.config_ip4() i.resolve_arp() # config 2M FIB entries def tearDown(self): """Run standard test teardown and log ``show ip arp``.""" super(TestIPv4, self).tearDown() def show_commands_at_teardown(self): self.logger.info(self.vapi.cli("show ip arp")) # info(self.vapi.cli("show ip fib")) # many entries def modify_packet(self, src_if, packet_size, pkt): """Add load, set destination IP and extend packet to required packet size for defined interface. :param VppInterface src_if: Interface to create packet for. :param int packet_size: Required packet size. :param Scapy pkt: Packet to be modified. """ dst_if_idx = int(packet_size / 10 % 2) dst_if = self.flows[src_if][dst_if_idx] info = self.create_packet_info(src_if, dst_if) payload = self.info_to_payload(info) p = pkt/Raw(payload) p[IP].dst = dst_if.remote_ip4 info.data = p.copy() if isinstance(src_if, VppSubInterface): p = src_if.add_dot1_layer(p) self.extend_packet(p, packet_size) return p def create_stream(self, src_if): """Create input packet stream for defined interface. :param VppInterface src_if: Interface to create packet stream for. """ hdr_ext = 4 if isinstance(src_if, VppSubInterface) else 0 pkt_tmpl = (Ether(dst=src_if.local_mac, src=src_if.remote_mac) / IP(src=src_if.remote_ip4) / UDP(sport=1234, dport=1234)) pkts = [self.modify_packet(src_if, i, pkt_tmpl) for i in moves.range(self.pg_if_packet_sizes[0], self.pg_if_packet_sizes[1], 10)] pkts_b = [self.modify_packet(src_if, i, pkt_tmpl) for i in moves.range(self.pg_if_packet_sizes[1] + hdr_ext, self.pg_if_packet_sizes[2] + hdr_ext, 50)] pkts.extend(pkts_b) return pkts def verify_capture(self, dst_if, capture): """Verify captured input packet stream for defined interface. :param VppInterface dst_if: Interface to verify captured packet stream for. :param list capture: Captured packet stream. """ self.logger.info("Verifying capture on interface %s" % dst_if.name) last_info = dict() for i in self.interfaces: last_info[i.sw_if_index] = None is_sub_if = False dst_sw_if_index = dst_if.sw_if_index if hasattr(dst_if, 'parent'): is_sub_if = True for packet in capture: if is_sub_if: # Check VLAN tags and Ethernet header packet = dst_if.remove_dot1_layer(packet) self.assertTrue(Dot1Q not in packet) try: ip = packet[IP] udp = packet[UDP] payload_info = self.payload_to_info(packet[Raw]) packet_index = payload_info.index self.assertEqual(payload_info.dst, dst_sw_if_index) self.logger.debug( "Got packet on port %s: src=%u (id=%u)" % (dst_if.name, payload_info.src, packet_index)) next_info = self.get_next_packet_info_for_interface2( payload_info.src, dst_sw_if_index, last_info[payload_info.src]) last_info[payload_info.src] = next_info self.assertTrue(next_info is not None) self.assertEqual(packet_index, next_info.index) saved_packet = next_info.data # Check standard fields self.assertEqual(ip.src, saved_packet[IP].src) self.assertEqual(ip.dst, saved_packet[IP].dst) self.assertEqual(udp.sport, saved_packet[UDP].sport) self.assertEqual(udp.dport, saved_packet[UDP].dport) except: self.logger.error(ppp("Unexpected or invalid packet:", packet)) raise for i in self.interfaces: remaining_packet = self.get_next_packet_info_for_interface2( i.sw_if_index, dst_sw_if_index, last_info[i.sw_if_index]) self.assertTrue(remaining_packet is None, "Interface %s: Packet expected from interface %s " "didn't arrive" % (dst_if.name, i.name)) def test_fib(self): """ IPv4 FIB test Test scenario: - Create IPv4 stream for pg0 interface - Create IPv4 tagged streams for pg1's and pg2's sub-interface. - Send and verify received packets on each interface. """ pkts = self.create_stream(self.pg0) self.pg0.add_stream(pkts) for i in self.sub_interfaces: pkts = self.create_stream(i) i.parent.add_stream(pkts) self.pg_enable_capture(self.pg_interfaces) self.pg_start() pkts = self.pg0.get_capture() self.verify_capture(self.pg0, pkts) for i in self.sub_interfaces: pkts = i.parent.get_capture() self.verify_capture(i, pkts) class TestIPv4IfAddrRoute(VppTestCase): """ IPv4 Interface Addr Route Test Case """ @classmethod def setUpClass(cls): super(TestIPv4IfAddrRoute, cls).setUpClass() @classmethod def tearDownClass(cls): super(TestIPv4IfAddrRoute, cls).tearDownClass() def setUp(self): super(TestIPv4IfAddrRoute, self).setUp() # create 1 pg interface self.create_pg_interfaces(range(1)) for i in self.pg_interfaces: i.admin_up() i.config_ip4() i.resolve_arp() def tearDown(self): super(TestIPv4IfAddrRoute, self).tearDown() for i in self.pg_interfaces: i.unconfig_ip4() i.admin_down() def test_ipv4_ifaddrs_same_prefix(self): """ IPv4 Interface Addresses Same Prefix test Test scenario: - Verify no route in FIB for prefix 10.10.10.0/24 - Configure IPv4 address 10.10.10.10/24 on an interface - Verify route in FIB for prefix 10.10.10.0/24 - Configure IPv4 address 10.10.10.20/24 on an interface - Delete 10.10.10.10/24 from interface - Verify route in FIB for prefix 10.10.10.0/24 - Delete 10.10.10.20/24 from interface - Verify no route in FIB for prefix 10.10.10.0/24 """ # create two addresses, verify route not present if_addr1 = VppIpInterfaceAddress(self, self.pg0, "10.10.10.10", 24) if_addr2 = VppIpInterfaceAddress(self, self.pg0, "10.10.10.20", 24) self.assertFalse(if_addr1.query_vpp_config()) # 10.10.10.10/24 self.assertFalse(find_route(self, "10.10.10.10", 32)) self.assertFalse(find_route(self, "10.10.10.20", 32)) self.assertFalse(find_route(self, "10.10.10.255", 32)) self.assertFalse(find_route(self, "10.10.10.0", 32)) # configure first address, verify route present if_addr1.add_vpp_config() self.assertTrue(if_addr1.query_vpp_config()) # 10.10.10.10/24 self.assertTrue(find_route(self, "10.10.10.10", 32)) self.assertFalse(find_route(self, "10.10.10.20", 32)) self.assertTrue(find_route(self, "10.10.10.255", 32)) self.assertTrue(find_route(self, "10.10.10.0", 32)) # configure second address, delete first, verify route not removed if_addr2.add_vpp_config() if_addr1.remove_vpp_config() self.assertFalse(if_addr1.query_vpp_config()) # 10.10.10.10/24 self.assertTrue(if_addr2.query_vpp_config()) # 10.10.10.20/24 self.assertFalse(find_route(self, "10.10.10.10", 32)) self.assertTrue(find_route(self, "10.10.10.20", 32)) self.assertTrue(find_route(self, "10.10.10.255", 32)) self.assertTrue(find_route(self, "10.10.10.0", 32)) # delete second address, verify route removed if_addr2.remove_vpp_config() self.assertFalse(if_addr2.query_vpp_config()) # 10.10.10.20/24 self.assertFalse(find_route(self, "10.10.10.10", 32)) self.assertFalse(find_route(self, "10.10.10.20", 32)) self.assertFalse(find_route(self, "10.10.10.255", 32)) self.assertFalse(find_route(self, "10.10.10.0", 32)) def test_ipv4_ifaddr_route(self): """ IPv4 Interface Address Route test Test scenario: - Create loopback - Configure IPv4 address on loopback - Verify that address is not in the FIB - Bring loopback up - Verify that address is in the FIB now - Bring loopback down - Verify that address is not in the FIB anymore - Bring loopback up - Configure IPv4 address on loopback - Verify that address is in the FIB now """ # create a loopback and configure IPv4 loopbacks = self.create_loopback_interfaces(1) lo_if = self.lo_interfaces[0] lo_if.local_ip4_prefix_len = 32 lo_if.config_ip4() # The intf was down when addr was added -> entry not in FIB fib4_dump = self.vapi.ip_route_dump(0) self.assertFalse(lo_if.is_ip4_entry_in_fib_dump(fib4_dump)) # When intf is brought up, entry is added lo_if.admin_up() fib4_dump = self.vapi.ip_route_dump(0) self.assertTrue(lo_if.is_ip4_entry_in_fib_dump(fib4_dump)) # When intf is brought down, entry is removed lo_if.admin_down() fib4_dump = self.vapi.ip_route_dump(0) self.assertFalse(lo_if.is_ip4_entry_in_fib_dump(fib4_dump)) # Remove addr, bring up interface, re-add -> entry in FIB lo_if.unconfig_ip4() lo_if.admin_up() lo_if.config_ip4() fib4_dump = self.vapi.ip_route_dump(0) self.assertTrue(lo_if.is_ip4_entry_in_fib_dump(fib4_dump)) class TestICMPEcho(VppTestCase): """ ICMP Echo Test Case """ @classmethod def setUpClass(cls): super(TestICMPEcho, cls).setUpClass() @classmethod def tearDownClass(cls): super(TestICMPEcho, cls).tearDownClass() def setUp(self): super(TestICMPEcho, self).setUp() # create 1 pg interface self.create_pg_interfaces(range(1)) for i in self.pg_interfaces: i.admin_up() i.config_ip4() i.resolve_arp() def tearDown(self): super(TestICMPEcho, self).tearDown() for i in self.pg_interfaces: i.unconfig_ip4() i.admin_down() def test_icmp_echo(self): """ VPP replies to ICMP Echo Request Test scenario: - Receive ICMP Echo Request message on pg0 interface. - Check outgoing ICMP Echo Reply message on pg0 interface. """ icmp_id = 0xb icmp_seq = 5 icmp_load = b'\x0a' * 18 p_echo_request = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) / IP(src=self.pg0.remote_ip4, dst=self.pg0.local_ip4) / ICMP(id=icmp_id, seq=icmp_seq) / Raw(load=icmp_load)) self.pg0.add_stream(p_echo_request) self.pg_enable_capture(self.pg_interfaces) self.pg_start() rx = self.pg0.get_capture(1) rx = rx[0] ether = rx[Ether] ipv4 = rx[IP] icmp = rx[ICMP] self.assertEqual(ether.src, self.pg0.local_mac) self.assertEqual(ether.dst, self.pg0.remote_mac) self.assertEqual(ipv4.src, self.pg0.local_ip4) self.assertEqual(ipv4.dst, self.pg0.remote_ip4) self.assertEqual(icmptypes[icmp.type], "echo-reply") self.assertEqual(icmp.id, icmp_id) self.assertEqual(icmp.seq, icmp_seq) self.assertEqual(icmp[Raw].load, icmp_load) class TestIPv4FibCrud(VppTestCase): """ FIB - add/update/delete - ip4 routes Test scenario: - add 1k, - del 100, - add new 1k, - del 1.5k ..note:: Python API is too slow to add many routes, needs replacement. """ def config_fib_many_to_one(self, start_dest_addr, next_hop_addr, count, start=0): """ :param start_dest_addr: :param next_hop_addr: :param count: :return list: added ips with 32 prefix """ routes = [] for i in range(count): r = VppIpRoute(self, start_dest_addr % (i + start), 32, [VppRoutePath(next_hop_addr, 0xffffffff)]) r.add_vpp_config() routes.append(r) return routes def unconfig_fib_many_to_one(self, start_dest_addr, next_hop_addr, count, start=0): routes = [] for i in range(count): r = VppIpRoute(self, start_dest_addr % (i + start), 32, [VppRoutePath(next_hop_addr, 0xffffffff)]) r.remove_vpp_config() routes.append(r) return routes def create_stream(self, src_if, dst_if, routes, count): pkts = [] for _ in range(count): dst_addr = random.choice(routes).prefix.network_address info = self.create_packet_info(src_if, dst_if) payload = self.info_to_payload(info) p = (Ether(dst=src_if.local_mac, src=src_if.remote_mac) / IP(src=src_if.remote_ip4, dst=str(dst_addr)) / UDP(sport=1234, dport=1234) / Raw(payload)) info.data = p.copy() self.extend_packet(p, random.choice(self.pg_if_packet_sizes)) pkts.append(p) return pkts def _find_ip_match(self, find_in, pkt): for p in find_in: if self.payload_to_info(p[Raw]) == \ self.payload_to_info(pkt[Raw]): if p[IP].src != pkt[IP].src: break if p[IP].dst != pkt[IP].dst: break if p[UDP].sport != pkt[UDP].sport: break if p[UDP].dport != pkt[UDP].dport: break return p return None def verify_capture(self, dst_interface, received_pkts, expected_pkts): self.assertEqual(len(received_pkts), len(expected_pkts)) to_verify = list(expected_pkts) for p in received_pkts: self.assertEqual(p.src, dst_interface.local_mac) self.assertEqual(p.dst, dst_interface.remote_mac) x = self._find_ip_match(to_verify, p) to_verify.remove(x) self.assertListEqual(to_verify, []) def verify_route_dump(self, routes): for r in routes: self.assertTrue(find_route(self, r.prefix.network_address, r.prefix.prefixlen)) def verify_not_in_route_dump(self, routes): for r in routes: self.assertFalse(find_route(self, r.prefix.network_address, r.prefix.prefixlen)) @classmethod def setUpClass(cls): """ #. Create and initialize 3 pg interfaces. #. initialize class attributes configured_routes and deleted_routes to store information between tests. """ super(TestIPv4FibCrud, cls).setUpClass() try: # create 3 pg interfaces cls.create_pg_interfaces(range(3)) cls.interfaces = list(cls.pg_interfaces) # setup all interfaces for i in cls.interfaces: i.admin_up() i.config_ip4() i.resolve_arp() cls.configured_routes = [] cls.deleted_routes = [] cls.pg_if_packet_sizes = [64, 512, 1518, 9018] except Exception: super(TestIPv4FibCrud, cls).tearDownClass() raise @classmethod def tearDownClass(cls): super(TestIPv4FibCrud, cls).tearDownClass() def setUp(self): super(TestIPv4FibCrud, self).setUp() self.reset_packet_infos() self.configured_routes = [] self.deleted_routes = [] def test_1_add_routes(self): """ Add 1k routes """ # add 100 routes check with traffic script. self.configured_routes.extend(self.config_fib_many_to_one( "10.0.0.%d", self.pg0.remote_ip4, 100)) self.verify_route_dump(self.configured_routes) self.stream_1 = self.create_stream( self.pg1, self.pg0, self.configured_routes, 100) self.stream_2 = self.create_stream( self.pg2, self.pg0, self.configured_routes, 100) self.pg1.add_stream(self.stream_1) self.pg2.add_stream(self.stream_2) self.pg_enable_capture(self.pg_interfaces) self.pg_start() pkts = self.pg0.get_capture(len(self.stream_1) + len(self.stream_2)) self.verify_capture(self.pg0, pkts, self.stream_1 + self.stream_2) def test_2_del_routes(self): """ Delete 100 routes - delete 10 routes check with traffic script. """ # config 1M FIB entries self.configured_routes.extend(self.config_fib_many_to_one( "10.0.0.%d", self.pg0.remote_ip4, 100)) self.deleted_routes.extend(self.unconfig_fib_many_to_one( "10.0.0.%d", self.pg0.remote_ip4, 10, start=10)) for x in self.deleted_routes: self.configured_routes.remove(x) self.verify_route_dump(self.configured_routes) self.stream_1 = self.create_stream( self.pg1, self.pg0, self.configured_routes, 100) self.stream_2 = self.create_stream( self.pg2, self.pg0, self.configured_routes, 100) self.stream_3 = self.create_stream( self.pg1, self.pg0, self.deleted_routes, 100) self.stream_4 = self.create_stream( self.pg2, self.pg0, self.deleted_routes, 100) self.pg1.add_stream(self.stream_1 + self.stream_3) self.pg2.add_stream(self.stream_2 + self.stream_4) self.pg_enable_capture(self.pg_interfaces) self.pg_start() pkts = self.pg0.get_capture(len(self.stream_1) + len(self.stream_2)) self.verify_capture(self.pg0, pkts, self.stream_1 + self.stream_2) def test_3_add_new_routes(self): """ Add 1k routes - re-add 5 routes check with traffic script. - add 100 routes check with traffic script. """ # config 1M FIB entries self.configured_routes.extend(self.config_fib_many_to_one( "10.0.0.%d", self.pg0.remote_ip4, 100)) self.deleted_routes.extend(self.unconfig_fib_many_to_one( "10.0.0.%d", self.pg0.remote_ip4, 10, start=10)) for x in self.deleted_routes: self.configured_routes.remove(x) tmp = self.config_fib_many_to_one( "10.0.0.%d", self.pg0.remote_ip4, 5, start=10) self.configured_routes.extend(tmp) for x in tmp: self.deleted_routes.remove(x) self.configured_routes.extend(self.config_fib_many_to_one( "10.0.1.%d", self.pg0.remote_ip4, 100)) self.verify_route_dump(self.configured_routes) self.stream_1 = self.create_stream( self.pg1, self.pg0, self.configured_routes, 300) self.stream_2 = self.create_stream( self.pg2, self.pg0, self.configured_routes, 300) self.stream_3 = self.create_stream( self.pg1, self.pg0, self.deleted_routes, 100) self.stream_4 = self.create_stream( self.pg2, self.pg0, self.deleted_routes, 100) self.pg1.add_stream(self.stream_1 + self.stream_3) self.pg2.add_stream(self.stream_2 + self.stream_4) self.pg_enable_capture(self.pg_interfaces) self.pg_start() pkts = self.pg0.get_capture(len(self.stream_1) + len(self.stream_2)) self.verify_capture(self.pg0, pkts, self.stream_1 + self.stream_2) # delete 5 routes check with traffic script. # add 100 routes check with traffic script. self.deleted_routes.extend(self.unconfig_fib_many_to_one( "10.0.0.%d", self.pg0.remote_ip4, 15)) self.deleted_routes.extend(self.unconfig_fib_many_to_one( "10.0.0.%d", self.pg0.remote_ip4, 85)) self.deleted_routes.extend(self.unconfig_fib_many_to_one( "10.0.1.%d", self.pg0.remote_ip4, 100)) self.verify_not_in_route_dump(self.deleted_routes) class TestIPNull(VppTestCase): """ IPv4 routes via NULL """ @classmethod def setUpClass(cls): super(TestIPNull, cls).setUpClass() @classmethod def tearDownClass(cls): super(TestIPNull, cls).tearDownClass() def setUp(self): super(TestIPNull, self).setUp() # create 2 pg interfaces self.create_pg_interfaces(range(2)) for i in self.pg_interfaces: i.admin_up() i.config_ip4() i.resolve_arp() def tearDown(self): super(TestIPNull, self).tearDown() for i in self.pg_interfaces: i.unconfig_ip4() i.admin_down() def test_ip_null(self): """ IP NULL route """ # # A route via IP NULL that will reply with ICMP unreachables # ip_unreach = VppIpRoute( self, "10.0.0.1", 32, [VppRoutePath("0.0.0.0", 0xffffffff, type=FibPathType.FIB_PATH_TYPE_ICMP_UNREACH)]) ip_unreach.add_vpp_config() p_unreach = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) / IP(src=self.pg0.remote_ip4, dst="10.0.0.1") / UDP(sport=1234, dport=1234) / Raw(b'\xa5' * 100)) self.pg0.add_stream(p_unreach) self.pg_enable_capture(self.pg_interfaces) self.pg_start() rx = self.pg0.get_capture(1) rx = rx[0] icmp = rx[ICMP] self.assertEqual(icmptypes[icmp.type], "dest-unreach") self.assertEqual(icmpcodes[icmp.type][icmp.code], "host-unreachable") self.assertEqual(icmp.src, self.pg0.remote_ip4) self.assertEqual(icmp.dst, "10.0.0.1") # # ICMP replies are rate limited. so sit and spin. # self.sleep(1) # # A route via IP NULL that will reply with ICMP prohibited # ip_prohibit = VppIpRoute( self, "10.0.0.2", 32, [VppRoutePath("0.0.0.0", 0xffffffff, type=FibPathType.FIB_PATH_TYPE_ICMP_PROHIBIT)]) ip_prohibit.add_vpp_config() p_prohibit = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) / IP(src=self.pg0.remote_ip4, dst="10.0.0.2") / UDP(sport=1234, dport=1234) / Raw(b'\xa5' * 100)) self.pg0.add_stream(p_prohibit) self.pg_enable_capture(self.pg_interfaces) self.pg_start() rx = self.pg0.get_capture(1) rx = rx[0] icmp = rx[ICMP] self.assertEqual(icmptypes[icmp.type], "dest-unreach") self.assertEqual(icmpcodes[icmp.type][icmp.code], "host-prohibited") self.assertEqual(icmp.src, self.pg0.remote_ip4) self.assertEqual(icmp.dst, "10.0.0.2") def test_ip_drop(self): """ IP Drop Routes """ p = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) / IP(src=self.pg0.remote_ip4, dst="1.1.1.1") / UDP(sport=1234, dport=1234) / Raw(b'\xa5' * 100)) r1 = VppIpRoute(self, "1.1.1.0", 24, [VppRoutePath(self.pg1.remote_ip4, self.pg1.sw_if_index)]) r1.add_vpp_config() rx = self.send_and_expect(self.pg0, p * NUM_PKTS, self.pg1) # # insert a more specific as a drop # r2 = VppIpRoute(self, "1.1.1.1", 32, [VppRoutePath("0.0.0.0", 0xffffffff, type=FibPathType.FIB_PATH_TYPE_DROP)]) r2.add_vpp_config() self.send_and_assert_no_replies(self.pg0, p * NUM_PKTS, "Drop Route") r2.remove_vpp_config() rx = self.send_and_expect(self.pg0, p * NUM_PKTS, self.pg1) class TestIPDisabled(VppTestCase): """ IPv4 disabled """ @classmethod def setUpClass(cls): super(TestIPDisabled, cls).setUpClass() @classmethod def tearDownClass(cls): super(TestIPDisabled, cls).tearDownClass() def setUp(self): super(TestIPDisabled, self).setUp() # create 2 pg interfaces self.create_pg_interfaces(range(2)) # PG0 is IP enalbed self.pg0.admin_up() self.pg0.config_ip4() self.pg0.resolve_arp() # PG 1 is not IP enabled self.pg1.admin_up() def tearDown(self): super(TestIPDisabled, self).tearDown() for } 
/*
 *------------------------------------------------------------------
 * Copyright (c) 2017 Cisco and/or its affiliates.
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at:
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 *------------------------------------------------------------------
 */

#include <memory>
#include <stdio.h>
#include <unistd.h>
#include <assert.h>
#include <setjmp.h>
#include <check.h>
#include <vapi/vapi.hpp>
#include <vapi/vpe.api.vapi.hpp>
#include <vapi/interface.api.vapi.hpp>
#include <fake.api.vapi.hpp>

DEFINE_VAPI_MSG_IDS_VPE_API_JSON;
DEFINE_VAPI_MSG_IDS_INTERFACE_API_JSON;
DEFINE_VAPI_MSG_IDS_FAKE_API_JSON;

static char *app_name = nullptr;
static char *api_prefix = nullptr;
static const int max_outstanding_requests = 32;
static const int response_queue_size = 32;

#define WAIT_FOR_RESPONSE(param, ret)      \
  do                                       \
    {                                      \
      ret = con.wait_for_response (param); \
    }                                      \
  while (ret == VAPI_EAGAIN)

using namespace vapi;

void verify_show_version_reply (const Show_version_reply &r)
{
  auto &p = r.get_payload ();
  printf ("show_version_reply: program: `%s', version: `%s', build directory: "
          "`%s', build date: `%s'\n",
          p.program, p.version, p.build_directory, p.build_date);
  ck_assert_str_eq ("vpe", (char *)p.program);
}

Connection con;

void setup (void)
{
  vapi_error_e rv = con.connect (
      app_name, api_prefix, max_outstanding_requests, response_queue_size);
  ck_assert_int_eq (VAPI_OK, rv);
}

void teardown (void)
{
  con.disconnect ();
}

START_TEST (test_show_version_1)
{
  printf ("--- Show version by reading response associated to request ---\n");
  Show_version sv (con);
  vapi_error_e rv = sv.execute ();
  ck_assert_int_eq (VAPI_OK, rv);
  WAIT_FOR_RESPONSE (sv, rv);
  ck_assert_int_eq (VAPI_OK, rv);
  auto &r = sv.get_response ();
  verify_show_version_reply (r);
}

END_TEST;

struct Show_version_cb
{
  Show_version_cb () : called{0} {};
  int called;
  vapi_error_e operator() (Show_version &sv)
  {
    auto &r = sv.get_response ();
    verify_show_version_reply (r);
    ++called;
    return VAPI_OK;
  }
};

START_TEST (test_show_version_2)
{
  printf ("--- Show version by getting a callback ---\n");
  Show_version_cb cb;
  Show_version sv (con, std::ref (cb));
  vapi_error_e rv = sv.execute ();
  ck_assert_int_eq (VAPI_OK, rv);
  con.dispatch (sv);
  ck_assert_int_eq (1, cb.called);
}

END_TEST;

START_TEST (test_loopbacks_1)
{
  printf ("--- Create/delete loopbacks by waiting for response ---\n");
  const auto num_ifs = 5;
  u8 mac_addresses[num_ifs][6];
  memset (&mac_addresses, 0, sizeof (mac_addresses));
  u32 sw_if_indexes[num_ifs];
  memset (&sw_if_indexes, 0xff, sizeof (sw_if_indexes));
  for (int i = 0; i < num_ifs; ++i)
    {
      memcpy (&mac_addresses[i], "\1\2\3\4\5\6", 6);
      mac_addresses[i][5] = i;
    }
  for (int i = 0; i < num_ifs; ++i)
    {
      Create_loopback cl (con);
      auto &p = cl.get_request ().get_payload ();
      memcpy (p.mac_address, mac_addresses[i], sizeof (p.mac_address));
      auto e = cl.execute ();
      ck_assert_int_eq (VAPI_OK, e);
      vapi_error_e rv;
      WAIT_FOR_RESPONSE (cl, rv);
      ck_assert_int_eq (VAPI_OK, rv);
      auto &rp = cl.get_response ().get_payload ();
      ck_assert_int_eq (0, rp.retval);
      sw_if_indexes[i] = rp.sw_if_index;
    }
  for (int i = 0; i < num_ifs; ++i)
    {
      printf ("Created loopback with MAC %02x:%02x:%02x:%02x:%02x:%02x --> "
              "sw_if_index %u\n",
              mac_addresses[i][0], mac_addresses[i][1], mac_addresses[i][2],
              mac_addresses[i][3], mac_addresses[i][4], mac_addresses[i][5],
              sw_if_indexes[i]);
    }

  { // new context
    bool seen[num_ifs] = {0};
    Sw_interface_dump d (con);
    auto &p = d.get_request ().get_payload ();
    auto rv = d.execute ();
    ck_assert_int_eq (VAPI_OK, rv);
    WAIT_FOR_RESPONSE (d, rv);
    ck_assert_int_eq (VAPI_OK, rv);
    auto &rs = d.get_result_set ();
    for (auto &r : rs)
      {
        auto &p = r.get_payload ();
        for (int i = 0; i < num_ifs; ++i)
          {
            if (sw_if_indexes[i] == p.sw_if_index)
              {
                ck_assert_int_eq (0, seen[i]);
                seen[i] = true;
              }
          }
      }
    for (int i = 0; i < num_ifs; ++i)
      {
        ck_assert_int_eq (1, seen[i]);
      }
  }

  for (int i = 0; i < num_ifs; ++i)
    {
      Delete_loopback dl (con);
      dl.get_request ().get_payload ().sw_if_index = sw_if_indexes[i];
      auto rv = dl.execute ();
      ck_assert_int_eq (VAPI_OK, rv);
      WAIT_FOR_RESPONSE (dl, rv);
      ck_assert_int_eq (VAPI_OK, rv);
      auto &response = dl.get_response ();
      auto rp = response.get_payload ();
      ck_assert_int_eq (0, rp.retval);
      printf ("Deleted loopback with sw_if_index %u\n", sw_if_indexes[i]);
    }

  { // new context
    Sw_interface_dump d (con);
    auto &p = d.get_request ().get_payload ();
    auto rv = d.execute ();
    ck_assert_int_eq (VAPI_OK, rv);
    WAIT_FOR_RESPONSE (d, rv);
    ck_assert_int_eq (VAPI_OK, rv);
    auto &rs = d.get_result_set ();
    for (auto &r : rs)
      {
        auto &p = r.get_payload ();
        for (int i = 0; i < num_ifs; ++i)
          {
            ck_assert_int_ne (sw_if_indexes[i], p.sw_if_index);
          }
      }
  }
}

END_TEST;

struct Create_loopback_cb
{
  Create_loopback_cb () : called{0}, sw_if_index{0} {};
  int called;
  u32 sw_if_index;
  bool seen;
  vapi_error_e operator() (Create_loopback &cl)
  {
    auto &r = cl.get_response ();
    sw_if_index = r.get_payload ().sw_if_index;
    ++called;
    return VAPI_OK;
  }
};

struct Delete_loopback_cb
{
  Delete_loopback_cb () : called{0}, sw_if_index{0} {};
  int called;
  u32 sw_if_index;
  bool seen;
  vapi_error_e operator() (Delete_loopback &dl)
  {
    auto &r = dl.get_response ();
    ck_assert_int_eq (0, r.get_payload ().retval);
    ++called;
    return VAPI_OK;
  }
};

template <int num_ifs> struct Sw_interface_dump_cb
{
  Sw_interface_dump_cb (std::array<Create_loopback_cb, num_ifs> &cbs)
      : called{0}, cbs{cbs} {};
  int called;
  std::array<Create_loopback_cb, num_ifs> &cbs;
  vapi_error_e operator() (Sw_interface_dump &d)
  {
    for (auto &y : cbs)
      {
        y.seen = false;
      }
    for (auto &x : d.get_result_set ())
      {
        auto &p = x.get_payload ();
        for (auto &y : cbs)
          {
            if (p.sw_if_index == y.sw_if_index)
              {
                y.seen = true;
              }
          }
      }
    for (auto &y : cbs)
      {
        ck_assert_int_eq (true, y.seen);
      }
    ++called;
    return VAPI_OK;
  }
};

START_TEST (test_loopbacks_2)
{
  printf ("--- Create/delete loopbacks by getting a callback ---\n");
  const auto num_ifs = 5;
  u8 mac_addresses[num_ifs][6];
  memset (&mac_addresses, 0, sizeof (mac_addresses));
  for (int i = 0; i < num_ifs; ++i)
    {
      memcpy (&mac_addresses[i], "\1\2\3\4\5\6", 6);
      mac_addresses[i][5] = i;
    }
  std::array<Create_loopback_cb, num_ifs> ccbs;
  std::array<std::unique_ptr<Create_loopback>, num_ifs> clcs;
  for (int i = 0; i < num_ifs; ++i)
    {
      Create_loopback *cl = new Create_loopback (con, std::ref (ccbs[i]));
      clcs[i].reset (cl);
      auto &p = cl->get_request ().get_payload ();
      memcpy (p.mac_address, mac_addresses[i], sizeof (p.mac_address));
      auto e = cl->execute ();
      ck_assert_int_eq (VAPI_OK, e);
    }
  con.dispatch ();
  for (int i = 0; i < num_ifs; ++i)
    {
      ck_assert_int_eq (1, ccbs[i].called);
      printf ("Created loopback with MAC %02x:%02x:%02x:%02x:%02x:%02x --> "
              "sw_if_index %u\n",
              mac_addresses[i][0], mac_addresses[i][1], mac_addresses[i][2],
              mac_addresses[i][3], mac_addresses[i][4], mac_addresses[i][5],
              ccbs[i].sw_if_index);
    }

  Sw_interface_dump_cb<num_ifs> swdcb (ccbs);
  Sw_interface_dump d (con, std::ref (swdcb));
  auto &p = d.get_request ().get_payload ();
  auto rv = d.execute ();
  ck_assert_int_eq (VAPI_OK, rv);
  WAIT_FOR_RESPONSE (d, rv);
  ck_assert_int_eq (VAPI_OK, rv);
  ck_assert_int_ne (0, swdcb.called);
  std::array<Delete_loopback_cb, num_ifs> dcbs;
  std::array<std::unique_ptr<Delete_loopback>, num_ifs> dlcs;
  for (int i = 0; i < num_ifs; ++i)
    {
      Delete_loopback *dl = new Delete_loopback (con, std::ref (dcbs[i]));
      dlcs[i].reset (dl);
      auto &p = dl->get_request ().get_payload ();
      p.sw_if_index = ccbs[i].sw_if_index;
      dcbs[i].sw_if_index = ccbs[i].sw_if_index;
      aut