X-Git-Url: https://gerrit.fd.io/r/gitweb?a=blobdiff_plain;f=test%2Ftest_acl_plugin_macip.py;h=fa051093fdf8b1f41c9deecb734fca5b31575a57;hb=90cf21b5d8fd2d3e531e841dcd752311df5f8a50;hp=d27458d7df07fdf0c75d8b56b4db7f74cef820a2;hpb=8daa80a4adfd82a19017c2c12554a8a43dddccd7;p=vpp.git

diff --git a/test/test_acl_plugin_macip.py b/test/test_acl_plugin_macip.py
index d27458d7df0..fa051093fdf 100644
--- a/test/test_acl_plugin_macip.py
+++ b/test/test_acl_plugin_macip.py
@@ -1,12 +1,16 @@
 #!/usr/bin/env python
+from __future__ import print_function
 """ACL plugin - MACIP tests
 """
+import binascii
+import ipaddress
 import random
+from socket import inet_ntop, inet_pton, AF_INET, AF_INET6
+from struct import pack, unpack
 import re
 import unittest
 
-from socket import inet_ntop, inet_pton, AF_INET, AF_INET6
-from struct import *
+import scapy.compat
 from scapy.packet import Raw
 from scapy.layers.l2 import Ether
 from scapy.layers.inet import IP, UDP
@@ -14,8 +18,9 @@ from scapy.layers.inet6 import IPv6
 
 from framework import VppTestCase, VppTestRunner, running_extended_tests
 from vpp_lo_interface import VppLoInterface
-from vpp_papi_provider import L2_VTR_OP
-from vpp_sub_interface import VppSubInterface, VppDot1QSubint, VppDot1ADSubint
+from vpp_l2 import L2_PORT_TYPE
+from vpp_sub_interface import L2_VTR_OP, VppSubInterface, VppDot1QSubint, \
+    VppDot1ADSubint
 
 
 class MethodHolder(VppTestCase):
@@ -63,7 +68,7 @@ class MethodHolder(VppTestCase):
         try:
             # create 4 pg interfaces, 1 loopback interface
             cls.create_pg_interfaces(range(4))
-            cls.create_loopback_interfaces(range(1))
+            cls.create_loopback_interfaces(1)
 
             # create 2 subinterfaces
             cls.subifs = [
@@ -90,15 +95,16 @@ class MethodHolder(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)
+                rx_sw_if_index=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)
+                rx_sw_if_index=cls.pg0.sw_if_index, bd_id=cls.bd_id)
             cls.vapi.sw_interface_set_l2_bridge(
-                cls.pg1.sw_if_index, bd_id=cls.bd_id)
+                rx_sw_if_index=cls.pg1.sw_if_index, bd_id=cls.bd_id)
             cls.vapi.sw_interface_set_l2_bridge(
-                cls.subifs[0].sw_if_index, bd_id=cls.bd_id)
+                rx_sw_if_index=cls.subifs[0].sw_if_index, bd_id=cls.bd_id)
             cls.vapi.sw_interface_set_l2_bridge(
-                cls.subifs[1].sw_if_index, bd_id=cls.bd_id)
+                rx_sw_if_index=cls.subifs[1].sw_if_index, bd_id=cls.bd_id)
 
             # Configure IPv4/6 addresses on loop interface and routed interface
             cls.loop0.config_ip4()
@@ -142,9 +148,13 @@ class MethodHolder(VppTestCase):
             cls.subifs[3].remote_hosts = cls.loop0.remote_hosts[175:]
 
         except Exception:
-            super(TestMACIP, cls).tearDownClass()
+            super(MethodHolder, cls).tearDownClass()
             raise
 
+    @classmethod
+    def tearDownClass(cls):
+        super(MethodHolder, cls).tearDownClass()
+
     def setUp(self):
         super(MethodHolder, self).setUp()
         self.reset_packet_infos()
@@ -155,43 +165,46 @@ class MethodHolder(VppTestCase):
         Show various debug prints after each test.
         """
         super(MethodHolder, self).tearDown()
-        if not self.vpp_dead:
-            self.logger.info(self.vapi.ppcli("show interface address"))
-            self.logger.info(self.vapi.ppcli("show hardware"))
-            self.logger.info(self.vapi.ppcli("sh acl-plugin macip acl"))
-            self.logger.info(self.vapi.ppcli("sh acl-plugin macip interface"))
-            self.logger.info(self.vapi.ppcli("sh classify tables verbose"))
-            self.logger.info(self.vapi.ppcli("sh acl-plugin acl"))
-            self.logger.info(self.vapi.ppcli("sh acl-plugin interface"))
-            self.logger.info(self.vapi.ppcli("sh acl-plugin tables"))
-            # print self.vapi.ppcli("show interface address")
-            # print self.vapi.ppcli("show hardware")
-            # print self.vapi.ppcli("sh acl-plugin macip interface")
-            # print self.vapi.ppcli("sh acl-plugin macip acl")
+
+    def show_commands_at_teardown(self):
+        self.logger.info(self.vapi.ppcli("show interface address"))
+        self.logger.info(self.vapi.ppcli("show hardware"))
+        self.logger.info(self.vapi.ppcli("sh acl-plugin macip acl"))
+        self.logger.info(self.vapi.ppcli("sh acl-plugin macip interface"))
+        self.logger.info(self.vapi.ppcli("sh classify tables verbose"))
+        self.logger.info(self.vapi.ppcli("sh acl-plugin acl"))
+        self.logger.info(self.vapi.ppcli("sh acl-plugin interface"))
+        self.logger.info(self.vapi.ppcli("sh acl-plugin tables"))
+        # print(self.vapi.ppcli("show interface address"))
+        # print(self.vapi.ppcli("show hardware"))
+        # print(self.vapi.ppcli("sh acl-plugin macip interface"))
+        # print(self.vapi.ppcli("sh acl-plugin macip acl"))
         self.delete_acls()
 
     def macip_acl_dump_debug(self):
         acls = self.vapi.macip_acl_dump()
         if self.DEBUG:
             for acl in acls:
-                print "ACL #"+str(acl.acl_index)
+                print("ACL #"+str(acl.acl_index))
                 for r in acl.r:
                     rule = "ACTION"
                     if r.is_permit == 1:
                         rule = "PERMIT"
                     elif r.is_permit == 0:
                         rule = "DENY  "
-                    print "    IP6" if r.is_ipv6 else "    IP4", \
-                          rule, \
-                          r.src_mac.encode('hex'), \
-                          r.src_mac_mask.encode('hex'),\
-                          unpack('<16B', r.src_ip_addr), \
-                          r.src_ip_prefix_len
+                    print("    IP6" if r.is_ipv6 else "    IP4",
+                          rule,
+                          binascii.hexlify(r.src_mac),
+                          binascii.hexlify(r.src_mac_mask),
+                          unpack('<16B', r.src_ip_addr),
+                          r.src_ip_prefix_len)
         return acls
 
     def create_rules(self, mac_type=EXACT_MAC, ip_type=EXACT_IP,
-                     acl_count=1, rules_count=[1]):
+                     acl_count=1, rules_count=None):
         acls = []
+        if rules_count is None:
+            rules_count = [1]
         src_mac = int("220000dead00", 16)
         for acl in range(2, (acl_count+1) * 2):
             rules = []
@@ -233,7 +246,8 @@ class MethodHolder(VppTestCase):
                     mac = ':'.join(re.findall('..', '{:02x}'.format(
                         src_mac))[:3])+":00:00:00"
                 else:
-                    mac = ':'.join(re.findall('..', '{:02x}'.format(src_mac)))
+                    mac = ':'.join(re.findall(
+                        '..', '{:02x}'.format(src_mac)))
 
                 if ip_type == self.EXACT_IP:
                     ip4[3] = random.randint(100, 200)
@@ -243,7 +257,7 @@ class MethodHolder(VppTestCase):
                     ip4[3] = 0
                     ip6[8] = random.randint(100, 200)
                     ip6[15] = 0
-                ip_pack = ''
+                ip_pack = b''
                 for j in range(0, len(ip)):
                     ip_pack += pack('<B', int(ip[j]))
 
@@ -251,8 +265,9 @@ class MethodHolder(VppTestCase):
                          'is_ipv6': is_ip6,
                          'src_ip_addr': ip_pack,
                          'src_ip_prefix_len': ip_len,
-                         'src_mac': mac.replace(':', '').decode('hex'),
-                         'src_mac_mask': mask.replace(':', '').decode('hex')})
+                         'src_mac': binascii.unhexlify(mac.replace(':', '')),
+                         'src_mac_mask': binascii.unhexlify(
+                             mask.replace(':', ''))})
                 rules.append(rule)
                 if ip_type == self.WILD_IP:
                     break
@@ -270,7 +285,7 @@ class MethodHolder(VppTestCase):
     def verify_macip_acls(self, acl_count, rules_count, expected_count=2):
         reply = self.macip_acl_dump_debug()
         for acl in range(2, (acl_count+1) * 2):
-            self.assertEqual(reply[acl - 2].count, rules_count[acl/2-1])
+            self.assertEqual(reply[acl - 2].count, rules_count[acl//2-1])
 
         self.vapi.macip_acl_interface_get()
 
@@ -287,13 +302,10 @@ class MethodHolder(VppTestCase):
         reply = self.vapi.macip_acl_dump()
         self.assertEqual(len(reply), 0)
 
-        intf_acls = self.vapi.ppcli("sh acl-plugin interface").split(
-            "\nsw_if_index")
+        intf_acls = self.vapi.acl_interface_list_dump()
         for i_a in intf_acls:
-            ia = i_a.split(":")
-            if len(ia) == 3:
-                sw_if_index = int(ia[0])
-                acl_index = int(ia[2])
+            sw_if_index = i_a.sw_if_index
+            for acl_index in i_a.acls:
                 self.vapi.acl_interface_add_del(sw_if_index, acl_index, 0)
                 self.vapi.acl_del(acl_index)
 
@@ -408,30 +420,32 @@ class MethodHolder(VppTestCase):
                         sub_ip[15] = random.randint(200, 255)
                     elif ip_type == self.SUBNET_IP:
                         if denyIP:
-                            sub_ip[2] = str(int(sub_ip[2]) + 1)
+                            sub_ip[2] = int(sub_ip[2]) + 1
                         sub_ip[14] = random.randint(100, 199)
                         sub_ip[15] = random.randint(200, 255)
-                    src_ip6 = inet_ntop(AF_INET6, str(bytearray(sub_ip)))
+                    packed_src_ip6 = b''.join(
+                        [scapy.compat.chb(x) for x in sub_ip])
+                    src_ip6 = inet_ntop(AF_INET6, packed_src_ip6)
                 packet /= IPv6(src=src_ip6, dst=dst_ip6)
             else:
                 if ip_type != self.EXACT_IP:
                     sub_ip = ip_rule.split('.')
                     if ip_type == self.WILD_IP:
-                        sub_ip[0] = str(random.randint(1, 49))
-                        sub_ip[1] = str(random.randint(50, 99))
-                        sub_ip[2] = str(random.randint(100, 199))
-                        sub_ip[3] = str(random.randint(200, 255))
+                        sub_ip[0] = random.randint(1, 49)
+                        sub_ip[1] = random.randint(50, 99)
+                        sub_ip[2] = random.randint(100, 199)
+                        sub_ip[3] = random.randint(200, 255)
                     elif ip_type == self.SUBNET_IP:
                         if denyIP:
-                            sub_ip[1] = str(int(sub_ip[1])+1)
-                        sub_ip[2] = str(random.randint(100, 199))
-                        sub_ip[3] = str(random.randint(200, 255))
-                    src_ip4 = ".".join(sub_ip)
+                            sub_ip[1] = int(sub_ip[1])+1
+                        sub_ip[2] = random.randint(100, 199)
+                        sub_ip[3] = random.randint(200, 255)
+                    src_ip4 = '.'.join(['{!s}'.format(x) for x in sub_ip])
                 packet /= IP(src=src_ip4, dst=dst_ip4, frag=0, flags=0)
 
             packet /= UDP(sport=src_port, dport=dst_port)/Raw(payload)
 
-            packet[Raw].load += " mac:"+src_mac
+            packet[Raw].load += b" mac:%s" % scapy.compat.raw(src_mac)
 
             size = self.pg_if_packet_sizes[p % len(self.pg_if_packet_sizes)]
             if isinstance(src_if, VppSubInterface):
@@ -483,7 +497,9 @@ class MethodHolder(VppTestCase):
                         sub_ip = list(unpack('<16B', inet_pton(AF_INET6, ip)))
                         for i in range(8, 16):
                             sub_ip[i] = 0
-                        ip = inet_ntop(AF_INET6, str(bytearray(sub_ip)))
+                        packed_ip = b''.join(
+                            [scapy.compat.chb(x) for x in sub_ip])
+                        ip = inet_ntop(AF_INET6, packed_ip)
             else:
                 if ip_type == self.WILD_IP:
                     ip = "0.0.0.0"
@@ -538,8 +554,9 @@ class MethodHolder(VppTestCase):
                     'is_ipv6': is_ip6,
                     'src_ip_addr': ip_rule,
                     'src_ip_prefix_len': prefix_len,
-                    'src_mac': mac_rule.replace(':', '').decode('hex'),
-                    'src_mac_mask': mac_mask.replace(':', '').decode('hex')})
+                    'src_mac': binascii.unhexlify(mac_rule.replace(':', '')),
+                    'src_mac_mask': binascii.unhexlify(
+                        mac_mask.replace(':', ''))})
                 macip_rules.append(macip_rule)
 
         # deny all other packets
@@ -569,26 +586,27 @@ class MethodHolder(VppTestCase):
         # p_l3 = IPv6 if is_ip6 else IP
         # if self.DEBUG:
         #     for p in stream:
-        #         print p[Ether].src, p[Ether].dst, p[p_l3].src, p[p_l3].dst
+        #         print(p[Ether].src, p[Ether].dst, p[p_l3].src, p[p_l3].dst)
         #
         # acls = self.macip_acl_dump_debug()
 
         # TODO : verify
         # for acl in acls:
         #     for r in acl.r:
-        #         print r.src_mac.encode('hex'), \
-        #               r.src_mac_mask.encode('hex'),\
+        #         print(binascii.hexlify(r.src_mac), \
+        #               binascii.hexlify(r.src_mac_mask),\
         #               unpack('<16B', r.src_ip_addr), \
-        #               r.src_ip_prefix_len
+        #               r.src_ip_prefix_len)
         #
         # for p in capture:
-        #     print p[Ether].src, p[Ether].dst, p[p_l3].src, p[p_l3].dst
-        #     data = p[Raw].load.split(':',1)[1]
-        #     print p[p_l3].src, data
+        #     print(p[Ether].src, p[Ether].dst, p[p_l3].src, p[p_l3].dst
+        #     data = p[Raw].load.split(':',1)[1])
+        #     print(p[p_l3].src, data)
 
     def run_traffic(self, mac_type, ip_type, traffic, is_ip6, packets,
                     do_not_expected_capture=False, tags=None,
-                    apply_rules=True, isMACIP=True, permit_tags=PERMIT_TAGS):
+                    apply_rules=True, isMACIP=True, permit_tags=PERMIT_TAGS,
+                    try_replace=False):
         self.reset_packet_infos()
 
         if tags is None:
@@ -650,6 +668,15 @@ class MethodHolder(VppTestCase):
             self.vapi.macip_acl_interface_add_del(
                 sw_if_index=tx_if.sw_if_index,
                 acl_index=0)
+        if try_replace:
+            if isMACIP:
+                reply = self.vapi.macip_acl_add_replace(
+                                                   test_dict['macip_rules'],
+                                                   acl_index)
+            else:
+                reply = self.vapi.acl_add_replace(acl_index=acl_index,
+                                                  r=test_dict['acl_rules'])
+            self.assertEqual(reply.retval, 0)
 
         if not isinstance(src_if, VppSubInterface):
             tx_if.add_stream(test_dict['stream'])
@@ -686,6 +713,14 @@ class MethodHolder(VppTestCase):
 class TestMACIP_IP4(MethodHolder):
     """MACIP with IP4 traffic"""
 
+    @classmethod
+    def setUpClass(cls):
+        super(TestMACIP_IP4, cls).setUpClass()
+
+    @classmethod
+    def tearDownClass(cls):
+        super(TestMACIP_IP4, cls).tearDownClass()
+
     def test_acl_bridged_ip4_exactMAC_exactIP(self):
         """ IP4 MACIP exactMAC|exactIP ACL bridged traffic
         """
@@ -812,21 +847,22 @@ class TestMACIP_IP4(MethodHolder):
         """ MACIP replace ACL with IP4 traffic
         """
         self.run_traffic(self.OUI_MAC, self.SUBNET_IP,
-                         self.BRIDGED, self.IS_IP4, 9)
-
-        r = self.create_rules()
-        # replace acls #2, #3 with new
-        reply = self.vapi.macip_acl_add_replace(r[0], 0)
-        self.assertEqual(reply.retval, 0)
-        self.assertEqual(reply.acl_index, 0)
-
+                         self.BRIDGED, self.IS_IP4, 9, try_replace=True)
         self.run_traffic(self.EXACT_MAC, self.EXACT_IP,
-                         self.BRIDGED, self.IS_IP4, 9, True)
+                         self.BRIDGED, self.IS_IP4, 9, try_replace=True)
 
 
 class TestMACIP_IP6(MethodHolder):
     """MACIP with IP6 traffic"""
 
+    @classmethod
+    def setUpClass(cls):
+        super(TestMACIP_IP6, cls).setUpClass()
+
+    @classmethod
+    def tearDownClass(cls):
+        super(TestMACIP_IP6, cls).tearDownClass()
+
     def test_acl_bridged_ip6_exactMAC_exactIP(self):
         """ IP6 MACIP exactMAC|exactIP ACL bridged traffic
         """
@@ -957,21 +993,22 @@ class TestMACIP_IP6(MethodHolder):
         """ MACIP replace ACL with IP6 traffic
         """
         self.run_traffic(self.OUI_MAC, self.SUBNET_IP,
-                         self.BRIDGED, self.IS_IP6, 9)
-
-        r = self.create_rules()
-        # replace acls #2, #3 with new
-        reply = self.vapi.macip_acl_add_replace(r[0], 0)
-        self.assertEqual(reply.retval, 0)
-        self.assertEqual(reply.acl_index, 0)
-
+                         self.BRIDGED, self.IS_IP6, 9, try_replace=True)
         self.run_traffic(self.EXACT_MAC, self.EXACT_IP,
-                         self.BRIDGED, self.IS_IP6, 9, True)
+                         self.BRIDGED, self.IS_IP6, 9, try_replace=True)
 
 
 class TestMACIP(MethodHolder):
     """MACIP Tests"""
 
+    @classmethod
+    def setUpClass(cls):
+        super(TestMACIP, cls).setUpClass()
+
+    @classmethod
+    def tearDownClass(cls):
+        super(TestMACIP, cls).tearDownClass()
+
     def test_acl_1_2(self):
         """ MACIP ACL with 2 entries
         """
@@ -1087,8 +1124,8 @@ class TestMACIP(MethodHolder):
         self.apply_macip_rules(self.create_rules(acl_count=3,
                                                  rules_count=[3, 5, 4]))
 
-        intf.append(VppLoInterface(self, 0))
-        intf.append(VppLoInterface(self, 1))
+        intf.append(VppLoInterface(self))
+        intf.append(VppLoInterface(self))
 
         sw_if_index0 = intf[0].sw_if_index
         self.vapi.macip_acl_interface_add_del(sw_if_index0, 1)
@@ -1110,8 +1147,8 @@ class TestMACIP(MethodHolder):
         self.assertEqual(reply.acls[sw_if_index0], 4294967295)
         self.assertEqual(reply.acls[sw_if_index1], 0)
 
-        intf.append(VppLoInterface(self, 2))
-        intf.append(VppLoInterface(self, 3))
+        intf.append(VppLoInterface(self))
+        intf.append(VppLoInterface(self))
         sw_if_index2 = intf[2].sw_if_index
         sw_if_index3 = intf[3].sw_if_index
         self.vapi.macip_acl_interface_add_del(sw_if_index2, 1)
@@ -1122,6 +1159,12 @@ class TestMACIP(MethodHolder):
         self.assertEqual(reply.acls[sw_if_index1], 0)
         self.assertEqual(reply.acls[sw_if_index2], 1)
         self.assertEqual(reply.acls[sw_if_index3], 1)
+        self.logger.info("MACIP ACL on multiple interfaces:")
+        self.logger.info(self.vapi.ppcli("sh acl-plugin macip acl"))
+        self.logger.info(self.vapi.ppcli("sh acl-plugin macip acl index 1234"))
+        self.logger.info(self.vapi.ppcli("sh acl-plugin macip acl index 1"))
+        self.logger.info(self.vapi.ppcli("sh acl-plugin macip acl index 0"))
+        self.logger.info(self.vapi.ppcli("sh acl-plugin macip interface"))
 
         intf[2].remove_vpp_config()
         intf[1].remove_vpp_config()
@@ -1142,6 +1185,14 @@ class TestMACIP(MethodHolder):
 class TestACL_dot1q_bridged(MethodHolder):
     """ACL on dot1q bridged subinterfaces Tests"""
 
+    @classmethod
+    def setUpClass(cls):
+        super(TestACL_dot1q_bridged, cls).setUpClass()
+
+    @classmethod
+    def tearDownClass(cls):
+        super(TestACL_dot1q_bridged, cls).tearDownClass()
+
     def test_acl_bridged_ip4_subif_dot1q(self):
         """ IP4 ACL SubIf Dot1Q bridged traffic"""
         self.run_traffic(self.EXACT_MAC, self.EXACT_IP, self.BRIDGED,
@@ -1156,6 +1207,14 @@ class TestACL_dot1q_bridged(MethodHolder):
 class TestACL_dot1ad_bridged(MethodHolder):
     """ACL on dot1ad bridged subinterfaces Tests"""
 
+    @classmethod
+    def setUpClass(cls):
+        super(TestACL_dot1ad_bridged, cls).setUpClass()
+
+    @classmethod
+    def tearDownClass(cls):
+        super(TestACL_dot1ad_bridged, cls).tearDownClass()
+
     def test_acl_bridged_ip4_subif_dot1ad(self):
         """ IP4 ACL SubIf Dot1AD bridged traffic"""
         self.run_traffic(self.EXACT_MAC, self.EXACT_IP, self.BRIDGED,
@@ -1170,6 +1229,14 @@ class TestACL_dot1ad_bridged(MethodHolder):
 class TestACL_dot1q_routed(MethodHolder):
     """ACL on dot1q routed subinterfaces Tests"""
 
+    @classmethod
+    def setUpClass(cls):
+        super(TestACL_dot1q_routed, cls).setUpClass()
+
+    @classmethod
+    def tearDownClass(cls):
+        super(TestACL_dot1q_routed, cls).tearDownClass()
+
     def test_acl_routed_ip4_subif_dot1q(self):
         """ IP4 ACL SubIf Dot1Q routed traffic"""
         self.run_traffic(self.EXACT_MAC, self.EXACT_IP, self.ROUTED,
@@ -1196,6 +1263,14 @@ class TestACL_dot1q_routed(MethodHolder):
 class TestACL_dot1ad_routed(MethodHolder):
     """ACL on dot1ad routed subinterfaces Tests"""
 
+    @classmethod
+    def setUpClass(cls):
+        super(TestACL_dot1ad_routed, cls).setUpClass()
+
+    @classmethod
+    def tearDownClass(cls):
+        super(TestACL_dot1ad_routed, cls).tearDownClass()
+
     def test_acl_routed_ip6_subif_dot1ad(self):
         """ IP6 ACL SubIf Dot1AD routed traffic"""
         self.run_traffic(self.EXACT_MAC, self.EXACT_IP, self.ROUTED,