tests: move plugin tests to src/plugins/*/test

- Relocate plugin tests for 'make test' into
  src/plugins/*/test so that plugin test cases
  are co-located with the plugin source code.

Type: refactor

Signed-off-by: Dave Wallace <dwallacelf@gmail.com>
Change-Id: I503e6a43528e14981799b735fa65674155713f67
Signed-off-by: Dave Wallace <dwallacelf@gmail.com>

1 files changed, 1519 insertions, 0 deletions

diff --git a/src/plugins/acl/test/test_acl_plugin.py b/src/plugins/acl/test/test_acl_plugin.py
new file mode 100644
index 00000000000..eca02316bf6
--- /dev/null
+++ b/src/plugins/acl/test/test_acl_plugin.py
@@ -0,0 +1,1519 @@
+#!/usr/bin/env python
+"""ACL plugin Test Case HLD:
+"""
+
+import unittest
+import random
+
+from scapy.packet import Raw
+from scapy.layers.l2 import Ether
+from scapy.layers.inet import IP, TCP, UDP, ICMP
+from scapy.layers.inet6 import IPv6, ICMPv6EchoRequest
+from scapy.layers.inet6 import IPv6ExtHdrFragment
+from framework import VppTestCase, VppTestRunner
+from util import Host, ppp
+
+from vpp_lo_interface import VppLoInterface
+
+
+class TestACLplugin(VppTestCase):
+    """ ACL plugin Test Case """
+
+    # traffic types
+    IP = 0
+    ICMP = 1
+
+    # IP version
+    IPRANDOM = -1
+    IPV4 = 0
+    IPV6 = 1
+
+    # rule types
+    DENY = 0
+    PERMIT = 1
+
+    # supported protocols
+    proto = [[6, 17], [1, 58]]
+    proto_map = {1: 'ICMP', 58: 'ICMPv6EchoRequest', 6: 'TCP', 17: 'UDP'}
+    ICMPv4 = 0
+    ICMPv6 = 1
+    TCP = 0
+    UDP = 1
+    PROTO_ALL = 0
+
+    # port ranges
+    PORTS_ALL = -1
+    PORTS_RANGE = 0
+    PORTS_RANGE_2 = 1
+    udp_sport_from = 10
+    udp_sport_to = udp_sport_from + 5
+    udp_dport_from = 20000
+    udp_dport_to = udp_dport_from + 5000
+    tcp_sport_from = 30
+    tcp_sport_to = tcp_sport_from + 5
+    tcp_dport_from = 40000
+    tcp_dport_to = tcp_dport_from + 5000
+
+    udp_sport_from_2 = 90
+    udp_sport_to_2 = udp_sport_from_2 + 5
+    udp_dport_from_2 = 30000
+    udp_dport_to_2 = udp_dport_from_2 + 5000
+    tcp_sport_from_2 = 130
+    tcp_sport_to_2 = tcp_sport_from_2 + 5
+    tcp_dport_from_2 = 20000
+    tcp_dport_to_2 = tcp_dport_from_2 + 5000
+
+    icmp4_type = 8  # echo request
+    icmp4_code = 3
+    icmp6_type = 128  # echo request
+    icmp6_code = 3
+
+    icmp4_type_2 = 8
+    icmp4_code_from_2 = 5
+    icmp4_code_to_2 = 20
+    icmp6_type_2 = 128
+    icmp6_code_from_2 = 8
+    icmp6_code_to_2 = 42
+
+    # Test variables
+    bd_id = 1
+
+    @classmethod
+    def setUpClass(cls):
+        """
+        Perform standard class setup (defined by class method setUpClass in
+        class VppTestCase) before running the test case, set test case related
+        variables and configure VPP.
+        """
+        super(TestACLplugin, cls).setUpClass()
+
+        try:
+            # Create 2 pg interfaces
+            cls.create_pg_interfaces(range(2))
+
+            # Packet flows mapping pg0 -> pg1, pg2 etc.
+            cls.flows = dict()
+            cls.flows[cls.pg0] = [cls.pg1]
+
+            # Packet sizes
+            cls.pg_if_packet_sizes = [64, 512, 1518, 9018]
+
+            # Create BD with MAC learning and unknown unicast flooding disabled
+            # and put interfaces to this BD
+            cls.vapi.bridge_domain_add_del(bd_id=cls.bd_id, uu_flood=1,
+                                           learn=1)
+            for pg_if in cls.pg_interfaces:
+                cls.vapi.sw_interface_set_l2_bridge(
+                    rx_sw_if_index=pg_if.sw_if_index, bd_id=cls.bd_id)
+
+            # Set up all interfaces
+            for i in cls.pg_interfaces:
+                i.admin_up()
+
+            # Mapping between packet-generator index and lists of test hosts
+            cls.hosts_by_pg_idx = dict()
+            for pg_if in cls.pg_interfaces:
+                cls.hosts_by_pg_idx[pg_if.sw_if_index] = []
+
+            # Create list of deleted hosts
+            cls.deleted_hosts_by_pg_idx = dict()
+            for pg_if in cls.pg_interfaces:
+                cls.deleted_hosts_by_pg_idx[pg_if.sw_if_index] = []
+
+            # warm-up the mac address tables
+            # self.warmup_test()
+            count = 16
+            start = 0
+            n_int = len(cls.pg_interfaces)
+            macs_per_if = count / n_int
+            i = -1
+            for pg_if in cls.pg_interfaces:
+                i += 1
+                start_nr = macs_per_if * i + start
+                end_nr = count + start if i == (n_int - 1) \
+                    else macs_per_if * (i + 1) + start
+                hosts = cls.hosts_by_pg_idx[pg_if.sw_if_index]
+                for j in range(start_nr, end_nr):
+                    host = Host(
+                        "00:00:00:ff:%02x:%02x" % (pg_if.sw_if_index, j),
+                        "172.17.1%02x.%u" % (pg_if.sw_if_index, j),
+                        "2017:dead:%02x::%u" % (pg_if.sw_if_index, j))
+                    hosts.append(host)
+
+        except Exception:
+            super(TestACLplugin, cls).tearDownClass()
+            raise
+
+    @classmethod
+    def tearDownClass(cls):
+        super(TestACLplugin, cls).tearDownClass()
+
+    def setUp(self):
+        super(TestACLplugin, self).setUp()
+        self.reset_packet_infos()
+
+    def tearDown(self):
+        """
+        Show various debug prints after each test.
+        """
+        super(TestACLplugin, self).tearDown()
+
+    def show_commands_at_teardown(self):
+        cli = "show vlib graph l2-input-feat-arc"
+        self.logger.info(self.vapi.ppcli(cli))
+        cli = "show vlib graph l2-input-feat-arc-end"
+        self.logger.info(self.vapi.ppcli(cli))
+        cli = "show vlib graph l2-output-feat-arc"
+        self.logger.info(self.vapi.ppcli(cli))
+        cli = "show vlib graph l2-output-feat-arc-end"
+        self.logger.info(self.vapi.ppcli(cli))
+        self.logger.info(self.vapi.ppcli("show l2fib verbose"))
+        self.logger.info(self.vapi.ppcli("show acl-plugin acl"))
+        self.logger.info(self.vapi.ppcli("show acl-plugin interface"))
+        self.logger.info(self.vapi.ppcli("show acl-plugin tables"))
+        self.logger.info(self.vapi.ppcli("show bridge-domain %s detail"
+                                         % self.bd_id))
+
+    def create_rule(self, ip=0, permit_deny=0, ports=PORTS_ALL, proto=-1,
+                    s_prefix=0, s_ip='\x00\x00\x00\x00',
+                    d_prefix=0, d_ip='\x00\x00\x00\x00'):
+        if proto == -1:
+            return
+        if ports == self.PORTS_ALL:
+            sport_from = 0
+            dport_from = 0
+            sport_to = 65535 if proto != 1 and proto != 58 else 255
+            dport_to = sport_to
+        elif ports == self.PORTS_RANGE:
+            if proto == 1:
+                sport_from = self.icmp4_type
+                sport_to = self.icmp4_type
+                dport_from = self.icmp4_code
+                dport_to = self.icmp4_code
+            elif proto == 58:
+                sport_from = self.icmp6_type
+                sport_to = self.icmp6_type
+                dport_from = self.icmp6_code
+                dport_to = self.icmp6_code
+            elif proto == self.proto[self.IP][self.TCP]:
+                sport_from = self.tcp_sport_from
+                sport_to = self.tcp_sport_to
+                dport_from = self.tcp_dport_from
+                dport_to = self.tcp_dport_to
+            elif proto == self.proto[self.IP][self.UDP]:
+                sport_from = self.udp_sport_from
+                sport_to = self.udp_sport_to
+                dport_from = self.udp_dport_from
+                dport_to = self.udp_dport_to
+        elif ports == self.PORTS_RANGE_2:
+            if proto == 1:
+                sport_from = self.icmp4_type_2
+                sport_to = self.icmp4_type_2
+                dport_from = self.icmp4_code_from_2
+                dport_to = self.icmp4_code_to_2
+            elif proto == 58:
+                sport_from = self.icmp6_type_2
+                sport_to = self.icmp6_type_2
+                dport_from = self.icmp6_code_from_2
+                dport_to = self.icmp6_code_to_2
+            elif proto == self.proto[self.IP][self.TCP]:
+                sport_from = self.tcp_sport_from_2
+                sport_to = self.tcp_sport_to_2
+                dport_from = self.tcp_dport_from_2
+                dport_to = self.tcp_dport_to_2
+            elif proto == self.proto[self.IP][self.UDP]:
+                sport_from = self.udp_sport_from_2
+                sport_to = self.udp_sport_to_2
+                dport_from = self.udp_dport_from_2
+                dport_to = self.udp_dport_to_2
+        else:
+            sport_from = ports
+            sport_to = ports
+            dport_from = ports
+            dport_to = ports
+
+        rule = ({'is_permit': permit_deny, 'is_ipv6': ip, 'proto': proto,
+                 'srcport_or_icmptype_first': sport_from,
+                 'srcport_or_icmptype_last': sport_to,
+                 'src_ip_prefix_len': s_prefix,
+                 'src_ip_addr': s_ip,
+                 'dstport_or_icmpcode_first': dport_from,
+                 'dstport_or_icmpcode_last': dport_to,
+                 'dst_ip_prefix_len': d_prefix,
+                 'dst_ip_addr': d_ip})
+        return rule
+
+    def apply_rules(self, rules, tag=b''):
+        reply = self.vapi.acl_add_replace(acl_index=4294967295, r=rules,
+                                          tag=tag)
+        self.logger.info("Dumped ACL: " + str(
+            self.vapi.acl_dump(reply.acl_index)))
+        # Apply a ACL on the interface as inbound
+        for i in self.pg_interfaces:
+            self.vapi.acl_interface_set_acl_list(sw_if_index=i.sw_if_index,
+                                                 n_input=1,
+                                                 acls=[reply.acl_index])
+        return reply.acl_index
+
+    def apply_rules_to(self, rules, tag=b'', sw_if_index=0xFFFFFFFF):
+        reply = self.vapi.acl_add_replace(acl_index=4294967295, r=rules,
+                                          tag=tag)
+        self.logger.info("Dumped ACL: " + str(
+            self.vapi.acl_dump(reply.acl_index)))
+        # Apply a ACL on the interface as inbound
+        self.vapi.acl_interface_set_acl_list(sw_if_index=sw_if_index,
+                                             n_input=1,
+                                             acls=[reply.acl_index])
+        return reply.acl_index
+
+    def etype_whitelist(self, whitelist, n_input):
+        # Apply whitelists on all the interfaces
+        for i in self.pg_interfaces:
+            # checkstyle can't read long names. Help them.
+            fun = self.vapi.acl_interface_set_etype_whitelist
+            fun(sw_if_index=i.sw_if_index, n_input=n_input,
+                whitelist=whitelist)
+        return
+
+    def create_upper_layer(self, packet_index, proto, ports=0):
+        p = self.proto_map[proto]
+        if p == 'UDP':
+            if ports == 0:
+                return UDP(sport=random.randint(self.udp_sport_from,
+                                                self.udp_sport_to),
+                           dport=random.randint(self.udp_dport_from,
+                                                self.udp_dport_to))
+            else:
+                return UDP(sport=ports, dport=ports)
+        elif p == 'TCP':
+            if ports == 0:
+                return TCP(sport=random.randint(self.tcp_sport_from,
+                                                self.tcp_sport_to),
+                           dport=random.randint(self.tcp_dport_from,
+                                                self.tcp_dport_to))
+            else:
+                return TCP(sport=ports, dport=ports)
+        return ''
+
+    def create_stream(self, src_if, packet_sizes, traffic_type=0, ipv6=0,
+                      proto=-1, ports=0, fragments=False,
+                      pkt_raw=True, etype=-1):
+        """
+        Create input packet stream for defined interface using hosts or
+        deleted_hosts list.
+
+        :param object src_if: Interface to create packet stream for.
+        :param list packet_sizes: List of required packet sizes.
+        :param traffic_type: 1: ICMP packet, 2: IPv6 with EH, 0: otherwise.
+        :return: Stream of packets.
+        """
+        pkts = []
+        if self.flows.__contains__(src_if):
+            src_hosts = self.hosts_by_pg_idx[src_if.sw_if_index]
+            for dst_if in self.flows[src_if]:
+                dst_hosts = self.hosts_by_pg_idx[dst_if.sw_if_index]
+                n_int = len(dst_hosts) * len(src_hosts)
+                for i in range(0, n_int):
+                    dst_host = dst_hosts[i / len(src_hosts)]
+                    src_host = src_hosts[i % len(src_hosts)]
+                    pkt_info = self.create_packet_info(src_if, dst_if)
+                    if ipv6 == 1:
+                        pkt_info.ip = 1
+                    elif ipv6 == 0:
+                        pkt_info.ip = 0
+                    else:
+                        pkt_info.ip = random.choice([0, 1])
+                    if proto == -1:
+                        pkt_info.proto = random.choice(self.proto[self.IP])
+                    else:
+                        pkt_info.proto = proto
+                    payload = self.info_to_payload(pkt_info)
+                    p = Ether(dst=dst_host.mac, src=src_host.mac)
+                    if etype > 0:
+                        p = Ether(dst=dst_host.mac,
+                                  src=src_host.mac,
+                                  type=etype)
+                    if pkt_info.ip:
+                        p /= IPv6(dst=dst_host.ip6, src=src_host.ip6)
+                        if fragments:
+                            p /= IPv6ExtHdrFragment(offset=64, m=1)
+                    else:
+                        if fragments:
+                            p /= IP(src=src_host.ip4, dst=dst_host.ip4,
+                                    flags=1, frag=64)
+                        else:
+                            p /= IP(src=src_host.ip4, dst=dst_host.ip4)
+                    if traffic_type == self.ICMP:
+                        if pkt_info.ip:
+                            p /= ICMPv6EchoRequest(type=self.icmp6_type,
+                                                   code=self.icmp6_code)
+                        else:
+                            p /= ICMP(type=self.icmp4_type,
+                                      code=self.icmp4_code)
+                    else:
+                        p /= self.create_upper_layer(i, pkt_info.proto, ports)
+                    if pkt_raw:
+                        p /= Raw(payload)
+                        pkt_info.data = p.copy()
+                    if pkt_raw:
+                        size = random.choice(packet_sizes)
+                        self.extend_packet(p, size)
+                    pkts.append(p)
+        return pkts
+
+    def verify_capture(self, pg_if, capture,
+                       traffic_type=0, ip_type=0, etype=-1):
+        """
+        Verify captured input packet stream for defined interface.
+
+        :param object pg_if: Interface to verify captured packet stream for.
+        :param list capture: Captured packet stream.
+        :param traffic_type: 1: ICMP packet, 2: IPv6 with EH, 0: otherwise.
+        """
+        last_info = dict()
+        for i in self.pg_interfaces:
+            last_info[i.sw_if_index] = None
+        dst_sw_if_index = pg_if.sw_if_index
+        for packet in capture:
+            if etype > 0:
+                if packet[Ether].type != etype:
+                    self.logger.error(ppp("Unexpected ethertype in packet:",
+                                          packet))
+                else:
+                    continue
+            try:
+                # Raw data for ICMPv6 are stored in ICMPv6EchoRequest.data
+                if traffic_type == self.ICMP and ip_type == self.IPV6:
+                    payload_info = self.payload_to_info(
+                        packet[ICMPv6EchoRequest], 'data')
+                    payload = packet[ICMPv6EchoRequest]
+                else:
+                    payload_info = self.payload_to_info(packet[Raw])
+                    payload = packet[self.proto_map[payload_info.proto]]
+            except:
+                self.logger.error(ppp("Unexpected or invalid packet "
+                                      "(outside network):", packet))
+                raise
+
+            if ip_type != 0:
+                self.assertEqual(payload_info.ip, ip_type)
+            if traffic_type == self.ICMP:
+                try:
+                    if payload_info.ip == 0:
+                        self.assertEqual(payload.type, self.icmp4_type)
+                        self.assertEqual(payload.code, self.icmp4_code)
+                    else:
+                        self.assertEqual(payload.type, self.icmp6_type)
+                        self.assertEqual(payload.code, self.icmp6_code)
+                except:
+                    self.logger.error(ppp("Unexpected or invalid packet "
+                                          "(outside network):", packet))
+                    raise
+            else:
+                try:
+                    ip_version = IPv6 if payload_info.ip == 1 else IP
+
+                    ip = packet[ip_version]
+                    packet_index = payload_info.index
+
+                    self.assertEqual(payload_info.dst, dst_sw_if_index)
+                    self.logger.debug("Got packet on port %s: src=%u (id=%u)" %
+                                      (pg_if.name, payload_info.src,
+                                       packet_index))
+                    next_info = self.get_next_packet_info_for_interface2(
+                        payload_info.src, dst_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)
+                    saved_packet = next_info.data
+                    # Check standard fields
+                    self.assertEqual(ip.src, saved_packet[ip_version].src)
+                    self.assertEqual(ip.dst, saved_packet[ip_version].dst)
+                    p = self.proto_map[payload_info.proto]
+                    if p == 'TCP':
+                        tcp = packet[TCP]
+                        self.assertEqual(tcp.sport, saved_packet[
+                            TCP].sport)
+                        self.assertEqual(tcp.dport, saved_packet[
+                            TCP].dport)
+                    elif p == 'UDP':
+                        udp = packet[UDP]
+                        self.assertEqual(udp.sport, saved_packet[
+                            UDP].sport)
+                        self.assertEqual(udp.dport, saved_packet[
+                            UDP].dport)
+                except:
+                    self.logger.error(ppp("Unexpected or invalid packet:",
+                                          packet))
+                    raise
+        for i in self.pg_interfaces:
+            remaining_packet = self.get_next_packet_info_for_interface2(
+                i, dst_sw_if_index, last_info[i.sw_if_index])
+            self.assertTrue(
+                remaining_packet is None,
+                "Port %u: Packet expected from source %u didn't arrive" %
+                (dst_sw_if_index, i.sw_if_index))
+
+    def run_traffic_no_check(self):
+        # Test
+        # Create incoming packet streams for packet-generator interfaces
+        for i in self.pg_interfaces:
+            if self.flows.__contains__(i):
+                pkts = self.create_stream(i, self.pg_if_packet_sizes)
+                if len(pkts) > 0:
+                    i.add_stream(pkts)
+
+        # Enable packet capture and start packet sending
+        self.pg_enable_capture(self.pg_interfaces)
+        self.pg_start()
+
+    def run_verify_test(self, traffic_type=0, ip_type=0, proto=-1, ports=0,
+                        frags=False, pkt_raw=True, etype=-1):
+        # Test
+        # Create incoming packet streams for packet-generator interfaces
+        pkts_cnt = 0
+        for i in self.pg_interfaces:
+            if self.flows.__contains__(i):
+                pkts = self.create_stream(i, self.pg_if_packet_sizes,
+                                          traffic_type, ip_type, proto, ports,
+                                          frags, pkt_raw, etype)
+                if len(pkts) > 0:
+                    i.add_stream(pkts)
+                    pkts_cnt += len(pkts)
+
+        # Enable packet capture and start packet sendingself.IPV
+        self.pg_enable_capture(self.pg_interfaces)
+        self.pg_start()
+        self.logger.info("sent packets count: %d" % pkts_cnt)
+
+        # Verify
+        # Verify outgoing packet streams per packet-generator interface
+        for src_if in self.pg_interfaces:
+            if self.flows.__contains__(src_if):
+                for dst_if in self.flows[src_if]:
+                    capture = dst_if.get_capture(pkts_cnt)
+                    self.logger.info("Verifying capture on interface %s" %
+                                     dst_if.name)
+                    self.verify_capture(dst_if, capture,
+                                        traffic_type, ip_type, etype)
+
+    def run_verify_negat_test(self, traffic_type=0, ip_type=0, proto=-1,
+                              ports=0, frags=False, etype=-1):
+        # Test
+        pkts_cnt = 0
+        self.reset_packet_infos()
+        for i in self.pg_interfaces:
+            if self.flows.__contains__(i):
+                pkts = self.create_stream(i, self.pg_if_packet_sizes,
+                                          traffic_type, ip_type, proto, ports,
+                                          frags, True, etype)
+                if len(pkts) > 0:
+                    i.add_stream(pkts)
+                    pkts_cnt += len(pkts)
+
+        # Enable packet capture and start packet sending
+        self.pg_enable_capture(self.pg_interfaces)
+        self.pg_start()
+        self.logger.info("sent packets count: %d" % pkts_cnt)
+
+        # Verify
+        # Verify outgoing packet streams per packet-generator interface
+        for src_if in self.pg_interfaces:
+            if self.flows.__contains__(src_if):
+                for dst_if in self.flows[src_if]:
+                    self.logger.info("Verifying capture on interface %s" %
+                                     dst_if.name)
+                    capture = dst_if.get_capture(0)
+                    self.assertEqual(len(capture), 0)
+
+    def test_0000_warmup_test(self):
+        """ ACL plugin version check; learn MACs
+        """
+        reply = self.vapi.papi.acl_plugin_get_version()
+        self.assertEqual(reply.major, 1)
+        self.logger.info("Working with ACL plugin version: %d.%d" % (
+            reply.major, reply.minor))
+        # minor version changes are non breaking
+        # self.assertEqual(reply.minor, 0)
+
+    def test_0001_acl_create(self):
+        """ ACL create/delete test
+        """
+
+        self.logger.info("ACLP_TEST_START_0001")
+        # Add an ACL
+        r = [{'is_permit': 1, 'is_ipv6': 0, 'proto': 17,
+              'srcport_or_icmptype_first': 1234,
+              'srcport_or_icmptype_last': 1235,
+              'src_ip_prefix_len': 0,
+              'src_ip_addr': b'\x00\x00\x00\x00',
+              'dstport_or_icmpcode_first': 1234,
+              'dstport_or_icmpcode_last': 1234,
+              'dst_ip_addr': b'\x00\x00\x00\x00',
+              'dst_ip_prefix_len': 0}]
+        # Test 1: add a new ACL
+        reply = self.vapi.acl_add_replace(acl_index=4294967295, r=r,
+                                          tag=b"permit 1234")
+        self.assertEqual(reply.retval, 0)
+        # The very first ACL gets #0
+        self.assertEqual(reply.acl_index, 0)
+        first_acl = reply.acl_index
+        rr = self.vapi.acl_dump(reply.acl_index)
+        self.logger.info("Dumped ACL: " + str(rr))
+        self.assertEqual(len(rr), 1)
+        # We should have the same number of ACL entries as we had asked
+        self.assertEqual(len(rr[0].r), len(r))
+        # The rules should be the same. But because the submitted and returned
+        # are different types, we need to iterate over rules and keys to get
+        # to basic values.
+        for i_rule in range(0, len(r) - 1):
+            for rule_key in r[i_rule]:
+                self.assertEqual(rr[0].r[i_rule][rule_key],
+                                 r[i_rule][rule_key])
+
+        # Add a deny-1234 ACL
+        r_deny = [{'is_permit': 0, 'is_ipv6': 0, 'proto': 17,
+                   'srcport_or_icmptype_first': 1234,
+                   'srcport_or_icmptype_last': 1235,
+                   'src_ip_prefix_len': 0,
+                   'src_ip_addr': b'\x00\x00\x00\x00',
+                   'dstport_or_icmpcode_first': 1234,
+                   'dstport_or_icmpcode_last': 1234,
+                   'dst_ip_addr': b'\x00\x00\x00\x00',
+                   'dst_ip_prefix_len': 0},
+                  {'is_permit': 1, 'is_ipv6': 0, 'proto': 17,
+                   'srcport_or_icmptype_first': 0,
+                   'srcport_or_icmptype_last': 0,
+                   'src_ip_prefix_len': 0,
+                   'src_ip_addr': b'\x00\x00\x00\x00',
+                   'dstport_or_icmpcode_first': 0,
+                   'dstport_or_icmpcode_last': 0,
+                   'dst_ip_addr': b'\x00\x00\x00\x00',
+                   'dst_ip_prefix_len': 0}]
+
+        reply = self.vapi.acl_add_replace(acl_index=4294967295, r=r_deny,
+                                          tag=b"deny 1234;permit all")
+        self.assertEqual(reply.retval, 0)
+        # The second ACL gets #1
+        self.assertEqual(reply.acl_index, 1)
+        second_acl = reply.acl_index
+
+        # Test 2: try to modify a nonexistent ACL
+        reply = self.vapi.acl_add_replace(acl_index=432, r=r,
+                                          tag=b"FFFF:FFFF", expected_retval=-6)
+        self.assertEqual(reply.retval, -6)
+        # The ACL number should pass through
+        self.assertEqual(reply.acl_index, 432)
+        # apply an ACL on an interface inbound, try to delete ACL, must fail
+        self.vapi.acl_interface_set_acl_list(sw_if_index=self.pg0.sw_if_index,
+                                             n_input=1,
+                                             acls=[first_acl])
+        reply = self.vapi.acl_del(acl_index=first_acl, expected_retval=-142)
+        # Unapply an ACL and then try to delete it - must be ok
+        self.vapi.acl_interface_set_acl_list(sw_if_index=self.pg0.sw_if_index,
+                                             n_input=0,
+                                             acls=[])
+        reply = self.vapi.acl_del(acl_index=first_acl, expected_retval=0)
+
+        # apply an ACL on an interface outbound, try to delete ACL, must fail
+        self.vapi.acl_interface_set_acl_list(sw_if_index=self.pg0.sw_if_index,
+                                             n_input=0,
+                                             acls=[second_acl])
+        reply = self.vapi.acl_del(acl_index=second_acl, expected_retval=-143)
+        # Unapply the ACL and then try to delete it - must be ok
+        self.vapi.acl_interface_set_acl_list(sw_if_index=self.pg0.sw_if_index,
+                                             n_input=0,
+                                             acls=[])
+        reply = self.vapi.acl_del(acl_index=second_acl, expected_retval=0)
+
+        # try to apply a nonexistent ACL - must fail
+        self.vapi.acl_interface_set_acl_list(sw_if_index=self.pg0.sw_if_index,
+                                             n_input=1,
+                                             acls=[first_acl],
+                                             expected_retval=-6)
+
+        self.logger.info("ACLP_TEST_FINISH_0001")
+
+    def test_0002_acl_permit_apply(self):
+        """ permit ACL apply test
+        """
+        self.logger.info("ACLP_TEST_START_0002")
+
+        rules = []
+        rules.append(self.create_rule(self.IPV4, self.PERMIT,
+                     0, self.proto[self.IP][self.UDP]))
+        rules.append(self.create_rule(self.IPV4, self.PERMIT,
+                     0, self.proto[self.IP][self.TCP]))
+
+        # Apply rules
+        acl_idx = self.apply_rules(rules, b"permit per-flow")
+
+        # enable counters
+        reply = self.vapi.papi.acl_stats_intf_counters_enable(enable=1)
+
+        # Traffic should still pass
+        self.run_verify_test(self.IP, self.IPV4, -1)
+
+        matches = self.statistics.get_counter('/acl/%d/matches' % acl_idx)
+        self.logger.info("stat segment counters: %s" % repr(matches))
+        cli = "show acl-plugin acl"
+        self.logger.info(self.vapi.ppcli(cli))
+        cli = "show acl-plugin tables"
+        self.logger.info(self.vapi.ppcli(cli))
+
+        total_hits = matches[0][0]['packets'] + matches[0][1]['packets']
+        self.assertEqual(total_hits, 64)
+
+        # disable counters
+        reply = self.vapi.papi.acl_stats_intf_counters_enable(enable=0)
+
+        self.logger.info("ACLP_TEST_FINISH_0002")
+
+    def test_0003_acl_deny_apply(self):
+        """ deny ACL apply test
+        """
+        self.logger.info("ACLP_TEST_START_0003")
+        # Add a deny-flows ACL
+        rules = []
+        rules.append(self.create_rule(self.IPV4, self.DENY,
+                     self.PORTS_ALL, self.proto[self.IP][self.UDP]))
+        # Permit ip any any in the end
+        rules.append(self.create_rule(self.IPV4, self.PERMIT,
+                                      self.PORTS_ALL, 0))
+
+        # Apply rules
+        acl_idx = self.apply_rules(rules, b"deny per-flow;permit all")
+
+        # enable counters
+        reply = self.vapi.papi.acl_stats_intf_counters_enable(enable=1)
+
+        # Traffic should not pass
+        self.run_verify_negat_test(self.IP, self.IPV4,
+                                   self.proto[self.IP][self.UDP])
+
+        matches = self.statistics.get_counter('/acl/%d/matches' % acl_idx)
+        self.logger.info("stat segment counters: %s" % repr(matches))
+        cli = "show acl-plugin acl"
+        self.logger.info(self.vapi.ppcli(cli))
+        cli = "show acl-plugin tables"
+        self.logger.info(self.vapi.ppcli(cli))
+        self.assertEqual(matches[0][0]['packets'], 64)
+        # disable counters
+        reply = self.vapi.papi.acl_stats_intf_counters_enable(enable=0)
+        self.logger.info("ACLP_TEST_FINISH_0003")
+        # self.assertEqual(, 0)
+
+    def test_0004_vpp624_permit_icmpv4(self):
+        """ VPP_624 permit ICMPv4
+        """
+        self.logger.info("ACLP_TEST_START_0004")
+
+        # Add an ACL
+        rules = []
+        rules.append(self.create_rule(self.IPV4, self.PERMIT, self.PORTS_RANGE,
+                                      self.proto[self.ICMP][self.ICMPv4]))
+        # deny ip any any in the end
+        rules.append(self.create_rule(self.IPV4, self.DENY, self.PORTS_ALL, 0))
+
+        # Apply rules
+        self.apply_rules(rules, b"permit icmpv4")
+
+        # Traffic should still pass
+        self.run_verify_test(self.ICMP, self.IPV4,
+                             self.proto[self.ICMP][self.ICMPv4])
+
+        self.logger.info("ACLP_TEST_FINISH_0004")
+
+    def test_0005_vpp624_permit_icmpv6(self):
+        """ VPP_624 permit ICMPv6
+        """
+        self.logger.info("ACLP_TEST_START_0005")
+
+        # Add an ACL
+        rules = []
+        rules.append(self.create_rule(self.IPV6, self.PERMIT, self.PORTS_RANGE,
+                                      self.proto[self.ICMP][self.ICMPv6]))
+        # deny ip any any in the end
+        rules.append(self.create_rule(self.IPV6, self.DENY, self.PORTS_ALL, 0))
+
+        # Apply rules
+        self.apply_rules(rules, b"permit icmpv6")
+
+        # Traffic should still pass
+        self.run_verify_test(self.ICMP, self.IPV6,
+                             self.proto[self.ICMP][self.ICMPv6])
+
+        self.logger.info("ACLP_TEST_FINISH_0005")
+
+    def test_0006_vpp624_deny_icmpv4(self):
+        """ VPP_624 deny ICMPv4
+        """
+        self.logger.info("ACLP_TEST_START_0006")
+        # Add an ACL
+        rules = []
+        rules.append(self.create_rule(self.IPV4, self.DENY, self.PORTS_RANGE,
+                                      self.proto[self.ICMP][self.ICMPv4]))
+        # permit ip any any in the end
+        rules.append(self.create_rule(self.IPV4, self.PERMIT,
+                                      self.PORTS_ALL, 0))
+
+        # Apply rules
+        self.apply_rules(rules, b"deny icmpv4")
+
+        # Traffic should not pass
+        self.run_verify_negat_test(self.ICMP, self.IPV4, 0)
+
+        self.logger.info("ACLP_TEST_FINISH_0006")
+
+    def test_0007_vpp624_deny_icmpv6(self):
+        """ VPP_624 deny ICMPv6
+        """
+        self.logger.info("ACLP_TEST_START_0007")
+        # Add an ACL
+        rules = []
+        rules.append(self.create_rule(self.IPV6, self.DENY, self.PORTS_RANGE,
+                                      self.proto[self.ICMP][self.ICMPv6]))
+        # deny ip any any in the end
+        rules.append(self.create_rule(self.IPV6, self.PERMIT,
+                                      self.PORTS_ALL, 0))
+
+        # Apply rules
+        self.apply_rules(rules, b"deny icmpv6")
+
+        # Traffic should not pass
+        self.run_verify_negat_test(self.ICMP, self.IPV6, 0)
+
+        self.logger.info("ACLP_TEST_FINISH_0007")
+
+    def test_0008_tcp_permit_v4(self):
+        """ permit TCPv4
+        """
+        self.logger.info("ACLP_TEST_START_0008")
+
+        # Add an ACL
+        rules = []
+        rules.append(self.create_rule(self.IPV4, self.PERMIT, self.PORTS_RANGE,
+                     self.proto[self.IP][self.TCP]))
+        # deny ip any any in the end
+        rules.append(self.create_rule(self.IPV4, self.DENY, self.PORTS_ALL, 0))
+
+        # Apply rules
+        self.apply_rules(rules, b"permit ipv4 tcp")
+
+        # Traffic should still pass
+        self.run_verify_test(self.IP, self.IPV4, self.proto[self.IP][self.TCP])
+
+        self.logger.info("ACLP_TEST_FINISH_0008")
+
+    def test_0009_tcp_permit_v6(self):
+        """ permit TCPv6
+        """
+        self.logger.info("ACLP_TEST_START_0009")
+
+        # Add an ACL
+        rules = []
+        rules.append(self.create_rule(self.IPV6, self.PERMIT, self.PORTS_RANGE,
+                                      self.proto[self.IP][self.TCP]))
+        # deny ip any any in the end
+        rules.append(self.create_rule(self.IPV6, self.DENY, self.PORTS_ALL, 0))
+
+        # Apply rules
+        self.apply_rules(rules, b"permit ip6 tcp")
+
+        # Traffic should still pass
+        self.run_verify_test(self.IP, self.IPV6, self.proto[self.IP][self.TCP])
+
+        self.logger.info("ACLP_TEST_FINISH_0008")
+
+    def test_0010_udp_permit_v4(self):
+        """ permit UDPv4
+        """
+        self.logger.info("ACLP_TEST_START_0010")
+
+        # Add an ACL
+        rules = []
+        rules.append(self.create_rule(self.IPV4, self.PERMIT, self.PORTS_RANGE,
+                                      self.proto[self.IP][self.UDP]))
+        # deny ip any any in the end
+        rules.append(self.create_rule(self.IPV4, self.DENY, self.PORTS_ALL, 0))
+
+        # Apply rules
+        self.apply_rules(rules, b"permit ipv udp")
+
+        # Traffic should still pass
+        self.run_verify_test(self.IP, self.IPV4, self.proto[self.IP][self.UDP])
+
+        self.logger.info("ACLP_TEST_FINISH_0010")
+
+    def test_0011_udp_permit_v6(self):
+        """ permit UDPv6
+        """
+        self.logger.info("ACLP_TEST_START_0011")
+
+        # Add an ACL
+        rules = []
+        rules.append(self.create_rule(self.IPV6, self.PERMIT, self.PORTS_RANGE,
+                                      self.proto[self.IP][self.UDP]))
+        # deny ip any any in the end
+        rules.append(self.create_rule(self.IPV6, self.DENY, self.PORTS_ALL, 0))
+
+        # Apply rules
+        self.apply_rules(rules, b"permit ip6 udp")
+
+        # Traffic should still pass
+        self.run_verify_test(self.IP, self.IPV6, self.proto[self.IP][self.UDP])
+
+        self.logger.info("ACLP_TEST_FINISH_0011")
+
+    def test_0012_tcp_deny(self):
+        """ deny TCPv4/v6
+        """
+        self.logger.info("ACLP_TEST_START_0012")
+
+        # Add an ACL
+        rules = []
+        rules.append(self.create_rule(self.IPV4, self.DENY, self.PORTS_RANGE,
+                                      self.proto[self.IP][self.TCP]))
+        rules.append(self.create_rule(self.IPV6, self.DENY, self.PORTS_RANGE,
+                                      self.proto[self.IP][self.TCP]))
+        # permit ip any any in the end
+        rules.append(self.create_rule(self.IPV4, self.PERMIT,
+                                      self.PORTS_ALL, 0))
+        rules.append(self.create_rule(self.IPV6, self.PERMIT,
+                                      self.PORTS_ALL, 0))
+
+        # Apply rules
+        self.apply_rules(rules, b"deny ip4/ip6 tcp")
+
+        # Traffic should not pass
+        self.run_verify_negat_test(self.IP, self.IPRANDOM,
+                                   self.proto[self.IP][self.TCP])
+
+        self.logger.info("ACLP_TEST_FINISH_0012")
+
+    def test_0013_udp_deny(self):
+        """ deny UDPv4/v6
+        """
+        self.logger.info("ACLP_TEST_START_0013")
+
+        # Add an ACL
+        rules = []
+        rules.append(self.create_rule(self.IPV4, self.DENY, self.PORTS_RANGE,
+                                      self.proto[self.IP][self.UDP]))
+        rules.append(self.create_rule(self.IPV6, self.DENY, self.PORTS_RANGE,
+                                      self.proto[self.IP][self.UDP]))
+        # permit ip any any in the end
+        rules.append(self.create_rule(self.IPV4, self.PERMIT,
+                                      self.PORTS_ALL, 0))
+        rules.append(self.create_rule(self.IPV6, self.PERMIT,
+                                      self.PORTS_ALL, 0))
+
+        # Apply rules
+        self.apply_rules(rules, b"deny ip4/ip6 udp")
+
+        # Traffic should not pass
+        self.run_verify_negat_test(self.IP, self.IPRANDOM,
+                                   self.proto[self.IP][self.UDP])
+
+        self.logger.info("ACLP_TEST_FINISH_0013")
+
+    def test_0014_acl_dump(self):
+        """ verify add/dump acls
+        """
+        self.logger.info("ACLP_TEST_START_0014")
+
+        r = [[self.IPV4, self.PERMIT, 1234, self.proto[self.IP][self.TCP]],
+             [self.IPV4, self.PERMIT, 2345, self.proto[self.IP][self.UDP]],
+             [self.IPV4, self.PERMIT, 0, self.proto[self.IP][self.TCP]],
+             [self.IPV4, self.PERMIT, 0, self.proto[self.IP][self.UDP]],
+             [self.IPV4, self.PERMIT, 5, self.proto[self.ICMP][self.ICMPv4]],
+             [self.IPV6, self.PERMIT, 4321, self.proto[self.IP][self.TCP]],
+             [self.IPV6, self.PERMIT, 5432, self.proto[self.IP][self.UDP]],
+             [self.IPV6, self.PERMIT, 0, self.proto[self.IP][self.TCP]],
+             [self.IPV6, self.PERMIT, 0, self.proto[self.IP][self.UDP]],
+             [self.IPV6, self.PERMIT, 6, self.proto[self.ICMP][self.ICMPv6]],
+             [self.IPV4, self.DENY, self.PORTS_ALL, 0],
+             [self.IPV4, self.DENY, 1234, self.proto[self.IP][self.TCP]],
+             [self.IPV4, self.DENY, 2345, self.proto[self.IP][self.UDP]],
+             [self.IPV4, self.DENY, 5, self.proto[self.ICMP][self.ICMPv4]],
+             [self.IPV6, self.DENY, 4321, self.proto[self.IP][self.TCP]],
+             [self.IPV6, self.DENY, 5432, self.proto[self.IP][self.UDP]],
+             [self.IPV6, self.DENY, 6, self.proto[self.ICMP][self.ICMPv6]],
+             [self.IPV6, self.DENY, self.PORTS_ALL, 0]
+             ]
+
+        # Add and verify new ACLs
+        rules = []
+        for i in range(len(r)):
+            rules.append(self.create_rule(r[i][0], r[i][1], r[i][2], r[i][3]))
+
+        reply = self.vapi.acl_add_replace(acl_index=4294967295, r=rules)
+        result = self.vapi.acl_dump(reply.acl_index)
+
+        i = 0
+        for drules in result:
+            for dr in drules.r:
+                self.assertEqual(dr.is_ipv6, r[i][0])
+                self.assertEqual(dr.is_permit, r[i][1])
+                self.assertEqual(dr.proto, r[i][3])
+
+                if r[i][2] > 0:
+                    self.assertEqual(dr.srcport_or_icmptype_first, r[i][2])
+                else:
+                    if r[i][2] < 0:
+                        self.assertEqual(dr.srcport_or_icmptype_first, 0)
+                        self.assertEqual(dr.srcport_or_icmptype_last, 65535)
+                    else:
+                        if dr.proto == self.proto[self.IP][self.TCP]:
+                            self.assertGreater(dr.srcport_or_icmptype_first,
+                                               self.tcp_sport_from-1)
+                            self.assertLess(dr.srcport_or_icmptype_first,
+                                            self.tcp_sport_to+1)
+                            self.assertGreater(dr.dstport_or_icmpcode_last,
+                                               self.tcp_dport_from-1)
+                            self.assertLess(dr.dstport_or_icmpcode_last,
+                                            self.tcp_dport_to+1)
+                        elif dr.proto == self.proto[self.IP][self.UDP]:
+                            self.assertGreater(dr.srcport_or_icmptype_first,
+                                               self.udp_sport_from-1)
+                            self.assertLess(dr.srcport_or_icmptype_first,
+                                            self.udp_sport_to+1)
+                            self.assertGreater(dr.dstport_or_icmpcode_last,
+                                               self.udp_dport_from-1)
+                            self.assertLess(dr.dstport_or_icmpcode_last,
+                                            self.udp_dport_to+1)
+                i += 1
+
+        self.logger.info("ACLP_TEST_FINISH_0014")
+
+    def test_0015_tcp_permit_port_v4(self):
+        """ permit single TCPv4
+        """
+        self.logger.info("ACLP_TEST_START_0015")
+
+        port = random.randint(0, 65535)
+        # Add an ACL
+        rules = []
+        rules.append(self.create_rule(self.IPV4, self.PERMIT, port,
+                                      self.proto[self.IP][self.TCP]))
+        # deny ip any any in the end
+        rules.append(self.create_rule(self.IPV4, self.DENY, self.PORTS_ALL, 0))
+
+        # Apply rules
+        self.apply_rules(rules, b"permit ip4 tcp %d" % port)
+
+        # Traffic should still pass
+        self.run_verify_test(self.IP, self.IPV4,
+                             self.proto[self.IP][self.TCP], port)
+
+        self.logger.info("ACLP_TEST_FINISH_0015")
+
+    def test_0016_udp_permit_port_v4(self):
+        """ permit single UDPv4
+        """
+        self.logger.info("ACLP_TEST_START_0016")
+
+        port = random.randint(0, 65535)
+        # Add an ACL
+        rules = []
+        rules.append(self.create_rule(self.IPV4, self.PERMIT, port,
+                                      self.proto[self.IP][self.UDP]))
+        # deny ip any any in the end
+        rules.append(self.create_rule(self.IPV4, self.DENY, self.PORTS_ALL, 0))
+
+        # Apply rules
+        self.apply_rules(rules, b"permit ip4 tcp %d" % port)
+
+        # Traffic should still pass
+        self.run_verify_test(self.IP, self.IPV4,
+                             self.proto[self.IP][self.UDP], port)
+
+        self.logger.info("ACLP_TEST_FINISH_0016")
+
+    def test_0017_tcp_permit_port_v6(self):
+        """ permit single TCPv6
+        """
+        self.logger.info("ACLP_TEST_START_0017")
+
+        port = random.randint(0, 65535)
+        # Add an ACL
+        rules = []
+        rules.append(self.create_rule(self.IPV6, self.PERMIT, port,
+                                      self.proto[self.IP][self.TCP]))
+        # deny ip any any in the end
+        rules.append(self.create_rule(self.IPV6, self.DENY, self.PORTS_ALL, 0))
+
+        # Apply rules
+        self.apply_rules(rules, b"permit ip4 tcp %d" % port)
+
+        # Traffic should still pass
+        self.run_verify_test(self.IP, self.IPV6,
+                             self.proto[self.IP][self.TCP], port)
+
+        self.logger.info("ACLP_TEST_FINISH_0017")
+
+    def test_0018_udp_permit_port_v6(self):
+        """ permit single UPPv6
+        """
+        self.logger.info("ACLP_TEST_START_0018")
+
+        port = random.randint(0, 65535)
+        # Add an ACL
+        rules = []
+        rules.append(self.create_rule(self.IPV6, self.PERMIT, port,
+                                      self.proto[self.IP][self.UDP]))
+        # deny ip any any in the end
+        rules.append(self.create_rule(self.IPV6, self.DENY,
+                                      self.PORTS_ALL, 0))
+
+        # Apply rules
+        self.apply_rules(rules, b"permit ip4 tcp %d" % port)
+
+        # Traffic should still pass
+        self.run_verify_test(self.IP, self.IPV6,
+                             self.proto[self.IP][self.UDP], port)
+
+        self.logger.info("ACLP_TEST_FINISH_0018")
+
+    def test_0019_udp_deny_port(self):
+        """ deny single TCPv4/v6
+        """
+        self.logger.info("ACLP_TEST_START_0019")
+
+        port = random.randint(0, 65535)
+        # Add an ACL
+        rules = []
+        rules.append(self.create_rule(self.IPV4, self.DENY, port,
+                                      self.proto[self.IP][self.TCP]))
+        rules.append(self.create_rule(self.IPV6, self.DENY, port,
+                                      self.proto[self.IP][self.TCP]))
+        # Permit ip any any in the end
+        rules.append(self.create_rule(self.IPV4, self.PERMIT,
+                                      self.PORTS_ALL, 0))
+        rules.append(self.create_rule(self.IPV6, self.PERMIT,
+                                      self.PORTS_ALL, 0))
+
+        # Apply rules
+        self.apply_rules(rules, b"deny ip4/ip6 udp %d" % port)
+
+        # Traffic should not pass
+        self.run_verify_negat_test(self.IP, self.IPRANDOM,
+                                   self.proto[self.IP][self.TCP], port)
+
+        self.logger.info("ACLP_TEST_FINISH_0019")
+
+    def test_0020_udp_deny_port(self):
+        """ deny single UDPv4/v6
+        """
+        self.logger.info("ACLP_TEST_START_0020")
+
+        port = random.randint(0, 65535)
+        # Add an ACL
+        rules = []
+        rules.append(self.create_rule(self.IPV4, self.DENY, port,
+                                      self.proto[self.IP][self.UDP]))
+        rules.append(self.create_rule(self.IPV6, self.DENY, port,
+                                      self.proto[self.IP][self.UDP]))
+        # Permit ip any any in the end
+        rules.append(self.create_rule(self.IPV4, self.PERMIT,
+                                      self.PORTS_ALL, 0))
+        rules.append(self.create_rule(self.IPV6, self.PERMIT,
+                                      self.PORTS_ALL, 0))
+
+        # Apply rules
+        self.apply_rules(rules, b"deny ip4/ip6 udp %d" % port)
+
+        # Traffic should not pass
+        self.run_verify_negat_test(self.IP, self.IPRANDOM,
+                                   self.proto[self.IP][self.UDP], port)
+
+        self.logger.info("ACLP_TEST_FINISH_0020")
+
+    def test_0021_udp_deny_port_verify_fragment_deny(self):
+        """ deny single UDPv4/v6, permit ip any, verify non-initial fragment
+        blocked
+        """
+        self.logger.info("ACLP_TEST_START_0021")
+
+        port = random.randint(0, 65535)
+        # Add an ACL
+        rules = []
+        rules.append(self.create_rule(self.IPV4, self.DENY, port,
+                                      self.proto[self.IP][self.UDP]))
+        rules.append(self.create_rule(self.IPV6, self.DENY, port,
+                                      self.proto[self.IP][self.UDP]))
+        # deny ip any any in the end
+        rules.append(self.create_rule(self.IPV4, self.PERMIT,
+                                      self.PORTS_ALL, 0))
+        rules.append(self.create_rule(self.IPV6, self.PERMIT,
+                                      self.PORTS_ALL, 0))
+
+        # Apply rules
+        self.apply_rules(rules, b"deny ip4/ip6 udp %d" % port)
+
+        # Traffic should not pass
+        self.run_verify_negat_test(self.IP, self.IPRANDOM,
+                                   self.proto[self.IP][self.UDP], port, True)
+
+        self.logger.info("ACLP_TEST_FINISH_0021")
+
+    def test_0022_zero_length_udp_ipv4(self):
+        """ VPP-687 zero length udp ipv4 packet"""
+        self.logger.info("ACLP_TEST_START_0022")
+
+        port = random.randint(0, 65535)
+        # Add an ACL
+        rules = []
+        rules.append(self.create_rule(self.IPV4, self.PERMIT, port,
+                                      self.proto[self.IP][self.UDP]))
+        # deny ip any any in the end
+        rules.append(
+            self.create_rule(self.IPV4, self.DENY, self.PORTS_ALL, 0))
+
+        # Apply rules
+        self.apply_rules(rules, b"permit empty udp ip4 %d" % port)
+
+        # Traffic should still pass
+        # Create incoming packet streams for packet-generator interfaces
+        pkts_cnt = 0
+        pkts = self.create_stream(self.pg0, self.pg_if_packet_sizes,
+                                  self.IP, self.IPV4,
+                                  self.proto[self.IP][self.UDP], port,
+                                  False, False)
+        if len(pkts) > 0:
+            self.pg0.add_stream(pkts)
+            pkts_cnt += len(pkts)
+
+        # Enable packet capture and start packet sendingself.IPV
+        self.pg_enable_capture(self.pg_interfaces)
+        self.pg_start()
+
+        self.pg1.get_capture(pkts_cnt)
+
+        self.logger.info("ACLP_TEST_FINISH_0022")
+
+    def test_0023_zero_length_udp_ipv6(self):
+        """ VPP-687 zero length udp ipv6 packet"""
+        self.logger.info("ACLP_TEST_START_0023")
+
+        port = random.randint(0, 65535)
+        # Add an ACL
+        rules = []
+        rules.append(self.create_rule(self.IPV6, self.PERMIT, port,
+                                      self.proto[self.IP][self.UDP]))
+        # deny ip any any in the end
+        rules.append(self.create_rule(self.IPV6, self.DENY, self.PORTS_ALL, 0))
+
+        # Apply rules
+        self.apply_rules(rules, b"permit empty udp ip6 %d" % port)
+
+        # Traffic should still pass
+        # Create incoming packet streams for packet-generator interfaces
+        pkts_cnt = 0
+        pkts = self.create_stream(self.pg0, self.pg_if_packet_sizes,
+                                  self.IP, self.IPV6,
+                                  self.proto[self.IP][self.UDP], port,
+                                  False, False)
+        if len(pkts) > 0:
+            self.pg0.add_stream(pkts)
+            pkts_cnt += len(pkts)
+
+        # Enable packet capture and start packet sendingself.IPV
+        self.pg_enable_capture(self.pg_interfaces)
+        self.pg_start()
+
+        # Verify outgoing packet streams per packet-generator interface
+        self.pg1.get_capture(pkts_cnt)
+
+        self.logger.info("ACLP_TEST_FINISH_0023")
+
+    def test_0108_tcp_permit_v4(self):
+        """ permit TCPv4 + non-match range
+        """
+        self.logger.info("ACLP_TEST_START_0108")
+
+        # Add an ACL
+        rules = []
+        rules.append(self.create_rule(self.IPV4, self.DENY, self.PORTS_RANGE_2,
+                     self.proto[self.IP][self.TCP]))
+        rules.append(self.create_rule(self.IPV4, self.PERMIT, self.PORTS_RANGE,
+                     self.proto[self.IP][self.TCP]))
+        # deny ip any any in the end
+        rules.append(self.create_rule(self.IPV4, self.DENY, self.PORTS_ALL, 0))
+
+        # Apply rules
+        self.apply_rules(rules, b"permit ipv4 tcp")
+
+        # Traffic should still pass
+        self.run_verify_test(self.IP, self.IPV4, self.proto[self.IP][self.TCP])
+
+        self.logger.info("ACLP_TEST_FINISH_0108")
+
+    def test_0109_tcp_permit_v6(self):
+        """ permit TCPv6 + non-match range
+        """
+        self.logger.info("ACLP_TEST_START_0109")
+
+        # Add an ACL
+        rules = []
+        rules.append(self.create_rule(self.IPV6, self.DENY, self.PORTS_RANGE_2,
+                                      self.proto[self.IP][self.TCP]))
+        rules.append(self.create_rule(self.IPV6, self.PERMIT, self.PORTS_RANGE,
+                                      self.proto[self.IP][self.TCP]))
+        # deny ip any any in the end
+        rules.append(self.create_rule(self.IPV6, self.DENY, self.PORTS_ALL, 0))
+
+        # Apply rules
+        self.apply_rules(rules, b"permit ip6 tcp")
+
+        # Traffic should still pass
+        self.run_verify_test(self.IP, self.IPV6, self.proto[self.IP][self.TCP])
+
+        self.logger.info("ACLP_TEST_FINISH_0109")
+
+    def test_0110_udp_permit_v4(self):
+        """ permit UDPv4 + non-match range
+        """
+        self.logger.info("ACLP_TEST_START_0110")
+
+        # Add an ACL
+        rules = []
+        rules.append(self.create_rule(self.IPV4, self.DENY, self.PORTS_RANGE_2,
+                                      self.proto[self.IP][self.UDP]))
+        rules.append(self.create_rule(self.IPV4, self.PERMIT, self.PORTS_RANGE,
+                                      self.proto[self.IP][self.UDP]))
+        # deny ip any any in the end
+        rules.append(self.create_rule(self.IPV4, self.DENY, self.PORTS_ALL, 0))
+
+        # Apply rules
+        self.apply_rules(rules, b"permit ipv4 udp")
+
+        # Traffic should still pass
+        self.run_verify_test(self.IP, self.IPV4, self.proto[self.IP][self.UDP])
+
+        self.logger.info("ACLP_TEST_FINISH_0110")
+
+    def test_0111_udp_permit_v6(self):
+        """ permit UDPv6 + non-match range
+        """
+        self.logger.info("ACLP_TEST_START_0111")
+
+        # Add an ACL
+        rules = []
+        rules.append(self.create_rule(self.IPV6, self.DENY, self.PORTS_RANGE_2,
+                                      self.proto[self.IP][self.UDP]))
+        rules.append(self.create_rule(self.IPV6, self.PERMIT, self.PORTS_RANGE,
+                                      self.proto[self.IP][self.UDP]))
+        # deny ip any any in the end
+        rules.append(self.create_rule(self.IPV6, self.DENY, self.PORTS_ALL, 0))
+
+        # Apply rules
+        self.apply_rules(rules, b"permit ip6 udp")
+
+        # Traffic should still pass
+        self.run_verify_test(self.IP, self.IPV6, self.proto[self.IP][self.UDP])
+
+        self.logger.info("ACLP_TEST_FINISH_0111")
+
+    def test_0112_tcp_deny(self):
+        """ deny TCPv4/v6 + non-match range
+        """
+        self.logger.info("ACLP_TEST_START_0112")
+
+        # Add an ACL
+        rules = []
+        rules.append(self.create_rule(self.IPV4, self.PERMIT,
+                                      self.PORTS_RANGE_2,
+                                      self.proto[self.IP][self.TCP]))
+        rules.append(self.create_rule(self.IPV6, self.PERMIT,
+                                      self.PORTS_RANGE_2,
+                                      self.proto[self.IP][self.TCP]))
+        rules.append(self.create_rule(self.IPV4, self.DENY, self.PORTS_RANGE,
+                                      self.proto[self.IP][self.TCP]))
+        rules.append(self.create_rule(self.IPV6, self.DENY, self.PORTS_RANGE,
+                                      self.proto[self.IP][self.TCP]))
+        # permit ip any any in the end
+        rules.append(self.create_rule(self.IPV4, self.PERMIT,
+                                      self.PORTS_ALL, 0))
+        rules.append(self.create_rule(self.IPV6, self.PERMIT,
+                                      self.PORTS_ALL, 0))
+
+        # Apply rules
+        self.apply_rules(rules, b"deny ip4/ip6 tcp")
+
+        # Traffic should not pass
+        self.run_verify_negat_test(self.IP, self.IPRANDOM,
+                                   self.proto[self.IP][self.TCP])
+
+        self.logger.info("ACLP_TEST_FINISH_0112")
+
+    def test_0113_udp_deny(self):
+        """ deny UDPv4/v6 + non-match range
+        """
+        self.logger.info("ACLP_TEST_START_0113")
+
+        # Add an ACL
+        rules = []
+        rules.append(self.create_rule(self.IPV4, self.PERMIT,
+                                      self.PORTS_RANGE_2,
+                                      self.proto[self.IP][self.UDP]))
+        rules.append(self.create_rule(self.IPV6, self.PERMIT,
+                                      self.PORTS_RANGE_2,
+                                      self.proto[self.IP][self.UDP]))
+        rules.append(self.create_rule(self.IPV4, self.DENY, self.PORTS_RANGE,
+                                      self.proto[self.IP][self.UDP]))
+        rules.append(self.create_rule(self.IPV6, self.DENY, self.PORTS_RANGE,
+                                      self.proto[self.IP][self.UDP]))
+        # permit ip any any in the end
+        rules.append(self.create_rule(self.IPV4, self.PERMIT,
+                                      self.PORTS_ALL, 0))
+        rules.append(self.create_rule(self.IPV6, self.PERMIT,
+                                      self.PORTS_ALL, 0))
+
+        # Apply rules
+        self.apply_rules(rules, b"deny ip4/ip6 udp")
+
+        # Traffic should not pass
+        self.run_verify_negat_test(self.IP, self.IPRANDOM,
+                                   self.proto[self.IP][self.UDP])
+
+        self.logger.info("ACLP_TEST_FINISH_0113")
+
+    def test_0300_tcp_permit_v4_etype_aaaa(self):
+        """ permit TCPv4, send 0xAAAA etype
+        """
+        self.logger.info("ACLP_TEST_START_0300")
+
+        # Add an ACL
+        rules = []
+        rules.append(self.create_rule(self.IPV4, self.DENY, self.PORTS_RANGE_2,
+                     self.proto[self.IP][self.TCP]))
+        rules.append(self.create_rule(self.IPV4, self.PERMIT, self.PORTS_RANGE,
+                     self.proto[self.IP][self.TCP]))
+        # deny ip any any in the end
+        rules.append(self.create_rule(self.IPV4, self.DENY, self.PORTS_ALL, 0))
+
+        # Apply rules
+        self.apply_rules(rules, b"permit ipv4 tcp")
+
+        # Traffic should still pass also for an odd ethertype
+        self.run_verify_test(self.IP, self.IPV4, self.proto[self.IP][self.TCP],
+                             0, False, True, 0xaaaa)
+        self.logger.info("ACLP_TEST_FINISH_0300")
+
+    def test_0305_tcp_permit_v4_etype_blacklist_aaaa(self):
+        """ permit TCPv4, whitelist 0x0BBB ethertype, send 0xAAAA-blocked
+        """
+        self.logger.info("ACLP_TEST_START_0305")
+
+        # Add an ACL
+        rules = []
+        rules.append(self.create_rule(self.IPV4, self.DENY, self.PORTS_RANGE_2,
+                     self.proto[self.IP][self.TCP]))
+        rules.append(self.create_rule(self.IPV4, self.PERMIT, self.PORTS_RANGE,
+                     self.proto[self.IP][self.TCP]))
+        # deny ip any any in the end
+        rules.append(self.create_rule(self.IPV4, self.DENY, self.PORTS_ALL, 0))
+
+        # Apply rules
+        self.apply_rules(rules, b"permit ipv4 tcp")
+
+        # whitelist the 0xbbbb etype - so the 0xaaaa should be blocked
+        self.etype_whitelist([0xbbb], 1)
+
+        # The oddball ethertype should be blocked
+        self.run_verify_negat_test(self.IP, self.IPV4,
+                                   self.proto[self.IP][self.TCP],
+                                   0, False, 0xaaaa)
+
+        # remove the whitelist
+        self.etype_whitelist([], 0)
+
+        self.logger.info("ACLP_TEST_FINISH_0305")
+
+    def test_0306_tcp_permit_v4_etype_blacklist_aaaa(self):
+        """ permit TCPv4, whitelist 0x0BBB ethertype, send 0x0BBB - pass
+        """
+        self.logger.info("ACLP_TEST_START_0306")
+
+        # Add an ACL
+        rules = []
+        rules.append(self.create_rule(self.IPV4, self.DENY, self.PORTS_RANGE_2,
+                     self.proto[self.IP][self.TCP]))
+        rules.append(self.create_rule(self.IPV4, self.PERMIT, self.PORTS_RANGE,
+                     self.proto[self.IP][self.TCP]))
+        # deny ip any any in the end
+        rules.append(self.create_rule(self.IPV4, self.DENY, self.PORTS_ALL, 0))
+
+        # Apply rules
+        self.apply_rules(rules, b"permit ipv4 tcp")
+
+        # whitelist the 0xbbbb etype - so the 0xaaaa should be blocked
+        self.etype_whitelist([0xbbb], 1)
+
+        # The whitelisted traffic, should pass
+        self.run_verify_test(self.IP, self.IPV4, self.proto[self.IP][self.TCP],
+                             0, False, True, 0x0bbb)
+
+        # remove the whitelist, the previously blocked 0xAAAA should pass now
+        self.etype_whitelist([], 0)
+
+        self.logger.info("ACLP_TEST_FINISH_0306")
+
+    def test_0307_tcp_permit_v4_etype_blacklist_aaaa(self):
+        """ permit TCPv4, whitelist 0x0BBB, remove, send 0xAAAA - pass
+        """
+        self.logger.info("ACLP_TEST_START_0307")
+
+        # Add an ACL
+        rules = []
+        rules.append(self.create_rule(self.IPV4, self.DENY, self.PORTS_RANGE_2,
+                     self.proto[self.IP][self.TCP]))
+        rules.append(self.create_rule(self.IPV4, self.PERMIT, self.PORTS_RANGE,
+                     self.proto[self.IP][self.TCP]))
+        # deny ip any any in the end
+        rules.append(self.create_rule(self.IPV4, self.DENY, self.PORTS_ALL, 0))
+
+        # Apply rules
+        self.apply_rules(rules, b"permit ipv4 tcp")
+
+        # whitelist the 0xbbbb etype - so the 0xaaaa should be blocked
+        self.etype_whitelist([0xbbb], 1)
+        # remove the whitelist, the previously blocked 0xAAAA should pass now
+        self.etype_whitelist([], 0)
+
+        # The whitelisted traffic, should pass
+        self.run_verify_test(self.IP, self.IPV4, self.proto[self.IP][self.TCP],
+                             0, False, True, 0xaaaa)
+
+        self.logger.info("ACLP_TEST_FINISH_0306")
+
+    def test_0315_del_intf(self):
+        """ apply an acl and delete the interface
+        """
+        self.logger.info("ACLP_TEST_START_0315")
+
+        # Add an ACL
+        rules = []
+        rules.append(self.create_rule(self.IPV4, self.DENY, self.PORTS_RANGE_2,
+                     self.proto[self.IP][self.TCP]))
+        rules.append(self.create_rule(self.IPV4, self.PERMIT, self.PORTS_RANGE,
+                     self.proto[self.IP][self.TCP]))
+        # deny ip any any in the end
+        rules.append(self.create_rule(self.IPV4, self.DENY, self.PORTS_ALL, 0))
+
+        # create an interface
+        intf = []
+        intf.append(VppLoInterface(self))
+
+        # Apply rules
+        self.apply_rules_to(rules, b"permit ipv4 tcp", intf[0].sw_if_index)
+
+        # Remove the interface
+        intf[0].remove_vpp_config()
+
+        self.logger.info("ACLP_TEST_FINISH_0315")
+
+if __name__ == '__main__':
+    unittest.main(testRunner=VppTestRunner)