6 from framework import VppTestRunner
7 from template_ipsec import IPSecIPv4Fwd
8 from template_ipsec import IPSecIPv6Fwd
9 from test_ipsec_esp import TemplateIpsecEsp
12 def debug_signal_handler(signal, frame):
20 signal.signal(signal.SIGINT, debug_signal_handler)
23 class SpdFastPathInbound(IPSecIPv4Fwd):
24 # In test cases derived from this class, packets in IPv4 FWD path
25 # are configured to go through IPSec inbound SPD policy lookup.
26 # Note that order in which the rules are applied is
27 # PROTECT, BYPASS, DISCARD. Therefore BYPASS rules take
28 # precedence over DISCARD.
30 # Override setUpConstants to enable inbound fast path in config
32 def setUpConstants(cls):
33 super(SpdFastPathInbound, cls).setUpConstants()
34 cls.vpp_cmdline.extend(["ipsec", "{", "ipv4-inbound-spd-fast-path on", "}"])
35 cls.logger.info("VPP modified cmdline is %s" % " ".join(cls.vpp_cmdline))
38 class SpdFastPathInboundProtect(TemplateIpsecEsp):
40 def setUpConstants(cls):
41 super(SpdFastPathInboundProtect, cls).setUpConstants()
42 cls.vpp_cmdline.extend(["ipsec", "{", "ipv4-inbound-spd-fast-path on", "}"])
43 cls.logger.info("VPP modified cmdline is %s" % " ".join(cls.vpp_cmdline))
47 super(SpdFastPathInboundProtect, cls).setUpClass()
50 def tearDownClass(cls):
51 super(SpdFastPathInboundProtect, cls).tearDownClass()
54 super(SpdFastPathInboundProtect, self).setUp()
57 self.unconfig_network()
58 super(SpdFastPathInboundProtect, self).tearDown()
61 class SpdFastPathIPv6Inbound(IPSecIPv6Fwd):
62 # In test cases derived from this class, packets in IPvr6 FWD path
63 # are configured to go through IPSec inbound SPD policy lookup.
64 # Note that order in which the rules are applied is
65 # PROTECT, BYPASS, DISCARD. Therefore BYPASS rules take
66 # precedence over DISCARDi.
68 # Override setUpConstants to enable inbound fast path in config
70 def setUpConstants(cls):
71 super(SpdFastPathIPv6Inbound, cls).setUpConstants()
72 cls.vpp_cmdline.extend(["ipsec", "{", "ipv6-inbound-spd-fast-path on", "}"])
73 cls.logger.info("VPP modified cmdline is %s" % " ".join(cls.vpp_cmdline))
76 class SpdFastPathIPv6InboundProtect(TemplateIpsecEsp):
78 def setUpConstants(cls):
79 super(SpdFastPathIPv6InboundProtect, cls).setUpConstants()
80 cls.vpp_cmdline.extend(["ipsec", "{", "ipv6-inbound-spd-fast-path on", "}"])
81 cls.logger.info("VPP modified cmdline is %s" % " ".join(cls.vpp_cmdline))
85 super(SpdFastPathIPv6InboundProtect, cls).setUpClass()
88 def tearDownClass(cls):
89 super(SpdFastPathIPv6InboundProtect, cls).tearDownClass()
92 super(SpdFastPathIPv6InboundProtect, self).setUp()
95 self.unconfig_network()
96 super(SpdFastPathIPv6InboundProtect, self).tearDown()
99 class IPSec4SpdTestCaseBypass(SpdFastPathInbound):
100 """ IPSec/IPv4 inbound: Policy mode test case with fast path \
103 def test_ipsec_spd_inbound_bypass(self):
104 # In this test case, packets in IPv4 FWD path are configured
105 # to go through IPSec inbound SPD policy lookup.
107 # 2 inbound SPD rules (1 HIGH and 1 LOW) are added.
108 # - High priority rule action is set to DISCARD.
109 # - Low priority rule action is set to BYPASS.
111 # Since BYPASS rules take precedence over DISCARD
112 # (the order being PROTECT, BYPASS, DISCARD) we expect the
113 # BYPASS rule to match and traffic to be correctly forwarded.
114 self.create_interfaces(2)
117 self.spd_create_and_intf_add(1, [self.pg1, self.pg0])
120 # bypass rule should take precedence over discard rule,
121 # even though it's lower priority, because for input policies
122 # matching PROTECT policies precedes matching BYPASS policies
123 # which preceeds matching for DISCARD policies.
124 # Any hit stops the process.
125 policy_0 = self.spd_add_rem_policy( # inbound, priority 10
132 policy_type="bypass",
134 local_ip_start=self.pg1.remote_ip4,
135 local_ip_stop=self.pg1.remote_ip4,
136 remote_ip_start=self.pg0.remote_ip4,
137 remote_ip_stop=self.pg0.remote_ip4,
139 policy_1 = self.spd_add_rem_policy( # inbound, priority 15
146 policy_type="discard",
148 local_ip_start=self.pg1.remote_ip4,
149 local_ip_stop=self.pg1.remote_ip4,
150 remote_ip_start=self.pg0.remote_ip4,
151 remote_ip_stop=self.pg0.remote_ip4,
154 # create output rule so we can capture forwarded packets
155 policy_2 = self.spd_add_rem_policy( # outbound, priority 10
162 policy_type="bypass",
165 # create the packet stream
166 packets = self.create_stream(self.pg0, self.pg1, pkt_count)
167 # add the stream to the source interface
168 self.pg0.add_stream(packets)
169 self.pg1.enable_capture()
172 # check capture on pg1
173 capture = self.pg1.get_capture()
174 for packet in capture:
176 self.logger.debug(ppp("SPD Add - Got packet:", packet))
178 self.logger.error(ppp("Unexpected or invalid packet:", packet))
180 self.logger.debug("SPD: Num packets: %s", len(capture.res))
182 # verify captured packets
183 self.verify_capture(self.pg0, self.pg1, capture)
184 # verify all policies matched the expected number of times
185 self.verify_policy_match(pkt_count, policy_0)
186 self.verify_policy_match(0, policy_1)
187 self.verify_policy_match(pkt_count, policy_2)
190 class IPSec4SpdTestCaseDiscard(SpdFastPathInbound):
191 """ IPSec/IPv4 inbound: Policy mode test case with fast path \
194 def test_ipsec_spd_inbound_discard(self):
195 # In this test case, packets in IPv4 FWD path are configured
196 # to go through IPSec inbound SPD policy lookup.
198 # Rule action is set to DISCARD.
200 self.create_interfaces(2)
203 self.spd_create_and_intf_add(1, [self.pg1, self.pg0])
206 # bypass rule should take precedence over discard rule,
207 # even though it's lower priority
208 policy_0 = self.spd_add_rem_policy( # inbound, priority 10
215 policy_type="discard",
218 # create output rule so we can capture forwarded packets
219 policy_1 = self.spd_add_rem_policy( # outbound, priority 10
226 policy_type="bypass",
229 # create the packet stream
230 packets = self.create_stream(self.pg0, self.pg1, pkt_count)
231 # add the stream to the source interface
232 self.pg0.add_stream(packets)
233 self.pg1.enable_capture()
236 # check capture on pg1
237 capture = self.pg1.assert_nothing_captured()
239 # verify all policies matched the expected number of times
240 self.verify_policy_match(pkt_count, policy_0)
241 self.verify_policy_match(0, policy_1)
244 class IPSec4SpdTestCaseProtect(SpdFastPathInboundProtect):
245 """ IPSec/IPv4 inbound: Policy mode test case with fast path \
250 super(IPSec4SpdTestCaseProtect, cls).setUpClass()
253 def tearDownClass(cls):
254 super(IPSec4SpdTestCaseProtect, cls).tearDownClass()
257 super(IPSec4SpdTestCaseProtect, self).setUp()
260 super(IPSec4SpdTestCaseProtect, self).tearDown()
262 def test_ipsec_spd_inbound_protect(self):
263 # In this test case, encrypted packets in IPv4
264 # PROTECT path are configured
265 # to go through IPSec inbound SPD policy lookup.
269 p = self.params[socket.AF_INET]
270 send_pkts = self.gen_encrypt_pkts(
274 src=self.tra_if.remote_ip4,
275 dst=self.tra_if.local_ip4,
277 payload_size=payload_size,
279 recv_pkts = self.send_and_expect(self.tra_if, send_pkts, self.tra_if)
281 self.logger.info(self.vapi.ppcli("show error"))
282 self.logger.info(self.vapi.ppcli("show ipsec all"))
284 pkts = p.tra_sa_in.get_stats()["packets"]
288 "incorrect SA in counts: expected %d != %d" % (pkt_count, pkts),
290 pkts = p.tra_sa_out.get_stats()["packets"]
294 "incorrect SA out counts: expected %d != %d" % (pkt_count, pkts),
296 self.assertEqual(p.tra_sa_out.get_err("lost"), 0)
297 self.assertEqual(p.tra_sa_in.get_err("lost"), 0)
300 class IPSec4SpdTestCaseAddIPRange(SpdFastPathInbound):
301 """ IPSec/IPv4 inbound: Policy mode test case with fast path \
302 (add ips range with any port rule)"""
304 def test_ipsec_spd_inbound_add(self):
305 # In this test case, packets in IPv4 FWD path are configured
306 # to go through IPSec inbound SPD policy lookup.
307 # 2 SPD bypass rules (1 for inbound and 1 for outbound) are added.
308 # Traffic sent on pg0 interface should match fast path priority
309 # rule and should be sent out on pg1 interface.
310 self.create_interfaces(2)
312 s_ip_s1 = ipaddress.ip_address(self.pg0.remote_ip4)
313 s_ip_e1 = ipaddress.ip_address(int(s_ip_s1) + 5)
314 d_ip_s1 = ipaddress.ip_address(self.pg1.remote_ip4)
315 d_ip_e1 = ipaddress.ip_address(int(d_ip_s1) + 0)
317 s_ip_s0 = ipaddress.ip_address(self.pg0.remote_ip4)
318 s_ip_e0 = ipaddress.ip_address(int(s_ip_s0) + 6)
319 d_ip_s0 = ipaddress.ip_address(self.pg1.remote_ip4)
320 d_ip_e0 = ipaddress.ip_address(int(d_ip_s0) + 0)
321 self.spd_create_and_intf_add(1, [self.pg1, self.pg0])
323 policy_0 = self.spd_add_rem_policy( # inbound fast path, priority 10
330 policy_type="bypass",
332 local_ip_start=d_ip_s0,
333 local_ip_stop=d_ip_e0,
334 remote_ip_start=s_ip_s0,
335 remote_ip_stop=s_ip_e0,
337 policy_1 = self.spd_add_rem_policy( # outbound, priority 5
344 policy_type="bypass",
346 local_ip_start=s_ip_s1,
347 local_ip_stop=s_ip_e1,
348 remote_ip_start=d_ip_s1,
349 remote_ip_stop=d_ip_e1,
352 # create the packet stream
353 packets = self.create_stream(self.pg0, self.pg1, pkt_count)
354 # add the stream to the source interface + enable capture
355 self.pg0.add_stream(packets)
356 self.pg0.enable_capture()
357 self.pg1.enable_capture()
358 # start the packet generator
361 capture = self.pg1.get_capture()
362 for packet in capture:
364 self.logger.debug(ppp("SPD - Got packet:", packet))
366 self.logger.error(ppp("Unexpected or invalid packet:", packet))
368 self.logger.debug("SPD: Num packets: %s", len(capture.res))
370 # assert nothing captured on pg0
371 self.pg0.assert_nothing_captured()
372 # verify captured packets
373 self.verify_capture(self.pg0, self.pg1, capture)
374 # verify all policies matched the expected number of times
375 self.verify_policy_match(pkt_count, policy_0)
376 self.verify_policy_match(pkt_count, policy_1)
379 class IPSec4SpdTestCaseAddAll(SpdFastPathInbound):
380 """ IPSec/IPv4 inbound: Policy mode test case with fast path \
381 (add all ips ports rule)"""
383 def test_ipsec_spd_inbound_add(self):
384 # In this test case, packets in IPv4 FWD path are configured
385 # to go through IPSec inbound SPD policy lookup.
386 # 2 SPD rules (1 HIGH and 1 LOW) are added.
387 # Low priority rule action is set to BYPASS all ips.
388 # High priority rule action is set to DISCARD all ips.
389 # Traffic not sent on pg0 interface when HIGH discard priority rule is added.
391 # rule is added and send the same traffic to pg0, this time expect
392 # the traffic is bypassed as bypass takes priority over discard.
393 self.create_interfaces(2)
395 self.spd_create_and_intf_add(1, [self.pg0, self.pg1])
397 policy_0 = self.spd_add_rem_policy( # inbound, priority 20
404 policy_type="discard",
408 policy_1 = self.spd_add_rem_policy( # inbound, priority 20
415 policy_type="bypass",
419 # create the packet stream
420 packets = self.create_stream(self.pg0, self.pg1, pkt_count)
421 # add the stream to the source interface + enable capture
422 self.pg0.add_stream(packets)
423 self.pg0.enable_capture()
424 self.pg1.enable_capture()
425 # start the packet generator
427 # assert nothing captured on pg0 and pg1
428 self.pg0.assert_nothing_captured()
429 self.pg1.assert_nothing_captured()
431 policy_2 = self.spd_add_rem_policy( # inbound, priority 10
438 policy_type="bypass",
442 # create the packet stream
443 packets = self.create_stream(self.pg0, self.pg1, pkt_count)
444 # add the stream to the source interface + enable capture
445 self.pg0.add_stream(packets)
446 self.pg0.enable_capture()
447 self.pg1.enable_capture()
448 # start the packet generator
451 capture = self.pg1.get_capture(expected_count=pkt_count)
452 for packet in capture:
454 self.logger.debug(ppp("SPD - Got packet:", packet))
456 self.logger.error(ppp("Unexpected or invalid packet:", packet))
458 self.logger.debug("SPD: Num packets: %s", len(capture.res))
460 # assert nothing captured on pg0
461 self.pg0.assert_nothing_captured()
462 # verify all policies matched the expected number of times
463 self.verify_policy_match(pkt_count, policy_2)
466 class IPSec4SpdTestCaseRemove(SpdFastPathInbound):
467 """ IPSec/IPv4 inbound: Policy mode test case with fast path \
470 def test_ipsec_spd_inbound_remove(self):
471 # In this test case, packets in IPv4 FWD path are configured
472 # to go through IPSec inbound SPD policy lookup.
473 # 2 SPD rules (1 HIGH and 1 LOW) are added.
474 # High priority rule action is set to BYPASS.
475 # Low priority rule action is set to DISCARD.
476 # High priority rule is then removed.
477 # Traffic sent on pg0 interface should match low priority
478 # rule and should be discarded after SPD lookup.
479 self.create_interfaces(2)
481 self.spd_create_and_intf_add(1, [self.pg0, self.pg1])
482 policy_0 = self.spd_add_rem_policy( # inbound, priority 10
489 policy_type="bypass",
491 policy_1 = self.spd_add_rem_policy( # inbound, priority 5
498 policy_type="discard",
501 policy_out = self.spd_add_rem_policy( # outbound, priority 10
508 policy_type="bypass",
511 # create the packet stream
512 packets = self.create_stream(self.pg0, self.pg1, pkt_count)
513 # add the stream to the source interface + enable capture
514 self.pg0.add_stream(packets)
515 self.pg0.enable_capture()
516 self.pg1.enable_capture()
517 # start the packet generator
520 capture = self.pg1.get_capture()
521 for packet in capture:
523 self.logger.debug(ppp("SPD - Got packet:", packet))
525 self.logger.error(ppp("Unexpected or invalid packet:", packet))
528 # assert nothing captured on pg0
529 self.pg0.assert_nothing_captured()
530 # verify capture on pg1
531 self.logger.debug("SPD: Num packets: %s", len(capture.res))
532 self.verify_capture(self.pg0, self.pg1, capture)
533 # verify all policies matched the expected number of times
534 self.verify_policy_match(pkt_count, policy_0)
535 self.verify_policy_match(0, policy_1)
536 # now remove the bypass rule
537 self.spd_add_rem_policy( # outbound, priority 10
544 policy_type="bypass",
548 # resend the same packets
549 self.pg0.add_stream(packets)
550 self.pg0.enable_capture() # flush the old captures
551 self.pg1.enable_capture()
553 # assert nothing captured on pg0
554 self.pg0.assert_nothing_captured()
555 # all packets will be dropped by SPD rule
556 self.pg1.assert_nothing_captured()
557 # verify all policies matched the expected number of times
558 self.verify_policy_match(pkt_count, policy_0)
559 self.verify_policy_match(pkt_count, policy_1)
562 class IPSec4SpdTestCaseReadd(SpdFastPathInbound):
563 """ IPSec/IPv4 inbound: Policy mode test case with fast path \
564 (add, remove, re-add)"""
566 def test_ipsec_spd_inbound_readd(self):
567 # In this test case, packets in IPv4 FWD path are configured
568 # to go through IPSec outbound SPD policy lookup.
569 # 2 SPD rules (1 HIGH and 1 LOW) are added.
570 # High priority rule action is set to BYPASS.
571 # Low priority rule action is set to DISCARD.
572 # Traffic sent on pg0 interface should match high priority
573 # rule and should be sent out on pg1 interface.
574 # High priority rule is then removed.
575 # Traffic sent on pg0 interface should match low priority
576 # rule and should be discarded after SPD lookup.
577 # Readd high priority rule.
578 # Traffic sent on pg0 interface should match high priority
579 # rule and should be sent out on pg1 interface.
580 self.create_interfaces(2)
582 self.spd_create_and_intf_add(1, [self.pg0, self.pg1])
583 policy_0 = self.spd_add_rem_policy( # inbound, priority 10
590 policy_type="bypass",
592 policy_1 = self.spd_add_rem_policy( # inbound, priority 5
599 policy_type="discard",
601 policy_2 = self.spd_add_rem_policy( # outbound, priority 10
608 policy_type="bypass",
611 # create the packet stream
612 packets = self.create_stream(self.pg0, self.pg1, pkt_count)
613 # add the stream to the source interface + enable capture
614 self.pg0.add_stream(packets)
615 self.pg0.enable_capture()
616 self.pg1.enable_capture()
617 # start the packet generator
620 capture = self.pg1.get_capture()
621 for packet in capture:
623 self.logger.debug(ppp("SPD - Got packet:", packet))
625 self.logger.error(ppp("Unexpected or invalid packet:", packet))
627 self.logger.debug("SPD: Num packets: %s", len(capture.res))
629 # assert nothing captured on pg0
630 self.pg0.assert_nothing_captured()
631 # verify capture on pg1
632 self.verify_capture(self.pg0, self.pg1, capture)
633 # verify all policies matched the expected number of times
634 self.verify_policy_match(pkt_count, policy_0)
635 self.verify_policy_match(0, policy_1)
636 # remove the bypass rule, leaving only the discard rule
637 self.spd_add_rem_policy( # inbound, priority 10
644 policy_type="bypass",
648 # resend the same packets
649 self.pg0.add_stream(packets)
650 self.pg0.enable_capture() # flush the old captures
651 self.pg1.enable_capture()
654 # assert nothing captured on pg0
655 self.pg0.assert_nothing_captured()
656 # all packets will be dropped by SPD rule
657 self.pg1.assert_nothing_captured()
658 # verify all policies matched the expected number of times
659 self.verify_policy_match(pkt_count, policy_0)
660 self.verify_policy_match(pkt_count, policy_1)
662 # now readd the bypass rule
663 policy_0 = self.spd_add_rem_policy( # outbound, priority 10
670 policy_type="bypass",
673 # resend the same packets
674 self.pg0.add_stream(packets)
675 self.pg0.enable_capture() # flush the old captures
676 self.pg1.enable_capture()
680 capture = self.pg1.get_capture(pkt_count)
681 for packet in capture:
683 self.logger.debug(ppp("SPD - Got packet:", packet))
685 self.logger.error(ppp("Unexpected or invalid packet:", packet))
687 self.logger.debug("SPD: Num packets: %s", len(capture.res))
689 # assert nothing captured on pg0
690 self.pg0.assert_nothing_captured()
691 # verify captured packets
692 self.verify_capture(self.pg0, self.pg1, capture)
693 # verify all policies matched the expected number of times
694 self.verify_policy_match(pkt_count, policy_0)
695 self.verify_policy_match(pkt_count, policy_1)
698 class IPSec4SpdTestCaseMultiple(SpdFastPathInbound):
699 """ IPSec/IPv4 inbound: Policy mode test case with fast path \
700 (multiple interfaces, multiple rules)"""
702 def test_ipsec_spd_inbound_multiple(self):
703 # In this test case, packets in IPv4 FWD path are configured to go
704 # through IPSec outbound SPD policy lookup.
705 # Multiples rules on multiple interfaces are tested at the same time.
706 # 3x interfaces are configured, binding the same SPD to each.
707 # Each interface has 2 SPD rules (1 BYPASS and 1 DISCARD).
708 # On pg0 & pg1, the BYPASS rule is HIGH priority
709 # On pg2, the DISCARD rule is HIGH priority
710 # Traffic should be received on pg0 & pg1 and dropped on pg2.
711 self.create_interfaces(3)
713 # bind SPD to all interfaces
714 self.spd_create_and_intf_add(1, self.pg_interfaces)
715 # add rules on all interfaces
716 policy_01 = self.spd_add_rem_policy( # inbound, priority 10
723 policy_type="bypass",
725 policy_02 = self.spd_add_rem_policy( # inbound, priority 5
732 policy_type="discard",
735 policy_11 = self.spd_add_rem_policy( # inbound, priority 10
742 policy_type="bypass",
744 policy_12 = self.spd_add_rem_policy( # inbound, priority 5
751 policy_type="discard",
754 policy_21 = self.spd_add_rem_policy( # inbound, priority 5
761 policy_type="bypass",
763 policy_22 = self.spd_add_rem_policy( # inbound, priority 10
770 policy_type="discard",
773 # interfaces bound to an SPD, will by default drop outbound
774 # traffic with no matching policies. add catch-all outbound
775 # bypass rule to SPD:
776 self.spd_add_rem_policy( # outbound, all interfaces
783 policy_type="bypass",
787 # create the packet streams
788 packets0 = self.create_stream(self.pg0, self.pg1, pkt_count)
789 packets1 = self.create_stream(self.pg1, self.pg2, pkt_count)
790 packets2 = self.create_stream(self.pg2, self.pg0, pkt_count)
791 # add the streams to the source interfaces
792 self.pg0.add_stream(packets0)
793 self.pg1.add_stream(packets1)
794 self.pg2.add_stream(packets2)
795 # enable capture on all interfaces
796 for pg in self.pg_interfaces:
798 # start the packet generator
803 for pg in [self.pg1, self.pg2]: # we are expecting captures on pg1/pg2
804 if_caps.append(pg.get_capture())
805 for packet in if_caps[-1]:
807 self.logger.debug(ppp("SPD - Got packet:", packet))
809 self.logger.error(ppp("Unexpected or invalid packet:", packet))
811 self.logger.debug("SPD: Num packets: %s", len(if_caps[0].res))
812 self.logger.debug("SPD: Num packets: %s", len(if_caps[1].res))
814 # verify captures that matched BYPASS rule
815 self.verify_capture(self.pg0, self.pg1, if_caps[0])
816 self.verify_capture(self.pg1, self.pg2, if_caps[1])
817 # verify that traffic to pg0 matched BYPASS rule
818 # although DISCARD rule had higher prioriy and was not dropped
819 self.verify_policy_match(pkt_count, policy_21)
821 # verify all packets that were expected to match rules, matched
823 self.verify_policy_match(pkt_count, policy_01)
824 self.verify_policy_match(0, policy_02)
826 self.verify_policy_match(pkt_count, policy_11)
827 self.verify_policy_match(0, policy_12)
829 self.verify_policy_match(0, policy_22)
832 class IPSec6SpdTestCaseProtect(SpdFastPathIPv6InboundProtect):
833 """ IPSec/IPv6 inbound: Policy mode test case with fast path \
838 super(IPSec6SpdTestCaseProtect, cls).setUpClass()
841 def tearDownClass(cls):
842 super(IPSec6SpdTestCaseProtect, cls).tearDownClass()
845 super(IPSec6SpdTestCaseProtect, self).setUp()
848 super(IPSec6SpdTestCaseProtect, self).tearDown()
850 def test_ipsec6_spd_inbound_protect(self):
853 p = self.params[socket.AF_INET6]
854 send_pkts = self.gen_encrypt_pkts6(
858 src=self.tra_if.remote_ip6,
859 dst=self.tra_if.local_ip6,
861 payload_size=payload_size,
863 recv_pkts = self.send_and_expect(self.tra_if, send_pkts, self.tra_if)
865 self.logger.info(self.vapi.ppcli("show error"))
866 self.logger.info(self.vapi.ppcli("show ipsec all"))
867 pkts = p.tra_sa_in.get_stats()["packets"]
871 "incorrect SA in counts: expected %d != %d" % (pkt_count, pkts),
873 pkts = p.tra_sa_out.get_stats()["packets"]
877 "incorrect SA out counts: expected %d != %d" % (pkt_count, pkts),
879 self.assertEqual(p.tra_sa_out.get_err("lost"), 0)
880 self.assertEqual(p.tra_sa_in.get_err("lost"), 0)
883 if __name__ == "__main__":
884 unittest.main(testRunner=VppTestRunner)