6 from asfframework import VppTestRunner
7 from template_ipsec import IPSecIPv4Fwd
8 from template_ipsec import IPSecIPv6Fwd
9 from test_ipsec_esp import TemplateIpsecEsp
10 from template_ipsec import SpdFastPathTemplate
13 def debug_signal_handler(signal, frame):
21 signal.signal(signal.SIGINT, debug_signal_handler)
24 class SpdFastPathInbound(SpdFastPathTemplate):
25 # In test cases derived from this class, packets in IPv4 FWD path
26 # are configured to go through IPSec inbound SPD policy lookup.
27 # Note that order in which the rules are applied is
28 # PROTECT, BYPASS, DISCARD. Therefore BYPASS rules take
29 # precedence over DISCARD.
31 # Override setUpConstants to enable inbound fast path in config
33 def setUpConstants(cls):
34 super(SpdFastPathInbound, cls).setUpConstants()
35 cls.vpp_cmdline.extend(["ipsec", "{", "ipv4-inbound-spd-fast-path on", "}"])
36 cls.logger.info("VPP modified cmdline is %s" % " ".join(cls.vpp_cmdline))
39 class SpdFastPathInboundProtect(TemplateIpsecEsp):
41 def setUpConstants(cls):
42 super(SpdFastPathInboundProtect, cls).setUpConstants()
43 cls.vpp_cmdline.extend(["ipsec", "{", "ipv4-inbound-spd-fast-path on", "}"])
44 cls.logger.info("VPP modified cmdline is %s" % " ".join(cls.vpp_cmdline))
48 super(SpdFastPathInboundProtect, cls).setUpClass()
51 def tearDownClass(cls):
52 super(SpdFastPathInboundProtect, cls).tearDownClass()
55 super(SpdFastPathInboundProtect, self).setUp()
58 self.unconfig_network()
59 super(SpdFastPathInboundProtect, self).tearDown()
62 class SpdFastPathIPv6Inbound(IPSecIPv6Fwd):
63 # In test cases derived from this class, packets in IPvr6 FWD path
64 # are configured to go through IPSec inbound SPD policy lookup.
65 # Note that order in which the rules are applied is
66 # PROTECT, BYPASS, DISCARD. Therefore BYPASS rules take
67 # precedence over DISCARDi.
69 # Override setUpConstants to enable inbound fast path in config
71 def setUpConstants(cls):
72 super(SpdFastPathIPv6Inbound, cls).setUpConstants()
73 cls.vpp_cmdline.extend(["ipsec", "{", "ipv6-inbound-spd-fast-path on", "}"])
74 cls.logger.info("VPP modified cmdline is %s" % " ".join(cls.vpp_cmdline))
77 class SpdFastPathIPv6InboundProtect(TemplateIpsecEsp):
79 def setUpConstants(cls):
80 super(SpdFastPathIPv6InboundProtect, cls).setUpConstants()
81 cls.vpp_cmdline.extend(["ipsec", "{", "ipv6-inbound-spd-fast-path on", "}"])
82 cls.logger.info("VPP modified cmdline is %s" % " ".join(cls.vpp_cmdline))
86 super(SpdFastPathIPv6InboundProtect, cls).setUpClass()
89 def tearDownClass(cls):
90 super(SpdFastPathIPv6InboundProtect, cls).tearDownClass()
93 super(SpdFastPathIPv6InboundProtect, self).setUp()
96 self.unconfig_network()
97 super(SpdFastPathIPv6InboundProtect, self).tearDown()
100 class IPSec4SpdTestCaseBypass(SpdFastPathInbound):
101 """ IPSec/IPv4 inbound: Policy mode test case with fast path \
104 def test_ipsec_spd_inbound_bypass(self):
105 # In this test case, packets in IPv4 FWD path are configured
106 # to go through IPSec inbound SPD policy lookup.
108 # 2 inbound SPD rules (1 HIGH and 1 LOW) are added.
109 # - High priority rule action is set to DISCARD.
110 # - Low priority rule action is set to BYPASS.
112 # Since BYPASS rules take precedence over DISCARD
113 # (the order being PROTECT, BYPASS, DISCARD) we expect the
114 # BYPASS rule to match and traffic to be correctly forwarded.
115 self.create_interfaces(2)
118 self.spd_create_and_intf_add(1, [self.pg1, self.pg0])
121 # bypass rule should take precedence over discard rule,
122 # even though it's lower priority, because for input policies
123 # matching PROTECT policies precedes matching BYPASS policies
124 # which preceeds matching for DISCARD policies.
125 # Any hit stops the process.
126 policy_0 = self.spd_add_rem_policy( # inbound, priority 10
133 policy_type="bypass",
135 local_ip_start=self.pg1.remote_ip4,
136 local_ip_stop=self.pg1.remote_ip4,
137 remote_ip_start=self.pg0.remote_ip4,
138 remote_ip_stop=self.pg0.remote_ip4,
140 policy_1 = self.spd_add_rem_policy( # inbound, priority 15
147 policy_type="discard",
149 local_ip_start=self.pg1.remote_ip4,
150 local_ip_stop=self.pg1.remote_ip4,
151 remote_ip_start=self.pg0.remote_ip4,
152 remote_ip_stop=self.pg0.remote_ip4,
155 # create output rule so we can capture forwarded packets
156 policy_2 = self.spd_add_rem_policy( # outbound, priority 10
163 policy_type="bypass",
166 # create the packet stream
167 packets = self.create_stream(self.pg0, self.pg1, pkt_count)
168 # add the stream to the source interface
169 self.pg0.add_stream(packets)
170 self.pg1.enable_capture()
173 # check capture on pg1
174 capture = self.pg1.get_capture()
175 for packet in capture:
177 self.logger.debug(ppp("SPD Add - Got packet:", packet))
179 self.logger.error(ppp("Unexpected or invalid packet:", packet))
181 self.logger.debug("SPD: Num packets: %s", len(capture.res))
183 # verify captured packets
184 self.verify_capture(self.pg0, self.pg1, capture)
185 # verify all policies matched the expected number of times
186 self.verify_policy_match(pkt_count, policy_0)
187 self.verify_policy_match(0, policy_1)
188 self.verify_policy_match(pkt_count, policy_2)
191 class IPSec4SpdTestCaseDiscard(SpdFastPathInbound):
192 """ IPSec/IPv4 inbound: Policy mode test case with fast path \
195 def test_ipsec_spd_inbound_discard(self):
196 # In this test case, packets in IPv4 FWD path are configured
197 # to go through IPSec inbound SPD policy lookup.
199 # Rule action is set to DISCARD.
201 self.create_interfaces(2)
204 self.spd_create_and_intf_add(1, [self.pg1, self.pg0])
207 # bypass rule should take precedence over discard rule,
208 # even though it's lower priority
209 policy_0 = self.spd_add_rem_policy( # inbound, priority 10
216 policy_type="discard",
219 # create output rule so we can capture forwarded packets
220 policy_1 = self.spd_add_rem_policy( # outbound, priority 10
227 policy_type="bypass",
230 # create the packet stream
231 packets = self.create_stream(self.pg0, self.pg1, pkt_count)
232 # add the stream to the source interface
233 self.pg0.add_stream(packets)
234 self.pg1.enable_capture()
237 # check capture on pg1
238 capture = self.pg1.assert_nothing_captured()
240 # verify all policies matched the expected number of times
241 self.verify_policy_match(pkt_count, policy_0)
242 self.verify_policy_match(0, policy_1)
245 class IPSec4SpdTestCaseProtect(SpdFastPathInboundProtect):
246 """ IPSec/IPv4 inbound: Policy mode test case with fast path \
251 super(IPSec4SpdTestCaseProtect, cls).setUpClass()
254 def tearDownClass(cls):
255 super(IPSec4SpdTestCaseProtect, cls).tearDownClass()
258 super(IPSec4SpdTestCaseProtect, self).setUp()
261 super(IPSec4SpdTestCaseProtect, self).tearDown()
263 def test_ipsec_spd_inbound_protect(self):
264 # In this test case, encrypted packets in IPv4
265 # PROTECT path are configured
266 # to go through IPSec inbound SPD policy lookup.
270 p = self.params[socket.AF_INET]
271 send_pkts = self.gen_encrypt_pkts(
275 src=self.tra_if.remote_ip4,
276 dst=self.tra_if.local_ip4,
278 payload_size=payload_size,
280 recv_pkts = self.send_and_expect(self.tra_if, send_pkts, self.tra_if)
282 self.logger.info(self.vapi.ppcli("show error"))
283 self.logger.info(self.vapi.ppcli("show ipsec all"))
285 pkts = p.tra_sa_in.get_stats()["packets"]
289 "incorrect SA in counts: expected %d != %d" % (pkt_count, pkts),
291 pkts = p.tra_sa_out.get_stats()["packets"]
295 "incorrect SA out counts: expected %d != %d" % (pkt_count, pkts),
297 self.assertEqual(p.tra_sa_out.get_err("lost"), 0)
298 self.assertEqual(p.tra_sa_in.get_err("lost"), 0)
301 class IPSec4SpdTestCaseAddIPRange(SpdFastPathInbound):
302 """ IPSec/IPv4 inbound: Policy mode test case with fast path \
303 (add ips range with any port rule)"""
305 def test_ipsec_spd_inbound_add(self):
306 # In this test case, packets in IPv4 FWD path are configured
307 # to go through IPSec inbound SPD policy lookup.
308 # 2 SPD bypass rules (1 for inbound and 1 for outbound) are added.
309 # Traffic sent on pg0 interface should match fast path priority
310 # rule and should be sent out on pg1 interface.
311 self.create_interfaces(2)
313 s_ip_s1 = ipaddress.ip_address(self.pg0.remote_ip4)
314 s_ip_e1 = ipaddress.ip_address(int(s_ip_s1) + 5)
315 d_ip_s1 = ipaddress.ip_address(self.pg1.remote_ip4)
316 d_ip_e1 = ipaddress.ip_address(int(d_ip_s1) + 0)
318 s_ip_s0 = ipaddress.ip_address(self.pg0.remote_ip4)
319 s_ip_e0 = ipaddress.ip_address(int(s_ip_s0) + 6)
320 d_ip_s0 = ipaddress.ip_address(self.pg1.remote_ip4)
321 d_ip_e0 = ipaddress.ip_address(int(d_ip_s0) + 0)
322 self.spd_create_and_intf_add(1, [self.pg1, self.pg0])
324 policy_0 = self.spd_add_rem_policy( # inbound fast path, priority 10
331 policy_type="bypass",
333 local_ip_start=d_ip_s0,
334 local_ip_stop=d_ip_e0,
335 remote_ip_start=s_ip_s0,
336 remote_ip_stop=s_ip_e0,
338 policy_1 = self.spd_add_rem_policy( # outbound, priority 5
345 policy_type="bypass",
347 local_ip_start=s_ip_s1,
348 local_ip_stop=s_ip_e1,
349 remote_ip_start=d_ip_s1,
350 remote_ip_stop=d_ip_e1,
353 # create the packet stream
354 packets = self.create_stream(self.pg0, self.pg1, pkt_count)
355 # add the stream to the source interface + enable capture
356 self.pg0.add_stream(packets)
357 self.pg0.enable_capture()
358 self.pg1.enable_capture()
359 # start the packet generator
362 capture = self.pg1.get_capture()
363 for packet in capture:
365 self.logger.debug(ppp("SPD - Got packet:", packet))
367 self.logger.error(ppp("Unexpected or invalid packet:", packet))
369 self.logger.debug("SPD: Num packets: %s", len(capture.res))
371 # assert nothing captured on pg0
372 self.pg0.assert_nothing_captured()
373 # verify captured packets
374 self.verify_capture(self.pg0, self.pg1, capture)
375 # verify all policies matched the expected number of times
376 self.verify_policy_match(pkt_count, policy_0)
377 self.verify_policy_match(pkt_count, policy_1)
380 class IPSec4SpdTestCaseAddAll(SpdFastPathInbound):
381 """ IPSec/IPv4 inbound: Policy mode test case with fast path \
382 (add all ips ports rule)"""
384 def test_ipsec_spd_inbound_add(self):
385 # In this test case, packets in IPv4 FWD path are configured
386 # to go through IPSec inbound SPD policy lookup.
387 # 2 SPD rules (1 HIGH and 1 LOW) are added.
388 # Low priority rule action is set to BYPASS all ips.
389 # High priority rule action is set to DISCARD all ips.
390 # Traffic not sent on pg0 interface when HIGH discard priority rule is added.
392 # rule is added and send the same traffic to pg0, this time expect
393 # the traffic is bypassed as bypass takes priority over discard.
394 self.create_interfaces(2)
396 self.spd_create_and_intf_add(1, [self.pg0, self.pg1])
398 policy_0 = self.spd_add_rem_policy( # inbound, priority 20
405 policy_type="discard",
409 policy_1 = self.spd_add_rem_policy( # inbound, priority 20
416 policy_type="bypass",
420 # create the packet stream
421 packets = self.create_stream(self.pg0, self.pg1, pkt_count)
422 # add the stream to the source interface + enable capture
423 self.pg0.add_stream(packets)
424 self.pg0.enable_capture()
425 self.pg1.enable_capture()
426 # start the packet generator
428 # assert nothing captured on pg0 and pg1
429 self.pg0.assert_nothing_captured()
430 self.pg1.assert_nothing_captured()
432 policy_2 = self.spd_add_rem_policy( # inbound, priority 10
439 policy_type="bypass",
443 # create the packet stream
444 packets = self.create_stream(self.pg0, self.pg1, pkt_count)
445 # add the stream to the source interface + enable capture
446 self.pg0.add_stream(packets)
447 self.pg0.enable_capture()
448 self.pg1.enable_capture()
449 # start the packet generator
452 capture = self.pg1.get_capture(expected_count=pkt_count)
453 for packet in capture:
455 self.logger.debug(ppp("SPD - Got packet:", packet))
457 self.logger.error(ppp("Unexpected or invalid packet:", packet))
459 self.logger.debug("SPD: Num packets: %s", len(capture.res))
461 # assert nothing captured on pg0
462 self.pg0.assert_nothing_captured()
463 # verify all policies matched the expected number of times
464 self.verify_policy_match(pkt_count, policy_2)
467 class IPSec4SpdTestCaseRemove(SpdFastPathInbound):
468 """ IPSec/IPv4 inbound: Policy mode test case with fast path \
471 def test_ipsec_spd_inbound_remove(self):
472 # In this test case, packets in IPv4 FWD path are configured
473 # to go through IPSec inbound SPD policy lookup.
474 # 2 SPD rules (1 HIGH and 1 LOW) are added.
475 # High priority rule action is set to BYPASS.
476 # Low priority rule action is set to DISCARD.
477 # High priority rule is then removed.
478 # Traffic sent on pg0 interface should match low priority
479 # rule and should be discarded after SPD lookup.
480 self.create_interfaces(2)
482 self.spd_create_and_intf_add(1, [self.pg0, self.pg1])
483 policy_0 = self.spd_add_rem_policy( # inbound, priority 10
490 policy_type="bypass",
492 policy_1 = self.spd_add_rem_policy( # inbound, priority 5
499 policy_type="discard",
502 policy_out = self.spd_add_rem_policy( # outbound, priority 10
509 policy_type="bypass",
512 # create the packet stream
513 packets = self.create_stream(self.pg0, self.pg1, pkt_count)
514 # add the stream to the source interface + enable capture
515 self.pg0.add_stream(packets)
516 self.pg0.enable_capture()
517 self.pg1.enable_capture()
518 # start the packet generator
521 capture = self.pg1.get_capture()
522 for packet in capture:
524 self.logger.debug(ppp("SPD - Got packet:", packet))
526 self.logger.error(ppp("Unexpected or invalid packet:", packet))
529 # assert nothing captured on pg0
530 self.pg0.assert_nothing_captured()
531 # verify capture on pg1
532 self.logger.debug("SPD: Num packets: %s", len(capture.res))
533 self.verify_capture(self.pg0, self.pg1, capture)
534 # verify all policies matched the expected number of times
535 self.verify_policy_match(pkt_count, policy_0)
536 self.verify_policy_match(0, policy_1)
537 # now remove the bypass rule
538 self.spd_add_rem_policy( # outbound, priority 10
545 policy_type="bypass",
549 # resend the same packets
550 self.pg0.add_stream(packets)
551 self.pg0.enable_capture() # flush the old captures
552 self.pg1.enable_capture()
554 # assert nothing captured on pg0
555 self.pg0.assert_nothing_captured()
556 # all packets will be dropped by SPD rule
557 self.pg1.assert_nothing_captured()
558 # verify all policies matched the expected number of times
559 self.verify_policy_match(pkt_count, policy_0)
560 self.verify_policy_match(pkt_count, policy_1)
563 class IPSec4SpdTestCaseReadd(SpdFastPathInbound):
564 """ IPSec/IPv4 inbound: Policy mode test case with fast path \
565 (add, remove, re-add)"""
567 def test_ipsec_spd_inbound_readd(self):
568 # In this test case, packets in IPv4 FWD path are configured
569 # to go through IPSec outbound SPD policy lookup.
570 # 2 SPD rules (1 HIGH and 1 LOW) are added.
571 # High priority rule action is set to BYPASS.
572 # Low priority rule action is set to DISCARD.
573 # Traffic sent on pg0 interface should match high priority
574 # rule and should be sent out on pg1 interface.
575 # High priority rule is then removed.
576 # Traffic sent on pg0 interface should match low priority
577 # rule and should be discarded after SPD lookup.
578 # Readd high priority rule.
579 # Traffic sent on pg0 interface should match high priority
580 # rule and should be sent out on pg1 interface.
581 self.create_interfaces(2)
583 self.spd_create_and_intf_add(1, [self.pg0, self.pg1])
584 policy_0 = self.spd_add_rem_policy( # inbound, priority 10
591 policy_type="bypass",
593 policy_1 = self.spd_add_rem_policy( # inbound, priority 5
600 policy_type="discard",
602 policy_2 = self.spd_add_rem_policy( # outbound, priority 10
609 policy_type="bypass",
612 # create the packet stream
613 packets = self.create_stream(self.pg0, self.pg1, pkt_count)
614 # add the stream to the source interface + enable capture
615 self.pg0.add_stream(packets)
616 self.pg0.enable_capture()
617 self.pg1.enable_capture()
618 # start the packet generator
621 capture = self.pg1.get_capture()
622 for packet in capture:
624 self.logger.debug(ppp("SPD - Got packet:", packet))
626 self.logger.error(ppp("Unexpected or invalid packet:", packet))
628 self.logger.debug("SPD: Num packets: %s", len(capture.res))
630 # assert nothing captured on pg0
631 self.pg0.assert_nothing_captured()
632 # verify capture on pg1
633 self.verify_capture(self.pg0, self.pg1, capture)
634 # verify all policies matched the expected number of times
635 self.verify_policy_match(pkt_count, policy_0)
636 self.verify_policy_match(0, policy_1)
637 # remove the bypass rule, leaving only the discard rule
638 self.spd_add_rem_policy( # inbound, priority 10
645 policy_type="bypass",
649 # resend the same packets
650 self.pg0.add_stream(packets)
651 self.pg0.enable_capture() # flush the old captures
652 self.pg1.enable_capture()
655 # assert nothing captured on pg0
656 self.pg0.assert_nothing_captured()
657 # all packets will be dropped by SPD rule
658 self.pg1.assert_nothing_captured()
659 # verify all policies matched the expected number of times
660 self.verify_policy_match(pkt_count, policy_0)
661 self.verify_policy_match(pkt_count, policy_1)
663 # now readd the bypass rule
664 policy_0 = self.spd_add_rem_policy( # outbound, priority 10
671 policy_type="bypass",
674 # resend the same packets
675 self.pg0.add_stream(packets)
676 self.pg0.enable_capture() # flush the old captures
677 self.pg1.enable_capture()
681 capture = self.pg1.get_capture(pkt_count)
682 for packet in capture:
684 self.logger.debug(ppp("SPD - Got packet:", packet))
686 self.logger.error(ppp("Unexpected or invalid packet:", packet))
688 self.logger.debug("SPD: Num packets: %s", len(capture.res))
690 # assert nothing captured on pg0
691 self.pg0.assert_nothing_captured()
692 # verify captured packets
693 self.verify_capture(self.pg0, self.pg1, capture)
694 # verify all policies matched the expected number of times
695 self.verify_policy_match(pkt_count, policy_0)
696 self.verify_policy_match(pkt_count, policy_1)
699 class IPSec4SpdTestCaseMultiple(SpdFastPathInbound):
700 """ IPSec/IPv4 inbound: Policy mode test case with fast path \
701 (multiple interfaces, multiple rules)"""
703 def test_ipsec_spd_inbound_multiple(self):
704 # In this test case, packets in IPv4 FWD path are configured to go
705 # through IPSec outbound SPD policy lookup.
706 # Multiples rules on multiple interfaces are tested at the same time.
707 # 3x interfaces are configured, binding the same SPD to each.
708 # Each interface has 2 SPD rules (1 BYPASS and 1 DISCARD).
709 # On pg0 & pg1, the BYPASS rule is HIGH priority
710 # On pg2, the DISCARD rule is HIGH priority
711 # Traffic should be received on pg0 & pg1 and dropped on pg2.
712 self.create_interfaces(3)
714 # bind SPD to all interfaces
715 self.spd_create_and_intf_add(1, self.pg_interfaces)
716 # add rules on all interfaces
717 policy_01 = self.spd_add_rem_policy( # inbound, priority 10
724 policy_type="bypass",
726 policy_02 = self.spd_add_rem_policy( # inbound, priority 5
733 policy_type="discard",
736 policy_11 = self.spd_add_rem_policy( # inbound, priority 10
743 policy_type="bypass",
745 policy_12 = self.spd_add_rem_policy( # inbound, priority 5
752 policy_type="discard",
755 policy_21 = self.spd_add_rem_policy( # inbound, priority 5
762 policy_type="bypass",
764 policy_22 = self.spd_add_rem_policy( # inbound, priority 10
771 policy_type="discard",
774 # interfaces bound to an SPD, will by default drop outbound
775 # traffic with no matching policies. add catch-all outbound
776 # bypass rule to SPD:
777 self.spd_add_rem_policy( # outbound, all interfaces
784 policy_type="bypass",
788 # create the packet streams
789 packets0 = self.create_stream(self.pg0, self.pg1, pkt_count)
790 packets1 = self.create_stream(self.pg1, self.pg2, pkt_count)
791 packets2 = self.create_stream(self.pg2, self.pg0, pkt_count)
792 # add the streams to the source interfaces
793 self.pg0.add_stream(packets0)
794 self.pg1.add_stream(packets1)
795 self.pg2.add_stream(packets2)
796 # enable capture on all interfaces
797 for pg in self.pg_interfaces:
799 # start the packet generator
804 for pg in [self.pg1, self.pg2]: # we are expecting captures on pg1/pg2
805 if_caps.append(pg.get_capture())
806 for packet in if_caps[-1]:
808 self.logger.debug(ppp("SPD - Got packet:", packet))
810 self.logger.error(ppp("Unexpected or invalid packet:", packet))
812 self.logger.debug("SPD: Num packets: %s", len(if_caps[0].res))
813 self.logger.debug("SPD: Num packets: %s", len(if_caps[1].res))
815 # verify captures that matched BYPASS rule
816 self.verify_capture(self.pg0, self.pg1, if_caps[0])
817 self.verify_capture(self.pg1, self.pg2, if_caps[1])
818 # verify that traffic to pg0 matched BYPASS rule
819 # although DISCARD rule had higher prioriy and was not dropped
820 self.verify_policy_match(pkt_count, policy_21)
822 # verify all packets that were expected to match rules, matched
824 self.verify_policy_match(pkt_count, policy_01)
825 self.verify_policy_match(0, policy_02)
827 self.verify_policy_match(pkt_count, policy_11)
828 self.verify_policy_match(0, policy_12)
830 self.verify_policy_match(0, policy_22)
833 class IPSec6SpdTestCaseProtect(SpdFastPathIPv6InboundProtect):
834 """ IPSec/IPv6 inbound: Policy mode test case with fast path \
839 super(IPSec6SpdTestCaseProtect, cls).setUpClass()
842 def tearDownClass(cls):
843 super(IPSec6SpdTestCaseProtect, cls).tearDownClass()
846 super(IPSec6SpdTestCaseProtect, self).setUp()
849 super(IPSec6SpdTestCaseProtect, self).tearDown()
851 def test_ipsec6_spd_inbound_protect(self):
854 p = self.params[socket.AF_INET6]
855 send_pkts = self.gen_encrypt_pkts6(
859 src=self.tra_if.remote_ip6,
860 dst=self.tra_if.local_ip6,
862 payload_size=payload_size,
864 recv_pkts = self.send_and_expect(self.tra_if, send_pkts, self.tra_if)
866 self.logger.info(self.vapi.ppcli("show error"))
867 self.logger.info(self.vapi.ppcli("show ipsec all"))
868 pkts = p.tra_sa_in.get_stats()["packets"]
872 "incorrect SA in counts: expected %d != %d" % (pkt_count, pkts),
874 pkts = p.tra_sa_out.get_stats()["packets"]
878 "incorrect SA out counts: expected %d != %d" % (pkt_count, pkts),
880 self.assertEqual(p.tra_sa_out.get_err("lost"), 0)
881 self.assertEqual(p.tra_sa_in.get_err("lost"), 0)
884 if __name__ == "__main__":
885 unittest.main(testRunner=VppTestRunner)