+ def test_tun_handoff_66_police(self):
+ """ESP 6o6 tunnel with policer worker hand-off test"""
+ self.vapi.cli("clear errors")
+ self.vapi.cli("clear ipsec sa")
+
+ N_PKTS = 15
+ p = self.params[socket.AF_INET6]
+
+ action_tx = PolicerAction(
+ VppEnum.vl_api_sse2_qos_action_type_t.SSE2_QOS_ACTION_API_TRANSMIT, 0
+ )
+ policer = VppPolicer(
+ self,
+ "pol1",
+ 80,
+ 0,
+ 1000,
+ 0,
+ conform_action=action_tx,
+ exceed_action=action_tx,
+ violate_action=action_tx,
+ )
+ policer.add_vpp_config()
+
+ # Start policing on tun
+ policer.apply_vpp_config(p.tun_if.sw_if_index, Dir.RX, True)
+
+ for pol_bind in [1, 0]:
+ policer.bind_vpp_config(pol_bind, True)
+
+ # inject alternately on worker 0 and 1.
+ for worker in [0, 1, 0, 1]:
+ send_pkts = self.gen_encrypt_pkts6(
+ p,
+ p.scapy_tun_sa,
+ self.tun_if,
+ src=p.remote_tun_if_host,
+ dst=self.pg1.remote_ip6,
+ count=N_PKTS,
+ )
+ recv_pkts = self.send_and_expect(
+ self.tun_if, send_pkts, self.pg1, worker=worker
+ )
+ self.verify_decrypted6(p, recv_pkts)
+ self.logger.debug(self.vapi.cli("show trace max 100"))
+
+ stats = policer.get_stats()
+ stats0 = policer.get_stats(worker=0)
+ stats1 = policer.get_stats(worker=1)
+
+ if pol_bind == 1:
+ # First pass: Worker 1, should have done all the policing
+ self.assertEqual(stats, stats1)
+
+ # Worker 0, should have handed everything off
+ self.assertEqual(stats0["conform_packets"], 0)
+ self.assertEqual(stats0["exceed_packets"], 0)
+ self.assertEqual(stats0["violate_packets"], 0)
+ else:
+ # Second pass: both workers should have policed equal amounts
+ self.assertGreater(stats1["conform_packets"], 0)
+ self.assertEqual(stats1["exceed_packets"], 0)
+ self.assertGreater(stats1["violate_packets"], 0)
+
+ self.assertGreater(stats0["conform_packets"], 0)
+ self.assertEqual(stats0["exceed_packets"], 0)
+ self.assertGreater(stats0["violate_packets"], 0)
+
+ self.assertEqual(
+ stats0["conform_packets"] + stats0["violate_packets"],
+ stats1["conform_packets"] + stats1["violate_packets"],
+ )
+
+ policer.apply_vpp_config(p.tun_if.sw_if_index, Dir.RX, False)
+ policer.remove_vpp_config()
+
+
+class TestIpsec4TunIfEspHandoff(TemplateIpsec4TunIfEsp, IpsecTun4HandoffTests):
+ """Ipsec ESP 4 Handoff tests"""