PAPI: Add MACAddress object wrapper for vl_api_mac_address_t

[vpp.git] / test / test_ip4_irb.py

diff --git a/test/test_ip4_irb.py b/test/test_ip4_irb.py
index cf2bb15..8c3830c 100644 (file)
--- a/test/test_ip4_irb.py
+++ b/test/test_ip4_irb.py
@@ -31,6 +31,8 @@ from scapy.layers.l2 import Ether
 from scapy.layers.inet import IP, UDP
 
 from framework import VppTestCase, VppTestRunner
+from vpp_papi import MACAddress
+from vpp_papi_provider import L2_PORT_TYPE
 
 
 class TestIpIrb(VppTestCase):
@@ -54,7 +56,7 @@ class TestIpIrb(VppTestCase):
 
         # create 3 pg interfaces, 1 loopback interface
         cls.create_pg_interfaces(range(3))
-        cls.create_loopback_interfaces(range(1))
+        cls.create_loopback_interfaces(1)
 
         cls.interfaces = list(cls.pg_interfaces)
         cls.interfaces.extend(cls.lo_interfaces)
@@ -64,7 +66,8 @@ class TestIpIrb(VppTestCase):
 
         # Create BD with MAC learning enabled and put interfaces to this BD
         cls.vapi.sw_interface_set_l2_bridge(
-            cls.loop0.sw_if_index, bd_id=cls.bd_id, bvi=1)
+            cls.loop0.sw_if_index, bd_id=cls.bd_id,
+            port_type=L2_PORT_TYPE.BVI)
         cls.vapi.sw_interface_set_l2_bridge(
             cls.pg0.sw_if_index, bd_id=cls.bd_id)
         cls.vapi.sw_interface_set_l2_bridge(
@@ -103,8 +106,7 @@ class TestIpIrb(VppTestCase):
         pkts = []
         for i in range(0, 257):
             remote_dst_host = choice(dst_ip_if.remote_hosts)
-            info = self.create_packet_info(
-                src_ip_if.sw_if_index, dst_ip_if.sw_if_index)
+            info = self.create_packet_info(src_ip_if, dst_ip_if)
             payload = self.info_to_payload(info)
             p = (Ether(dst=src_ip_if.local_mac, src=src_ip_if.remote_mac) /
                  IP(src=src_ip_if.remote_ip4,
@@ -121,14 +123,13 @@ class TestIpIrb(VppTestCase):
                                packet_sizes):
         pkts = []
         for i in range(0, 257):
-            info = self.create_packet_info(
-                src_ip_if.sw_if_index, dst_ip_if.sw_if_index)
+            info = self.create_packet_info(src_ip_if, dst_ip_if)
             payload = self.info_to_payload(info)
 
             host = choice(src_l2_if.remote_hosts)
 
             p = (Ether(src=host.mac,
-                       dst = src_ip_if.local_mac) /
+                       dst=src_ip_if.local_mac) /
                  IP(src=host.ip4,
                     dst=dst_ip_if.remote_ip4) /
                  UDP(sport=1234, dport=1234) /
@@ -152,7 +153,6 @@ class TestIpIrb(VppTestCase):
             ip = packet[IP]
             udp = packet[IP][UDP]
             payload_info = self.payload_to_info(str(packet[IP][UDP][Raw]))
-            packet_index = payload_info.index
 
             self.assertEqual(payload_info.dst, dst_ip_sw_if_index)
 
@@ -231,35 +231,45 @@ class TestIpIrb(VppTestCase):
         self.pg_enable_capture(self.pg_interfaces)
         self.pg_start()
 
-        rcvd1 = self.pg0.get_capture()
-        rcvd2 = self.pg1.get_capture()
+        packet_count = self.get_packet_count_for_if_idx(self.loop0.sw_if_index)
+
+        rcvd1 = self.pg0.get_capture(packet_count)
+        rcvd2 = self.pg1.get_capture(packet_count)
 
         self.verify_capture(self.loop0, self.pg2, rcvd1)
         self.verify_capture(self.loop0, self.pg2, rcvd2)
 
         self.assertListEqual(rcvd1.res, rcvd2.res)
 
-    def test_ip4_irb_2(self):
-        """ IPv4 IRB test 2
-
-        Test scenario:
-            - ip traffic from pg0 and pg1 ends on pg2
-        """
-
+    def send_and_verify_l2_to_ip(self):
         stream1 = self.create_stream_l2_to_ip(
             self.pg0, self.loop0, self.pg2, self.pg_if_packet_sizes)
         stream2 = self.create_stream_l2_to_ip(
             self.pg1, self.loop0, self.pg2, self.pg_if_packet_sizes)
+        self.vapi.cli("clear trace")
         self.pg0.add_stream(stream1)
         self.pg1.add_stream(stream2)
 
         self.pg_enable_capture(self.pg_interfaces)
         self.pg_start()
 
-        rcvd = self.pg2.get_capture()
+        rcvd = self.pg2.get_capture(514)
         self.verify_capture_l2_to_ip(self.pg2, self.loop0, rcvd)
 
-        self.assertEqual(len(stream1) + len(stream2), len(rcvd.res))
+    def test_ip4_irb_2(self):
+        """ IPv4 IRB test 2
+
+        Test scenario:
+            - ip traffic from pg0 and pg1 ends on pg2
+        """
+        self.send_and_verify_l2_to_ip()
+
+        # change the BVI's mac and resed traffic
+        self.loop0.set_mac(MACAddress("00:00:00:11:11:33"))
+
+        self.send_and_verify_l2_to_ip()
+        # check it wasn't flooded
+        self.pg1.assert_nothing_captured(remark="UU Flood")
 
 
 if __name__ == '__main__':