tests: add ip6 punt policer handoff test

Add a test for ip6 punt policer thread handoff. A child class is
created that uses common punt test setup but is configured to have 2
worker threads.

Type: test
Signed-off-by: Brian Russell <brian@graphiant.com>
Change-Id: I65c44f5d79455129accd7cf405c15920a5a9b678

1 files changed, 52 insertions, 1 deletions

diff --git a/test/test_ip6.py b/test/test_ip6.py
index a9244bd3d60..2aab4defa64 100644
--- a/test/test_ip6.py
+++ b/test/test_ip6.py
@@ -31,7 +31,7 @@ from vpp_ip_route import VppIpRoute, VppRoutePath, find_route, VppIpMRoute, \
 from vpp_neighbor import find_nbr, VppNeighbor
 from vpp_pg_interface import is_ipv6_misc
 from vpp_sub_interface import VppSubInterface, VppDot1QSubint
-from vpp_policer import VppPolicer
+from vpp_policer import VppPolicer, PolicerAction
 from ipaddress import IPv6Network, IPv6Address
 
 AF_INET6 = socket.AF_INET6
@@ -2309,6 +2309,57 @@ class TestIP6Punt(IP6PuntSetup, VppTestCase):
         self.assertEqual(str(punts[2].punt.nh), '::')
 
 
+class TestIP6PuntHandoff(IP6PuntSetup, VppTestCase):
+    """ IPv6 Punt Police/Redirect """
+    worker_config = "workers 2"
+
+    def setUp(self):
+        super(TestIP6PuntHandoff, self).setUp()
+        super(TestIP6PuntHandoff, self).punt_setup()
+
+    def tearDown(self):
+        super(TestIP6PuntHandoff, self).punt_teardown()
+        super(TestIP6PuntHandoff, self).tearDown()
+
+    def test_ip_punt(self):
+        """ IP6 punt policer thread handoff """
+        pkts = self.pkt * NUM_PKTS
+
+        #
+        # Configure a punt redirect via pg1.
+        #
+        nh_addr = self.pg1.remote_ip6
+        ip_punt_redirect = VppIpPuntRedirect(self, self.pg0.sw_if_index,
+                                             self.pg1.sw_if_index, nh_addr)
+        ip_punt_redirect.add_vpp_config()
+
+        action_tx = PolicerAction(
+            VppEnum.vl_api_sse2_qos_action_type_t.SSE2_QOS_ACTION_API_TRANSMIT,
+            0)
+        #
+        # This policer drops no packets, we are just
+        # testing that they get to the right thread.
+        #
+        policer = VppPolicer(self, "ip6-punt", 400, 0, 10, 0, 1,
+                             0, 0, False, action_tx, action_tx, action_tx)
+        policer.add_vpp_config()
+        ip_punt_policer = VppIpPuntPolicer(self, policer.policer_index,
+                                           is_ip6=True)
+        ip_punt_policer.add_vpp_config()
+
+        for worker in [0, 1]:
+            self.send_and_expect(self.pg0, pkts, self.pg1, worker=worker)
+            if worker == 0:
+                self.logger.debug(self.vapi.cli("show trace max 100"))
+
+        #
+        # Clean up
+        #
+        ip_punt_policer.remove_vpp_config()
+        policer.remove_vpp_config()
+        ip_punt_redirect.remove_vpp_config()
+
+
 class TestIPDeag(VppTestCase):
     """ IPv6 Deaggregate Routes """