tests: refactor asf framework code

[vpp.git] / test / test_acl_plugin_l2l3.py

diff --git a/test/test_acl_plugin_l2l3.py b/test/test_acl_plugin_l2l3.py
index 26b562e..b272299 100644 (file)
--- a/test/test_acl_plugin_l2l3.py
+++ b/test/test_acl_plugin_l2l3.py
@@ -1,4 +1,4 @@
-#!/usr/bin/env python
+#!/usr/bin/env python3

 """ACL IRB Test Case HLD:
 
 **config**
 """ACL IRB Test Case HLD:
 
 **config**


@@ -23,24 +23,31 @@
 
 """
 
 
 """
 

+import copy

 import unittest
 import unittest

-from socket import inet_pton, AF_INET, AF_INET6
-from random import choice, shuffle
-from pprint import pprint
+from socket import AF_INET, AF_INET6
+from random import shuffle
+from ipaddress import ip_network
+from config import config

 
 

+import scapy.compat

 from scapy.packet import Raw
 from scapy.layers.l2 import Ether
 from scapy.layers.inet import IP, UDP, ICMP, TCP
 from scapy.packet import Raw
 from scapy.layers.l2 import Ether
 from scapy.layers.inet import IP, UDP, ICMP, TCP

-from scapy.layers.inet6 import IPv6, ICMPv6Unknown, ICMPv6EchoRequest
-from scapy.layers.inet6 import ICMPv6EchoReply, IPv6ExtHdrRouting
+from scapy.layers.inet6 import IPv6, ICMPv6Unknown
+from scapy.layers.inet6 import IPv6ExtHdrRouting

 from scapy.layers.inet6 import IPv6ExtHdrFragment
 
 from scapy.layers.inet6 import IPv6ExtHdrFragment
 

-from framework import VppTestCase, VppTestRunner
-import time
+from framework import VppTestCase
+from asfframework import VppTestRunner
+from vpp_l2 import L2_PORT_TYPE

 
 

+from vpp_acl import AclRule, VppAcl, VppAclInterface

 
 

-class TestIpIrb(VppTestCase):
-    """IRB Test Case"""
+
+@unittest.skipIf("acl" in config.excluded_plugins, "Exclude ACL plugin tests")
+class TestACLpluginL2L3(VppTestCase):
+    """TestACLpluginL2L3 Test Case"""

 
     @classmethod
     def setUpClass(cls):
 
     @classmethod
     def setUpClass(cls):


@@ -52,7 +59,7 @@ class TestIpIrb(VppTestCase):
         #. Loopback BVI interface has remote hosts, one half of hosts are
            behind pg0 second behind pg1.
         """
         #. Loopback BVI interface has remote hosts, one half of hosts are
            behind pg0 second behind pg1.
         """

-        super(TestIpIrb, cls).setUpClass()
+        super(TestACLpluginL2L3, cls).setUpClass()

 
         cls.pg_if_packet_sizes = [64, 512, 1518, 9018]  # packet sizes
         cls.bd_id = 10
 
         cls.pg_if_packet_sizes = [64, 512, 1518, 9018]  # packet sizes
         cls.bd_id = 10


@@ -70,11 +77,16 @@ class TestIpIrb(VppTestCase):
 
         # Create BD with MAC learning enabled and put interfaces to this BD
         cls.vapi.sw_interface_set_l2_bridge(
 
         # 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.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.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
+        )

 
         # Configure IPv4 addresses on loopback interface and routed interface
         cls.loop0.config_ip4()
 
         # Configure IPv4 addresses on loopback interface and routed interface
         cls.loop0.config_ip4()


@@ -92,44 +104,69 @@ class TestIpIrb(VppTestCase):
 
         cls.WITHOUT_EH = False
         cls.WITH_EH = True
 
         cls.WITHOUT_EH = False
         cls.WITH_EH = True

+        cls.STATELESS_ICMP = False
+        cls.STATEFUL_ICMP = True

 
         # Loopback BVI interface has remote hosts, one half of hosts are behind
         # pg0 second behind pg1
         half = cls.remote_hosts_count // 2
         cls.pg0.remote_hosts = cls.loop0.remote_hosts[:half]
         cls.pg1.remote_hosts = cls.loop0.remote_hosts[half:]
 
         # Loopback BVI interface has remote hosts, one half of hosts are behind
         # pg0 second behind pg1
         half = cls.remote_hosts_count // 2
         cls.pg0.remote_hosts = cls.loop0.remote_hosts[:half]
         cls.pg1.remote_hosts = cls.loop0.remote_hosts[half:]

+        reply = cls.vapi.papi.acl_stats_intf_counters_enable(enable=1)
+
+    @classmethod
+    def tearDownClass(cls):
+        reply = cls.vapi.papi.acl_stats_intf_counters_enable(enable=0)
+        super(TestACLpluginL2L3, cls).tearDownClass()

 
     def tearDown(self):
         """Run standard test teardown and log ``show l2patch``,
         ``show l2fib verbose``,``show bridge-domain <bd_id> detail``,
 
     def tearDown(self):
         """Run standard test teardown and log ``show l2patch``,
         ``show l2fib verbose``,``show bridge-domain <bd_id> detail``,

-        ``show ip arp``.
+        ``show ip neighbors``.

         """
         """

-        super(TestIpIrb, self).tearDown()
-        if not self.vpp_dead:
-            self.logger.info(self.vapi.cli("show l2patch"))
-            self.logger.info(self.vapi.cli("show classify tables"))
-            self.logger.info(self.vapi.cli("show vlib graph"))
-            self.logger.info(self.vapi.cli("show l2fib verbose"))
-            self.logger.info(self.vapi.cli("show bridge-domain %s detail" %
-                                           self.bd_id))
-            self.logger.info(self.vapi.cli("show ip arp"))
-            self.logger.info(self.vapi.cli("show ip6 neighbors"))
-            self.logger.info(self.vapi.cli("show acl-plugin sessions"))
-            self.logger.info(self.vapi.cli("show acl-plugin acl"))
-            self.logger.info(self.vapi.cli("show acl-plugin interface"))
-            self.logger.info(self.vapi.cli("show acl-plugin tables"))
-
-    def create_stream(self, src_ip_if, dst_ip_if, reverse, packet_sizes,
-                      is_ip6, expect_blocked, expect_established,
-                      add_extension_header):
+        super(TestACLpluginL2L3, self).tearDown()
+
+    def show_commands_at_teardown(self):
+        self.logger.info(self.vapi.cli("show l2patch"))
+        self.logger.info(self.vapi.cli("show classify tables"))
+        self.logger.info(self.vapi.cli("show l2fib verbose"))
+        self.logger.info(self.vapi.cli("show bridge-domain %s detail" % self.bd_id))
+        self.logger.info(self.vapi.cli("show ip neighbors"))
+        cmd = "show acl-plugin sessions verbose 1"
+        self.logger.info(self.vapi.cli(cmd))
+        self.logger.info(self.vapi.cli("show acl-plugin acl"))
+        self.logger.info(self.vapi.cli("show acl-plugin interface"))
+        self.logger.info(self.vapi.cli("show acl-plugin tables"))
+
+    def create_stream(
+        self,
+        src_ip_if,
+        dst_ip_if,
+        reverse,
+        packet_sizes,
+        is_ip6,
+        expect_blocked,
+        expect_established,
+        add_extension_header,
+        icmp_stateful=False,
+    ):

         pkts = []
         rules = []
         permit_rules = []
         permit_and_reflect_rules = []
         total_packet_count = 8
         for i in range(0, total_packet_count):
         pkts = []
         rules = []
         permit_rules = []
         permit_and_reflect_rules = []
         total_packet_count = 8
         for i in range(0, total_packet_count):

-            modulo = (i//2) % 2
-            can_reflect_this_packet = (modulo == 0)
+            modulo = (i // 2) % 2
+            icmp_type_delta = i % 2
+            icmp_code = i
+            is_udp_packet = modulo == 0
+            if is_udp_packet and icmp_stateful:
+                continue
+            is_reflectable_icmp = (
+                icmp_stateful and icmp_type_delta == 0 and not is_udp_packet
+            )
+            is_reflected_icmp = is_reflectable_icmp and expect_established
+            can_reflect_this_packet = is_udp_packet or is_reflectable_icmp

             is_permit = i % 2
             remote_dst_index = i % len(dst_ip_if.remote_hosts)
             remote_dst_host = dst_ip_if.remote_hosts[remote_dst_index]
             is_permit = i % 2
             remote_dst_index = i % len(dst_ip_if.remote_hosts)
             remote_dst_host = dst_ip_if.remote_hosts[remote_dst_index]


@@ -138,9 +175,9 @@ class TestIpIrb(VppTestCase):
                 payload = self.info_to_payload(info)
             else:
                 to_be_blocked = False
                 payload = self.info_to_payload(info)
             else:
                 to_be_blocked = False

-                if (expect_blocked and not expect_established):
+                if expect_blocked and not expect_established:

                     to_be_blocked = True
                     to_be_blocked = True

-                if (not can_reflect_this_packet):
+                if not can_reflect_this_packet:

                     to_be_blocked = True
                 if to_be_blocked:
                     payload = "to be blocked"
                     to_be_blocked = True
                 if to_be_blocked:
                     payload = "to be blocked"


@@ -148,7 +185,7 @@ class TestIpIrb(VppTestCase):
                     info = self.create_packet_info(src_ip_if, dst_ip_if)
                     payload = self.info_to_payload(info)
             if reverse:
                     info = self.create_packet_info(src_ip_if, dst_ip_if)
                     payload = self.info_to_payload(info)
             if reverse:

-                dst_mac = 'de:ad:00:00:00:00'
+                dst_mac = "de:ad:00:00:00:00"

                 src_mac = remote_dst_host._mac
                 dst_ip6 = src_ip_if.remote_ip6
                 src_ip6 = remote_dst_host.ip6
                 src_mac = remote_dst_host._mac
                 dst_ip6 = src_ip_if.remote_ip6
                 src_ip6 = remote_dst_host.ip6


@@ -165,52 +202,66 @@ class TestIpIrb(VppTestCase):
                 dst_ip4 = remote_dst_host.ip4
                 src_l4 = 1234 + i
                 dst_l4 = 4321 + i
                 dst_ip4 = remote_dst_host.ip4
                 src_l4 = 1234 + i
                 dst_l4 = 4321 + i

+            if is_reflected_icmp:
+                icmp_type_delta = 1

 
             # default ULP should be something we do not use in tests
             ulp_l4 = TCP(sport=src_l4, dport=dst_l4)
             # potentially a chain of protocols leading to ULP
             ulp = ulp_l4
 
 
             # default ULP should be something we do not use in tests
             ulp_l4 = TCP(sport=src_l4, dport=dst_l4)
             # potentially a chain of protocols leading to ULP
             ulp = ulp_l4
 

-            if can_reflect_this_packet:
+            if is_udp_packet:

                 if is_ip6:
                     ulp_l4 = UDP(sport=src_l4, dport=dst_l4)
                     if add_extension_header:
                         # prepend some extension headers
                 if is_ip6:
                     ulp_l4 = UDP(sport=src_l4, dport=dst_l4)
                     if add_extension_header:
                         # prepend some extension headers

-                        ulp = (IPv6ExtHdrRouting() / IPv6ExtHdrRouting() /
-                               IPv6ExtHdrFragment(offset=0, m=1) / ulp_l4)
+                        ulp = (
+                            IPv6ExtHdrRouting()
+                            / IPv6ExtHdrRouting()
+                            / IPv6ExtHdrFragment(offset=0, m=1)
+                            / ulp_l4
+                        )

                         # uncomment below to test invalid ones
                         # ulp = IPv6ExtHdrRouting(len = 200) / ulp_l4
                     else:
                         ulp = ulp_l4
                         # uncomment below to test invalid ones
                         # ulp = IPv6ExtHdrRouting(len = 200) / ulp_l4
                     else:
                         ulp = ulp_l4

-                    p = (Ether(dst=dst_mac, src=src_mac) /
-                         IPv6(src=src_ip6, dst=dst_ip6) /
-                         ulp /
-                         Raw(payload))
+                    p = (
+                        Ether(dst=dst_mac, src=src_mac)
+                        / IPv6(src=src_ip6, dst=dst_ip6)
+                        / ulp
+                        / Raw(payload)
+                    )

                 else:
                     ulp_l4 = UDP(sport=src_l4, dport=dst_l4)
                     # IPv4 does not allow extension headers,
                     # but we rather make it a first fragment
                     flags = 1 if add_extension_header else 0
                     ulp = ulp_l4
                 else:
                     ulp_l4 = UDP(sport=src_l4, dport=dst_l4)
                     # IPv4 does not allow extension headers,
                     # but we rather make it a first fragment
                     flags = 1 if add_extension_header else 0
                     ulp = ulp_l4

-                    p = (Ether(dst=dst_mac, src=src_mac) /
-                         IP(src=src_ip4, dst=dst_ip4, frag=0, flags=flags) /
-                         ulp /
-                         Raw(payload))
+                    p = (
+                        Ether(dst=dst_mac, src=src_mac)
+                        / IP(src=src_ip4, dst=dst_ip4, frag=0, flags=flags)
+                        / ulp
+                        / Raw(payload)
+                    )

             elif modulo == 1:
                 if is_ip6:
             elif modulo == 1:
                 if is_ip6:

-                    ulp_l4 = ICMPv6Unknown(type=128 + (i % 2), code=i % 2)
+                    ulp_l4 = ICMPv6Unknown(type=128 + icmp_type_delta, code=icmp_code)

                     ulp = ulp_l4
                     ulp = ulp_l4

-                    p = (Ether(dst=dst_mac, src=src_mac) /
-                         IPv6(src=src_ip6, dst=dst_ip6) /
-                         ulp /
-                         Raw(payload))
+                    p = (
+                        Ether(dst=dst_mac, src=src_mac)
+                        / IPv6(src=src_ip6, dst=dst_ip6)
+                        / ulp
+                        / Raw(payload)
+                    )

                 else:
                 else:

-                    ulp_l4 = ICMP(type=8 + (i % 2), code=i % 2)
+                    ulp_l4 = ICMP(type=8 - 8 * icmp_type_delta, code=icmp_code)

                     ulp = ulp_l4
                     ulp = ulp_l4

-                    p = (Ether(dst=dst_mac, src=src_mac) /
-                         IP(src=src_ip4, dst=dst_ip4) /
-                         ulp /
-                         Raw(payload))
+                    p = (
+                        Ether(dst=dst_mac, src=src_mac)
+                        / IP(src=src_ip4, dst=dst_ip4)
+                        / ulp
+                        / Raw(payload)
+                    )

 
             if i % 2 == 1:
                 info.data = p.copy()
 
             if i % 2 == 1:
                 info.data = p.copy()


@@ -233,41 +284,41 @@ class TestIpIrb(VppTestCase):
                     rule_l4_sport = p[ICMPv6Unknown].type
                     rule_l4_dport = p[ICMPv6Unknown].code
             if p.haslayer(IPv6):
                     rule_l4_sport = p[ICMPv6Unknown].type
                     rule_l4_dport = p[ICMPv6Unknown].code
             if p.haslayer(IPv6):

-                rule_l4_proto = ulp_l4.overload_fields[IPv6]['nh']
+                rule_l4_proto = ulp_l4.overload_fields[IPv6]["nh"]

             else:
                 rule_l4_proto = p[IP].proto
 
             else:
                 rule_l4_proto = p[IP].proto
 

-            new_rule = {
-                        'is_permit': is_permit,
-                        'is_ipv6': p.haslayer(IPv6),
-                        'src_ip_addr': inet_pton(rule_family,
-                                                 p[rule_l3_layer].src),
-                        'src_ip_prefix_len': rule_prefix_len,
-                        'dst_ip_addr': inet_pton(rule_family,
-                                                 p[rule_l3_layer].dst),
-                        'dst_ip_prefix_len': rule_prefix_len,
-                        'srcport_or_icmptype_first': rule_l4_sport,
-                        'srcport_or_icmptype_last': rule_l4_sport,
-                        'dstport_or_icmpcode_first': rule_l4_dport,
-                        'dstport_or_icmpcode_last': rule_l4_dport,
-                        'proto': rule_l4_proto,
-                       }
+            new_rule = AclRule(
+                is_permit=is_permit,
+                proto=rule_l4_proto,
+                src_prefix=ip_network((p[rule_l3_layer].src, rule_prefix_len)),
+                dst_prefix=ip_network((p[rule_l3_layer].dst, rule_prefix_len)),
+                sport_from=rule_l4_sport,
+                sport_to=rule_l4_sport,
+                dport_from=rule_l4_dport,
+                dport_to=rule_l4_dport,
+            )
+

             rules.append(new_rule)
             rules.append(new_rule)

-            new_rule_permit = new_rule.copy()
-            new_rule_permit['is_permit'] = 1
+            new_rule_permit = copy.copy(new_rule)
+            new_rule_permit.is_permit = 1

             permit_rules.append(new_rule_permit)
 
             permit_rules.append(new_rule_permit)
 

-            new_rule_permit_and_reflect = new_rule.copy()
+            new_rule_permit_and_reflect = copy.copy(new_rule)

             if can_reflect_this_packet:
             if can_reflect_this_packet:

-                new_rule_permit_and_reflect['is_permit'] = 2
+                new_rule_permit_and_reflect.is_permit = 2

             else:
             else:

-                new_rule_permit_and_reflect['is_permit'] = is_permit
+                new_rule_permit_and_reflect.is_permit = is_permit
+

             permit_and_reflect_rules.append(new_rule_permit_and_reflect)
             permit_and_reflect_rules.append(new_rule_permit_and_reflect)

+            self.logger.info("create_stream pkt#%d: %s" % (i, payload))

 
 

-        return {'stream': pkts,
-                'rules': rules,
-                'permit_rules': permit_rules,
-                'permit_and_reflect_rules': permit_and_reflect_rules}
+        return {
+            "stream": pkts,
+            "rules": rules,
+            "permit_rules": permit_rules,
+            "permit_and_reflect_rules": permit_and_reflect_rules,
+        }

 
     def verify_capture(self, dst_ip_if, src_ip_if, capture, reverse):
         last_info = dict()
 
     def verify_capture(self, dst_ip_if, src_ip_if, capture, reverse):
         last_info = dict()


@@ -275,7 +326,6 @@ class TestIpIrb(VppTestCase):
             last_info[i.sw_if_index] = None
 
         dst_ip_sw_if_index = dst_ip_if.sw_if_index
             last_info[i.sw_if_index] = None
 
         dst_ip_sw_if_index = dst_ip_if.sw_if_index

-        return

 
         for packet in capture:
             l3 = IP if packet.haslayer(IP) else IPv6
 
         for packet in capture:
             l3 = IP if packet.haslayer(IP) else IPv6


@@ -291,21 +341,27 @@ class TestIpIrb(VppTestCase):
             # Scapy IPv6 stuff is too smart for its own good.
             # So we do this and coerce the ICMP into unknown type
             if packet.haslayer(UDP):
             # Scapy IPv6 stuff is too smart for its own good.
             # So we do this and coerce the ICMP into unknown type
             if packet.haslayer(UDP):

-                data = str(packet[UDP][Raw])
+                data = scapy.compat.raw(packet[UDP][Raw])

             else:
                 if l3 == IP:
             else:
                 if l3 == IP:

-                    data = str(ICMP(str(packet[l3].payload))[Raw])
+                    data = scapy.compat.raw(
+                        ICMP(scapy.compat.raw(packet[l3].payload))[Raw]
+                    )

                 else:
                 else:

-                    data = str(ICMPv6Unknown(str(packet[l3].payload)).msgbody)
+                    data = scapy.compat.raw(
+                        ICMPv6Unknown(scapy.compat.raw(packet[l3].payload)).msgbody
+                    )

             udp_or_icmp = packet[l3].payload
             udp_or_icmp = packet[l3].payload

-            payload_info = self.payload_to_info(data)
+            data_obj = Raw(data)
+            # FIXME: make framework believe we are on object
+            payload_info = self.payload_to_info(data_obj)

             packet_index = payload_info.index
 
             self.assertEqual(payload_info.dst, dst_ip_sw_if_index)
 
             next_info = self.get_next_packet_info_for_interface2(
             packet_index = payload_info.index
 
             self.assertEqual(payload_info.dst, dst_ip_sw_if_index)
 
             next_info = self.get_next_packet_info_for_interface2(

-                payload_info.src, dst_ip_sw_if_index,
-                last_info[payload_info.src])
+                payload_info.src, dst_ip_sw_if_index, last_info[payload_info.src]
+            )

             last_info[payload_info.src] = next_info
             self.assertTrue(next_info is not None)
             self.assertEqual(packet_index, next_info.index)
             last_info[payload_info.src] = next_info
             self.assertTrue(next_info is not None)
             self.assertEqual(packet_index, next_info.index)


@@ -325,204 +381,266 @@ class TestIpIrb(VppTestCase):
                 if l4 == UDP:
                     self.assertEqual(udp_or_icmp.sport, saved_packet[l4].sport)
                     self.assertEqual(udp_or_icmp.dport, saved_packet[l4].dport)
                 if l4 == UDP:
                     self.assertEqual(udp_or_icmp.sport, saved_packet[l4].sport)
                     self.assertEqual(udp_or_icmp.dport, saved_packet[l4].dport)

-            else:
-                print("Saved packet is none")

             # self.assertEqual(ip.dst, host.ip4)
 
             # UDP:
 
             # self.assertEqual(ip.dst, host.ip4)
 
             # UDP:
 

-    def applied_acl_shuffle(self, sw_if_index):
-        # first collect what ACLs are applied and what they look like
-        r = self.vapi.acl_interface_list_dump(sw_if_index=sw_if_index)
-        orig_applied_acls = r[0]
-
-        # we will collect these just to save and generate additional rulesets
-        orig_acls = []
-        for acl_num in orig_applied_acls.acls:
-            rr = self.vapi.acl_dump(acl_num)
-            orig_acls.append(rr[0])
+    def applied_acl_shuffle(self, acl_if):
+        saved_n_input = acl_if.n_input
+        # TOTO: maybe copy each one??
+        saved_acls = acl_if.acls

 
         # now create a list of all the rules in all ACLs
         all_rules = []
 
         # now create a list of all the rules in all ACLs
         all_rules = []

-        for old_acl in orig_acls:
-            for rule in old_acl.r:
-                all_rules.append(dict(rule._asdict()))
+        for old_acl in saved_acls:
+            for rule in old_acl.rules:
+                all_rules.append(rule)

 
         # Add a few ACLs made from shuffled rules
         shuffle(all_rules)
 
         # Add a few ACLs made from shuffled rules
         shuffle(all_rules)

-        reply = self.vapi.acl_add_replace(acl_index=4294967295,
-                                          r=all_rules[::2],
-                                          tag="shuffle 1. acl")
-        shuffle_acl_1 = reply.acl_index
+        acl1 = VppAcl(self, rules=all_rules[::2], tag="shuffle 1. acl")
+        acl1.add_vpp_config()
+

         shuffle(all_rules)
         shuffle(all_rules)

-        reply = self.vapi.acl_add_replace(acl_index=4294967295,
-                                          r=all_rules[::3],
-                                          tag="shuffle 2. acl")
-        shuffle_acl_2 = reply.acl_index
+        acl2 = VppAcl(self, rules=all_rules[::3], tag="shuffle 2. acl")
+        acl2.add_vpp_config()
+

         shuffle(all_rules)
         shuffle(all_rules)

-        reply = self.vapi.acl_add_replace(acl_index=4294967295,
-                                          r=all_rules[::2],
-                                          tag="shuffle 3. acl")
-        shuffle_acl_3 = reply.acl_index
+        acl3 = VppAcl(self, rules=all_rules[::2], tag="shuffle 3. acl")
+        acl3.add_vpp_config()

 
         # apply the shuffle ACLs in front
 
         # apply the shuffle ACLs in front

-        input_acls = [shuffle_acl_1, shuffle_acl_2]
-        output_acls = [shuffle_acl_1, shuffle_acl_2]
+        input_acls = [acl1, acl2]
+        output_acls = [acl1, acl2]

 
         # add the currently applied ACLs
 
         # add the currently applied ACLs

-        n_input = orig_applied_acls.n_input
-        input_acls.extend(orig_applied_acls.acls[:n_input])
-        output_acls.extend(orig_applied_acls.acls[n_input:])
+        n_input = acl_if.n_input
+        input_acls.extend(saved_acls[:n_input])
+        output_acls.extend(saved_acls[n_input:])

 
         # and the trailing shuffle ACL(s)
 
         # and the trailing shuffle ACL(s)

-        input_acls.extend([shuffle_acl_3])
-        output_acls.extend([shuffle_acl_3])
+        input_acls.extend([acl3])
+        output_acls.extend([acl3])

 
         # set the interface ACL list to the result
 
         # set the interface ACL list to the result

-        self.vapi.acl_interface_set_acl_list(sw_if_index=sw_if_index,
-                                             n_input=len(input_acls),
-                                             acls=input_acls + output_acls)
+        acl_if.n_input = len(input_acls)
+        acl_if.acls = input_acls + output_acls
+        acl_if.add_vpp_config()
+

         # change the ACLs a few times
         for i in range(1, 10):
             shuffle(all_rules)
         # change the ACLs a few times
         for i in range(1, 10):
             shuffle(all_rules)

-            reply = self.vapi.acl_add_replace(acl_index=shuffle_acl_1,
-                                              r=all_rules[::1+(i % 2)],
-                                              tag="shuffle 1. acl")
+            acl1.modify_vpp_config(all_rules[:: 1 + (i % 2)])
+

             shuffle(all_rules)
             shuffle(all_rules)

-            reply = self.vapi.acl_add_replace(acl_index=shuffle_acl_2,
-                                              r=all_rules[::1+(i % 3)],
-                                              tag="shuffle 2. acl")
+            acl2.modify_vpp_config(all_rules[:: 1 + (i % 3)])
+

             shuffle(all_rules)
             shuffle(all_rules)

-            reply = self.vapi.acl_add_replace(acl_index=shuffle_acl_2,
-                                              r=all_rules[::1+(i % 5)],
-                                              tag="shuffle 3. acl")
+            acl3.modify_vpp_config(all_rules[:: 1 + (i % 5)])

 
         # restore to how it was before and clean up
 
         # restore to how it was before and clean up

-        self.vapi.acl_interface_set_acl_list(sw_if_index=sw_if_index,
-                                             n_input=orig_applied_acls.n_input,
-                                             acls=orig_applied_acls.acls)
-        reply = self.vapi.acl_del(acl_index=shuffle_acl_1)
-        reply = self.vapi.acl_del(acl_index=shuffle_acl_2)
-        reply = self.vapi.acl_del(acl_index=shuffle_acl_3)
-
-    def create_acls_for_a_stream(self, stream_dict,
-                                 test_l2_action, is_reflect):
-        r = stream_dict['rules']
-        r_permit = stream_dict['permit_rules']
-        r_permit_reflect = stream_dict['permit_and_reflect_rules']
+        acl_if.n_input = saved_n_input
+        acl_if.acls = saved_acls
+        acl_if.add_vpp_config()
+
+        acl1.remove_vpp_config()
+        acl2.remove_vpp_config()
+        acl3.remove_vpp_config()
+
+    def create_acls_for_a_stream(self, stream_dict, test_l2_action, is_reflect):
+        r = stream_dict["rules"]
+        r_permit = stream_dict["permit_rules"]
+        r_permit_reflect = stream_dict["permit_and_reflect_rules"]

         r_action = r_permit_reflect if is_reflect else r
         r_action = r_permit_reflect if is_reflect else r

-        reply = self.vapi.acl_add_replace(acl_index=4294967295, r=r_action,
-                                          tag="act. acl")
-        action_acl_index = reply.acl_index
-        reply = self.vapi.acl_add_replace(acl_index=4294967295, r=r_permit,
-                                          tag="perm. acl")
-        permit_acl_index = reply.acl_index
-        return {'L2': action_acl_index if test_l2_action else permit_acl_index,
-                'L3': permit_acl_index if test_l2_action else action_acl_index,
-                'permit': permit_acl_index, 'action': action_acl_index}
-
-    def apply_acl_ip46_x_to_y(self, bridged_to_routed, test_l2_deny,
-                              is_ip6, is_reflect, add_eh):
-        """ Apply the ACLs
-        """
+        action_acl = VppAcl(self, rules=r_action, tag="act. acl")
+        action_acl.add_vpp_config()
+        permit_acl = VppAcl(self, rules=r_permit, tag="perm. acl")
+        permit_acl.add_vpp_config()
+
+        return {
+            "L2": action_acl if test_l2_action else permit_acl,
+            "L3": permit_acl if test_l2_action else action_acl,
+            "permit": permit_acl,
+            "action": action_acl,
+        }
+
+    def apply_acl_ip46_x_to_y(
+        self, bridged_to_routed, test_l2_deny, is_ip6, is_reflect, add_eh
+    ):
+        """Apply the ACLs"""

         self.reset_packet_infos()
         stream_dict = self.create_stream(
         self.reset_packet_infos()
         stream_dict = self.create_stream(

-                                         self.pg2, self.loop0,
-                                         bridged_to_routed,
-                                         self.pg_if_packet_sizes, is_ip6,
-                                         not is_reflect, False, add_eh)
-        stream = stream_dict['stream']
-        acl_idx = self.create_acls_for_a_stream(stream_dict, test_l2_deny,
-                                                is_reflect)
+            self.pg2,
+            self.loop0,
+            bridged_to_routed,
+            self.pg_if_packet_sizes,
+            is_ip6,
+            not is_reflect,
+            False,
+            add_eh,
+        )
+        stream = stream_dict["stream"]
+        acl_idx = self.create_acls_for_a_stream(stream_dict, test_l2_deny, is_reflect)

         n_input_l3 = 0 if bridged_to_routed else 1
         n_input_l2 = 1 if bridged_to_routed else 0
         n_input_l3 = 0 if bridged_to_routed else 1
         n_input_l2 = 1 if bridged_to_routed else 0

-        self.vapi.acl_interface_set_acl_list(sw_if_index=self.pg2.sw_if_index,
-                                             n_input=n_input_l3,
-                                             acls=[acl_idx['L3']])
-        self.vapi.acl_interface_set_acl_list(sw_if_index=self.pg0.sw_if_index,
-                                             n_input=n_input_l2,
-                                             acls=[acl_idx['L2']])
-        self.vapi.acl_interface_set_acl_list(sw_if_index=self.pg1.sw_if_index,
-                                             n_input=n_input_l2,
-                                             acls=[acl_idx['L2']])
-        self.applied_acl_shuffle(self.pg0.sw_if_index)
-        self.applied_acl_shuffle(self.pg2.sw_if_index)
-
-    def apply_acl_ip46_both_directions_reflect(self,
-                                               primary_is_bridged_to_routed,
-                                               reflect_on_l2, is_ip6, add_eh):
+
+        acl_if_pg2 = VppAclInterface(
+            self,
+            sw_if_index=self.pg2.sw_if_index,
+            n_input=n_input_l3,
+            acls=[acl_idx["L3"]],
+        )
+        acl_if_pg2.add_vpp_config()
+
+        acl_if_pg0 = VppAclInterface(
+            self,
+            sw_if_index=self.pg0.sw_if_index,
+            n_input=n_input_l2,
+            acls=[acl_idx["L2"]],
+        )
+        acl_if_pg0.add_vpp_config()
+
+        acl_if_pg1 = VppAclInterface(
+            self,
+            sw_if_index=self.pg1.sw_if_index,
+            n_input=n_input_l2,
+            acls=[acl_idx["L2"]],
+        )
+        acl_if_pg1.add_vpp_config()
+
+        self.applied_acl_shuffle(acl_if_pg0)
+        self.applied_acl_shuffle(acl_if_pg1)
+        return {"L2": acl_idx["L2"], "L3": acl_idx["L3"]}
+
+    def apply_acl_ip46_both_directions_reflect(
+        self, primary_is_bridged_to_routed, reflect_on_l2, is_ip6, add_eh, stateful_icmp
+    ):

         primary_is_routed_to_bridged = not primary_is_bridged_to_routed
         self.reset_packet_infos()
         primary_is_routed_to_bridged = not primary_is_bridged_to_routed
         self.reset_packet_infos()

-        stream_dict_fwd = self.create_stream(self.pg2, self.loop0,
-                                             primary_is_bridged_to_routed,
-                                             self.pg_if_packet_sizes, is_ip6,
-                                             False, False, add_eh)
-        acl_idx_fwd = self.create_acls_for_a_stream(stream_dict_fwd,
-                                                    reflect_on_l2, True)
-
-        stream_dict_rev = self.create_stream(self.pg2, self.loop0,
-                                             not primary_is_bridged_to_routed,
-                                             self.pg_if_packet_sizes, is_ip6,
-                                             True, True, add_eh)
+        stream_dict_fwd = self.create_stream(
+            self.pg2,
+            self.loop0,
+            primary_is_bridged_to_routed,
+            self.pg_if_packet_sizes,
+            is_ip6,
+            False,
+            False,
+            add_eh,
+            stateful_icmp,
+        )
+        acl_idx_fwd = self.create_acls_for_a_stream(
+            stream_dict_fwd, reflect_on_l2, True
+        )
+
+        stream_dict_rev = self.create_stream(
+            self.pg2,
+            self.loop0,
+            not primary_is_bridged_to_routed,
+            self.pg_if_packet_sizes,
+            is_ip6,
+            True,
+            True,
+            add_eh,
+            stateful_icmp,
+        )

         # We want the primary action to be "deny" rather than reflect
         # We want the primary action to be "deny" rather than reflect

-        acl_idx_rev = self.create_acls_for_a_stream(stream_dict_rev,
-                                                    reflect_on_l2, False)
+        acl_idx_rev = self.create_acls_for_a_stream(
+            stream_dict_rev, reflect_on_l2, False
+        )

 
         if primary_is_bridged_to_routed:
 
         if primary_is_bridged_to_routed:

-            inbound_l2_acl = acl_idx_fwd['L2']
+            inbound_l2_acl = acl_idx_fwd["L2"]

         else:
         else:

-            inbound_l2_acl = acl_idx_rev['L2']
+            inbound_l2_acl = acl_idx_rev["L2"]

 
         if primary_is_routed_to_bridged:
 
         if primary_is_routed_to_bridged:

-            outbound_l2_acl = acl_idx_fwd['L2']
+            outbound_l2_acl = acl_idx_fwd["L2"]

         else:
         else:

-            outbound_l2_acl = acl_idx_rev['L2']
+            outbound_l2_acl = acl_idx_rev["L2"]

 
         if primary_is_routed_to_bridged:
 
         if primary_is_routed_to_bridged:

-            inbound_l3_acl = acl_idx_fwd['L3']
+            inbound_l3_acl = acl_idx_fwd["L3"]

         else:
         else:

-            inbound_l3_acl = acl_idx_rev['L3']
+            inbound_l3_acl = acl_idx_rev["L3"]

 
         if primary_is_bridged_to_routed:
 
         if primary_is_bridged_to_routed:

-            outbound_l3_acl = acl_idx_fwd['L3']
+            outbound_l3_acl = acl_idx_fwd["L3"]

         else:
         else:

-            outbound_l3_acl = acl_idx_rev['L3']
-
-        self.vapi.acl_interface_set_acl_list(sw_if_index=self.pg2.sw_if_index,
-                                             n_input=1,
-                                             acls=[inbound_l3_acl,
-                                                   outbound_l3_acl])
-        self.vapi.acl_interface_set_acl_list(sw_if_index=self.pg0.sw_if_index,
-                                             n_input=1,
-                                             acls=[inbound_l2_acl,
-                                                   outbound_l2_acl])
-        self.vapi.acl_interface_set_acl_list(sw_if_index=self.pg1.sw_if_index,
-                                             n_input=1,
-                                             acls=[inbound_l2_acl,
-                                                   outbound_l2_acl])
-        self.applied_acl_shuffle(self.pg0.sw_if_index)
-        self.applied_acl_shuffle(self.pg2.sw_if_index)
-
-    def apply_acl_ip46_routed_to_bridged(self, test_l2_deny, is_ip6,
-                                         is_reflect, add_eh):
-        self.apply_acl_ip46_x_to_y(False, test_l2_deny, is_ip6,
-                                   is_reflect, add_eh)
-
-    def apply_acl_ip46_bridged_to_routed(self, test_l2_deny, is_ip6,
-                                         is_reflect, add_eh):
-        self.apply_acl_ip46_x_to_y(True, test_l2_deny, is_ip6,
-                                   is_reflect, add_eh)
-
-    def run_traffic_ip46_x_to_y(self, bridged_to_routed,
-                                test_l2_deny, is_ip6,
-                                is_reflect, is_established, add_eh):
+            outbound_l3_acl = acl_idx_rev["L3"]
+
+        acl_if_pg2 = VppAclInterface(
+            self,
+            sw_if_index=self.pg2.sw_if_index,
+            n_input=1,
+            acls=[inbound_l3_acl, outbound_l3_acl],
+        )
+        acl_if_pg2.add_vpp_config()
+
+        acl_if_pg0 = VppAclInterface(
+            self,
+            sw_if_index=self.pg0.sw_if_index,
+            n_input=1,
+            acls=[inbound_l2_acl, outbound_l2_acl],
+        )
+        acl_if_pg0.add_vpp_config()
+
+        acl_if_pg1 = VppAclInterface(
+            self,
+            sw_if_index=self.pg1.sw_if_index,
+            n_input=1,
+            acls=[inbound_l2_acl, outbound_l2_acl],
+        )
+        acl_if_pg1.add_vpp_config()
+
+        self.applied_acl_shuffle(acl_if_pg0)
+        self.applied_acl_shuffle(acl_if_pg2)
+
+    def apply_acl_ip46_routed_to_bridged(
+        self, test_l2_deny, is_ip6, is_reflect, add_eh
+    ):
+        return self.apply_acl_ip46_x_to_y(
+            False, test_l2_deny, is_ip6, is_reflect, add_eh
+        )
+
+    def apply_acl_ip46_bridged_to_routed(
+        self, test_l2_deny, is_ip6, is_reflect, add_eh
+    ):
+        return self.apply_acl_ip46_x_to_y(
+            True, test_l2_deny, is_ip6, is_reflect, add_eh
+        )
+
+    def verify_acl_packet_count(self, acl_idx, packet_count):
+        matches = self.statistics.get_counter("/acl/%d/matches" % acl_idx)
+        self.logger.info("stat seg for ACL %d: %s" % (acl_idx, repr(matches)))
+        total_count = 0
+        for m in matches:
+            for p in m:
+                total_count = total_count + p["packets"]
+        self.assertEqual(total_count, packet_count)
+
+    def run_traffic_ip46_x_to_y(
+        self,
+        bridged_to_routed,
+        test_l2_deny,
+        is_ip6,
+        is_reflect,
+        is_established,
+        add_eh,
+        stateful_icmp=False,
+    ):

         self.reset_packet_infos()
         self.reset_packet_infos()

-        stream_dict = self.create_stream(self.pg2, self.loop0,
-                                         bridged_to_routed,
-                                         self.pg_if_packet_sizes, is_ip6,
-                                         not is_reflect, is_established,
-                                         add_eh)
-        stream = stream_dict['stream']
+        stream_dict = self.create_stream(
+            self.pg2,
+            self.loop0,
+            bridged_to_routed,
+            self.pg_if_packet_sizes,
+            is_ip6,
+            not is_reflect,
+            is_established,
+            add_eh,
+            stateful_icmp,
+        )
+        stream = stream_dict["stream"]

 
         tx_if = self.pg0 if bridged_to_routed else self.pg2
         rx_if = self.pg2 if bridged_to_routed else self.pg0
 
         tx_if = self.pg0 if bridged_to_routed else self.pg2
         rx_if = self.pg2 if bridged_to_routed else self.pg0


@@ -533,51 +651,92 @@ class TestIpIrb(VppTestCase):
         packet_count = self.get_packet_count_for_if_idx(self.loop0.sw_if_index)
         rcvd1 = rx_if.get_capture(packet_count)
         self.verify_capture(self.loop0, self.pg2, rcvd1, bridged_to_routed)
         packet_count = self.get_packet_count_for_if_idx(self.loop0.sw_if_index)
         rcvd1 = rx_if.get_capture(packet_count)
         self.verify_capture(self.loop0, self.pg2, rcvd1, bridged_to_routed)

-
-    def run_traffic_ip46_routed_to_bridged(self, test_l2_deny, is_ip6,
-                                           is_reflect, is_established, add_eh):
-        self.run_traffic_ip46_x_to_y(False, test_l2_deny, is_ip6,
-                                     is_reflect, is_established, add_eh)
-
-    def run_traffic_ip46_bridged_to_routed(self, test_l2_deny, is_ip6,
-                                           is_reflect, is_established, add_eh):
-        self.run_traffic_ip46_x_to_y(True, test_l2_deny, is_ip6,
-                                     is_reflect, is_established, add_eh)
-
-    def run_test_ip46_routed_to_bridged(self, test_l2_deny,
-                                        is_ip6, is_reflect, add_eh):
-        self.apply_acl_ip46_routed_to_bridged(test_l2_deny,
-                                              is_ip6, is_reflect, add_eh)
-        self.run_traffic_ip46_routed_to_bridged(test_l2_deny, is_ip6,
-                                                is_reflect, False, add_eh)
-
-    def run_test_ip46_bridged_to_routed(self, test_l2_deny,
-                                        is_ip6, is_reflect, add_eh):
-        self.apply_acl_ip46_bridged_to_routed(test_l2_deny,
-                                              is_ip6, is_reflect, add_eh)
-        self.run_traffic_ip46_bridged_to_routed(test_l2_deny, is_ip6,
-                                                is_reflect, False, add_eh)
-
-    def run_test_ip46_routed_to_bridged_and_back(self, test_l2_action,
-                                                 is_ip6, add_eh):
-        self.apply_acl_ip46_both_directions_reflect(False, test_l2_action,
-                                                    is_ip6, add_eh)
-        self.run_traffic_ip46_routed_to_bridged(test_l2_action, is_ip6,
-                                                True, False, add_eh)
-        self.run_traffic_ip46_bridged_to_routed(test_l2_action, is_ip6,
-                                                False, True, add_eh)
-
-    def run_test_ip46_bridged_to_routed_and_back(self, test_l2_action,
-                                                 is_ip6, add_eh):
-        self.apply_acl_ip46_both_directions_reflect(True, test_l2_action,
-                                                    is_ip6, add_eh)
-        self.run_traffic_ip46_bridged_to_routed(test_l2_action, is_ip6,
-                                                True, False, add_eh)
-        self.run_traffic_ip46_routed_to_bridged(test_l2_action, is_ip6,
-                                                False, True, add_eh)
+        return len(stream)
+
+    def run_traffic_ip46_routed_to_bridged(
+        self,
+        test_l2_deny,
+        is_ip6,
+        is_reflect,
+        is_established,
+        add_eh,
+        stateful_icmp=False,
+    ):
+        return self.run_traffic_ip46_x_to_y(
+            False,
+            test_l2_deny,
+            is_ip6,
+            is_reflect,
+            is_established,
+            add_eh,
+            stateful_icmp,
+        )
+
+    def run_traffic_ip46_bridged_to_routed(
+        self,
+        test_l2_deny,
+        is_ip6,
+        is_reflect,
+        is_established,
+        add_eh,
+        stateful_icmp=False,
+    ):
+        return self.run_traffic_ip46_x_to_y(
+            True,
+            test_l2_deny,
+            is_ip6,
+            is_reflect,
+            is_established,
+            add_eh,
+            stateful_icmp,
+        )
+
+    def run_test_ip46_routed_to_bridged(self, test_l2_deny, is_ip6, is_reflect, add_eh):
+        acls = self.apply_acl_ip46_routed_to_bridged(
+            test_l2_deny, is_ip6, is_reflect, add_eh
+        )
+        pkts = self.run_traffic_ip46_routed_to_bridged(
+            test_l2_deny, is_ip6, is_reflect, False, add_eh
+        )
+        self.verify_acl_packet_count(acls["L3"].acl_index, pkts)
+
+    def run_test_ip46_bridged_to_routed(self, test_l2_deny, is_ip6, is_reflect, add_eh):
+        acls = self.apply_acl_ip46_bridged_to_routed(
+            test_l2_deny, is_ip6, is_reflect, add_eh
+        )
+        pkts = self.run_traffic_ip46_bridged_to_routed(
+            test_l2_deny, is_ip6, is_reflect, False, add_eh
+        )
+        self.verify_acl_packet_count(acls["L2"].acl_index, pkts)
+
+    def run_test_ip46_routed_to_bridged_and_back(
+        self, test_l2_action, is_ip6, add_eh, stateful_icmp=False
+    ):
+        self.apply_acl_ip46_both_directions_reflect(
+            False, test_l2_action, is_ip6, add_eh, stateful_icmp
+        )
+        self.run_traffic_ip46_routed_to_bridged(
+            test_l2_action, is_ip6, True, False, add_eh, stateful_icmp
+        )
+        self.run_traffic_ip46_bridged_to_routed(
+            test_l2_action, is_ip6, False, True, add_eh, stateful_icmp
+        )
+
+    def run_test_ip46_bridged_to_routed_and_back(
+        self, test_l2_action, is_ip6, add_eh, stateful_icmp=False
+    ):
+        self.apply_acl_ip46_both_directions_reflect(
+            True, test_l2_action, is_ip6, add_eh, stateful_icmp
+        )
+        self.run_traffic_ip46_bridged_to_routed(
+            test_l2_action, is_ip6, True, False, add_eh, stateful_icmp
+        )
+        self.run_traffic_ip46_routed_to_bridged(
+            test_l2_action, is_ip6, False, True, add_eh, stateful_icmp
+        )

 
     def test_0000_ip6_irb_1(self):
 
     def test_0000_ip6_irb_1(self):

-        """ ACL plugin prepare"""
+        """ACL plugin prepare"""

         if not self.vpp_dead:
             cmd = "set acl-plugin session timeout udp idle 2000"
             self.logger.info(self.vapi.ppcli(cmd))
         if not self.vpp_dead:
             cmd = "set acl-plugin session timeout udp idle 2000"
             self.logger.info(self.vapi.ppcli(cmd))


@@ -596,169 +755,188 @@ class TestIpIrb(VppTestCase):
             #    "set acl-plugin l2-datapath old"))
 
     def test_0001_ip6_irb_1(self):
             #    "set acl-plugin l2-datapath old"))
 
     def test_0001_ip6_irb_1(self):

-        """ ACL IPv6 routed -> bridged, L2 ACL deny"""
-        self.run_test_ip46_routed_to_bridged(True, True, False,
-                                             self.WITHOUT_EH)
+        """ACL IPv6 routed -> bridged, L2 ACL deny"""
+        self.run_test_ip46_routed_to_bridged(True, True, False, self.WITHOUT_EH)

 
     def test_0002_ip6_irb_1(self):
 
     def test_0002_ip6_irb_1(self):

-        """ ACL IPv6 routed -> bridged, L3 ACL deny"""
-        self.run_test_ip46_routed_to_bridged(False, True, False,
-                                             self.WITHOUT_EH)
+        """ACL IPv6 routed -> bridged, L3 ACL deny"""
+        self.run_test_ip46_routed_to_bridged(False, True, False, self.WITHOUT_EH)

 
     def test_0003_ip4_irb_1(self):
 
     def test_0003_ip4_irb_1(self):

-        """ ACL IPv4 routed -> bridged, L2 ACL deny"""
-        self.run_test_ip46_routed_to_bridged(True, False, False,
-                                             self.WITHOUT_EH)
+        """ACL IPv4 routed -> bridged, L2 ACL deny"""
+        self.run_test_ip46_routed_to_bridged(True, False, False, self.WITHOUT_EH)

 
     def test_0004_ip4_irb_1(self):
 
     def test_0004_ip4_irb_1(self):

-        """ ACL IPv4 routed -> bridged, L3 ACL deny"""
-        self.run_test_ip46_routed_to_bridged(False, False, False,
-                                             self.WITHOUT_EH)
+        """ACL IPv4 routed -> bridged, L3 ACL deny"""
+        self.run_test_ip46_routed_to_bridged(False, False, False, self.WITHOUT_EH)

 
     def test_0005_ip6_irb_1(self):
 
     def test_0005_ip6_irb_1(self):

-        """ ACL IPv6 bridged -> routed, L2 ACL deny """
-        self.run_test_ip46_bridged_to_routed(True, True, False,
-                                             self.WITHOUT_EH)
+        """ACL IPv6 bridged -> routed, L2 ACL deny"""
+        self.run_test_ip46_bridged_to_routed(True, True, False, self.WITHOUT_EH)

 
     def test_0006_ip6_irb_1(self):
 
     def test_0006_ip6_irb_1(self):

-        """ ACL IPv6 bridged -> routed, L3 ACL deny """
-        self.run_test_ip46_bridged_to_routed(False, True, False,
-                                             self.WITHOUT_EH)
+        """ACL IPv6 bridged -> routed, L3 ACL deny"""
+        self.run_test_ip46_bridged_to_routed(False, True, False, self.WITHOUT_EH)

 
     def test_0007_ip6_irb_1(self):
 
     def test_0007_ip6_irb_1(self):

-        """ ACL IPv4 bridged -> routed, L2 ACL deny """
-        self.run_test_ip46_bridged_to_routed(True, False, False,
-                                             self.WITHOUT_EH)
+        """ACL IPv4 bridged -> routed, L2 ACL deny"""
+        self.run_test_ip46_bridged_to_routed(True, False, False, self.WITHOUT_EH)

 
     def test_0008_ip6_irb_1(self):
 
     def test_0008_ip6_irb_1(self):

-        """ ACL IPv4 bridged -> routed, L3 ACL deny """
-        self.run_test_ip46_bridged_to_routed(False, False, False,
-                                             self.WITHOUT_EH)
+        """ACL IPv4 bridged -> routed, L3 ACL deny"""
+        self.run_test_ip46_bridged_to_routed(False, False, False, self.WITHOUT_EH)

 
     # Stateful ACL tests
     def test_0101_ip6_irb_1(self):
 
     # Stateful ACL tests
     def test_0101_ip6_irb_1(self):

-        """ ACL IPv6 routed -> bridged, L2 ACL permit+reflect"""
-        self.run_test_ip46_routed_to_bridged_and_back(True, True,
-                                                      self.WITHOUT_EH)
+        """ACL IPv6 routed -> bridged, L2 ACL permit+reflect"""
+        self.run_test_ip46_routed_to_bridged_and_back(True, True, self.WITHOUT_EH)

 
     def test_0102_ip6_irb_1(self):
 
     def test_0102_ip6_irb_1(self):

-        """ ACL IPv6 bridged -> routed, L2 ACL permit+reflect"""
-        self.run_test_ip46_bridged_to_routed_and_back(True, True,
-                                                      self.WITHOUT_EH)
+        """ACL IPv6 bridged -> routed, L2 ACL permit+reflect"""
+        self.run_test_ip46_bridged_to_routed_and_back(True, True, self.WITHOUT_EH)

 
     def test_0103_ip6_irb_1(self):
 
     def test_0103_ip6_irb_1(self):

-        """ ACL IPv4 routed -> bridged, L2 ACL permit+reflect"""
-        self.run_test_ip46_routed_to_bridged_and_back(True, False,
-                                                      self.WITHOUT_EH)
+        """ACL IPv4 routed -> bridged, L2 ACL permit+reflect"""
+        self.run_test_ip46_routed_to_bridged_and_back(True, False, self.WITHOUT_EH)

 
     def test_0104_ip6_irb_1(self):
 
     def test_0104_ip6_irb_1(self):

-        """ ACL IPv4 bridged -> routed, L2 ACL permit+reflect"""
-        self.run_test_ip46_bridged_to_routed_and_back(True, False,
-                                                      self.WITHOUT_EH)
+        """ACL IPv4 bridged -> routed, L2 ACL permit+reflect"""
+        self.run_test_ip46_bridged_to_routed_and_back(True, False, self.WITHOUT_EH)

 
     def test_0111_ip6_irb_1(self):
 
     def test_0111_ip6_irb_1(self):

-        """ ACL IPv6 routed -> bridged, L3 ACL permit+reflect"""
-        self.run_test_ip46_routed_to_bridged_and_back(False, True,
-                                                      self.WITHOUT_EH)
+        """ACL IPv6 routed -> bridged, L3 ACL permit+reflect"""
+        self.run_test_ip46_routed_to_bridged_and_back(False, True, self.WITHOUT_EH)

 
     def test_0112_ip6_irb_1(self):
 
     def test_0112_ip6_irb_1(self):

-        """ ACL IPv6 bridged -> routed, L3 ACL permit+reflect"""
-        self.run_test_ip46_bridged_to_routed_and_back(False, True,
-                                                      self.WITHOUT_EH)
+        """ACL IPv6 bridged -> routed, L3 ACL permit+reflect"""
+        self.run_test_ip46_bridged_to_routed_and_back(False, True, self.WITHOUT_EH)

 
     def test_0113_ip6_irb_1(self):
 
     def test_0113_ip6_irb_1(self):

-        """ ACL IPv4 routed -> bridged, L3 ACL permit+reflect"""
-        self.run_test_ip46_routed_to_bridged_and_back(False, False,
-                                                      self.WITHOUT_EH)
+        """ACL IPv4 routed -> bridged, L3 ACL permit+reflect"""
+        self.run_test_ip46_routed_to_bridged_and_back(False, False, self.WITHOUT_EH)

 
     def test_0114_ip6_irb_1(self):
 
     def test_0114_ip6_irb_1(self):

-        """ ACL IPv4 bridged -> routed, L3 ACL permit+reflect"""
-        self.run_test_ip46_bridged_to_routed_and_back(False, False,
-                                                      self.WITHOUT_EH)
+        """ACL IPv4 bridged -> routed, L3 ACL permit+reflect"""
+        self.run_test_ip46_bridged_to_routed_and_back(False, False, self.WITHOUT_EH)

 
     # A block of tests with extension headers
 
     def test_1001_ip6_irb_1(self):
 
     # A block of tests with extension headers
 
     def test_1001_ip6_irb_1(self):

-        """ ACL IPv6+EH routed -> bridged, L2 ACL deny"""
-        self.run_test_ip46_routed_to_bridged(True, True, False,
-                                             self.WITH_EH)
+        """ACL IPv6+EH routed -> bridged, L2 ACL deny"""
+        self.run_test_ip46_routed_to_bridged(True, True, False, self.WITH_EH)

 
     def test_1002_ip6_irb_1(self):
 
     def test_1002_ip6_irb_1(self):

-        """ ACL IPv6+EH routed -> bridged, L3 ACL deny"""
-        self.run_test_ip46_routed_to_bridged(False, True, False,
-                                             self.WITH_EH)
+        """ACL IPv6+EH routed -> bridged, L3 ACL deny"""
+        self.run_test_ip46_routed_to_bridged(False, True, False, self.WITH_EH)

 
     def test_1005_ip6_irb_1(self):
 
     def test_1005_ip6_irb_1(self):

-        """ ACL IPv6+EH bridged -> routed, L2 ACL deny """
-        self.run_test_ip46_bridged_to_routed(True, True, False,
-                                             self.WITH_EH)
+        """ACL IPv6+EH bridged -> routed, L2 ACL deny"""
+        self.run_test_ip46_bridged_to_routed(True, True, False, self.WITH_EH)

 
     def test_1006_ip6_irb_1(self):
 
     def test_1006_ip6_irb_1(self):

-        """ ACL IPv6+EH bridged -> routed, L3 ACL deny """
-        self.run_test_ip46_bridged_to_routed(False, True, False,
-                                             self.WITH_EH)
+        """ACL IPv6+EH bridged -> routed, L3 ACL deny"""
+        self.run_test_ip46_bridged_to_routed(False, True, False, self.WITH_EH)

 
     def test_1101_ip6_irb_1(self):
 
     def test_1101_ip6_irb_1(self):

-        """ ACL IPv6+EH routed -> bridged, L2 ACL permit+reflect"""
-        self.run_test_ip46_routed_to_bridged_and_back(True, True,
-                                                      self.WITH_EH)
+        """ACL IPv6+EH routed -> bridged, L2 ACL permit+reflect"""
+        self.run_test_ip46_routed_to_bridged_and_back(True, True, self.WITH_EH)

 
     def test_1102_ip6_irb_1(self):
 
     def test_1102_ip6_irb_1(self):

-        """ ACL IPv6+EH bridged -> routed, L2 ACL permit+reflect"""
-        self.run_test_ip46_bridged_to_routed_and_back(True, True,
-                                                      self.WITH_EH)
+        """ACL IPv6+EH bridged -> routed, L2 ACL permit+reflect"""
+        self.run_test_ip46_bridged_to_routed_and_back(True, True, self.WITH_EH)

 
     def test_1111_ip6_irb_1(self):
 
     def test_1111_ip6_irb_1(self):

-        """ ACL IPv6+EH routed -> bridged, L3 ACL permit+reflect"""
-        self.run_test_ip46_routed_to_bridged_and_back(False, True,
-                                                      self.WITH_EH)
+        """ACL IPv6+EH routed -> bridged, L3 ACL permit+reflect"""
+        self.run_test_ip46_routed_to_bridged_and_back(False, True, self.WITH_EH)

 
     def test_1112_ip6_irb_1(self):
 
     def test_1112_ip6_irb_1(self):

-        """ ACL IPv6+EH bridged -> routed, L3 ACL permit+reflect"""
-        self.run_test_ip46_bridged_to_routed_and_back(False, True,
-                                                      self.WITH_EH)
+        """ACL IPv6+EH bridged -> routed, L3 ACL permit+reflect"""
+        self.run_test_ip46_bridged_to_routed_and_back(False, True, self.WITH_EH)

 
     # IPv4 with "MF" bit set
 
     def test_1201_ip6_irb_1(self):
 
     # IPv4 with "MF" bit set
 
     def test_1201_ip6_irb_1(self):

-        """ ACL IPv4+MF routed -> bridged, L2 ACL deny"""
-        self.run_test_ip46_routed_to_bridged(True, False, False,
-                                             self.WITH_EH)
+        """ACL IPv4+MF routed -> bridged, L2 ACL deny"""
+        self.run_test_ip46_routed_to_bridged(True, False, False, self.WITH_EH)

 
     def test_1202_ip6_irb_1(self):
 
     def test_1202_ip6_irb_1(self):

-        """ ACL IPv4+MF routed -> bridged, L3 ACL deny"""
-        self.run_test_ip46_routed_to_bridged(False, False, False,
-                                             self.WITH_EH)
+        """ACL IPv4+MF routed -> bridged, L3 ACL deny"""
+        self.run_test_ip46_routed_to_bridged(False, False, False, self.WITH_EH)

 
     def test_1205_ip6_irb_1(self):
 
     def test_1205_ip6_irb_1(self):

-        """ ACL IPv4+MF bridged -> routed, L2 ACL deny """
-        self.run_test_ip46_bridged_to_routed(True, False, False,
-                                             self.WITH_EH)
+        """ACL IPv4+MF bridged -> routed, L2 ACL deny"""
+        self.run_test_ip46_bridged_to_routed(True, False, False, self.WITH_EH)

 
     def test_1206_ip6_irb_1(self):
 
     def test_1206_ip6_irb_1(self):

-        """ ACL IPv4+MF bridged -> routed, L3 ACL deny """
-        self.run_test_ip46_bridged_to_routed(False, False, False,
-                                             self.WITH_EH)
+        """ACL IPv4+MF bridged -> routed, L3 ACL deny"""
+        self.run_test_ip46_bridged_to_routed(False, False, False, self.WITH_EH)

 
     def test_1301_ip6_irb_1(self):
 
     def test_1301_ip6_irb_1(self):

-        """ ACL IPv4+MF routed -> bridged, L2 ACL permit+reflect"""
-        self.run_test_ip46_routed_to_bridged_and_back(True, False,
-                                                      self.WITH_EH)
+        """ACL IPv4+MF routed -> bridged, L2 ACL permit+reflect"""
+        self.run_test_ip46_routed_to_bridged_and_back(True, False, self.WITH_EH)

 
     def test_1302_ip6_irb_1(self):
 
     def test_1302_ip6_irb_1(self):

-        """ ACL IPv4+MF bridged -> routed, L2 ACL permit+reflect"""
-        self.run_test_ip46_bridged_to_routed_and_back(True, False,
-                                                      self.WITH_EH)
+        """ACL IPv4+MF bridged -> routed, L2 ACL permit+reflect"""
+        self.run_test_ip46_bridged_to_routed_and_back(True, False, self.WITH_EH)

 
     def test_1311_ip6_irb_1(self):
 
     def test_1311_ip6_irb_1(self):

-        """ ACL IPv4+MF routed -> bridged, L3 ACL permit+reflect"""
-        self.run_test_ip46_routed_to_bridged_and_back(False, False,
-                                                      self.WITH_EH)
+        """ACL IPv4+MF routed -> bridged, L3 ACL permit+reflect"""
+        self.run_test_ip46_routed_to_bridged_and_back(False, False, self.WITH_EH)

 
     def test_1312_ip6_irb_1(self):
 
     def test_1312_ip6_irb_1(self):

-        """ ACL IPv4+MF bridged -> routed, L3 ACL permit+reflect"""
-        self.run_test_ip46_bridged_to_routed_and_back(False, False,
-                                                      self.WITH_EH)
-
-if __name__ == '__main__':
+        """ACL IPv4+MF bridged -> routed, L3 ACL permit+reflect"""
+        self.run_test_ip46_bridged_to_routed_and_back(False, False, self.WITH_EH)
+
+    # Stateful ACL tests with stateful ICMP
+
+    def test_1401_ip6_irb_1(self):
+        """IPv6 routed -> bridged, L2 ACL permit+reflect, ICMP reflect"""
+        self.run_test_ip46_routed_to_bridged_and_back(
+            True, True, self.WITHOUT_EH, self.STATEFUL_ICMP
+        )
+
+    def test_1402_ip6_irb_1(self):
+        """IPv6 bridged -> routed, L2 ACL permit+reflect, ICMP reflect"""
+        self.run_test_ip46_bridged_to_routed_and_back(
+            True, True, self.WITHOUT_EH, self.STATEFUL_ICMP
+        )
+
+    def test_1403_ip4_irb_1(self):
+        """IPv4 routed -> bridged, L2 ACL permit+reflect, ICMP reflect"""
+        self.run_test_ip46_routed_to_bridged_and_back(
+            True, False, self.WITHOUT_EH, self.STATEFUL_ICMP
+        )
+
+    def test_1404_ip4_irb_1(self):
+        """IPv4 bridged -> routed, L2 ACL permit+reflect, ICMP reflect"""
+        self.run_test_ip46_bridged_to_routed_and_back(
+            True, False, self.WITHOUT_EH, self.STATEFUL_ICMP
+        )
+
+    def test_1411_ip6_irb_1(self):
+        """IPv6 routed -> bridged, L3 ACL permit+reflect, ICMP reflect"""
+        self.run_test_ip46_routed_to_bridged_and_back(
+            False, True, self.WITHOUT_EH, self.STATEFUL_ICMP
+        )
+
+    def test_1412_ip6_irb_1(self):
+        """IPv6 bridged -> routed, L3 ACL permit+reflect, ICMP reflect"""
+        self.run_test_ip46_bridged_to_routed_and_back(
+            False, True, self.WITHOUT_EH, self.STATEFUL_ICMP
+        )
+
+    def test_1413_ip4_irb_1(self):
+        """IPv4 routed -> bridged, L3 ACL permit+reflect, ICMP reflect"""
+        self.run_test_ip46_routed_to_bridged_and_back(
+            False, False, self.WITHOUT_EH, self.STATEFUL_ICMP
+        )
+
+    def test_1414_ip4_irb_1(self):
+        """IPv4 bridged -> routed, L3 ACL permit+reflect, ICMP reflect"""
+        self.run_test_ip46_bridged_to_routed_and_back(
+            False, False, self.WITHOUT_EH, self.STATEFUL_ICMP
+        )
+
+
+if __name__ == "__main__":

     unittest.main(testRunner=VppTestRunner)
     unittest.main(testRunner=VppTestRunner)