test/test_ip4.py

   1 #!/usr/bin/env python
   2
   3 import unittest
   4 import socket
   5
   6 from framework import VppTestCase, VppTestRunner
   7 from vpp_interface import VppInterface
   8 from vpp_sub_interface import VppSubInterface, VppDot1QSubint, VppDot1ADSubint
   9
  10 from scapy.packet import Raw
  11 from scapy.layers.l2 import Ether, Dot1Q
  12 from scapy.layers.inet import IP, UDP
  13
  14
  15 class TestIPv4(VppTestCase):
  16     """ IPv4 Test Case """
  17
  18     def setUp(self):
  19         """
  20         Perform test setup before test case.
  21
  22         **Config:**
  23             - create 3 pg interfaces
  24                 - untagged pg0 interface
  25                 - Dot1Q subinterface on pg1
  26                 - Dot1AD subinterface on pg2
  27             - setup interfaces:
  28                 - put it into UP state
  29                 - set IPv4 addresses
  30                 - resolve neighbor address using ARP
  31             - configure 200 fib entries
  32
  33         :ivar list interfaces: pg interfaces and subinterfaces.
  34         :ivar dict flows: IPv4 packet flows in test.
  35         :ivar list pg_if_packet_sizes: packet sizes in test.
  36         """
  37         super(TestIPv4, self).setUp()
  38
  39         # create 3 pg interfaces
  40         self.create_pg_interfaces(range(3))
  41
  42         # create 2 subinterfaces for pg1 and pg2
  43         self.sub_interfaces = [
  44             VppDot1QSubint(self, self.pg1, 100),
  45             VppDot1ADSubint(self, self.pg2, 200, 300, 400)]
  46
  47         # packet flows mapping pg0 -> pg1.sub, pg2.sub, etc.
  48         self.flows = dict()
  49         self.flows[self.pg0] = [self.pg1.sub_if, self.pg2.sub_if]
  50         self.flows[self.pg1.sub_if] = [self.pg0, self.pg2.sub_if]
  51         self.flows[self.pg2.sub_if] = [self.pg0, self.pg1.sub_if]
  52
  53         # packet sizes
  54         self.pg_if_packet_sizes = [64, 512, 1518, 9018]
  55         self.sub_if_packet_sizes = [64, 512, 1518 + 4, 9018 + 4]
  56
  57         self.interfaces = list(self.pg_interfaces)
  58         self.interfaces.extend(self.sub_interfaces)
  59
  60         # setup all interfaces
  61         for i in self.interfaces:
  62             i.admin_up()
  63             i.config_ip4()
  64             i.resolve_arp()
  65
  66         # config 2M FIB entries
  67         self.config_fib_entries(200)
  68
  69     def tearDown(self):
  70         """Run standard test teardown and log ``show ip arp``."""
  71         super(TestIPv4, self).tearDown()
  72         if not self.vpp_dead:
  73             self.logger.info(self.vapi.cli("show ip arp"))
  74             # info(self.vapi.cli("show ip fib"))  # many entries
  75
  76     def config_fib_entries(self, count):
  77         """For each interface add to the FIB table *count* routes to
  78         "10.0.0.1/32" destination with interface's local address as next-hop
  79         address.
  80
  81         :param int count: Number of FIB entries.
  82
  83         - *TODO:* check if the next-hop address shouldn't be remote address
  84           instead of local address.
  85         """
  86         n_int = len(self.interfaces)
  87         percent = 0
  88         counter = 0.0
  89         dest_addr = socket.inet_pton(socket.AF_INET, "10.0.0.1")
  90         dest_addr_len = 32
  91         for i in self.interfaces:
  92             next_hop_address = i.local_ip4n
  93             for j in range(count / n_int):
  94                 self.vapi.ip_add_del_route(
  95                     dest_addr, dest_addr_len, next_hop_address)
  96                 counter += 1
  97                 if counter / count * 100 > percent:
  98                     self.logger.info("Configure %d FIB entries .. %d%% done" %
  99                                      (count, percent))
 100                     percent += 1
 101
 102     def create_stream(self, src_if, packet_sizes):
 103         """Create input packet stream for defined interface.
 104
 105         :param VppInterface src_if: Interface to create packet stream for.
 106         :param list packet_sizes: Required packet sizes.
 107         """
 108         pkts = []
 109         for i in range(0, 257):
 110             dst_if = self.flows[src_if][i % 2]
 111             info = self.create_packet_info(
 112                 src_if.sw_if_index, dst_if.sw_if_index)
 113             payload = self.info_to_payload(info)
 114             p = (Ether(dst=src_if.local_mac, src=src_if.remote_mac) /
 115                  IP(src=src_if.remote_ip4, dst=dst_if.remote_ip4) /
 116                  UDP(sport=1234, dport=1234) /
 117                  Raw(payload))
 118             info.data = p.copy()
 119             if isinstance(src_if, VppSubInterface):
 120                 p = src_if.add_dot1_layer(p)
 121             size = packet_sizes[(i // 2) % len(packet_sizes)]
 122             self.extend_packet(p, size)
 123             pkts.append(p)
 124         return pkts
 125
 126     def verify_capture(self, dst_if, capture):
 127         """Verify captured input packet stream for defined interface.
 128
 129         :param VppInterface dst_if: Interface to verify captured packet stream
 130                                     for.
 131         :param list capture: Captured packet stream.
 132         """
 133         self.logger.info("Verifying capture on interface %s" % dst_if.name)
 134         last_info = dict()
 135         for i in self.interfaces:
 136             last_info[i.sw_if_index] = None
 137         is_sub_if = False
 138         dst_sw_if_index = dst_if.sw_if_index
 139         if hasattr(dst_if, 'parent'):
 140             is_sub_if = True
 141         for packet in capture:
 142             if is_sub_if:
 143                 # Check VLAN tags and Ethernet header
 144                 packet = dst_if.remove_dot1_layer(packet)
 145             self.assertTrue(Dot1Q not in packet)
 146             try:
 147                 ip = packet[IP]
 148                 udp = packet[UDP]
 149                 payload_info = self.payload_to_info(str(packet[Raw]))
 150                 packet_index = payload_info.index
 151                 self.assertEqual(payload_info.dst, dst_sw_if_index)
 152                 self.logger.debug("Got packet on port %s: src=%u (id=%u)" %
 153                                   (dst_if.name, payload_info.src, packet_index))
 154                 next_info = self.get_next_packet_info_for_interface2(
 155                     payload_info.src, dst_sw_if_index,
 156                     last_info[payload_info.src])
 157                 last_info[payload_info.src] = next_info
 158                 self.assertTrue(next_info is not None)
 159                 self.assertEqual(packet_index, next_info.index)
 160                 saved_packet = next_info.data
 161                 # Check standard fields
 162                 self.assertEqual(ip.src, saved_packet[IP].src)
 163                 self.assertEqual(ip.dst, saved_packet[IP].dst)
 164                 self.assertEqual(udp.sport, saved_packet[UDP].sport)
 165                 self.assertEqual(udp.dport, saved_packet[UDP].dport)
 166             except:
 167                 self.logger.error("Unexpected or invalid packet:")
 168                 self.logger.error(packet.show())
 169                 raise
 170         for i in self.interfaces:
 171             remaining_packet = self.get_next_packet_info_for_interface2(
 172                 i.sw_if_index, dst_sw_if_index, last_info[i.sw_if_index])
 173             self.assertTrue(
 174                 remaining_packet is None,
 175                 "Interface %s: Packet expected from interface %s didn't arrive" %
 176                 (dst_if.name, i.name))
 177
 178     def test_fib(self):
 179         """ IPv4 FIB test
 180
 181         Test scenario:
 182
 183             - Create IPv4 stream for pg0 interface
 184             - Create IPv4 tagged streams for pg1's and pg2's subinterface.
 185             - Send and verify received packets on each interface.
 186         """
 187
 188         pkts = self.create_stream(self.pg0, self.pg_if_packet_sizes)
 189         self.pg0.add_stream(pkts)
 190
 191         for i in self.sub_interfaces:
 192             pkts = self.create_stream(i, self.sub_if_packet_sizes)
 193             i.parent.add_stream(pkts)
 194
 195         self.pg_enable_capture(self.pg_interfaces)
 196         self.pg_start()
 197
 198         pkts = self.pg0.get_capture()
 199         self.verify_capture(self.pg0, pkts)
 200
 201         for i in self.sub_interfaces:
 202             pkts = i.parent.get_capture()
 203             self.verify_capture(i, pkts)
 204
 205
 206 if __name__ == '__main__':
 207     unittest.main(testRunner=VppTestRunner)