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