test/test_ip4_irb.py

   1 #!/usr/bin/env python
   2 """IRB Test Case HLD:
   3
   4 **config**
   5     - L2 MAC learning enabled in l2bd
   6     - 2 routed interfaces untagged, bvi (Bridge Virtual Interface)
   7     - 2 bridged interfaces in l2bd with bvi
   8
   9 **test**
  10     - sending ip4 eth pkts between routed interfaces
  11         - 2 routed interfaces
  12         - 2 bridged interfaces
  13
  14     - 64B, 512B, 1518B, 9200B (ether_size)
  15
  16     - burst of pkts per interface
  17         - 257pkts per burst
  18         - routed pkts hitting different FIB entries
  19         - bridged pkts hitting different MAC entries
  20
  21 **verify**
  22     - all packets received correctly
  23
  24 """
  25
  26 import unittest
  27 from random import choice
  28
  29 from scapy.packet import Raw
  30 from scapy.layers.l2 import Ether
  31 from scapy.layers.inet import IP, UDP
  32
  33 from framework import VppTestCase, VppTestRunner
  34
  35
  36 class TestIpIrb(VppTestCase):
  37     """IRB Test Case"""
  38
  39     @classmethod
  40     def setUpClass(cls):
  41         """
  42         #. Create BD with MAC learning enabled and put interfaces to this BD.
  43         #. Configure IPv4 addresses on loopback interface and routed interface.
  44         #. Configure MAC address binding to IPv4 neighbors on loop0.
  45         #. Configure MAC address on pg2.
  46         #. Loopback BVI interface has remote hosts, one half of hosts are
  47            behind pg0 second behind pg1.
  48         """
  49         super(TestIpIrb, cls).setUpClass()
  50
  51         cls.pg_if_packet_sizes = [64, 512, 1518, 9018]  # packet sizes
  52         cls.bd_id = 10
  53         cls.remote_hosts_count = 250
  54
  55         # create 3 pg interfaces, 1 loopback interface
  56         cls.create_pg_interfaces(range(3))
  57         cls.create_loopback_interfaces(1)
  58
  59         cls.interfaces = list(cls.pg_interfaces)
  60         cls.interfaces.extend(cls.lo_interfaces)
  61
  62         for i in cls.interfaces:
  63             i.admin_up()
  64
  65         # Create BD with MAC learning enabled and put interfaces to this BD
  66         cls.vapi.sw_interface_set_l2_bridge(
  67             cls.loop0.sw_if_index, bd_id=cls.bd_id, bvi=1)
  68         cls.vapi.sw_interface_set_l2_bridge(
  69             cls.pg0.sw_if_index, bd_id=cls.bd_id)
  70         cls.vapi.sw_interface_set_l2_bridge(
  71             cls.pg1.sw_if_index, bd_id=cls.bd_id)
  72
  73         # Configure IPv4 addresses on loopback interface and routed interface
  74         cls.loop0.config_ip4()
  75         cls.pg2.config_ip4()
  76
  77         # Configure MAC address binding to IPv4 neighbors on loop0
  78         cls.loop0.generate_remote_hosts(cls.remote_hosts_count)
  79         cls.loop0.configure_ipv4_neighbors()
  80         # configure MAC address on pg2
  81         cls.pg2.resolve_arp()
  82
  83         # Loopback BVI interface has remote hosts, one half of hosts are behind
  84         # pg0 second behind pg1
  85         half = cls.remote_hosts_count // 2
  86         cls.pg0.remote_hosts = cls.loop0.remote_hosts[:half]
  87         cls.pg1.remote_hosts = cls.loop0.remote_hosts[half:]
  88
  89     def tearDown(self):
  90         """Run standard test teardown and log ``show l2patch``,
  91         ``show l2fib verbose``,``show bridge-domain <bd_id> detail``,
  92         ``show ip arp``.
  93         """
  94         super(TestIpIrb, self).tearDown()
  95         if not self.vpp_dead:
  96             self.logger.info(self.vapi.cli("show l2patch"))
  97             self.logger.info(self.vapi.cli("show l2fib verbose"))
  98             self.logger.info(self.vapi.cli("show bridge-domain %s detail" %
  99                                            self.bd_id))
 100             self.logger.info(self.vapi.cli("show ip arp"))
 101
 102     def create_stream(self, src_ip_if, dst_ip_if, packet_sizes):
 103         pkts = []
 104         for i in range(0, 257):
 105             remote_dst_host = choice(dst_ip_if.remote_hosts)
 106             info = self.create_packet_info(src_ip_if, dst_ip_if)
 107             payload = self.info_to_payload(info)
 108             p = (Ether(dst=src_ip_if.local_mac, src=src_ip_if.remote_mac) /
 109                  IP(src=src_ip_if.remote_ip4,
 110                     dst=remote_dst_host.ip4) /
 111                  UDP(sport=1234, dport=1234) /
 112                  Raw(payload))
 113             info.data = p.copy()
 114             size = packet_sizes[(i // 2) % len(packet_sizes)]
 115             self.extend_packet(p, size)
 116             pkts.append(p)
 117         return pkts
 118
 119     def create_stream_l2_to_ip(self, src_l2_if, src_ip_if, dst_ip_if,
 120                                packet_sizes):
 121         pkts = []
 122         for i in range(0, 257):
 123             info = self.create_packet_info(src_ip_if, dst_ip_if)
 124             payload = self.info_to_payload(info)
 125
 126             host = choice(src_l2_if.remote_hosts)
 127
 128             p = (Ether(src=host.mac,
 129                        dst=src_ip_if.local_mac) /
 130                  IP(src=host.ip4,
 131                     dst=dst_ip_if.remote_ip4) /
 132                  UDP(sport=1234, dport=1234) /
 133                  Raw(payload))
 134
 135             info.data = p.copy()
 136             size = packet_sizes[(i // 2) % len(packet_sizes)]
 137             self.extend_packet(p, size)
 138
 139             pkts.append(p)
 140         return pkts
 141
 142     def verify_capture_l2_to_ip(self, dst_ip_if, src_ip_if, capture):
 143         last_info = dict()
 144         for i in self.interfaces:
 145             last_info[i.sw_if_index] = None
 146
 147         dst_ip_sw_if_index = dst_ip_if.sw_if_index
 148
 149         for packet in capture:
 150             ip = packet[IP]
 151             udp = packet[IP][UDP]
 152             payload_info = self.payload_to_info(str(packet[IP][UDP][Raw]))
 153
 154             self.assertEqual(payload_info.dst, dst_ip_sw_if_index)
 155
 156             next_info = self.get_next_packet_info_for_interface2(
 157                 payload_info.src, dst_ip_sw_if_index,
 158                 last_info[payload_info.src])
 159             last_info[payload_info.src] = next_info
 160             self.assertTrue(next_info is not None)
 161             saved_packet = next_info.data
 162             self.assertTrue(next_info is not None)
 163
 164             # MAC: src, dst
 165             self.assertEqual(packet.src, dst_ip_if.local_mac)
 166             self.assertEqual(packet.dst, dst_ip_if.remote_mac)
 167
 168             # IP: src, dst
 169             host = src_ip_if.host_by_ip4(ip.src)
 170             self.assertIsNotNone(host)
 171             self.assertEqual(ip.dst, saved_packet[IP].dst)
 172             self.assertEqual(ip.dst, dst_ip_if.remote_ip4)
 173
 174             # UDP:
 175             self.assertEqual(udp.sport, saved_packet[UDP].sport)
 176             self.assertEqual(udp.dport, saved_packet[UDP].dport)
 177
 178     def verify_capture(self, dst_ip_if, src_ip_if, capture):
 179         last_info = dict()
 180         for i in self.interfaces:
 181             last_info[i.sw_if_index] = None
 182
 183         dst_ip_sw_if_index = dst_ip_if.sw_if_index
 184
 185         for packet in capture:
 186             ip = packet[IP]
 187             udp = packet[IP][UDP]
 188             payload_info = self.payload_to_info(str(packet[IP][UDP][Raw]))
 189             packet_index = payload_info.index
 190
 191             self.assertEqual(payload_info.dst, dst_ip_sw_if_index)
 192
 193             next_info = self.get_next_packet_info_for_interface2(
 194                 payload_info.src, dst_ip_sw_if_index,
 195                 last_info[payload_info.src])
 196             last_info[payload_info.src] = next_info
 197             self.assertTrue(next_info is not None)
 198             self.assertEqual(packet_index, next_info.index)
 199             saved_packet = next_info.data
 200             self.assertTrue(next_info is not None)
 201
 202             # MAC: src, dst
 203             self.assertEqual(packet.src, dst_ip_if.local_mac)
 204             host = dst_ip_if.host_by_mac(packet.dst)
 205
 206             # IP: src, dst
 207             self.assertEqual(ip.src, src_ip_if.remote_ip4)
 208             self.assertEqual(ip.dst, saved_packet[IP].dst)
 209             self.assertEqual(ip.dst, host.ip4)
 210
 211             # UDP:
 212             self.assertEqual(udp.sport, saved_packet[UDP].sport)
 213             self.assertEqual(udp.dport, saved_packet[UDP].dport)
 214
 215     def test_ip4_irb_1(self):
 216         """ IPv4 IRB test 1
 217
 218         Test scenario:
 219             - ip traffic from pg2 interface must ends in both pg0 and pg1
 220             - arp entry present in loop0 interface for destination IP
 221             - no l2 entree configured, pg0 and pg1 are same
 222         """
 223
 224         stream = self.create_stream(
 225             self.pg2, self.loop0, self.pg_if_packet_sizes)
 226         self.pg2.add_stream(stream)
 227
 228         self.pg_enable_capture(self.pg_interfaces)
 229         self.pg_start()
 230
 231         packet_count = self.get_packet_count_for_if_idx(self.loop0.sw_if_index)
 232
 233         rcvd1 = self.pg0.get_capture(packet_count)
 234         rcvd2 = self.pg1.get_capture(packet_count)
 235
 236         self.verify_capture(self.loop0, self.pg2, rcvd1)
 237         self.verify_capture(self.loop0, self.pg2, rcvd2)
 238
 239         self.assertListEqual(rcvd1.res, rcvd2.res)
 240
 241     def test_ip4_irb_2(self):
 242         """ IPv4 IRB test 2
 243
 244         Test scenario:
 245             - ip traffic from pg0 and pg1 ends on pg2
 246         """
 247
 248         stream1 = self.create_stream_l2_to_ip(
 249             self.pg0, self.loop0, self.pg2, self.pg_if_packet_sizes)
 250         stream2 = self.create_stream_l2_to_ip(
 251             self.pg1, self.loop0, self.pg2, self.pg_if_packet_sizes)
 252         self.pg0.add_stream(stream1)
 253         self.pg1.add_stream(stream2)
 254
 255         self.pg_enable_capture(self.pg_interfaces)
 256         self.pg_start()
 257
 258         rcvd = self.pg2.get_capture()
 259         self.verify_capture_l2_to_ip(self.pg2, self.loop0, rcvd)
 260
 261
 262 if __name__ == '__main__':
 263     unittest.main(testRunner=VppTestRunner)