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