5 from socket import AF_INET6
7 from framework import VppTestCase, VppTestRunner
8 from vpp_ip_route import VppIpRoute, VppRoutePath, FibPathProto, VppIpTable
9 from vpp_srv6 import SRv6LocalSIDBehaviors, VppSRv6LocalSID, VppSRv6Policy, \
10 SRv6PolicyType, VppSRv6Steering, SRv6PolicySteeringTypes
13 from scapy.packet import Raw
14 from scapy.layers.l2 import Ether, Dot1Q
15 from scapy.layers.inet6 import IPv6, UDP, IPv6ExtHdrSegmentRouting
16 from scapy.layers.inet import IP, UDP
21 class TestSRv6(VppTestCase):
22 """ SRv6 Test Case """
26 super(TestSRv6, cls).setUpClass()
29 def tearDownClass(cls):
30 super(TestSRv6, cls).tearDownClass()
33 """ Perform test setup before each test case.
35 super(TestSRv6, self).setUp()
37 # packet sizes, inclusive L2 overhead
38 self.pg_packet_sizes = [64, 512, 1518, 9018]
41 self.reset_packet_infos()
44 """ Clean up test setup after each test case.
46 self.teardown_interfaces()
48 super(TestSRv6, self).tearDown()
50 def configure_interface(self,
52 ipv6=False, ipv4=False,
53 ipv6_table_id=0, ipv4_table_id=0):
54 """ Configure interface.
55 :param ipv6: configure IPv6 on interface
56 :param ipv4: configure IPv4 on interface
57 :param ipv6_table_id: FIB table_id for IPv6
58 :param ipv4_table_id: FIB table_id for IPv4
60 self.logger.debug("Configuring interface %s" % (interface.name))
62 self.logger.debug("Configuring IPv6")
63 interface.set_table_ip6(ipv6_table_id)
64 interface.config_ip6()
65 interface.resolve_ndp(timeout=5)
67 self.logger.debug("Configuring IPv4")
68 interface.set_table_ip4(ipv4_table_id)
69 interface.config_ip4()
70 interface.resolve_arp()
73 def setup_interfaces(self, ipv6=[], ipv4=[],
74 ipv6_table_id=[], ipv4_table_id=[]):
75 """ Create and configure interfaces.
77 :param ipv6: list of interface IPv6 capabilities
78 :param ipv4: list of interface IPv4 capabilities
79 :param ipv6_table_id: list of intf IPv6 FIB table_ids
80 :param ipv4_table_id: list of intf IPv4 FIB table_ids
81 :returns: List of created interfaces.
83 # how many interfaces?
88 self.logger.debug("Creating and configuring %d interfaces" % (count))
90 # fill up ipv6 and ipv4 lists if needed
91 # not enabled (False) is the default
93 ipv6 += (count - len(ipv6)) * [False]
95 ipv4 += (count - len(ipv4)) * [False]
97 # fill up table_id lists if needed
98 # table_id 0 (global) is the default
99 if len(ipv6_table_id) < count:
100 ipv6_table_id += (count - len(ipv6_table_id)) * [0]
101 if len(ipv4_table_id) < count:
102 ipv4_table_id += (count - len(ipv4_table_id)) * [0]
104 # create 'count' pg interfaces
105 self.create_pg_interfaces(range(count))
107 # setup all interfaces
108 for i in range(count):
109 intf = self.pg_interfaces[i]
110 self.configure_interface(intf,
112 ipv6_table_id[i], ipv4_table_id[i])
115 self.logger.debug(self.vapi.cli("show ip6 neighbors"))
117 self.logger.debug(self.vapi.cli("show ip4 neighbors"))
118 self.logger.debug(self.vapi.cli("show interface"))
119 self.logger.debug(self.vapi.cli("show hardware"))
121 return self.pg_interfaces
123 def teardown_interfaces(self):
124 """ Unconfigure and bring down interface.
126 self.logger.debug("Tearing down interfaces")
127 # tear down all interfaces
128 # AFAIK they cannot be deleted
129 for i in self.pg_interfaces:
130 self.logger.debug("Tear down interface %s" % (i.name))
136 @unittest.skipUnless(0, "PC to fix")
137 def test_SRv6_T_Encaps(self):
138 """ Test SRv6 Transit.Encaps behavior for IPv6.
140 # send traffic to one destination interface
141 # source and destination are IPv6 only
142 self.setup_interfaces(ipv6=[True, True])
144 # configure FIB entries
145 route = VppIpRoute(self, "a4::", 64,
146 [VppRoutePath(self.pg1.remote_ip6,
147 self.pg1.sw_if_index)])
148 route.add_vpp_config()
150 # configure encaps IPv6 source address
151 # needs to be done before SR Policy config
153 self.vapi.cli("set sr encaps source addr a3::")
156 # configure SRv6 Policy
157 # Note: segment list order: first -> last
158 sr_policy = VppSRv6Policy(
161 sr_type=SRv6PolicyType.SR_POLICY_TYPE_DEFAULT,
162 weight=1, fib_table=0,
163 segments=['a4::', 'a5::', 'a6::c7'],
165 sr_policy.add_vpp_config()
166 self.sr_policy = sr_policy
168 # log the sr policies
169 self.logger.info(self.vapi.cli("show sr policies"))
171 # steer IPv6 traffic to a7::/64 into SRv6 Policy
172 # use the bsid of the above self.sr_policy
173 pol_steering = VppSRv6Steering(
175 bsid=self.sr_policy.bsid,
176 prefix="a7::", mask_width=64,
177 traffic_type=SRv6PolicySteeringTypes.SR_STEER_IPV6,
178 sr_policy_index=0, table_id=0,
180 pol_steering.add_vpp_config()
182 # log the sr steering policies
183 self.logger.info(self.vapi.cli("show sr steering policies"))
186 count = len(self.pg_packet_sizes)
187 dst_inner = 'a7::1234'
190 # create IPv6 packets without SRH
191 packet_header = self.create_packet_header_IPv6(dst_inner)
192 # create traffic stream pg0->pg1
193 pkts.extend(self.create_stream(self.pg0, self.pg1, packet_header,
194 self.pg_packet_sizes, count))
196 # create IPv6 packets with SRH
197 # packets with segments-left 1, active segment a7::
198 packet_header = self.create_packet_header_IPv6_SRH(
199 sidlist=['a8::', 'a7::', 'a6::'],
201 # create traffic stream pg0->pg1
202 pkts.extend(self.create_stream(self.pg0, self.pg1, packet_header,
203 self.pg_packet_sizes, count))
205 # create IPv6 packets with SRH and IPv6
206 # packets with segments-left 1, active segment a7::
207 packet_header = self.create_packet_header_IPv6_SRH_IPv6(
209 sidlist=['a8::', 'a7::', 'a6::'],
211 # create traffic stream pg0->pg1
212 pkts.extend(self.create_stream(self.pg0, self.pg1, packet_header,
213 self.pg_packet_sizes, count))
215 # send packets and verify received packets
216 self.send_and_verify_pkts(self.pg0, pkts, self.pg1,
217 self.compare_rx_tx_packet_T_Encaps)
219 # log the localsid counters
220 self.logger.info(self.vapi.cli("show sr localsid"))
223 pol_steering.remove_vpp_config()
224 self.logger.info(self.vapi.cli("show sr steering policies"))
227 self.sr_policy.remove_vpp_config()
228 self.logger.info(self.vapi.cli("show sr policies"))
234 self.teardown_interfaces()
236 @unittest.skipUnless(0, "PC to fix")
237 def test_SRv6_T_Insert(self):
238 """ Test SRv6 Transit.Insert behavior (IPv6 only).
240 # send traffic to one destination interface
241 # source and destination are IPv6 only
242 self.setup_interfaces(ipv6=[True, True])
244 # configure FIB entries
245 route = VppIpRoute(self, "a4::", 64,
246 [VppRoutePath(self.pg1.remote_ip6,
247 self.pg1.sw_if_index)])
248 route.add_vpp_config()
250 # configure encaps IPv6 source address
251 # needs to be done before SR Policy config
253 self.vapi.cli("set sr encaps source addr a3::")
256 # configure SRv6 Policy
257 # Note: segment list order: first -> last
258 sr_policy = VppSRv6Policy(
261 sr_type=SRv6PolicyType.SR_POLICY_TYPE_DEFAULT,
262 weight=1, fib_table=0,
263 segments=['a4::', 'a5::', 'a6::c7'],
265 sr_policy.add_vpp_config()
266 self.sr_policy = sr_policy
268 # log the sr policies
269 self.logger.info(self.vapi.cli("show sr policies"))
271 # steer IPv6 traffic to a7::/64 into SRv6 Policy
272 # use the bsid of the above self.sr_policy
273 pol_steering = VppSRv6Steering(
275 bsid=self.sr_policy.bsid,
276 prefix="a7::", mask_width=64,
277 traffic_type=SRv6PolicySteeringTypes.SR_STEER_IPV6,
278 sr_policy_index=0, table_id=0,
280 pol_steering.add_vpp_config()
282 # log the sr steering policies
283 self.logger.info(self.vapi.cli("show sr steering policies"))
286 count = len(self.pg_packet_sizes)
287 dst_inner = 'a7::1234'
290 # create IPv6 packets without SRH
291 packet_header = self.create_packet_header_IPv6(dst_inner)
292 # create traffic stream pg0->pg1
293 pkts.extend(self.create_stream(self.pg0, self.pg1, packet_header,
294 self.pg_packet_sizes, count))
296 # create IPv6 packets with SRH
297 # packets with segments-left 1, active segment a7::
298 packet_header = self.create_packet_header_IPv6_SRH(
299 sidlist=['a8::', 'a7::', 'a6::'],
301 # create traffic stream pg0->pg1
302 pkts.extend(self.create_stream(self.pg0, self.pg1, packet_header,
303 self.pg_packet_sizes, count))
305 # send packets and verify received packets
306 self.send_and_verify_pkts(self.pg0, pkts, self.pg1,
307 self.compare_rx_tx_packet_T_Insert)
309 # log the localsid counters
310 self.logger.info(self.vapi.cli("show sr localsid"))
313 pol_steering.remove_vpp_config()
314 self.logger.info(self.vapi.cli("show sr steering policies"))
317 self.sr_policy.remove_vpp_config()
318 self.logger.info(self.vapi.cli("show sr policies"))
324 self.teardown_interfaces()
326 @unittest.skipUnless(0, "PC to fix")
327 def test_SRv6_T_Encaps_IPv4(self):
328 """ Test SRv6 Transit.Encaps behavior for IPv4.
330 # send traffic to one destination interface
331 # source interface is IPv4 only
332 # destination interface is IPv6 only
333 self.setup_interfaces(ipv6=[False, True], ipv4=[True, False])
335 # configure FIB entries
336 route = VppIpRoute(self, "a4::", 64,
337 [VppRoutePath(self.pg1.remote_ip6,
338 self.pg1.sw_if_index)])
339 route.add_vpp_config()
341 # configure encaps IPv6 source address
342 # needs to be done before SR Policy config
344 self.vapi.cli("set sr encaps source addr a3::")
347 # configure SRv6 Policy
348 # Note: segment list order: first -> last
349 sr_policy = VppSRv6Policy(
352 sr_type=SRv6PolicyType.SR_POLICY_TYPE_DEFAULT,
353 weight=1, fib_table=0,
354 segments=['a4::', 'a5::', 'a6::c7'],
356 sr_policy.add_vpp_config()
357 self.sr_policy = sr_policy
359 # log the sr policies
360 self.logger.info(self.vapi.cli("show sr policies"))
362 # steer IPv4 traffic to 7.1.1.0/24 into SRv6 Policy
363 # use the bsid of the above self.sr_policy
364 pol_steering = VppSRv6Steering(
366 bsid=self.sr_policy.bsid,
367 prefix="7.1.1.0", mask_width=24,
368 traffic_type=SRv6PolicySteeringTypes.SR_STEER_IPV4,
369 sr_policy_index=0, table_id=0,
371 pol_steering.add_vpp_config()
373 # log the sr steering policies
374 self.logger.info(self.vapi.cli("show sr steering policies"))
377 count = len(self.pg_packet_sizes)
378 dst_inner = '7.1.1.123'
381 # create IPv4 packets
382 packet_header = self.create_packet_header_IPv4(dst_inner)
383 # create traffic stream pg0->pg1
384 pkts.extend(self.create_stream(self.pg0, self.pg1, packet_header,
385 self.pg_packet_sizes, count))
387 # send packets and verify received packets
388 self.send_and_verify_pkts(self.pg0, pkts, self.pg1,
389 self.compare_rx_tx_packet_T_Encaps_IPv4)
391 # log the localsid counters
392 self.logger.info(self.vapi.cli("show sr localsid"))
395 pol_steering.remove_vpp_config()
396 self.logger.info(self.vapi.cli("show sr steering policies"))
399 self.sr_policy.remove_vpp_config()
400 self.logger.info(self.vapi.cli("show sr policies"))
406 self.teardown_interfaces()
408 @unittest.skip("VPP crashes after running this test")
409 def test_SRv6_T_Encaps_L2(self):
410 """ Test SRv6 Transit.Encaps behavior for L2.
412 # send traffic to one destination interface
413 # source interface is IPv4 only TODO?
414 # destination interface is IPv6 only
415 self.setup_interfaces(ipv6=[False, True], ipv4=[False, False])
417 # configure FIB entries
418 route = VppIpRoute(self, "a4::", 64,
419 [VppRoutePath(self.pg1.remote_ip6,
420 self.pg1.sw_if_index)])
421 route.add_vpp_config()
423 # configure encaps IPv6 source address
424 # needs to be done before SR Policy config
426 self.vapi.cli("set sr encaps source addr a3::")
429 # configure SRv6 Policy
430 # Note: segment list order: first -> last
431 sr_policy = VppSRv6Policy(
434 sr_type=SRv6PolicyType.SR_POLICY_TYPE_DEFAULT,
435 weight=1, fib_table=0,
436 segments=['a4::', 'a5::', 'a6::c7'],
438 sr_policy.add_vpp_config()
439 self.sr_policy = sr_policy
441 # log the sr policies
442 self.logger.info(self.vapi.cli("show sr policies"))
444 # steer L2 traffic into SRv6 Policy
445 # use the bsid of the above self.sr_policy
446 pol_steering = VppSRv6Steering(
448 bsid=self.sr_policy.bsid,
449 prefix="::", mask_width=0,
450 traffic_type=SRv6PolicySteeringTypes.SR_STEER_L2,
451 sr_policy_index=0, table_id=0,
452 sw_if_index=self.pg0.sw_if_index)
453 pol_steering.add_vpp_config()
455 # log the sr steering policies
456 self.logger.info(self.vapi.cli("show sr steering policies"))
459 count = len(self.pg_packet_sizes)
462 # create L2 packets without dot1q header
463 packet_header = self.create_packet_header_L2()
464 # create traffic stream pg0->pg1
465 pkts.extend(self.create_stream(self.pg0, self.pg1, packet_header,
466 self.pg_packet_sizes, count))
468 # create L2 packets with dot1q header
469 packet_header = self.create_packet_header_L2(vlan=123)
470 # create traffic stream pg0->pg1
471 pkts.extend(self.create_stream(self.pg0, self.pg1, packet_header,
472 self.pg_packet_sizes, count))
474 # send packets and verify received packets
475 self.send_and_verify_pkts(self.pg0, pkts, self.pg1,
476 self.compare_rx_tx_packet_T_Encaps_L2)
478 # log the localsid counters
479 self.logger.info(self.vapi.cli("show sr localsid"))
482 pol_steering.remove_vpp_config()
483 self.logger.info(self.vapi.cli("show sr steering policies"))
486 self.sr_policy.remove_vpp_config()
487 self.logger.info(self.vapi.cli("show sr policies"))
493 self.teardown_interfaces()
495 def test_SRv6_End(self):
496 """ Test SRv6 End (without PSP) behavior.
498 # send traffic to one destination interface
499 # source and destination interfaces are IPv6 only
500 self.setup_interfaces(ipv6=[True, True])
502 # configure FIB entries
503 route = VppIpRoute(self, "a4::", 64,
504 [VppRoutePath(self.pg1.remote_ip6,
505 self.pg1.sw_if_index)])
506 route.add_vpp_config()
508 # configure SRv6 localSID End without PSP behavior
509 localsid = VppSRv6LocalSID(
510 self, localsid={'addr': 'A3::0'},
511 behavior=SRv6LocalSIDBehaviors.SR_BEHAVIOR_END,
518 localsid.add_vpp_config()
520 self.logger.debug(self.vapi.cli("show sr localsid"))
522 # create IPv6 packets with SRH (SL=2, SL=1, SL=0)
523 # send one packet per SL value per packet size
524 # SL=0 packet with localSID End with USP needs 2nd SRH
525 count = len(self.pg_packet_sizes)
526 dst_inner = 'a4::1234'
529 # packets with segments-left 2, active segment a3::
530 packet_header = self.create_packet_header_IPv6_SRH_IPv6(
532 sidlist=['a5::', 'a4::', 'a3::'],
534 # create traffic stream pg0->pg1
535 pkts.extend(self.create_stream(self.pg0, self.pg1, packet_header,
536 self.pg_packet_sizes, count))
538 # packets with segments-left 1, active segment a3::
539 packet_header = self.create_packet_header_IPv6_SRH_IPv6(
541 sidlist=['a4::', 'a3::', 'a2::'],
543 # add to traffic stream pg0->pg1
544 pkts.extend(self.create_stream(self.pg0, self.pg1, packet_header,
545 self.pg_packet_sizes, count))
547 # TODO: test behavior with SL=0 packet (needs 2*SRH?)
549 # send packets and verify received packets
550 self.send_and_verify_pkts(self.pg0, pkts, self.pg1,
551 self.compare_rx_tx_packet_End)
553 # log the localsid counters
554 self.logger.info(self.vapi.cli("show sr localsid"))
556 # remove SRv6 localSIDs
557 localsid.remove_vpp_config()
563 self.teardown_interfaces()
565 def test_SRv6_End_with_PSP(self):
566 """ Test SRv6 End with PSP behavior.
568 # send traffic to one destination interface
569 # source and destination interfaces are IPv6 only
570 self.setup_interfaces(ipv6=[True, True])
572 # configure FIB entries
573 route = VppIpRoute(self, "a4::", 64,
574 [VppRoutePath(self.pg1.remote_ip6,
575 self.pg1.sw_if_index)])
576 route.add_vpp_config()
578 # configure SRv6 localSID End with PSP behavior
579 localsid = VppSRv6LocalSID(
580 self, localsid={'addr': 'A3::0'},
581 behavior=SRv6LocalSIDBehaviors.SR_BEHAVIOR_END,
588 localsid.add_vpp_config()
590 self.logger.debug(self.vapi.cli("show sr localsid"))
592 # create IPv6 packets with SRH (SL=2, SL=1)
593 # send one packet per SL value per packet size
594 # SL=0 packet with localSID End with PSP is dropped
595 count = len(self.pg_packet_sizes)
596 dst_inner = 'a4::1234'
599 # packets with segments-left 2, active segment a3::
600 packet_header = self.create_packet_header_IPv6_SRH_IPv6(
602 sidlist=['a5::', 'a4::', 'a3::'],
604 # create traffic stream pg0->pg1
605 pkts.extend(self.create_stream(self.pg0, self.pg1, packet_header,
606 self.pg_packet_sizes, count))
608 # packets with segments-left 1, active segment a3::
609 packet_header = self.create_packet_header_IPv6_SRH_IPv6(
611 sidlist=['a4::', 'a3::', 'a2::'],
613 # add to traffic stream pg0->pg1
614 pkts.extend(self.create_stream(self.pg0, self.pg1, packet_header,
615 self.pg_packet_sizes, count))
617 # send packets and verify received packets
618 self.send_and_verify_pkts(self.pg0, pkts, self.pg1,
619 self.compare_rx_tx_packet_End_PSP)
621 # log the localsid counters
622 self.logger.info(self.vapi.cli("show sr localsid"))
624 # remove SRv6 localSIDs
625 localsid.remove_vpp_config()
631 self.teardown_interfaces()
633 def test_SRv6_End_X(self):
634 """ Test SRv6 End.X (without PSP) behavior.
636 # create three interfaces (1 source, 2 destinations)
637 # source and destination interfaces are IPv6 only
638 self.setup_interfaces(ipv6=[True, True, True])
640 # configure FIB entries
641 # a4::/64 via pg1 and pg2
642 route = VppIpRoute(self, "a4::", 64,
643 [VppRoutePath(self.pg1.remote_ip6,
644 self.pg1.sw_if_index),
645 VppRoutePath(self.pg2.remote_ip6,
646 self.pg2.sw_if_index)])
647 route.add_vpp_config()
648 self.logger.debug(self.vapi.cli("show ip6 fib"))
650 # configure SRv6 localSID End.X without PSP behavior
651 # End.X points to interface pg1
652 localsid = VppSRv6LocalSID(
653 self, localsid={'addr': 'A3::C4'},
654 behavior=SRv6LocalSIDBehaviors.SR_BEHAVIOR_X,
656 nh_addr6=self.pg1.remote_ip6,
658 sw_if_index=self.pg1.sw_if_index,
661 localsid.add_vpp_config()
663 self.logger.debug(self.vapi.cli("show sr localsid"))
665 # create IPv6 packets with SRH (SL=2, SL=1)
666 # send one packet per SL value per packet size
667 # SL=0 packet with localSID End with PSP is dropped
668 count = len(self.pg_packet_sizes)
669 dst_inner = 'a4::1234'
672 # packets with segments-left 2, active segment a3::c4
673 packet_header = self.create_packet_header_IPv6_SRH_IPv6(
675 sidlist=['a5::', 'a4::', 'a3::c4'],
677 # create traffic stream pg0->pg1
678 pkts.extend(self.create_stream(self.pg0, self.pg1, packet_header,
679 self.pg_packet_sizes, count))
681 # packets with segments-left 1, active segment a3::c4
682 packet_header = self.create_packet_header_IPv6_SRH_IPv6(
684 sidlist=['a4::', 'a3::c4', 'a2::'],
686 # add to traffic stream pg0->pg1
687 pkts.extend(self.create_stream(self.pg0, self.pg1, packet_header,
688 self.pg_packet_sizes, count))
690 # send packets and verify received packets
691 # using same comparison function as End (no PSP)
692 self.send_and_verify_pkts(self.pg0, pkts, self.pg1,
693 self.compare_rx_tx_packet_End)
695 # assert nothing was received on the other interface (pg2)
696 self.pg2.assert_nothing_captured("mis-directed packet(s)")
698 # log the localsid counters
699 self.logger.info(self.vapi.cli("show sr localsid"))
701 # remove SRv6 localSIDs
702 localsid.remove_vpp_config()
708 self.teardown_interfaces()
710 def test_SRv6_End_X_with_PSP(self):
711 """ Test SRv6 End.X with PSP behavior.
713 # create three interfaces (1 source, 2 destinations)
714 # source and destination interfaces are IPv6 only
715 self.setup_interfaces(ipv6=[True, True, True])
717 # configure FIB entries
718 # a4::/64 via pg1 and pg2
719 route = VppIpRoute(self, "a4::", 64,
722 self.pg1.sw_if_index),
723 VppRoutePath(self.pg2.remote_ip6,
724 self.pg2.sw_if_index)])
725 route.add_vpp_config()
727 # configure SRv6 localSID End with PSP behavior
728 localsid = VppSRv6LocalSID(
729 self, localsid={'addr': 'A3::C4'},
730 behavior=SRv6LocalSIDBehaviors.SR_BEHAVIOR_X,
732 nh_addr6=self.pg1.remote_ip6,
734 sw_if_index=self.pg1.sw_if_index,
737 localsid.add_vpp_config()
739 self.logger.debug(self.vapi.cli("show sr localsid"))
741 # create IPv6 packets with SRH (SL=2, SL=1)
742 # send one packet per SL value per packet size
743 # SL=0 packet with localSID End with PSP is dropped
744 count = len(self.pg_packet_sizes)
745 dst_inner = 'a4::1234'
748 # packets with segments-left 2, active segment a3::
749 packet_header = self.create_packet_header_IPv6_SRH_IPv6(
751 sidlist=['a5::', 'a4::', 'a3::c4'],
753 # create traffic stream pg0->pg1
754 pkts.extend(self.create_stream(self.pg0, self.pg1, packet_header,
755 self.pg_packet_sizes, count))
757 # packets with segments-left 1, active segment a3::
758 packet_header = self.create_packet_header_IPv6_SRH_IPv6(
760 sidlist=['a4::', 'a3::c4', 'a2::'],
762 # add to traffic stream pg0->pg1
763 pkts.extend(self.create_stream(self.pg0, self.pg1, packet_header,
764 self.pg_packet_sizes, count))
766 # send packets and verify received packets
767 # using same comparison function as End with PSP
768 self.send_and_verify_pkts(self.pg0, pkts, self.pg1,
769 self.compare_rx_tx_packet_End_PSP)
771 # assert nothing was received on the other interface (pg2)
772 self.pg2.assert_nothing_captured("mis-directed packet(s)")
774 # log the localsid counters
775 self.logger.info(self.vapi.cli("show sr localsid"))
777 # remove SRv6 localSIDs
778 localsid.remove_vpp_config()
784 self.teardown_interfaces()
786 def test_SRv6_End_DX6(self):
787 """ Test SRv6 End.DX6 behavior.
789 # send traffic to one destination interface
790 # source and destination interfaces are IPv6 only
791 self.setup_interfaces(ipv6=[True, True])
793 # configure SRv6 localSID End.DX6 behavior
794 localsid = VppSRv6LocalSID(
795 self, localsid={'addr': 'A3::C4'},
796 behavior=SRv6LocalSIDBehaviors.SR_BEHAVIOR_DX6,
798 nh_addr6=self.pg1.remote_ip6,
800 sw_if_index=self.pg1.sw_if_index,
803 localsid.add_vpp_config()
805 self.logger.debug(self.vapi.cli("show sr localsid"))
807 # create IPv6 packets with SRH (SL=0)
808 # send one packet per packet size
809 count = len(self.pg_packet_sizes)
810 dst_inner = 'a4::1234' # inner header destination address
813 # packets with SRH, segments-left 0, active segment a3::c4
814 packet_header = self.create_packet_header_IPv6_SRH_IPv6(
816 sidlist=['a3::c4', 'a2::', 'a1::'],
818 # add to traffic stream pg0->pg1
819 pkts.extend(self.create_stream(self.pg0, self.pg1, packet_header,
820 self.pg_packet_sizes, count))
822 # packets without SRH, IPv6 in IPv6
823 # outer IPv6 dest addr is the localsid End.DX6
824 packet_header = self.create_packet_header_IPv6_IPv6(
827 # add to traffic stream pg0->pg1
828 pkts.extend(self.create_stream(self.pg0, self.pg1, packet_header,
829 self.pg_packet_sizes, count))
831 # send packets and verify received packets
832 self.send_and_verify_pkts(self.pg0, pkts, self.pg1,
833 self.compare_rx_tx_packet_End_DX6)
835 # log the localsid counters
836 self.logger.info(self.vapi.cli("show sr localsid"))
838 # remove SRv6 localSIDs
839 localsid.remove_vpp_config()
842 self.teardown_interfaces()
844 def test_SRv6_End_DT6(self):
845 """ Test SRv6 End.DT6 behavior.
847 # create three interfaces (1 source, 2 destinations)
848 # all interfaces are IPv6 only
849 # source interface in global FIB (0)
850 # destination interfaces in global and vrf
852 ipt = VppIpTable(self, vrf_1, is_ip6=True)
854 self.setup_interfaces(ipv6=[True, True, True],
855 ipv6_table_id=[0, 0, vrf_1])
857 # configure FIB entries
858 # a4::/64 is reachable
859 # via pg1 in table 0 (global)
860 # and via pg2 in table vrf_1
861 route0 = VppIpRoute(self, "a4::", 64,
862 [VppRoutePath(self.pg1.remote_ip6,
863 self.pg1.sw_if_index,
866 route0.add_vpp_config()
867 route1 = VppIpRoute(self, "a4::", 64,
868 [VppRoutePath(self.pg2.remote_ip6,
869 self.pg2.sw_if_index,
872 route1.add_vpp_config()
873 self.logger.debug(self.vapi.cli("show ip6 fib"))
875 # configure SRv6 localSID End.DT6 behavior
877 # fib_table: where the localsid is installed
878 # sw_if_index: in T-variants of localsid this is the vrf table_id
879 localsid = VppSRv6LocalSID(
880 self, localsid={'addr': 'A3::C4'},
881 behavior=SRv6LocalSIDBehaviors.SR_BEHAVIOR_DT6,
888 localsid.add_vpp_config()
890 self.logger.debug(self.vapi.cli("show sr localsid"))
892 # create IPv6 packets with SRH (SL=0)
893 # send one packet per packet size
894 count = len(self.pg_packet_sizes)
895 dst_inner = 'a4::1234' # inner header destination address
898 # packets with SRH, segments-left 0, active segment a3::c4
899 packet_header = self.create_packet_header_IPv6_SRH_IPv6(
901 sidlist=['a3::c4', 'a2::', 'a1::'],
903 # add to traffic stream pg0->pg1
904 pkts.extend(self.create_stream(self.pg0, self.pg2, packet_header,
905 self.pg_packet_sizes, count))
907 # packets without SRH, IPv6 in IPv6
908 # outer IPv6 dest addr is the localsid End.DT6
909 packet_header = self.create_packet_header_IPv6_IPv6(
912 # add to traffic stream pg0->pg1
913 pkts.extend(self.create_stream(self.pg0, self.pg2, packet_header,
914 self.pg_packet_sizes, count))
916 # send packets and verify received packets
917 # using same comparison function as End.DX6
918 self.send_and_verify_pkts(self.pg0, pkts, self.pg2,
919 self.compare_rx_tx_packet_End_DX6)
921 # assert nothing was received on the other interface (pg2)
922 self.pg1.assert_nothing_captured("mis-directed packet(s)")
924 # log the localsid counters
925 self.logger.info(self.vapi.cli("show sr localsid"))
927 # remove SRv6 localSIDs
928 localsid.remove_vpp_config()
934 self.teardown_interfaces()
936 def test_SRv6_End_DX4(self):
937 """ Test SRv6 End.DX4 behavior.
939 # send traffic to one destination interface
940 # source interface is IPv6 only
941 # destination interface is IPv4 only
942 self.setup_interfaces(ipv6=[True, False], ipv4=[False, True])
944 # configure SRv6 localSID End.DX4 behavior
945 localsid = VppSRv6LocalSID(
946 self, localsid={'addr': 'A3::C4'},
947 behavior=SRv6LocalSIDBehaviors.SR_BEHAVIOR_DX4,
948 nh_addr4=self.pg1.remote_ip4,
951 sw_if_index=self.pg1.sw_if_index,
954 localsid.add_vpp_config()
956 self.logger.debug(self.vapi.cli("show sr localsid"))
958 # send one packet per packet size
959 count = len(self.pg_packet_sizes)
960 dst_inner = '4.1.1.123' # inner header destination address
963 # packets with SRH, segments-left 0, active segment a3::c4
964 packet_header = self.create_packet_header_IPv6_SRH_IPv4(
966 sidlist=['a3::c4', 'a2::', 'a1::'],
968 # add to traffic stream pg0->pg1
969 pkts.extend(self.create_stream(self.pg0, self.pg1, packet_header,
970 self.pg_packet_sizes, count))
972 # packets without SRH, IPv4 in IPv6
973 # outer IPv6 dest addr is the localsid End.DX4
974 packet_header = self.create_packet_header_IPv6_IPv4(
977 # add to traffic stream pg0->pg1
978 pkts.extend(self.create_stream(self.pg0, self.pg1, packet_header,
979 self.pg_packet_sizes, count))
981 # send packets and verify received packets
982 self.send_and_verify_pkts(self.pg0, pkts, self.pg1,
983 self.compare_rx_tx_packet_End_DX4)
985 # log the localsid counters
986 self.logger.info(self.vapi.cli("show sr localsid"))
988 # remove SRv6 localSIDs
989 localsid.remove_vpp_config()
992 self.teardown_interfaces()
994 def test_SRv6_End_DT4(self):
995 """ Test SRv6 End.DT4 behavior.
997 # create three interfaces (1 source, 2 destinations)
998 # source interface is IPv6-only
999 # destination interfaces are IPv4 only
1000 # source interface in global FIB (0)
1001 # destination interfaces in global and vrf
1003 ipt = VppIpTable(self, vrf_1)
1004 ipt.add_vpp_config()
1005 self.setup_interfaces(ipv6=[True, False, False],
1006 ipv4=[False, True, True],
1007 ipv6_table_id=[0, 0, 0],
1008 ipv4_table_id=[0, 0, vrf_1])
1010 # configure FIB entries
1011 # 4.1.1.0/24 is reachable
1012 # via pg1 in table 0 (global)
1013 # and via pg2 in table vrf_1
1014 route0 = VppIpRoute(self, "4.1.1.0", 24,
1015 [VppRoutePath(self.pg1.remote_ip4,
1016 self.pg1.sw_if_index,
1019 route0.add_vpp_config()
1020 route1 = VppIpRoute(self, "4.1.1.0", 24,
1021 [VppRoutePath(self.pg2.remote_ip4,
1022 self.pg2.sw_if_index,
1023 nh_table_id=vrf_1)],
1025 route1.add_vpp_config()
1026 self.logger.debug(self.vapi.cli("show ip fib"))
1028 # configure SRv6 localSID End.DT6 behavior
1030 # fib_table: where the localsid is installed
1031 # sw_if_index: in T-variants of localsid: vrf table_id
1032 localsid = VppSRv6LocalSID(
1033 self, localsid={'addr': 'A3::C4'},
1034 behavior=SRv6LocalSIDBehaviors.SR_BEHAVIOR_DT4,
1041 localsid.add_vpp_config()
1043 self.logger.debug(self.vapi.cli("show sr localsid"))
1045 # create IPv6 packets with SRH (SL=0)
1046 # send one packet per packet size
1047 count = len(self.pg_packet_sizes)
1048 dst_inner = '4.1.1.123' # inner header destination address
1051 # packets with SRH, segments-left 0, active segment a3::c4
1052 packet_header = self.create_packet_header_IPv6_SRH_IPv4(
1054 sidlist=['a3::c4', 'a2::', 'a1::'],
1056 # add to traffic stream pg0->pg1
1057 pkts.extend(self.create_stream(self.pg0, self.pg2, packet_header,
1058 self.pg_packet_sizes, count))
1060 # packets without SRH, IPv6 in IPv6
1061 # outer IPv6 dest addr is the localsid End.DX4
1062 packet_header = self.create_packet_header_IPv6_IPv4(
1065 # add to traffic stream pg0->pg1
1066 pkts.extend(self.create_stream(self.pg0, self.pg2, packet_header,
1067 self.pg_packet_sizes, count))
1069 # send packets and verify received packets
1070 # using same comparison function as End.DX4
1071 self.send_and_verify_pkts(self.pg0, pkts, self.pg2,
1072 self.compare_rx_tx_packet_End_DX4)
1074 # assert nothing was received on the other interface (pg2)
1075 self.pg1.assert_nothing_captured("mis-directed packet(s)")
1077 # log the localsid counters
1078 self.logger.info(self.vapi.cli("show sr localsid"))
1080 # remove SRv6 localSIDs
1081 localsid.remove_vpp_config()
1083 # remove FIB entries
1086 # cleanup interfaces
1087 self.teardown_interfaces()
1089 def test_SRv6_End_DX2(self):
1090 """ Test SRv6 End.DX2 behavior.
1092 # send traffic to one destination interface
1093 # source interface is IPv6 only
1094 self.setup_interfaces(ipv6=[True, False], ipv4=[False, False])
1096 # configure SRv6 localSID End.DX2 behavior
1097 localsid = VppSRv6LocalSID(
1098 self, localsid={'addr': 'A3::C4'},
1099 behavior=SRv6LocalSIDBehaviors.SR_BEHAVIOR_DX2,
1103 sw_if_index=self.pg1.sw_if_index,
1106 localsid.add_vpp_config()
1108 self.logger.debug(self.vapi.cli("show sr localsid"))
1110 # send one packet per packet size
1111 count = len(self.pg_packet_sizes)
1114 # packets with SRH, segments-left 0, active segment a3::c4
1115 # L2 has no dot1q header
1116 packet_header = self.create_packet_header_IPv6_SRH_L2(
1117 sidlist=['a3::c4', 'a2::', 'a1::'],
1120 # add to traffic stream pg0->pg1
1121 pkts.extend(self.create_stream(self.pg0, self.pg1, packet_header,
1122 self.pg_packet_sizes, count))
1124 # packets with SRH, segments-left 0, active segment a3::c4
1125 # L2 has dot1q header
1126 packet_header = self.create_packet_header_IPv6_SRH_L2(
1127 sidlist=['a3::c4', 'a2::', 'a1::'],
1130 # add to traffic stream pg0->pg1
1131 pkts.extend(self.create_stream(self.pg0, self.pg1, packet_header,
1132 self.pg_packet_sizes, count))
1134 # packets without SRH, L2 in IPv6
1135 # outer IPv6 dest addr is the localsid End.DX2
1136 # L2 has no dot1q header
1137 packet_header = self.create_packet_header_IPv6_L2(
1140 # add to traffic stream pg0->pg1
1141 pkts.extend(self.create_stream(self.pg0, self.pg1, packet_header,
1142 self.pg_packet_sizes, count))
1144 # packets without SRH, L2 in IPv6
1145 # outer IPv6 dest addr is the localsid End.DX2
1146 # L2 has dot1q header
1147 packet_header = self.create_packet_header_IPv6_L2(
1150 # add to traffic stream pg0->pg1
1151 pkts.extend(self.create_stream(self.pg0, self.pg1, packet_header,
1152 self.pg_packet_sizes, count))
1154 # send packets and verify received packets
1155 self.send_and_verify_pkts(self.pg0, pkts, self.pg1,
1156 self.compare_rx_tx_packet_End_DX2)
1158 # log the localsid counters
1159 self.logger.info(self.vapi.cli("show sr localsid"))
1161 # remove SRv6 localSIDs
1162 localsid.remove_vpp_config()
1164 # cleanup interfaces
1165 self.teardown_interfaces()
1167 @unittest.skipUnless(0, "PC to fix")
1168 def test_SRv6_T_Insert_Classifier(self):
1169 """ Test SRv6 Transit.Insert behavior (IPv6 only).
1170 steer packets using the classifier
1172 # send traffic to one destination interface
1173 # source and destination are IPv6 only
1174 self.setup_interfaces(ipv6=[False, False, False, True, True])
1176 # configure FIB entries
1177 route = VppIpRoute(self, "a4::", 64,
1179 self.pg4.remote_ip6,
1180 self.pg4.sw_if_index)])
1181 route.add_vpp_config()
1183 # configure encaps IPv6 source address
1184 # needs to be done before SR Policy config
1186 self.vapi.cli("set sr encaps source addr a3::")
1189 # configure SRv6 Policy
1190 # Note: segment list order: first -> last
1191 sr_policy = VppSRv6Policy(
1194 sr_type=SRv6PolicyType.SR_POLICY_TYPE_DEFAULT,
1195 weight=1, fib_table=0,
1196 segments=['a4::', 'a5::', 'a6::c7'],
1198 sr_policy.add_vpp_config()
1199 self.sr_policy = sr_policy
1201 # log the sr policies
1202 self.logger.info(self.vapi.cli("show sr policies"))
1204 # add classify table
1205 # mask on dst ip address prefix a7::/8
1206 mask = '{!s:0<16}'.format('ff')
1207 r = self.vapi.classify_add_del_table(
1209 binascii.unhexlify(mask),
1210 match_n_vectors=(len(mask) - 1) // 32 + 1,
1211 current_data_flag=1,
1212 skip_n_vectors=2) # data offset
1213 self.assertIsNotNone(r, 'No response msg for add_del_table')
1214 table_index = r.new_table_index
1216 # add the source routing node as a ip6 inacl netxt node
1217 r = self.vapi.add_node_next('ip6-inacl',
1218 'sr-pl-rewrite-insert')
1219 inacl_next_node_index = r.node_index
1221 match = '{!s:0<16}'.format('a7')
1222 r = self.vapi.classify_add_del_session(
1225 binascii.unhexlify(match),
1226 hit_next_index=inacl_next_node_index,
1228 metadata=0) # sr policy index
1229 self.assertIsNotNone(r, 'No response msg for add_del_session')
1231 # log the classify table used in the steering policy
1232 self.logger.info(self.vapi.cli("show classify table"))
1234 r = self.vapi.input_acl_set_interface(
1236 sw_if_index=self.pg3.sw_if_index,
1237 ip6_table_index=table_index)
1238 self.assertIsNotNone(r,
1239 'No response msg for input_acl_set_interface')
1242 self.logger.info(self.vapi.cli("show inacl type ip6"))
1245 count = len(self.pg_packet_sizes)
1246 dst_inner = 'a7::1234'
1249 # create IPv6 packets without SRH
1250 packet_header = self.create_packet_header_IPv6(dst_inner)
1251 # create traffic stream pg3->pg4
1252 pkts.extend(self.create_stream(self.pg3, self.pg4, packet_header,
1253 self.pg_packet_sizes, count))
1255 # create IPv6 packets with SRH
1256 # packets with segments-left 1, active segment a7::
1257 packet_header = self.create_packet_header_IPv6_SRH(
1258 sidlist=['a8::', 'a7::', 'a6::'],
1260 # create traffic stream pg3->pg4
1261 pkts.extend(self.create_stream(self.pg3, self.pg4, packet_header,
1262 self.pg_packet_sizes, count))
1264 # send packets and verify received packets
1265 self.send_and_verify_pkts(self.pg3, pkts, self.pg4,
1266 self.compare_rx_tx_packet_T_Insert)
1268 # remove the interface l2 input feature
1269 r = self.vapi.input_acl_set_interface(
1271 sw_if_index=self.pg3.sw_if_index,
1272 ip6_table_index=table_index)
1273 self.assertIsNotNone(r,
1274 'No response msg for input_acl_set_interface')
1276 # log the ip6 inacl after cleaning
1277 self.logger.info(self.vapi.cli("show inacl type ip6"))
1279 # log the localsid counters
1280 self.logger.info(self.vapi.cli("show sr localsid"))
1282 # remove classifier SR steering
1283 # classifier_steering.remove_vpp_config()
1284 self.logger.info(self.vapi.cli("show sr steering policies"))
1286 # remove SR Policies
1287 self.sr_policy.remove_vpp_config()
1288 self.logger.info(self.vapi.cli("show sr policies"))
1290 # remove classify session and table
1291 r = self.vapi.classify_add_del_session(
1294 binascii.unhexlify(match))
1295 self.assertIsNotNone(r, 'No response msg for add_del_session')
1297 r = self.vapi.classify_add_del_table(
1299 binascii.unhexlify(mask),
1300 table_index=table_index)
1301 self.assertIsNotNone(r, 'No response msg for add_del_table')
1303 self.logger.info(self.vapi.cli("show classify table"))
1305 # remove FIB entries
1308 # cleanup interfaces
1309 self.teardown_interfaces()
1311 def compare_rx_tx_packet_T_Encaps(self, tx_pkt, rx_pkt):
1312 """ Compare input and output packet after passing T.Encaps
1314 :param tx_pkt: transmitted packet
1315 :param rx_pkt: received packet
1317 # T.Encaps updates the headers as follows:
1318 # SR Policy seglist (S3, S2, S1)
1319 # SR Policy source C
1322 # out: IPv6(C, S1)SRH(S3, S2, S1; SL=2)IPv6(A, B2)
1324 # in: IPv6(A, B2)SRH(B3, B2, B1; SL=1)
1325 # out: IPv6(C, S1)SRH(S3, S2, S1; SL=2)IPv6(a, B2)SRH(B3, B2, B1; SL=1)
1327 # get first (outer) IPv6 header of rx'ed packet
1328 rx_ip = rx_pkt.getlayer(IPv6)
1331 tx_ip = tx_pkt.getlayer(IPv6)
1333 # expected segment-list
1334 seglist = self.sr_policy.segments
1335 # reverse list to get order as in SRH
1336 tx_seglist = seglist[::-1]
1338 # get source address of SR Policy
1339 sr_policy_source = self.sr_policy.source
1341 # rx'ed packet should have SRH
1342 self.assertTrue(rx_pkt.haslayer(IPv6ExtHdrSegmentRouting))
1344 rx_srh = rx_pkt.getlayer(IPv6ExtHdrSegmentRouting)
1346 # received ip.src should be equal to SR Policy source
1347 self.assertEqual(rx_ip.src, sr_policy_source)
1348 # received ip.dst should be equal to expected sidlist[lastentry]
1349 self.assertEqual(rx_ip.dst, tx_seglist[-1])
1350 # rx'ed seglist should be equal to expected seglist
1351 self.assertEqual(rx_srh.addresses, tx_seglist)
1352 # segleft should be equal to size expected seglist-1
1353 self.assertEqual(rx_srh.segleft, len(tx_seglist)-1)
1354 # segleft should be equal to lastentry
1355 self.assertEqual(rx_srh.segleft, rx_srh.lastentry)
1357 # the whole rx'ed pkt beyond SRH should be equal to tx'ed pkt
1358 # except for the hop-limit field
1359 # -> update tx'ed hlim to the expected hlim
1360 tx_ip.hlim = tx_ip.hlim - 1
1362 self.assertEqual(rx_srh.payload, tx_ip)
1364 self.logger.debug("packet verification: SUCCESS")
1366 def compare_rx_tx_packet_T_Encaps_IPv4(self, tx_pkt, rx_pkt):
1367 """ Compare input and output packet after passing T.Encaps for IPv4
1369 :param tx_pkt: transmitted packet
1370 :param rx_pkt: received packet
1372 # T.Encaps for IPv4 updates the headers as follows:
1373 # SR Policy seglist (S3, S2, S1)
1374 # SR Policy source C
1377 # out: IPv6(C, S1)SRH(S3, S2, S1; SL=2)IPv4(A, B2)
1379 # get first (outer) IPv6 header of rx'ed packet
1380 rx_ip = rx_pkt.getlayer(IPv6)
1383 tx_ip = tx_pkt.getlayer(IP)
1385 # expected segment-list
1386 seglist = self.sr_policy.segments
1387 # reverse list to get order as in SRH
1388 tx_seglist = seglist[::-1]
1390 # get source address of SR Policy
1391 sr_policy_source = self.sr_policy.source
1393 # checks common to cases tx with and without SRH
1394 # rx'ed packet should have SRH and IPv4 header
1395 self.assertTrue(rx_pkt.haslayer(IPv6ExtHdrSegmentRouting))
1396 self.assertTrue(rx_ip.payload.haslayer(IP))
1398 rx_srh = rx_pkt.getlayer(IPv6ExtHdrSegmentRouting)
1400 # received ip.src should be equal to SR Policy source
1401 self.assertEqual(rx_ip.src, sr_policy_source)
1402 # received ip.dst should be equal to sidlist[lastentry]
1403 self.assertEqual(rx_ip.dst, tx_seglist[-1])
1404 # rx'ed seglist should be equal to seglist
1405 self.assertEqual(rx_srh.addresses, tx_seglist)
1406 # segleft should be equal to size seglist-1
1407 self.assertEqual(rx_srh.segleft, len(tx_seglist)-1)
1408 # segleft should be equal to lastentry
1409 self.assertEqual(rx_srh.segleft, rx_srh.lastentry)
1411 # the whole rx'ed pkt beyond SRH should be equal to tx'ed pkt
1412 # except for the ttl field and ip checksum
1413 # -> adjust tx'ed ttl to expected ttl
1414 tx_ip.ttl = tx_ip.ttl - 1
1415 # -> set tx'ed ip checksum to None and let scapy recompute
1417 # read back the pkt (with str()) to force computing these fields
1418 # probably other ways to accomplish this are possible
1419 tx_ip = IP(scapy.compat.raw(tx_ip))
1421 self.assertEqual(rx_srh.payload, tx_ip)
1423 self.logger.debug("packet verification: SUCCESS")
1425 def compare_rx_tx_packet_T_Encaps_L2(self, tx_pkt, rx_pkt):
1426 """ Compare input and output packet after passing T.Encaps for L2
1428 :param tx_pkt: transmitted packet
1429 :param rx_pkt: received packet
1431 # T.Encaps for L2 updates the headers as follows:
1432 # SR Policy seglist (S3, S2, S1)
1433 # SR Policy source C
1436 # out: IPv6(C, S1)SRH(S3, S2, S1; SL=2)L2
1438 # get first (outer) IPv6 header of rx'ed packet
1439 rx_ip = rx_pkt.getlayer(IPv6)
1442 tx_ether = tx_pkt.getlayer(Ether)
1444 # expected segment-list
1445 seglist = self.sr_policy.segments
1446 # reverse list to get order as in SRH
1447 tx_seglist = seglist[::-1]
1449 # get source address of SR Policy
1450 sr_policy_source = self.sr_policy.source
1452 # rx'ed packet should have SRH
1453 self.assertTrue(rx_pkt.haslayer(IPv6ExtHdrSegmentRouting))
1455 rx_srh = rx_pkt.getlayer(IPv6ExtHdrSegmentRouting)
1457 # received ip.src should be equal to SR Policy source
1458 self.assertEqual(rx_ip.src, sr_policy_source)
1459 # received ip.dst should be equal to sidlist[lastentry]
1460 self.assertEqual(rx_ip.dst, tx_seglist[-1])
1461 # rx'ed seglist should be equal to seglist
1462 self.assertEqual(rx_srh.addresses, tx_seglist)
1463 # segleft should be equal to size seglist-1
1464 self.assertEqual(rx_srh.segleft, len(tx_seglist)-1)
1465 # segleft should be equal to lastentry
1466 self.assertEqual(rx_srh.segleft, rx_srh.lastentry)
1467 # nh should be "No Next Header" (59)
1468 self.assertEqual(rx_srh.nh, 59)
1470 # the whole rx'ed pkt beyond SRH should be equal to tx'ed pkt
1471 self.assertEqual(Ether(scapy.compat.raw(rx_srh.payload)), tx_ether)
1473 self.logger.debug("packet verification: SUCCESS")
1475 def compare_rx_tx_packet_T_Insert(self, tx_pkt, rx_pkt):
1476 """ Compare input and output packet after passing T.Insert
1478 :param tx_pkt: transmitted packet
1479 :param rx_pkt: received packet
1481 # T.Insert updates the headers as follows:
1484 # out: IPv6(A, S1)SRH(B2, S3, S2, S1; SL=3)
1486 # in: IPv6(A, B2)SRH(B3, B2, B1; SL=1)
1487 # out: IPv6(A, S1)SRH(B2, S3, S2, S1; SL=3)SRH(B3, B2, B1; SL=1)
1489 # get first (outer) IPv6 header of rx'ed packet
1490 rx_ip = rx_pkt.getlayer(IPv6)
1495 rx_udp = rx_pkt[UDP]
1497 tx_ip = tx_pkt.getlayer(IPv6)
1500 # some packets have been tx'ed with an SRH, some without it
1501 # get SRH if tx'ed packet has it
1502 if tx_pkt.haslayer(IPv6ExtHdrSegmentRouting):
1503 tx_srh = tx_pkt.getlayer(IPv6ExtHdrSegmentRouting)
1504 tx_ip2 = tx_pkt.getlayer(IPv6, 2)
1505 tx_udp = tx_pkt[UDP]
1507 # expected segment-list (make copy of SR Policy segment list)
1508 seglist = self.sr_policy.segments[:]
1509 # expected seglist has initial dest addr as last segment
1510 seglist.append(tx_ip.dst)
1511 # reverse list to get order as in SRH
1512 tx_seglist = seglist[::-1]
1514 # get source address of SR Policy
1515 sr_policy_source = self.sr_policy.source
1517 # checks common to cases tx with and without SRH
1518 # rx'ed packet should have SRH and only one IPv6 header
1519 self.assertTrue(rx_pkt.haslayer(IPv6ExtHdrSegmentRouting))
1520 self.assertFalse(rx_ip.payload.haslayer(IPv6))
1522 rx_srh = rx_pkt.getlayer(IPv6ExtHdrSegmentRouting)
1524 # rx'ed ip.src should be equal to tx'ed ip.src
1525 self.assertEqual(rx_ip.src, tx_ip.src)
1526 # rx'ed ip.dst should be equal to sidlist[lastentry]
1527 self.assertEqual(rx_ip.dst, tx_seglist[-1])
1529 # rx'ed seglist should be equal to expected seglist
1530 self.assertEqual(rx_srh.addresses, tx_seglist)
1531 # segleft should be equal to size(expected seglist)-1
1532 self.assertEqual(rx_srh.segleft, len(tx_seglist)-1)
1533 # segleft should be equal to lastentry
1534 self.assertEqual(rx_srh.segleft, rx_srh.lastentry)
1536 if tx_srh: # packet was tx'ed with SRH
1537 # packet should have 2nd SRH
1538 self.assertTrue(rx_srh.payload.haslayer(IPv6ExtHdrSegmentRouting))
1540 rx_srh2 = rx_pkt.getlayer(IPv6ExtHdrSegmentRouting, 2)
1542 # rx'ed srh2.addresses should be equal to tx'ed srh.addresses
1543 self.assertEqual(rx_srh2.addresses, tx_srh.addresses)
1544 # rx'ed srh2.segleft should be equal to tx'ed srh.segleft
1545 self.assertEqual(rx_srh2.segleft, tx_srh.segleft)
1546 # rx'ed srh2.lastentry should be equal to tx'ed srh.lastentry
1547 self.assertEqual(rx_srh2.lastentry, tx_srh.lastentry)
1549 else: # packet was tx'ed without SRH
1550 # rx packet should have no other SRH
1551 self.assertFalse(rx_srh.payload.haslayer(IPv6ExtHdrSegmentRouting))
1553 # UDP layer should be unchanged
1554 self.assertEqual(rx_udp, tx_udp)
1556 self.logger.debug("packet verification: SUCCESS")
1558 def compare_rx_tx_packet_End(self, tx_pkt, rx_pkt):
1559 """ Compare input and output packet after passing End (without PSP)
1561 :param tx_pkt: transmitted packet
1562 :param rx_pkt: received packet
1564 # End (no PSP) updates the headers as follows:
1566 # in: IPv6(A, S1)SRH(S3, S2, S1; SL=2)
1567 # out: IPv6(A, S2)SRH(S3, S2, S1; SL=1)
1569 # get first (outer) IPv6 header of rx'ed packet
1570 rx_ip = rx_pkt.getlayer(IPv6)
1573 rx_udp = rx_pkt[UDP]
1575 tx_ip = tx_pkt.getlayer(IPv6)
1576 # we know the packet has been tx'ed
1577 # with an inner IPv6 header and an SRH
1578 tx_ip2 = tx_pkt.getlayer(IPv6, 2)
1579 tx_srh = tx_pkt.getlayer(IPv6ExtHdrSegmentRouting)
1580 tx_udp = tx_pkt[UDP]
1582 # common checks, regardless of tx segleft value
1583 # rx'ed packet should have 2nd IPv6 header
1584 self.assertTrue(rx_ip.payload.haslayer(IPv6))
1585 # get second (inner) IPv6 header
1586 rx_ip2 = rx_pkt.getlayer(IPv6, 2)
1588 if tx_ip.segleft > 0:
1589 # SRH should NOT have been popped:
1590 # End SID without PSP does not pop SRH if segleft>0
1591 self.assertTrue(rx_pkt.haslayer(IPv6ExtHdrSegmentRouting))
1592 rx_srh = rx_pkt.getlayer(IPv6ExtHdrSegmentRouting)
1594 # received ip.src should be equal to expected ip.src
1595 self.assertEqual(rx_ip.src, tx_ip.src)
1596 # sidlist should be unchanged
1597 self.assertEqual(rx_srh.addresses, tx_srh.addresses)
1598 # segleft should have been decremented
1599 self.assertEqual(rx_srh.segleft, tx_srh.segleft-1)
1600 # received ip.dst should be equal to sidlist[segleft]
1601 self.assertEqual(rx_ip.dst, rx_srh.addresses[rx_srh.segleft])
1602 # lastentry should be unchanged
1603 self.assertEqual(rx_srh.lastentry, tx_srh.lastentry)
1604 # inner IPv6 packet (ip2) should be unchanged
1605 self.assertEqual(rx_ip2.src, tx_ip2.src)
1606 self.assertEqual(rx_ip2.dst, tx_ip2.dst)
1607 # else: # tx_ip.segleft == 0
1608 # TODO: Does this work with 2 SRHs in ingress packet?
1610 # UDP layer should be unchanged
1611 self.assertEqual(rx_udp, tx_udp)
1613 self.logger.debug("packet verification: SUCCESS")
1615 def compare_rx_tx_packet_End_PSP(self, tx_pkt, rx_pkt):
1616 """ Compare input and output packet after passing End with PSP
1618 :param tx_pkt: transmitted packet
1619 :param rx_pkt: received packet
1621 # End (PSP) updates the headers as follows:
1622 # IPv6 + SRH (SL>1):
1623 # in: IPv6(A, S1)SRH(S3, S2, S1; SL=2)
1624 # out: IPv6(A, S2)SRH(S3, S2, S1; SL=1)
1625 # IPv6 + SRH (SL=1):
1626 # in: IPv6(A, S2)SRH(S3, S2, S1; SL=1)
1629 # get first (outer) IPv6 header of rx'ed packet
1630 rx_ip = rx_pkt.getlayer(IPv6)
1633 rx_udp = rx_pkt[UDP]
1635 tx_ip = tx_pkt.getlayer(IPv6)
1636 # we know the packet has been tx'ed
1637 # with an inner IPv6 header and an SRH
1638 tx_ip2 = tx_pkt.getlayer(IPv6, 2)
1639 tx_srh = tx_pkt.getlayer(IPv6ExtHdrSegmentRouting)
1640 tx_udp = tx_pkt[UDP]
1642 # common checks, regardless of tx segleft value
1643 self.assertTrue(rx_ip.payload.haslayer(IPv6))
1644 rx_ip2 = rx_pkt.getlayer(IPv6, 2)
1645 # inner IPv6 packet (ip2) should be unchanged
1646 self.assertEqual(rx_ip2.src, tx_ip2.src)
1647 self.assertEqual(rx_ip2.dst, tx_ip2.dst)
1649 if tx_ip.segleft > 1:
1650 # SRH should NOT have been popped:
1651 # End SID with PSP does not pop SRH if segleft>1
1652 # rx'ed packet should have SRH
1653 self.assertTrue(rx_pkt.haslayer(IPv6ExtHdrSegmentRouting))
1654 rx_srh = rx_pkt.getlayer(IPv6ExtHdrSegmentRouting)
1656 # received ip.src should be equal to expected ip.src
1657 self.assertEqual(rx_ip.src, tx_ip.src)
1658 # sidlist should be unchanged
1659 self.assertEqual(rx_srh.addresses, tx_srh.addresses)
1660 # segleft should have been decremented
1661 self.assertEqual(rx_srh.segleft, tx_srh.segleft-1)
1662 # received ip.dst should be equal to sidlist[segleft]
1663 self.assertEqual(rx_ip.dst, rx_srh.addresses[rx_srh.segleft])
1664 # lastentry should be unchanged
1665 self.assertEqual(rx_srh.lastentry, tx_srh.lastentry)
1667 else: # tx_ip.segleft <= 1
1668 # SRH should have been popped:
1669 # End SID with PSP and segleft=1 pops SRH
1670 # the two IPv6 headers are still present
1671 # outer IPv6 header has DA == last segment of popped SRH
1672 # SRH should not be present
1673 self.assertFalse(rx_pkt.haslayer(IPv6ExtHdrSegmentRouting))
1674 # outer IPv6 header ip.src should be equal to tx'ed ip.src
1675 self.assertEqual(rx_ip.src, tx_ip.src)
1676 # outer IPv6 header ip.dst should be = to tx'ed sidlist[segleft-1]
1677 self.assertEqual(rx_ip.dst, tx_srh.addresses[tx_srh.segleft-1])
1679 # UDP layer should be unchanged
1680 self.assertEqual(rx_udp, tx_udp)
1682 self.logger.debug("packet verification: SUCCESS")
1684 def compare_rx_tx_packet_End_DX6(self, tx_pkt, rx_pkt):
1685 """ Compare input and output packet after passing End.DX6
1687 :param tx_pkt: transmitted packet
1688 :param rx_pkt: received packet
1690 # End.DX6 updates the headers as follows:
1691 # IPv6 + SRH (SL=0):
1692 # in: IPv6(A, S3)SRH(S3, S2, S1; SL=0)IPv6(B, D)
1695 # in: IPv6(A, S3)IPv6(B, D)
1698 # get first (outer) IPv6 header of rx'ed packet
1699 rx_ip = rx_pkt.getlayer(IPv6)
1701 tx_ip = tx_pkt.getlayer(IPv6)
1702 tx_ip2 = tx_pkt.getlayer(IPv6, 2)
1704 # verify if rx'ed packet has no SRH
1705 self.assertFalse(rx_pkt.haslayer(IPv6ExtHdrSegmentRouting))
1707 # the whole rx_ip pkt should be equal to tx_ip2
1708 # except for the hlim field
1709 # -> adjust tx'ed hlim to expected hlim
1710 tx_ip2.hlim = tx_ip2.hlim - 1
1712 self.assertEqual(rx_ip, tx_ip2)
1714 self.logger.debug("packet verification: SUCCESS")
1716 def compare_rx_tx_packet_End_DX4(self, tx_pkt, rx_pkt):
1717 """ Compare input and output packet after passing End.DX4
1719 :param tx_pkt: transmitted packet
1720 :param rx_pkt: received packet
1722 # End.DX4 updates the headers as follows:
1723 # IPv6 + SRH (SL=0):
1724 # in: IPv6(A, S3)SRH(S3, S2, S1; SL=0)IPv4(B, D)
1727 # in: IPv6(A, S3)IPv4(B, D)
1730 # get IPv4 header of rx'ed packet
1731 rx_ip = rx_pkt.getlayer(IP)
1733 tx_ip = tx_pkt.getlayer(IPv6)
1734 tx_ip2 = tx_pkt.getlayer(IP)
1736 # verify if rx'ed packet has no SRH
1737 self.assertFalse(rx_pkt.haslayer(IPv6ExtHdrSegmentRouting))
1739 # the whole rx_ip pkt should be equal to tx_ip2
1740 # except for the ttl field and ip checksum
1741 # -> adjust tx'ed ttl to expected ttl
1742 tx_ip2.ttl = tx_ip2.ttl - 1
1743 # -> set tx'ed ip checksum to None and let scapy recompute
1744 tx_ip2.chksum = None
1745 # read back the pkt (with str()) to force computing these fields
1746 # probably other ways to accomplish this are possible
1747 tx_ip2 = IP(scapy.compat.raw(tx_ip2))
1749 self.assertEqual(rx_ip, tx_ip2)
1751 self.logger.debug("packet verification: SUCCESS")
1753 def compare_rx_tx_packet_End_DX2(self, tx_pkt, rx_pkt):
1754 """ Compare input and output packet after passing End.DX2
1756 :param tx_pkt: transmitted packet
1757 :param rx_pkt: received packet
1759 # End.DX2 updates the headers as follows:
1760 # IPv6 + SRH (SL=0):
1761 # in: IPv6(A, S3)SRH(S3, S2, S1; SL=0)L2
1767 # get IPv4 header of rx'ed packet
1768 rx_eth = rx_pkt.getlayer(Ether)
1770 tx_ip = tx_pkt.getlayer(IPv6)
1771 # we can't just get the 2nd Ether layer
1772 # get the Raw content and dissect it as Ether
1773 tx_eth1 = Ether(scapy.compat.raw(tx_pkt[Raw]))
1775 # verify if rx'ed packet has no SRH
1776 self.assertFalse(rx_pkt.haslayer(IPv6ExtHdrSegmentRouting))
1778 # the whole rx_eth pkt should be equal to tx_eth1
1779 self.assertEqual(rx_eth, tx_eth1)
1781 self.logger.debug("packet verification: SUCCESS")
1783 def create_stream(self, src_if, dst_if, packet_header, packet_sizes,
1785 """Create SRv6 input packet stream for defined interface.
1787 :param VppInterface src_if: Interface to create packet stream for
1788 :param VppInterface dst_if: destination interface of packet stream
1789 :param packet_header: Layer3 scapy packet headers,
1790 L2 is added when not provided,
1791 Raw(payload) with packet_info is added
1792 :param list packet_sizes: packet stream pckt sizes,sequentially applied
1793 to packets in stream have
1794 :param int count: number of packets in packet stream
1795 :return: list of packets
1797 self.logger.info("Creating packets")
1799 for i in range(0, count-1):
1800 payload_info = self.create_packet_info(src_if, dst_if)
1802 "Creating packet with index %d" % (payload_info.index))
1803 payload = self.info_to_payload(payload_info)
1804 # add L2 header if not yet provided in packet_header
1805 if packet_header.getlayer(0).name == 'Ethernet':
1806 p = (packet_header /
1809 p = (Ether(dst=src_if.local_mac, src=src_if.remote_mac) /
1812 size = packet_sizes[i % len(packet_sizes)]
1813 self.logger.debug("Packet size %d" % (size))
1814 self.extend_packet(p, size)
1815 # we need to store the packet with the automatic fields computed
1816 # read back the dumped packet (with str())
1817 # to force computing these fields
1818 # probably other ways are possible
1819 p = Ether(scapy.compat.raw(p))
1820 payload_info.data = p.copy()
1821 self.logger.debug(ppp("Created packet:", p))
1823 self.logger.info("Done creating packets")
1826 def send_and_verify_pkts(self, input, pkts, output, compare_func):
1827 """Send packets and verify received packets using compare_func
1829 :param input: ingress interface of DUT
1830 :param pkts: list of packets to transmit
1831 :param output: egress interface of DUT
1832 :param compare_func: function to compare in and out packets
1834 # add traffic stream to input interface
1835 input.add_stream(pkts)
1837 # enable capture on all interfaces
1838 self.pg_enable_capture(self.pg_interfaces)
1841 self.logger.info("Starting traffic")
1844 # get output capture
1845 self.logger.info("Getting packet capture")
1846 capture = output.get_capture()
1848 # assert nothing was captured on input interface
1849 input.assert_nothing_captured()
1851 # verify captured packets
1852 self.verify_captured_pkts(output, capture, compare_func)
1854 def create_packet_header_IPv6(self, dst):
1855 """Create packet header: IPv6 header, UDP header
1857 :param dst: IPv6 destination address
1859 IPv6 source address is 1234::1
1860 UDP source port and destination port are 1234
1863 p = (IPv6(src='1234::1', dst=dst) /
1864 UDP(sport=1234, dport=1234))
1867 def create_packet_header_IPv6_SRH(self, sidlist, segleft):
1868 """Create packet header: IPv6 header with SRH, UDP header
1870 :param list sidlist: segment list
1871 :param int segleft: segments-left field value
1873 IPv6 destination address is set to sidlist[segleft]
1874 IPv6 source addresses are 1234::1 and 4321::1
1875 UDP source port and destination port are 1234
1878 p = (IPv6(src='1234::1', dst=sidlist[segleft]) /
1879 IPv6ExtHdrSegmentRouting(addresses=sidlist) /
1880 UDP(sport=1234, dport=1234))
1883 def create_packet_header_IPv6_SRH_IPv6(self, dst, sidlist, segleft):
1884 """Create packet header: IPv6 encapsulated in SRv6:
1885 IPv6 header with SRH, IPv6 header, UDP header
1887 :param ipv6address dst: inner IPv6 destination address
1888 :param list sidlist: segment list of outer IPv6 SRH
1889 :param int segleft: segments-left field of outer IPv6 SRH
1891 Outer IPv6 destination address is set to sidlist[segleft]
1892 IPv6 source addresses are 1234::1 and 4321::1
1893 UDP source port and destination port are 1234
1896 p = (IPv6(src='1234::1', dst=sidlist[segleft]) /
1897 IPv6ExtHdrSegmentRouting(addresses=sidlist,
1898 segleft=segleft, nh=41) /
1899 IPv6(src='4321::1', dst=dst) /
1900 UDP(sport=1234, dport=1234))
1903 def create_packet_header_IPv6_IPv6(self, dst_inner, dst_outer):
1904 """Create packet header: IPv6 encapsulated in IPv6:
1905 IPv6 header, IPv6 header, UDP header
1907 :param ipv6address dst_inner: inner IPv6 destination address
1908 :param ipv6address dst_outer: outer IPv6 destination address
1910 IPv6 source addresses are 1234::1 and 4321::1
1911 UDP source port and destination port are 1234
1914 p = (IPv6(src='1234::1', dst=dst_outer) /
1915 IPv6(src='4321::1', dst=dst_inner) /
1916 UDP(sport=1234, dport=1234))
1919 def create_packet_header_IPv6_SRH_SRH_IPv6(self, dst, sidlist1, segleft1,
1920 sidlist2, segleft2):
1921 """Create packet header: IPv6 encapsulated in SRv6 with 2 SRH:
1922 IPv6 header with SRH, 2nd SRH, IPv6 header, UDP header
1924 :param ipv6address dst: inner IPv6 destination address
1925 :param list sidlist1: segment list of outer IPv6 SRH
1926 :param int segleft1: segments-left field of outer IPv6 SRH
1927 :param list sidlist2: segment list of inner IPv6 SRH
1928 :param int segleft2: segments-left field of inner IPv6 SRH
1930 Outer IPv6 destination address is set to sidlist[segleft]
1931 IPv6 source addresses are 1234::1 and 4321::1
1932 UDP source port and destination port are 1234
1935 p = (IPv6(src='1234::1', dst=sidlist1[segleft1]) /
1936 IPv6ExtHdrSegmentRouting(addresses=sidlist1,
1937 segleft=segleft1, nh=43) /
1938 IPv6ExtHdrSegmentRouting(addresses=sidlist2,
1939 segleft=segleft2, nh=41) /
1940 IPv6(src='4321::1', dst=dst) /
1941 UDP(sport=1234, dport=1234))
1944 def create_packet_header_IPv4(self, dst):
1945 """Create packet header: IPv4 header, UDP header
1947 :param dst: IPv4 destination address
1949 IPv4 source address is 123.1.1.1
1950 UDP source port and destination port are 1234
1953 p = (IP(src='123.1.1.1', dst=dst) /
1954 UDP(sport=1234, dport=1234))
1957 def create_packet_header_IPv6_IPv4(self, dst_inner, dst_outer):
1958 """Create packet header: IPv4 encapsulated in IPv6:
1959 IPv6 header, IPv4 header, UDP header
1961 :param ipv4address dst_inner: inner IPv4 destination address
1962 :param ipv6address dst_outer: outer IPv6 destination address
1964 IPv6 source address is 1234::1
1965 IPv4 source address is 123.1.1.1
1966 UDP source port and destination port are 1234
1969 p = (IPv6(src='1234::1', dst=dst_outer) /
1970 IP(src='123.1.1.1', dst=dst_inner) /
1971 UDP(sport=1234, dport=1234))
1974 def create_packet_header_IPv6_SRH_IPv4(self, dst, sidlist, segleft):
1975 """Create packet header: IPv4 encapsulated in SRv6:
1976 IPv6 header with SRH, IPv4 header, UDP header
1978 :param ipv4address dst: inner IPv4 destination address
1979 :param list sidlist: segment list of outer IPv6 SRH
1980 :param int segleft: segments-left field of outer IPv6 SRH
1982 Outer IPv6 destination address is set to sidlist[segleft]
1983 IPv6 source address is 1234::1
1984 IPv4 source address is 123.1.1.1
1985 UDP source port and destination port are 1234
1988 p = (IPv6(src='1234::1', dst=sidlist[segleft]) /
1989 IPv6ExtHdrSegmentRouting(addresses=sidlist,
1990 segleft=segleft, nh=4) /
1991 IP(src='123.1.1.1', dst=dst) /
1992 UDP(sport=1234, dport=1234))
1995 def create_packet_header_L2(self, vlan=0):
1996 """Create packet header: L2 header
1998 :param vlan: if vlan!=0 then add 802.1q header
2000 # Note: the dst addr ('00:55:44:33:22:11') is used in
2001 # the compare function compare_rx_tx_packet_T_Encaps_L2
2002 # to detect presence of L2 in SRH payload
2003 p = Ether(src='00:11:22:33:44:55', dst='00:55:44:33:22:11')
2004 etype = 0x8137 # IPX
2007 p /= Dot1Q(vlan=vlan, type=etype)
2012 def create_packet_header_IPv6_SRH_L2(self, sidlist, segleft, vlan=0):
2013 """Create packet header: L2 encapsulated in SRv6:
2014 IPv6 header with SRH, L2
2016 :param list sidlist: segment list of outer IPv6 SRH
2017 :param int segleft: segments-left field of outer IPv6 SRH
2018 :param vlan: L2 vlan; if vlan!=0 then add 802.1q header
2020 Outer IPv6 destination address is set to sidlist[segleft]
2021 IPv6 source address is 1234::1
2023 eth = Ether(src='00:11:22:33:44:55', dst='00:55:44:33:22:11')
2024 etype = 0x8137 # IPX
2027 eth /= Dot1Q(vlan=vlan, type=etype)
2031 p = (IPv6(src='1234::1', dst=sidlist[segleft]) /
2032 IPv6ExtHdrSegmentRouting(addresses=sidlist,
2033 segleft=segleft, nh=59) /
2037 def create_packet_header_IPv6_L2(self, dst_outer, vlan=0):
2038 """Create packet header: L2 encapsulated in IPv6:
2041 :param ipv6address dst_outer: outer IPv6 destination address
2042 :param vlan: L2 vlan; if vlan!=0 then add 802.1q header
2044 eth = Ether(src='00:11:22:33:44:55', dst='00:55:44:33:22:11')
2045 etype = 0x8137 # IPX
2048 eth /= Dot1Q(vlan=vlan, type=etype)
2052 p = (IPv6(src='1234::1', dst=dst_outer, nh=59) / eth)
2055 def get_payload_info(self, packet):
2056 """ Extract the payload_info from the packet
2058 # in most cases, payload_info is in packet[Raw]
2059 # but packet[Raw] gives the complete payload
2060 # (incl L2 header) for the T.Encaps L2 case
2062 payload_info = self.payload_to_info(packet[Raw])
2065 # remote L2 header from packet[Raw]:
2066 # take packet[Raw], convert it to an Ether layer
2067 # and then extract Raw from it
2068 payload_info = self.payload_to_info(
2069 Ether(scapy.compat.r(packet[Raw]))[Raw])
2073 def verify_captured_pkts(self, dst_if, capture, compare_func):
2075 Verify captured packet stream for specified interface.
2076 Compare ingress with egress packets using the specified compare fn
2078 :param dst_if: egress interface of DUT
2079 :param capture: captured packets
2080 :param compare_func: function to compare in and out packet
2082 self.logger.info("Verifying capture on interface %s using function %s"
2083 % (dst_if.name, compare_func.__name__))
2086 for i in self.pg_interfaces:
2087 last_info[i.sw_if_index] = None
2088 dst_sw_if_index = dst_if.sw_if_index
2090 for packet in capture:
2092 # extract payload_info from packet's payload
2093 payload_info = self.get_payload_info(packet)
2094 packet_index = payload_info.index
2096 self.logger.debug("Verifying packet with index %d"
2098 # packet should have arrived on the expected interface
2099 self.assertEqual(payload_info.dst, dst_sw_if_index)
2101 "Got packet on interface %s: src=%u (idx=%u)" %
2102 (dst_if.name, payload_info.src, packet_index))
2104 # search for payload_info with same src and dst if_index
2105 # this will give us the transmitted packet
2106 next_info = self.get_next_packet_info_for_interface2(
2107 payload_info.src, dst_sw_if_index,
2108 last_info[payload_info.src])
2109 last_info[payload_info.src] = next_info
2110 # next_info should not be None
2111 self.assertTrue(next_info is not None)
2112 # index of tx and rx packets should be equal
2113 self.assertEqual(packet_index, next_info.index)
2114 # data field of next_info contains the tx packet
2115 txed_packet = next_info.data
2117 self.logger.debug(ppp("Transmitted packet:",
2118 txed_packet)) # ppp=Pretty Print Packet
2120 self.logger.debug(ppp("Received packet:", packet))
2122 # compare rcvd packet with expected packet using compare_func
2123 compare_func(txed_packet, packet)
2126 self.logger.error(ppp("Unexpected or invalid packet:", packet))
2129 # have all expected packets arrived?
2130 for i in self.pg_interfaces:
2131 remaining_packet = self.get_next_packet_info_for_interface2(
2132 i.sw_if_index, dst_sw_if_index, last_info[i.sw_if_index])
2133 self.assertTrue(remaining_packet is None,
2134 "Interface %s: Packet expected from interface %s "
2135 "didn't arrive" % (dst_if.name, i.name))
2138 if __name__ == '__main__':
2139 unittest.main(testRunner=VppTestRunner)