aboutsummaryrefslogtreecommitdiffstats
path: root/test/test_srv6_as.py
diff options
context:
space:
mode:
authorKlement Sekera <klement.sekera@gmail.com>2022-04-26 19:02:15 +0200
committerOle Tr�an <otroan@employees.org>2022-05-10 18:52:08 +0000
commitd9b0c6fbf7aa5bd9af84264105b39c82028a4a29 (patch)
tree4f786cfd8ebc2443cb11e11b74c8657204068898 /test/test_srv6_as.py
parentf90348bcb4afd0af2611cefc43b17ef3042b511c (diff)
tests: replace pycodestyle with black
Drop pycodestyle for code style checking in favor of black. Black is much faster, stable PEP8 compliant code style checker offering also automatic formatting. It aims to be very stable and produce smallest diffs. It's used by many small and big projects. Running checkstyle with black takes a few seconds with a terse output. Thus, test-checkstyle-diff is no longer necessary. Expand scope of checkstyle to all python files in the repo, replacing test-checkstyle with checkstyle-python. Also, fixstyle-python is now available for automatic style formatting. Note: python virtualenv has been consolidated in test/Makefile, test/requirements*.txt which will eventually be moved to a central location. This is required to simply the automated generation of docker executor images in the CI. Type: improvement Change-Id: I022a326603485f58585e879ac0f697fceefbc9c8 Signed-off-by: Klement Sekera <klement.sekera@gmail.com> Signed-off-by: Dave Wallace <dwallacelf@gmail.com>
Diffstat (limited to 'test/test_srv6_as.py')
-rw-r--r--test/test_srv6_as.py380
1 files changed, 212 insertions, 168 deletions
diff --git a/test/test_srv6_as.py b/test/test_srv6_as.py
index 947633e7fab..87cafd1e5ed 100644
--- a/test/test_srv6_as.py
+++ b/test/test_srv6_as.py
@@ -6,8 +6,14 @@ from socket import AF_INET6
from framework import VppTestCase, VppTestRunner
from vpp_ip_route import VppIpRoute, VppRoutePath, FibPathProto, VppIpTable
-from vpp_srv6 import SRv6LocalSIDBehaviors, VppSRv6LocalSID, VppSRv6Policy, \
- SRv6PolicyType, VppSRv6Steering, SRv6PolicySteeringTypes
+from vpp_srv6 import (
+ SRv6LocalSIDBehaviors,
+ VppSRv6LocalSID,
+ VppSRv6Policy,
+ SRv6PolicyType,
+ VppSRv6Steering,
+ SRv6PolicySteeringTypes,
+)
import scapy.compat
from scapy.packet import Raw
@@ -19,7 +25,7 @@ from util import ppp
class TestSRv6As(VppTestCase):
- """ SRv6 Static Proxy plugin Test Case """
+ """SRv6 Static Proxy plugin Test Case"""
@classmethod
def setUpClass(self):
@@ -30,8 +36,7 @@ class TestSRv6As(VppTestCase):
super(TestSRv6As, cls).tearDownClass()
def setUp(self):
- """ Perform test setup before each test case.
- """
+ """Perform test setup before each test case."""
super(TestSRv6As, self).setUp()
# packet sizes, inclusive L2 overhead
@@ -41,17 +46,15 @@ class TestSRv6As(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(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
@@ -70,9 +73,8 @@ class TestSRv6As(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
@@ -107,9 +109,9 @@ class TestSRv6As(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"))
@@ -121,8 +123,7 @@ class TestSRv6As(VppTestCase):
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
@@ -134,75 +135,83 @@ class TestSRv6As(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)])
+ 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)
@@ -215,17 +224,18 @@ class TestSRv6As(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"))
@@ -234,12 +244,14 @@ class TestSRv6As(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"))
@@ -251,28 +263,37 @@ class TestSRv6As(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)])
+ 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)
@@ -285,16 +306,18 @@ class TestSRv6As(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"))
@@ -303,12 +326,14 @@ class TestSRv6As(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"))
@@ -320,28 +345,37 @@ class TestSRv6As(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)])
+ 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)
@@ -354,16 +388,18 @@ class TestSRv6As(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"))
@@ -372,12 +408,14 @@ class TestSRv6As(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"))
@@ -389,7 +427,7 @@ class TestSRv6As(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
@@ -417,7 +455,7 @@ class TestSRv6As(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
@@ -438,7 +476,7 @@ class TestSRv6As(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
@@ -467,7 +505,7 @@ class TestSRv6As(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
@@ -492,7 +530,7 @@ class TestSRv6As(VppTestCase):
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
@@ -520,7 +558,7 @@ class TestSRv6As(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" (143)
@@ -539,7 +577,7 @@ class TestSRv6As(VppTestCase):
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
@@ -564,7 +602,7 @@ class TestSRv6As(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
@@ -594,7 +632,7 @@ class TestSRv6As(VppTestCase):
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
@@ -616,8 +654,7 @@ class TestSRv6As(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
@@ -632,19 +669,19 @@ class TestSRv6As(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)
@@ -697,8 +734,7 @@ class TestSRv6As(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):
@@ -715,11 +751,12 @@ class TestSRv6As(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):
@@ -732,8 +769,7 @@ class TestSRv6As(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):
@@ -751,11 +787,12 @@ class TestSRv6As(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):
@@ -766,7 +803,7 @@ class TestSRv6As(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
@@ -786,7 +823,7 @@ class TestSRv6As(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
@@ -794,15 +831,15 @@ class TestSRv6As(VppTestCase):
else:
eth.type = etype
- p = (IPv6(src='1234::1', dst=sidlist[segleft]) /
- IPv6ExtHdrSegmentRouting(addresses=sidlist,
- segleft=segleft, nh=143) /
- 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
@@ -814,7 +851,8 @@ class TestSRv6As(VppTestCase):
# take packet[Raw], convert it to an Ether layer
# and then extract Raw from it
payload_info = self.payload_to_info(
- Ether(scapy.compat.raw(packet[Raw]))[Raw])
+ Ether(scapy.compat.raw(packet[Raw]))[Raw]
+ )
return payload_info
@@ -827,8 +865,10 @@ class TestSRv6As(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.__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:
@@ -841,19 +881,19 @@ class TestSRv6As(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)
@@ -862,8 +902,9 @@ class TestSRv6As(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))
@@ -877,11 +918,14 @@ class TestSRv6As(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)