X-Git-Url: https://gerrit.fd.io/r/gitweb?a=blobdiff_plain;f=test%2Ftest_srv6_as.py;h=87cafd1e5ed5e1d2f12db3dfa14aea9fb2e6b633;hb=051579d0f2bf5a408ed8439cfd28831e846389ab;hp=64ff957da4b19cb5f8a2ca63ef2a09b4887b2045;hpb=661f91fe0a6bd87040408d45d116b63c0811f4f9;p=vpp.git diff --git a/test/test_srv6_as.py b/test/test_srv6_as.py old mode 100755 new mode 100644 index 64ff957da4b..87cafd1e5ed --- a/test/test_srv6_as.py +++ b/test/test_srv6_as.py @@ -1,35 +1,43 @@ -#!/usr/bin/env python +#!/usr/bin/env python3 import unittest import binascii from socket import AF_INET6 from framework import VppTestCase, VppTestRunner -from vpp_ip_route import VppIpRoute, VppRoutePath, DpoProto, VppIpTable -from vpp_srv6 import SRv6LocalSIDBehaviors, VppSRv6LocalSID, VppSRv6Policy, \ - SRv6PolicyType, VppSRv6Steering, SRv6PolicySteeringTypes - +from vpp_ip_route import VppIpRoute, VppRoutePath, FibPathProto, VppIpTable +from vpp_srv6 import ( + SRv6LocalSIDBehaviors, + VppSRv6LocalSID, + VppSRv6Policy, + SRv6PolicyType, + VppSRv6Steering, + SRv6PolicySteeringTypes, +) + +import scapy.compat from scapy.packet import Raw from scapy.layers.l2 import Ether, Dot1Q from scapy.layers.inet6 import IPv6, UDP, IPv6ExtHdrSegmentRouting from scapy.layers.inet import IP, UDP -from scapy.utils import inet_pton, inet_ntop - from util import ppp -class TestSRv6(VppTestCase): - """ SRv6 Static Proxy plugin Test Case """ +class TestSRv6As(VppTestCase): + """SRv6 Static Proxy plugin Test Case""" @classmethod def setUpClass(self): - super(TestSRv6, self).setUpClass() + super(TestSRv6As, self).setUpClass() + + @classmethod + def tearDownClass(cls): + super(TestSRv6As, cls).tearDownClass() def setUp(self): - """ Perform test setup before each test case. - """ - super(TestSRv6, self).setUp() + """Perform test setup before each test case.""" + super(TestSRv6As, self).setUp() # packet sizes, inclusive L2 overhead self.pg_packet_sizes = [64, 512, 1518, 9018] @@ -38,17 +46,15 @@ class TestSRv6(VppTestCase): self.reset_packet_infos() def tearDown(self): - """ Clean up test setup after each test case. - """ + """Clean up test setup after each test case.""" self.teardown_interfaces() - super(TestSRv6, self).tearDown() + super(TestSRv6As, self).tearDown() - def configure_interface(self, - interface, - ipv6=False, ipv4=False, - ipv6_table_id=0, ipv4_table_id=0): - """ Configure interface. + def configure_interface( + self, interface, ipv6=False, ipv4=False, ipv6_table_id=0, ipv4_table_id=0 + ): + """Configure interface. :param ipv6: configure IPv6 on interface :param ipv4: configure IPv4 on interface :param ipv6_table_id: FIB table_id for IPv6 @@ -67,9 +73,8 @@ class TestSRv6(VppTestCase): interface.resolve_arp() interface.admin_up() - def setup_interfaces(self, ipv6=[], ipv4=[], - ipv6_table_id=[], ipv4_table_id=[]): - """ Create and configure interfaces. + def setup_interfaces(self, ipv6=[], ipv4=[], ipv6_table_id=[], ipv4_table_id=[]): + """Create and configure interfaces. :param ipv6: list of interface IPv6 capabilities :param ipv4: list of interface IPv4 capabilities @@ -104,22 +109,21 @@ class TestSRv6(VppTestCase): # setup all interfaces for i in range(count): intf = self.pg_interfaces[i] - self.configure_interface(intf, - ipv6[i], ipv4[i], - ipv6_table_id[i], ipv4_table_id[i]) + self.configure_interface( + intf, ipv6[i], ipv4[i], ipv6_table_id[i], ipv4_table_id[i] + ) if any(ipv6): self.logger.debug(self.vapi.cli("show ip6 neighbors")) if any(ipv4): - self.logger.debug(self.vapi.cli("show ip arp")) + self.logger.debug(self.vapi.cli("show ip4 neighbors")) self.logger.debug(self.vapi.cli("show interface")) self.logger.debug(self.vapi.cli("show hardware")) return self.pg_interfaces def teardown_interfaces(self): - """ Unconfigure and bring down interface. - """ + """Unconfigure and bring down interface.""" self.logger.debug("Tearing down interfaces") # tear down all interfaces # AFAIK they cannot be deleted @@ -131,77 +135,83 @@ class TestSRv6(VppTestCase): i.set_table_ip6(0) def test_SRv6_End_AS_IPv6_noSRH(self): - """ Test SRv6 End.AS behavior with IPv6 traffic and no SRH rewrite. - """ + """Test SRv6 End.AS behavior with IPv6 traffic and no SRH rewrite.""" self.run_SRv6_End_AS_IPv6( - sid_list=['a1::', 'a2::a6', 'a3::'], + sid_list=["a1::", "a2::a6", "a3::"], test_sid_index=1, - rewrite_src_addr='a2::') + rewrite_src_addr="a2::", + ) def test_SRv6_End_AS_IPv6_SRH(self): - """ Test SRv6 End.AS behavior with IPv6 traffic and SRH rewrite. - """ + """Test SRv6 End.AS behavior with IPv6 traffic and SRH rewrite.""" self.run_SRv6_End_AS_IPv6( - sid_list=['a1::a6', 'a2::', 'a3::'], + sid_list=["a1::a6", "a2::", "a3::"], test_sid_index=0, - rewrite_src_addr='a1::') + rewrite_src_addr="a1::", + ) def test_SRv6_End_AS_IPv4_noSRH(self): - """ Test SRv6 End.AS behavior with IPv4 traffic and no SRH rewrite. - """ + """Test SRv6 End.AS behavior with IPv4 traffic and no SRH rewrite.""" self.run_SRv6_End_AS_IPv4( - sid_list=['a1::', 'a2::a6', 'a3::'], + sid_list=["a1::", "a2::a6", "a3::"], test_sid_index=1, - rewrite_src_addr='a2::') + rewrite_src_addr="a2::", + ) def test_SRv6_End_AS_IPv4_SRH(self): - """ Test SRv6 End.AS behavior with IPv4 traffic and SRH rewrite. - """ + """Test SRv6 End.AS behavior with IPv4 traffic and SRH rewrite.""" self.run_SRv6_End_AS_IPv4( - sid_list=['a1::a6', 'a2::', 'a3::'], + sid_list=["a1::a6", "a2::", "a3::"], test_sid_index=0, - rewrite_src_addr='a1::') + rewrite_src_addr="a1::", + ) def test_SRv6_End_AS_L2_noSRH(self): - """ Test SRv6 End.AS behavior with L2 traffic and no SRH rewrite. - """ + """Test SRv6 End.AS behavior with L2 traffic and no SRH rewrite.""" self.run_SRv6_End_AS_L2( - sid_list=['a1::', 'a2::a6', 'a3::'], + sid_list=["a1::", "a2::a6", "a3::"], test_sid_index=1, - rewrite_src_addr='a2::') + rewrite_src_addr="a2::", + ) def test_SRv6_End_AS_L2_SRH(self): - """ Test SRv6 End.AS behavior with L2 traffic and SRH rewrite. - """ + """Test SRv6 End.AS behavior with L2 traffic and SRH rewrite.""" self.run_SRv6_End_AS_L2( - sid_list=['a1::a6', 'a2::', 'a3::'], + sid_list=["a1::a6", "a2::", "a3::"], test_sid_index=0, - rewrite_src_addr='a1::') + rewrite_src_addr="a1::", + ) def run_SRv6_End_AS_L2(self, sid_list, test_sid_index, rewrite_src_addr): - """ Run SRv6 End.AS test with L2 traffic. - """ + """Run SRv6 End.AS test with L2 traffic.""" self.rewrite_src_addr = rewrite_src_addr - self.rewrite_sid_list = sid_list[test_sid_index + 1::] + self.rewrite_sid_list = sid_list[test_sid_index + 1 : :] # send traffic to one destination interface # source and destination interfaces are IPv6 only self.setup_interfaces(ipv6=[True, False]) # configure route to next segment - route = VppIpRoute(self, sid_list[test_sid_index + 1], 128, - [VppRoutePath(self.pg0.remote_ip6, - self.pg0.sw_if_index, - proto=DpoProto.DPO_PROTO_IP6)], - is_ip6=1) + route = VppIpRoute( + self, + sid_list[test_sid_index + 1], + 128, + [VppRoutePath(self.pg0.remote_ip6, self.pg0.sw_if_index)], + ) route.add_vpp_config() # configure SRv6 localSID behavior - cli_str = "sr localsid address " + sid_list[test_sid_index] \ - + " behavior end.as" \ - + " oif " + self.pg1.name \ - + " iif " + self.pg1.name \ - + " src " + self.rewrite_src_addr + cli_str = ( + "sr localsid address " + + sid_list[test_sid_index] + + " behavior end.as" + + " oif " + + self.pg1.name + + " iif " + + self.pg1.name + + " src " + + self.rewrite_src_addr + ) for s in self.rewrite_sid_list: cli_str += " next " + s self.vapi.cli(cli_str) @@ -214,17 +224,18 @@ class TestSRv6(VppTestCase): # prepare L2 in SRv6 headers packet_header1 = self.create_packet_header_IPv6_SRH_L2( - sidlist=sid_list[::-1], - segleft=len(sid_list) - test_sid_index - 1, - vlan=0) + sidlist=sid_list[::-1], segleft=len(sid_list) - test_sid_index - 1, vlan=0 + ) # generate packets (pg0->pg1) - pkts1 = self.create_stream(self.pg0, self.pg1, packet_header1, - self.pg_packet_sizes, count) + pkts1 = self.create_stream( + self.pg0, self.pg1, packet_header1, self.pg_packet_sizes, count + ) # send packets and verify received packets - self.send_and_verify_pkts(self.pg0, pkts1, self.pg1, - self.compare_rx_tx_packet_End_AS_L2_out) + self.send_and_verify_pkts( + self.pg0, pkts1, self.pg1, self.compare_rx_tx_packet_End_AS_L2_out + ) # log the localsid counters self.logger.info(self.vapi.cli("show sr localsid")) @@ -233,12 +244,14 @@ class TestSRv6(VppTestCase): packet_header2 = self.create_packet_header_L2() # generate returning packets (pg1->pg0) - pkts2 = self.create_stream(self.pg1, self.pg0, packet_header2, - self.pg_packet_sizes, count) + pkts2 = self.create_stream( + self.pg1, self.pg0, packet_header2, self.pg_packet_sizes, count + ) # send packets and verify received packets - self.send_and_verify_pkts(self.pg1, pkts2, self.pg0, - self.compare_rx_tx_packet_End_AS_L2_in) + self.send_and_verify_pkts( + self.pg1, pkts2, self.pg0, self.compare_rx_tx_packet_End_AS_L2_in + ) # log the localsid counters self.logger.info(self.vapi.cli("show sr localsid")) @@ -250,30 +263,37 @@ class TestSRv6(VppTestCase): self.teardown_interfaces() def run_SRv6_End_AS_IPv6(self, sid_list, test_sid_index, rewrite_src_addr): - """ Run SRv6 End.AS test with IPv6 traffic. - """ + """Run SRv6 End.AS test with IPv6 traffic.""" self.rewrite_src_addr = rewrite_src_addr - self.rewrite_sid_list = sid_list[test_sid_index + 1::] + self.rewrite_sid_list = sid_list[test_sid_index + 1 : :] # send traffic to one destination interface # source and destination interfaces are IPv6 only self.setup_interfaces(ipv6=[True, True]) # configure route to next segment - route = VppIpRoute(self, sid_list[test_sid_index + 1], 128, - [VppRoutePath(self.pg0.remote_ip6, - self.pg0.sw_if_index, - proto=DpoProto.DPO_PROTO_IP6)], - is_ip6=1) + route = VppIpRoute( + self, + sid_list[test_sid_index + 1], + 128, + [VppRoutePath(self.pg0.remote_ip6, self.pg0.sw_if_index)], + ) route.add_vpp_config() # configure SRv6 localSID behavior - cli_str = "sr localsid address " + sid_list[test_sid_index] \ - + " behavior end.as" \ - + " nh " + self.pg1.remote_ip6 \ - + " oif " + self.pg1.name \ - + " iif " + self.pg1.name \ - + " src " + self.rewrite_src_addr + cli_str = ( + "sr localsid address " + + sid_list[test_sid_index] + + " behavior end.as" + + " nh " + + self.pg1.remote_ip6 + + " oif " + + self.pg1.name + + " iif " + + self.pg1.name + + " src " + + self.rewrite_src_addr + ) for s in self.rewrite_sid_list: cli_str += " next " + s self.vapi.cli(cli_str) @@ -286,16 +306,18 @@ class TestSRv6(VppTestCase): # prepare IPv6 in SRv6 headers packet_header1 = self.create_packet_header_IPv6_SRH_IPv6( - sidlist=sid_list[::-1], - segleft=len(sid_list) - test_sid_index - 1) + sidlist=sid_list[::-1], segleft=len(sid_list) - test_sid_index - 1 + ) # generate packets (pg0->pg1) - pkts1 = self.create_stream(self.pg0, self.pg1, packet_header1, - self.pg_packet_sizes, count) + pkts1 = self.create_stream( + self.pg0, self.pg1, packet_header1, self.pg_packet_sizes, count + ) # send packets and verify received packets - self.send_and_verify_pkts(self.pg0, pkts1, self.pg1, - self.compare_rx_tx_packet_End_AS_IPv6_out) + self.send_and_verify_pkts( + self.pg0, pkts1, self.pg1, self.compare_rx_tx_packet_End_AS_IPv6_out + ) # log the localsid counters self.logger.info(self.vapi.cli("show sr localsid")) @@ -304,12 +326,14 @@ class TestSRv6(VppTestCase): packet_header2 = self.create_packet_header_IPv6() # generate returning packets (pg1->pg0) - pkts2 = self.create_stream(self.pg1, self.pg0, packet_header2, - self.pg_packet_sizes, count) + pkts2 = self.create_stream( + self.pg1, self.pg0, packet_header2, self.pg_packet_sizes, count + ) # send packets and verify received packets - self.send_and_verify_pkts(self.pg1, pkts2, self.pg0, - self.compare_rx_tx_packet_End_AS_IPv6_in) + self.send_and_verify_pkts( + self.pg1, pkts2, self.pg0, self.compare_rx_tx_packet_End_AS_IPv6_in + ) # log the localsid counters self.logger.info(self.vapi.cli("show sr localsid")) @@ -321,30 +345,37 @@ class TestSRv6(VppTestCase): self.teardown_interfaces() def run_SRv6_End_AS_IPv4(self, sid_list, test_sid_index, rewrite_src_addr): - """ Run SRv6 End.AS test with IPv4 traffic. - """ + """Run SRv6 End.AS test with IPv4 traffic.""" self.rewrite_src_addr = rewrite_src_addr - self.rewrite_sid_list = sid_list[test_sid_index + 1::] + self.rewrite_sid_list = sid_list[test_sid_index + 1 : :] # send traffic to one destination interface # source and destination interfaces are IPv6 only self.setup_interfaces(ipv6=[True, False], ipv4=[True, True]) # configure route to next segment - route = VppIpRoute(self, sid_list[test_sid_index + 1], 128, - [VppRoutePath(self.pg0.remote_ip6, - self.pg0.sw_if_index, - proto=DpoProto.DPO_PROTO_IP6)], - is_ip6=1) + route = VppIpRoute( + self, + sid_list[test_sid_index + 1], + 128, + [VppRoutePath(self.pg0.remote_ip6, self.pg0.sw_if_index)], + ) route.add_vpp_config() # configure SRv6 localSID behavior - cli_str = "sr localsid address " + sid_list[test_sid_index] \ - + " behavior end.as" \ - + " nh " + self.pg1.remote_ip4 \ - + " oif " + self.pg1.name \ - + " iif " + self.pg1.name \ - + " src " + self.rewrite_src_addr + cli_str = ( + "sr localsid address " + + sid_list[test_sid_index] + + " behavior end.as" + + " nh " + + self.pg1.remote_ip4 + + " oif " + + self.pg1.name + + " iif " + + self.pg1.name + + " src " + + self.rewrite_src_addr + ) for s in self.rewrite_sid_list: cli_str += " next " + s self.vapi.cli(cli_str) @@ -357,16 +388,18 @@ class TestSRv6(VppTestCase): # prepare IPv4 in SRv6 headers packet_header1 = self.create_packet_header_IPv6_SRH_IPv4( - sidlist=sid_list[::-1], - segleft=len(sid_list) - test_sid_index - 1) + sidlist=sid_list[::-1], segleft=len(sid_list) - test_sid_index - 1 + ) # generate packets (pg0->pg1) - pkts1 = self.create_stream(self.pg0, self.pg1, packet_header1, - self.pg_packet_sizes, count) + pkts1 = self.create_stream( + self.pg0, self.pg1, packet_header1, self.pg_packet_sizes, count + ) # send packets and verify received packets - self.send_and_verify_pkts(self.pg0, pkts1, self.pg1, - self.compare_rx_tx_packet_End_AS_IPv4_out) + self.send_and_verify_pkts( + self.pg0, pkts1, self.pg1, self.compare_rx_tx_packet_End_AS_IPv4_out + ) # log the localsid counters self.logger.info(self.vapi.cli("show sr localsid")) @@ -375,12 +408,14 @@ class TestSRv6(VppTestCase): packet_header2 = self.create_packet_header_IPv4() # generate returning packets (pg1->pg0) - pkts2 = self.create_stream(self.pg1, self.pg0, packet_header2, - self.pg_packet_sizes, count) + pkts2 = self.create_stream( + self.pg1, self.pg0, packet_header2, self.pg_packet_sizes, count + ) # send packets and verify received packets - self.send_and_verify_pkts(self.pg1, pkts2, self.pg0, - self.compare_rx_tx_packet_End_AS_IPv4_in) + self.send_and_verify_pkts( + self.pg1, pkts2, self.pg0, self.compare_rx_tx_packet_End_AS_IPv4_in + ) # log the localsid counters self.logger.info(self.vapi.cli("show sr localsid")) @@ -392,7 +427,7 @@ class TestSRv6(VppTestCase): self.teardown_interfaces() def compare_rx_tx_packet_End_AS_IPv6_in(self, tx_pkt, rx_pkt): - """ Compare input and output packet after passing End.AS + """Compare input and output packet after passing End.AS :param tx_pkt: transmitted packet :param rx_pkt: received packet @@ -420,7 +455,7 @@ class TestSRv6(VppTestCase): # rx'ed seglist should be equal to expected seglist self.assertEqual(rx_srh.addresses, tx_seglist) # segleft should be equal to size expected seglist-1 - self.assertEqual(rx_srh.segleft, len(tx_seglist)-1) + self.assertEqual(rx_srh.segleft, len(tx_seglist) - 1) # segleft should be equal to lastentry self.assertEqual(rx_srh.segleft, rx_srh.lastentry) # get payload @@ -441,7 +476,7 @@ class TestSRv6(VppTestCase): self.logger.debug("packet verification: SUCCESS") def compare_rx_tx_packet_End_AS_IPv4_in(self, tx_pkt, rx_pkt): - """ Compare input and output packet after passing End.AS + """Compare input and output packet after passing End.AS :param tx_pkt: transmitted packet :param rx_pkt: received packet @@ -470,7 +505,7 @@ class TestSRv6(VppTestCase): # rx'ed seglist should be equal to seglist self.assertEqual(rx_srh.addresses, tx_seglist) # segleft should be equal to size seglist-1 - self.assertEqual(rx_srh.segleft, len(tx_seglist)-1) + self.assertEqual(rx_srh.segleft, len(tx_seglist) - 1) # segleft should be equal to lastentry self.assertEqual(rx_srh.segleft, rx_srh.lastentry) payload = rx_srh.payload @@ -488,14 +523,14 @@ class TestSRv6(VppTestCase): tx_ip.chksum = None # read back the pkt (with str()) to force computing these fields # probably other ways to accomplish this are possible - tx_ip = IP(str(tx_ip)) + tx_ip = IP(scapy.compat.raw(tx_ip)) self.assertEqual(payload, tx_ip) self.logger.debug("packet verification: SUCCESS") def compare_rx_tx_packet_End_AS_L2_in(self, tx_pkt, rx_pkt): - """ Compare input and output packet after passing End.AS + """Compare input and output packet after passing End.AS :param tx_pkt: transmitted packet :param rx_pkt: received packet @@ -523,11 +558,11 @@ class TestSRv6(VppTestCase): # rx'ed seglist should be equal to seglist self.assertEqual(rx_srh.addresses, tx_seglist) # segleft should be equal to size seglist-1 - self.assertEqual(rx_srh.segleft, len(tx_seglist)-1) + self.assertEqual(rx_srh.segleft, len(tx_seglist) - 1) # segleft should be equal to lastentry self.assertEqual(rx_srh.segleft, rx_srh.lastentry) - # nh should be "No Next Header" (59) - self.assertEqual(rx_srh.nh, 59) + # nh should be "No Next Header" (143) + self.assertEqual(rx_srh.nh, 143) # get payload payload = rx_srh.payload else: @@ -537,12 +572,12 @@ class TestSRv6(VppTestCase): payload = rx_ip.payload # the whole rx'ed pkt beyond SRH should be equal to tx'ed pkt - self.assertEqual(Ether(str(payload)), tx_ether) + self.assertEqual(Ether(scapy.compat.raw(payload)), tx_ether) self.logger.debug("packet verification: SUCCESS") def compare_rx_tx_packet_End_AS_IPv6_out(self, tx_pkt, rx_pkt): - """ Compare input and output packet after passing End.AS with IPv6 + """Compare input and output packet after passing End.AS with IPv6 :param tx_pkt: transmitted packet :param rx_pkt: received packet @@ -567,7 +602,7 @@ class TestSRv6(VppTestCase): self.logger.debug("packet verification: SUCCESS") def compare_rx_tx_packet_End_AS_IPv4_out(self, tx_pkt, rx_pkt): - """ Compare input and output packet after passing End.AS with IPv4 + """Compare input and output packet after passing End.AS with IPv4 :param tx_pkt: transmitted packet :param rx_pkt: received packet @@ -590,14 +625,14 @@ class TestSRv6(VppTestCase): tx_ip2.chksum = None # read back the pkt (with str()) to force computing these fields # probably other ways to accomplish this are possible - tx_ip2 = IP(str(tx_ip2)) + tx_ip2 = IP(scapy.compat.raw(tx_ip2)) self.assertEqual(rx_ip, tx_ip2) self.logger.debug("packet verification: SUCCESS") def compare_rx_tx_packet_End_AS_L2_out(self, tx_pkt, rx_pkt): - """ Compare input and output packet after passing End.AS with L2 + """Compare input and output packet after passing End.AS with L2 :param tx_pkt: transmitted packet :param rx_pkt: received packet @@ -609,7 +644,7 @@ class TestSRv6(VppTestCase): tx_ip = tx_pkt.getlayer(IPv6) # we can't just get the 2nd Ether layer # get the Raw content and dissect it as Ether - tx_eth1 = Ether(str(tx_pkt[Raw])) + tx_eth1 = Ether(scapy.compat.raw(tx_pkt[Raw])) # verify if rx'ed packet has no SRH self.assertFalse(rx_pkt.haslayer(IPv6ExtHdrSegmentRouting)) @@ -619,8 +654,7 @@ class TestSRv6(VppTestCase): self.logger.debug("packet verification: SUCCESS") - def create_stream(self, src_if, dst_if, packet_header, packet_sizes, - count): + def create_stream(self, src_if, dst_if, packet_header, packet_sizes, count): """Create SRv6 input packet stream for defined interface. :param VppInterface src_if: Interface to create packet stream for @@ -635,19 +669,19 @@ class TestSRv6(VppTestCase): """ self.logger.info("Creating packets") pkts = [] - for i in range(0, count-1): + for i in range(0, count - 1): payload_info = self.create_packet_info(src_if, dst_if) - self.logger.debug( - "Creating packet with index %d" % (payload_info.index)) + self.logger.debug("Creating packet with index %d" % (payload_info.index)) payload = self.info_to_payload(payload_info) # add L2 header if not yet provided in packet_header - if packet_header.getlayer(0).name == 'Ethernet': - p = (packet_header / - Raw(payload)) + if packet_header.getlayer(0).name == "Ethernet": + p = packet_header / Raw(payload) else: - p = (Ether(dst=src_if.local_mac, src=src_if.remote_mac) / - packet_header / - Raw(payload)) + p = ( + Ether(dst=src_if.local_mac, src=src_if.remote_mac) + / packet_header + / Raw(payload) + ) size = packet_sizes[i % len(packet_sizes)] self.logger.debug("Packet size %d" % (size)) self.extend_packet(p, size) @@ -655,7 +689,7 @@ class TestSRv6(VppTestCase): # read back the dumped packet (with str()) # to force computing these fields # probably other ways are possible - p = Ether(str(p)) + p = Ether(scapy.compat.raw(p)) payload_info.data = p.copy() self.logger.debug(ppp("Created packet:", p)) pkts.append(p) @@ -700,8 +734,7 @@ class TestSRv6(VppTestCase): UDP source port and destination port are 1234 """ - p = (IPv6(src='1234::1', dst='4321::1') / - UDP(sport=1234, dport=1234)) + p = IPv6(src="1234::1", dst="4321::1") / UDP(sport=1234, dport=1234) return p def create_packet_header_IPv6_SRH_IPv6(self, sidlist, segleft): @@ -718,11 +751,12 @@ class TestSRv6(VppTestCase): UDP source port and destination port are 1234 """ - p = (IPv6(src='5678::1', dst=sidlist[segleft]) / - IPv6ExtHdrSegmentRouting(addresses=sidlist, - segleft=segleft, nh=41) / - IPv6(src='1234::1', dst='4321::1') / - UDP(sport=1234, dport=1234)) + p = ( + IPv6(src="5678::1", dst=sidlist[segleft]) + / IPv6ExtHdrSegmentRouting(addresses=sidlist, segleft=segleft, nh=41) + / IPv6(src="1234::1", dst="4321::1") + / UDP(sport=1234, dport=1234) + ) return p def create_packet_header_IPv4(self): @@ -735,8 +769,7 @@ class TestSRv6(VppTestCase): UDP source port and destination port are 1234 """ - p = (IP(src='123.1.1.1', dst='124.1.1.1') / - UDP(sport=1234, dport=1234)) + p = IP(src="123.1.1.1", dst="124.1.1.1") / UDP(sport=1234, dport=1234) return p def create_packet_header_IPv6_SRH_IPv4(self, sidlist, segleft): @@ -754,11 +787,12 @@ class TestSRv6(VppTestCase): UDP source port and destination port are 1234 """ - p = (IPv6(src='1234::1', dst=sidlist[segleft]) / - IPv6ExtHdrSegmentRouting(addresses=sidlist, - segleft=segleft, nh=4) / - IP(src='123.1.1.1', dst='124.1.1.1') / - UDP(sport=1234, dport=1234)) + p = ( + IPv6(src="1234::1", dst=sidlist[segleft]) + / IPv6ExtHdrSegmentRouting(addresses=sidlist, segleft=segleft, nh=4) + / IP(src="123.1.1.1", dst="124.1.1.1") + / UDP(sport=1234, dport=1234) + ) return p def create_packet_header_L2(self, vlan=0): @@ -769,7 +803,7 @@ class TestSRv6(VppTestCase): # Note: the dst addr ('00:55:44:33:22:11') is used in # the compare function compare_rx_tx_packet_T_Encaps_L2 # to detect presence of L2 in SRH payload - p = Ether(src='00:11:22:33:44:55', dst='00:55:44:33:22:11') + p = Ether(src="00:11:22:33:44:55", dst="00:55:44:33:22:11") etype = 0x8137 # IPX if vlan: # add 802.1q layer @@ -789,7 +823,7 @@ class TestSRv6(VppTestCase): Outer IPv6 destination address is set to sidlist[segleft] IPv6 source address is 1234::1 """ - eth = Ether(src='00:11:22:33:44:55', dst='00:55:44:33:22:11') + eth = Ether(src="00:11:22:33:44:55", dst="00:55:44:33:22:11") etype = 0x8137 # IPX if vlan: # add 802.1q layer @@ -797,27 +831,28 @@ class TestSRv6(VppTestCase): else: eth.type = etype - p = (IPv6(src='1234::1', dst=sidlist[segleft]) / - IPv6ExtHdrSegmentRouting(addresses=sidlist, - segleft=segleft, nh=59) / - eth) + p = ( + IPv6(src="1234::1", dst=sidlist[segleft]) + / IPv6ExtHdrSegmentRouting(addresses=sidlist, segleft=segleft, nh=143) + / eth + ) return p def get_payload_info(self, packet): - """ Extract the payload_info from the packet - """ + """Extract the payload_info from the packet""" # in most cases, payload_info is in packet[Raw] # but packet[Raw] gives the complete payload # (incl L2 header) for the T.Encaps L2 case try: - payload_info = self.payload_to_info(str(packet[Raw])) + payload_info = self.payload_to_info(packet[Raw]) except: # remote L2 header from packet[Raw]: # take packet[Raw], convert it to an Ether layer # and then extract Raw from it payload_info = self.payload_to_info( - str(Ether(str(packet[Raw]))[Raw])) + Ether(scapy.compat.raw(packet[Raw]))[Raw] + ) return payload_info @@ -830,8 +865,10 @@ class TestSRv6(VppTestCase): :param capture: captured packets :param compare_func: function to compare in and out packet """ - self.logger.info("Verifying capture on interface %s using function %s" - % (dst_if.name, compare_func.func_name)) + self.logger.info( + "Verifying capture on interface %s using function %s" + % (dst_if.name, compare_func.__name__) + ) last_info = dict() for i in self.pg_interfaces: @@ -844,19 +881,19 @@ class TestSRv6(VppTestCase): payload_info = self.get_payload_info(packet) packet_index = payload_info.index - self.logger.debug("Verifying packet with index %d" - % (packet_index)) + self.logger.debug("Verifying packet with index %d" % (packet_index)) # packet should have arrived on the expected interface self.assertEqual(payload_info.dst, dst_sw_if_index) self.logger.debug( - "Got packet on interface %s: src=%u (idx=%u)" % - (dst_if.name, payload_info.src, packet_index)) + "Got packet on interface %s: src=%u (idx=%u)" + % (dst_if.name, payload_info.src, packet_index) + ) # search for payload_info with same src and dst if_index # this will give us the transmitted packet next_info = self.get_next_packet_info_for_interface2( - payload_info.src, dst_sw_if_index, - last_info[payload_info.src]) + payload_info.src, dst_sw_if_index, last_info[payload_info.src] + ) last_info[payload_info.src] = next_info # next_info should not be None self.assertTrue(next_info is not None) @@ -865,8 +902,9 @@ class TestSRv6(VppTestCase): # data field of next_info contains the tx packet txed_packet = next_info.data - self.logger.debug(ppp("Transmitted packet:", - txed_packet)) # ppp=Pretty Print Packet + self.logger.debug( + ppp("Transmitted packet:", txed_packet) + ) # ppp=Pretty Print Packet self.logger.debug(ppp("Received packet:", packet)) @@ -880,11 +918,14 @@ class TestSRv6(VppTestCase): # have all expected packets arrived? for i in self.pg_interfaces: remaining_packet = self.get_next_packet_info_for_interface2( - i.sw_if_index, dst_sw_if_index, last_info[i.sw_if_index]) - self.assertTrue(remaining_packet is None, - "Interface %s: Packet expected from interface %s " - "didn't arrive" % (dst_if.name, i.name)) + i.sw_if_index, dst_sw_if_index, last_info[i.sw_if_index] + ) + self.assertTrue( + remaining_packet is None, + "Interface %s: Packet expected from interface %s " + "didn't arrive" % (dst_if.name, i.name), + ) -if __name__ == '__main__': +if __name__ == "__main__": unittest.main(testRunner=VppTestRunner)