test/test_ip4_irb.py

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