hs-test: add more asserts

[vpp.git] / test / test_srv6_as.py

diff --git a/test/test_srv6_as.py b/test/test_srv6_as.py
old mode 100755 (executable)
new mode 100644 (file)
index 1ff7906..645cf33
--- a/test/test_srv6_as.py
+++ b/test/test_srv6_as.py
@@ -1,35 +1,34 @@
-#!/usr/bin/env python
+#!/usr/bin/env python3

 
 import unittest
 
 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 framework import VppTestCase
+from asfframework import VppTestRunner
+from vpp_ip_route import VppIpRoute, VppRoutePath

 
 

+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.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
 
 
 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):
 
     @classmethod
     def setUpClass(self):

-        super(TestSRv6, self).setUpClass()
+        super(TestSRv6As, self).setUpClass()
+
+    @classmethod
+    def tearDownClass(cls):
+        super(TestSRv6As, cls).tearDownClass()

 
     def setUp(self):
 
     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]
 
         # packet sizes, inclusive L2 overhead
         self.pg_packet_sizes = [64, 512, 1518, 9018]


@@ -38,17 +37,15 @@ class TestSRv6(VppTestCase):
         self.reset_packet_infos()
 
     def tearDown(self):
         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()
 
         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
         :param ipv6: configure IPv6 on interface
         :param ipv4: configure IPv4 on interface
         :param ipv6_table_id: FIB table_id for IPv6


@@ -67,9 +64,8 @@ class TestSRv6(VppTestCase):
             interface.resolve_arp()
         interface.admin_up()
 
             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
 
         :param ipv6: list of interface IPv6 capabilities
         :param ipv4: list of interface IPv4 capabilities


@@ -104,22 +100,21 @@ class TestSRv6(VppTestCase):
         # setup all interfaces
         for i in range(count):
             intf = self.pg_interfaces[i]
         # 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):
 
         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):
         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
         self.logger.debug("Tearing down interfaces")
         # tear down all interfaces
         # AFAIK they cannot be deleted


@@ -131,77 +126,83 @@ class TestSRv6(VppTestCase):
             i.set_table_ip6(0)
 
     def test_SRv6_End_AS_IPv6_noSRH(self):
             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(
         self.run_SRv6_End_AS_IPv6(

-            sid_list=['a1::', 'a2::a6', 'a3::'],
+            sid_list=["a1::", "a2::a6", "a3::"],

             test_sid_index=1,
             test_sid_index=1,

-            rewrite_src_addr='a2::')
+            rewrite_src_addr="a2::",
+        )

 
     def test_SRv6_End_AS_IPv6_SRH(self):
 
     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(
         self.run_SRv6_End_AS_IPv6(

-            sid_list=['a1::a6', 'a2::', 'a3::'],
+            sid_list=["a1::a6", "a2::", "a3::"],

             test_sid_index=0,
             test_sid_index=0,

-            rewrite_src_addr='a1::')
+            rewrite_src_addr="a1::",
+        )

 
     def test_SRv6_End_AS_IPv4_noSRH(self):
 
     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(
         self.run_SRv6_End_AS_IPv4(

-            sid_list=['a1::', 'a2::a6', 'a3::'],
+            sid_list=["a1::", "a2::a6", "a3::"],

             test_sid_index=1,
             test_sid_index=1,

-            rewrite_src_addr='a2::')
+            rewrite_src_addr="a2::",
+        )

 
     def test_SRv6_End_AS_IPv4_SRH(self):
 
     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(
         self.run_SRv6_End_AS_IPv4(

-            sid_list=['a1::a6', 'a2::', 'a3::'],
+            sid_list=["a1::a6", "a2::", "a3::"],

             test_sid_index=0,
             test_sid_index=0,

-            rewrite_src_addr='a1::')
+            rewrite_src_addr="a1::",
+        )

 
     def test_SRv6_End_AS_L2_noSRH(self):
 
     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(
         self.run_SRv6_End_AS_L2(

-            sid_list=['a1::', 'a2::a6', 'a3::'],
+            sid_list=["a1::", "a2::a6", "a3::"],

             test_sid_index=1,
             test_sid_index=1,

-            rewrite_src_addr='a2::')
+            rewrite_src_addr="a2::",
+        )

 
     def test_SRv6_End_AS_L2_SRH(self):
 
     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(
         self.run_SRv6_End_AS_L2(

-            sid_list=['a1::a6', 'a2::', 'a3::'],
+            sid_list=["a1::a6", "a2::", "a3::"],

             test_sid_index=0,
             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):
 
     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_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
 
         # 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
         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)
         for s in self.rewrite_sid_list:
             cli_str += " next " + s
         self.vapi.cli(cli_str)


@@ -214,17 +215,18 @@ class TestSRv6(VppTestCase):
 
         # prepare L2 in SRv6 headers
         packet_header1 = self.create_packet_header_IPv6_SRH_L2(
 
         # 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)
 
         # 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
 
         # 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"))
 
         # log the localsid counters
         self.logger.info(self.vapi.cli("show sr localsid"))


@@ -233,12 +235,14 @@ class TestSRv6(VppTestCase):
         packet_header2 = self.create_packet_header_L2()
 
         # generate returning packets (pg1->pg0)
         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
 
         # 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"))
 
         # log the localsid counters
         self.logger.info(self.vapi.cli("show sr localsid"))


@@ -250,30 +254,37 @@ class TestSRv6(VppTestCase):
         self.teardown_interfaces()
 
     def run_SRv6_End_AS_IPv6(self, sid_list, test_sid_index, rewrite_src_addr):
         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_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
 
         # 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
         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)
         for s in self.rewrite_sid_list:
             cli_str += " next " + s
         self.vapi.cli(cli_str)


@@ -286,16 +297,18 @@ class TestSRv6(VppTestCase):
 
         # prepare IPv6 in SRv6 headers
         packet_header1 = self.create_packet_header_IPv6_SRH_IPv6(
 
         # 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)
 
         # 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
 
         # 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"))
 
         # log the localsid counters
         self.logger.info(self.vapi.cli("show sr localsid"))


@@ -304,12 +317,14 @@ class TestSRv6(VppTestCase):
         packet_header2 = self.create_packet_header_IPv6()
 
         # generate returning packets (pg1->pg0)
         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
 
         # 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"))
 
         # log the localsid counters
         self.logger.info(self.vapi.cli("show sr localsid"))


@@ -321,30 +336,37 @@ class TestSRv6(VppTestCase):
         self.teardown_interfaces()
 
     def run_SRv6_End_AS_IPv4(self, sid_list, test_sid_index, rewrite_src_addr):
         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_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
 
         # 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
         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)
         for s in self.rewrite_sid_list:
             cli_str += " next " + s
         self.vapi.cli(cli_str)


@@ -357,16 +379,18 @@ class TestSRv6(VppTestCase):
 
         # prepare IPv4 in SRv6 headers
         packet_header1 = self.create_packet_header_IPv6_SRH_IPv4(
 
         # 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)
 
         # 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
 
         # 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"))
 
         # log the localsid counters
         self.logger.info(self.vapi.cli("show sr localsid"))


@@ -375,12 +399,14 @@ class TestSRv6(VppTestCase):
         packet_header2 = self.create_packet_header_IPv4()
 
         # generate returning packets (pg1->pg0)
         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
 
         # 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"))
 
         # log the localsid counters
         self.logger.info(self.vapi.cli("show sr localsid"))


@@ -392,7 +418,7 @@ class TestSRv6(VppTestCase):
         self.teardown_interfaces()
 
     def compare_rx_tx_packet_End_AS_IPv6_in(self, tx_pkt, rx_pkt):
         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
 
         :param tx_pkt: transmitted packet
         :param rx_pkt: received packet


@@ -420,7 +446,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
             # 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
             # segleft should be equal to lastentry
             self.assertEqual(rx_srh.segleft, rx_srh.lastentry)
             # get payload


@@ -441,7 +467,7 @@ class TestSRv6(VppTestCase):
         self.logger.debug("packet verification: SUCCESS")
 
     def compare_rx_tx_packet_End_AS_IPv4_in(self, tx_pkt, rx_pkt):
         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
 
         :param tx_pkt: transmitted packet
         :param rx_pkt: received packet


@@ -470,7 +496,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
             # 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
             # segleft should be equal to lastentry
             self.assertEqual(rx_srh.segleft, rx_srh.lastentry)
             payload = rx_srh.payload


@@ -488,14 +514,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.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):
 
         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
 
         :param tx_pkt: transmitted packet
         :param rx_pkt: received packet


@@ -523,11 +549,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
             # 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)
             # 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:
             # get payload
             payload = rx_srh.payload
         else:


@@ -537,12 +563,12 @@ class TestSRv6(VppTestCase):
             payload = rx_ip.payload
 
         # the whole rx'ed pkt beyond SRH should be equal to tx'ed pkt
             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):
 
         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
 
         :param tx_pkt: transmitted packet
         :param rx_pkt: received packet


@@ -567,7 +593,7 @@ class TestSRv6(VppTestCase):
         self.logger.debug("packet verification: SUCCESS")
 
     def compare_rx_tx_packet_End_AS_IPv4_out(self, tx_pkt, rx_pkt):
         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
 
         :param tx_pkt: transmitted packet
         :param rx_pkt: received packet


@@ -590,14 +616,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.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):
 
         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
 
         :param tx_pkt: transmitted packet
         :param rx_pkt: received packet


@@ -609,7 +635,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_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))
 
         # verify if rx'ed packet has no SRH
         self.assertFalse(rx_pkt.haslayer(IPv6ExtHdrSegmentRouting))


@@ -619,8 +645,7 @@ class TestSRv6(VppTestCase):
 
         self.logger.debug("packet verification: SUCCESS")
 
 
         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
         """Create SRv6 input packet stream for defined interface.
 
         :param VppInterface src_if: Interface to create packet stream for


@@ -635,19 +660,19 @@ class TestSRv6(VppTestCase):
         """
         self.logger.info("Creating packets")
         pkts = []
         """
         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)
             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
             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:
             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)
             size = packet_sizes[i % len(packet_sizes)]
             self.logger.debug("Packet size %d" % (size))
             self.extend_packet(p, size)


@@ -655,7 +680,7 @@ class TestSRv6(VppTestCase):
             # read back the dumped packet (with str())
             # to force computing these fields
             # probably other ways are possible
             # 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)
             payload_info.data = p.copy()
             self.logger.debug(ppp("Created packet:", p))
             pkts.append(p)


@@ -700,8 +725,7 @@ class TestSRv6(VppTestCase):
         UDP source port and destination port are 1234
         """
 
         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):
         return p
 
     def create_packet_header_IPv6_SRH_IPv6(self, sidlist, segleft):


@@ -718,11 +742,12 @@ class TestSRv6(VppTestCase):
         UDP source port and destination port are 1234
         """
 
         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):
         return p
 
     def create_packet_header_IPv4(self):


@@ -735,8 +760,7 @@ class TestSRv6(VppTestCase):
         UDP source port and destination port are 1234
         """
 
         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):
         return p
 
     def create_packet_header_IPv6_SRH_IPv4(self, sidlist, segleft):


@@ -754,11 +778,12 @@ class TestSRv6(VppTestCase):
         UDP source port and destination port are 1234
         """
 
         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):
         return p
 
     def create_packet_header_L2(self, vlan=0):


@@ -769,7 +794,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
         # 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
         etype = 0x8137  # IPX
         if vlan:
             # add 802.1q layer


@@ -789,7 +814,7 @@ class TestSRv6(VppTestCase):
         Outer IPv6 destination address is set to sidlist[segleft]
         IPv6 source address is 1234::1
         """
         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
         etype = 0x8137  # IPX
         if vlan:
             # add 802.1q layer


@@ -797,15 +822,15 @@ class TestSRv6(VppTestCase):
         else:
             eth.type = etype
 
         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):
         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
         # 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


@@ -817,7 +842,8 @@ class TestSRv6(VppTestCase):
             # take packet[Raw], convert it to an Ether layer
             # and then extract Raw from it
             payload_info = self.payload_to_info(
             # take packet[Raw], convert it to an Ether layer
             # and then extract Raw from it
             payload_info = self.payload_to_info(

-                Ether(str(packet[Raw]))[Raw])
+                Ether(scapy.compat.raw(packet[Raw]))[Raw]
+            )

 
         return payload_info
 
 
         return payload_info
 


@@ -830,8 +856,10 @@ class TestSRv6(VppTestCase):
         :param capture: captured packets
         :param compare_func: function to compare in and out packet
         """
         :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:
 
         last_info = dict()
         for i in self.pg_interfaces:


@@ -844,19 +872,19 @@ class TestSRv6(VppTestCase):
                 payload_info = self.get_payload_info(packet)
                 packet_index = payload_info.index
 
                 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(
                 # 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(
 
                 # 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)
                 last_info[payload_info.src] = next_info
                 # next_info should not be None
                 self.assertTrue(next_info is not None)


@@ -865,8 +893,9 @@ class TestSRv6(VppTestCase):
                 # data field of next_info contains the tx packet
                 txed_packet = next_info.data
 
                 # 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))
 
 
                 self.logger.debug(ppp("Received packet:", packet))
 


@@ -880,11 +909,14 @@ class TestSRv6(VppTestCase):
         # have all expected packets arrived?
         for i in self.pg_interfaces:
             remaining_packet = self.get_next_packet_info_for_interface2(
         # 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)
     unittest.main(testRunner=VppTestRunner)