aboutsummaryrefslogtreecommitdiffstats
path: root/test/test_srv6_mobile.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_mobile.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_mobile.py')
-rw-r--r--test/test_srv6_mobile.py141
1 files changed, 76 insertions, 65 deletions
diff --git a/test/test_srv6_mobile.py b/test/test_srv6_mobile.py
index e617acef9ab..33190b84ba0 100644
--- a/test/test_srv6_mobile.py
+++ b/test/test_srv6_mobile.py
@@ -8,7 +8,7 @@ from scapy.all import *
class TestSRv6EndMGTP4E(VppTestCase):
- """ SRv6 End.M.GTP4.E (SRv6 -> GTP-U) """
+ """SRv6 End.M.GTP4.E (SRv6 -> GTP-U)"""
@classmethod
def setUpClass(cls):
@@ -37,14 +37,12 @@ class TestSRv6EndMGTP4E(VppTestCase):
ip4_dst = IPv4Address(str(self.ip4_dst))
# 32bit prefix + 32bit IPv4 DA + 8bit + 32bit TEID + 24bit
- dst = b'\xaa' * 4 + ip4_dst.packed + \
- b'\x11' + b'\xbb' * 4 + b'\x11' * 3
+ dst = b"\xaa" * 4 + ip4_dst.packed + b"\x11" + b"\xbb" * 4 + b"\x11" * 3
ip6_dst = IPv6Address(dst)
ip4_src = IPv4Address(str(self.ip4_src))
# 64bit prefix + 32bit IPv4 SA + 16 bit port + 16bit
- src = b'\xcc' * 8 + ip4_src.packed + \
- b'\xdd' * 2 + b'\x11' * 2
+ src = b"\xcc" * 8 + ip4_src.packed + b"\xdd" * 2 + b"\x11" * 2
ip6_src = IPv6Address(src)
self.logger.info("ip4 dst: {}".format(ip4_dst))
@@ -54,24 +52,26 @@ class TestSRv6EndMGTP4E(VppTestCase):
pkts = list()
for d, s in inner:
- pkt = (Ether() /
- IPv6(dst=str(ip6_dst), src=str(ip6_src)) /
- IPv6ExtHdrSegmentRouting() /
- IPv6(dst=d, src=s) /
- UDP(sport=1000, dport=23))
+ pkt = (
+ Ether()
+ / IPv6(dst=str(ip6_dst), src=str(ip6_src))
+ / IPv6ExtHdrSegmentRouting()
+ / IPv6(dst=d, src=s)
+ / UDP(sport=1000, dport=23)
+ )
self.logger.info(pkt.show2(dump=True))
pkts.append(pkt)
return pkts
def test_srv6_mobile(self):
- """ test_srv6_mobile """
+ """test_srv6_mobile"""
pkts = self.create_packets([("A::1", "B::1"), ("C::1", "D::1")])
self.vapi.cli(
- "sr localsid address {} behavior end.m.gtp4.e "
- .format(pkts[0]['IPv6'].dst) +
- "v4src_position 64 fib-table 0")
+ "sr localsid address {} behavior end.m.gtp4.e ".format(pkts[0]["IPv6"].dst)
+ + "v4src_position 64 fib-table 0"
+ )
self.logger.info(self.vapi.cli("show sr localsid"))
self.vapi.cli("clear errors")
@@ -89,11 +89,11 @@ class TestSRv6EndMGTP4E(VppTestCase):
self.logger.info(pkt.show2(dump=True))
self.assertEqual(pkt[IP].dst, self.ip4_dst)
self.assertEqual(pkt[IP].src, self.ip4_src)
- self.assertEqual(pkt[GTP_U_Header].teid, 0xbbbbbbbb)
+ self.assertEqual(pkt[GTP_U_Header].teid, 0xBBBBBBBB)
class TestSRv6TMGTP4D(VppTestCase):
- """ SRv6 T.M.GTP4.D (GTP-U -> SRv6) """
+ """SRv6 T.M.GTP4.D (GTP-U -> SRv6)"""
@classmethod
def setUpClass(cls):
@@ -133,26 +133,29 @@ class TestSRv6TMGTP4D(VppTestCase):
pkts = list()
for d, s in inner:
- pkt = (Ether() /
- IP(dst=str(ip4_dst), src=str(ip4_src)) /
- UDP(sport=2152, dport=2152) /
- GTP_U_Header(gtp_type="g_pdu", teid=200) /
- IPv6(dst=d, src=s) /
- UDP(sport=1000, dport=23))
+ pkt = (
+ Ether()
+ / IP(dst=str(ip4_dst), src=str(ip4_src))
+ / UDP(sport=2152, dport=2152)
+ / GTP_U_Header(gtp_type="g_pdu", teid=200)
+ / IPv6(dst=d, src=s)
+ / UDP(sport=1000, dport=23)
+ )
self.logger.info(pkt.show2(dump=True))
pkts.append(pkt)
return pkts
def test_srv6_mobile(self):
- """ test_srv6_mobile """
+ """test_srv6_mobile"""
pkts = self.create_packets([("A::1", "B::1"), ("C::1", "D::1")])
self.vapi.cli("set sr encaps source addr A1::1")
self.vapi.cli("sr policy add bsid D4:: next D2:: next D3::")
self.vapi.cli(
- "sr policy add bsid D5:: behavior t.m.gtp4.d D4::/32 " +
- "v6src_prefix C1::/64 nhtype ipv6 fib-table 0 drop-in")
+ "sr policy add bsid D5:: behavior t.m.gtp4.d D4::/32 "
+ + "v6src_prefix C1::/64 nhtype ipv6 fib-table 0 drop-in"
+ )
self.vapi.cli("sr steer l3 {}/32 via bsid D5::".format(self.ip4_dst))
self.vapi.cli("ip route add D2::/32 via {}".format(self.ip6_dst))
@@ -172,15 +175,18 @@ class TestSRv6TMGTP4D(VppTestCase):
for pkt in capture:
self.logger.info(pkt.show2(dump=True))
- self.logger.info("GTP4.D Address={}".format(
- str(pkt[IPv6ExtHdrSegmentRouting].addresses[0])))
+ self.logger.info(
+ "GTP4.D Address={}".format(
+ str(pkt[IPv6ExtHdrSegmentRouting].addresses[0])
+ )
+ )
self.assertEqual(
- str(pkt[IPv6ExtHdrSegmentRouting].addresses[0]),
- "d4:0:101:101::c800:0")
+ str(pkt[IPv6ExtHdrSegmentRouting].addresses[0]), "d4:0:101:101::c800:0"
+ )
class TestSRv6EndMGTP6E(VppTestCase):
- """ SRv6 End.M.GTP6.E """
+ """SRv6 End.M.GTP6.E"""
@classmethod
def setUpClass(cls):
@@ -205,42 +211,42 @@ class TestSRv6EndMGTP6E(VppTestCase):
def create_packets(self, inner):
# 64bit prefix + 8bit QFI + 32bit TEID + 24bit
- dst = b'\xaa' * 8 + b'\x00' + \
- b'\xbb' * 4 + b'\x00' * 3
+ dst = b"\xaa" * 8 + b"\x00" + b"\xbb" * 4 + b"\x00" * 3
ip6_dst = IPv6Address(dst)
self.ip6_dst = ip6_dst
- src = b'\xcc' * 8 + \
- b'\xdd' * 4 + b'\x11' * 4
+ src = b"\xcc" * 8 + b"\xdd" * 4 + b"\x11" * 4
ip6_src = IPv6Address(src)
self.ip6_src = ip6_src
pkts = list()
for d, s in inner:
- pkt = (Ether() /
- IPv6(dst=str(ip6_dst),
- src=str(ip6_src)) /
- IPv6ExtHdrSegmentRouting(segleft=1,
- lastentry=0,
- tag=0,
- addresses=["a1::1"]) /
- IPv6(dst=d, src=s) / UDP(sport=1000, dport=23))
+ pkt = (
+ Ether()
+ / IPv6(dst=str(ip6_dst), src=str(ip6_src))
+ / IPv6ExtHdrSegmentRouting(
+ segleft=1, lastentry=0, tag=0, addresses=["a1::1"]
+ )
+ / IPv6(dst=d, src=s)
+ / UDP(sport=1000, dport=23)
+ )
self.logger.info(pkt.show2(dump=True))
pkts.append(pkt)
return pkts
def test_srv6_mobile(self):
- """ test_srv6_mobile """
+ """test_srv6_mobile"""
pkts = self.create_packets([("A::1", "B::1"), ("C::1", "D::1")])
self.vapi.cli(
- "sr localsid prefix {}/64 behavior end.m.gtp6.e fib-table 0"
- .format(pkts[0]['IPv6'].dst))
- self.vapi.cli(
- "ip route add a1::/64 via {}".format(self.ip6_nhop))
+ "sr localsid prefix {}/64 behavior end.m.gtp6.e fib-table 0".format(
+ pkts[0]["IPv6"].dst
+ )
+ )
+ self.vapi.cli("ip route add a1::/64 via {}".format(self.ip6_nhop))
self.logger.info(self.vapi.cli("show sr localsid"))
self.vapi.cli("clear errors")
@@ -258,11 +264,11 @@ class TestSRv6EndMGTP6E(VppTestCase):
self.logger.info(pkt.show2(dump=True))
self.assertEqual(pkt[IPv6].dst, "a1::1")
self.assertEqual(pkt[IPv6].src, str(self.ip6_src))
- self.assertEqual(pkt[GTP_U_Header].teid, 0xbbbbbbbb)
+ self.assertEqual(pkt[GTP_U_Header].teid, 0xBBBBBBBB)
class TestSRv6EndMGTP6D(VppTestCase):
- """ SRv6 End.M.GTP6.D """
+ """SRv6 End.M.GTP6.D"""
@classmethod
def setUpClass(cls):
@@ -299,26 +305,29 @@ class TestSRv6EndMGTP6D(VppTestCase):
pkts = list()
for d, s in inner:
- pkt = (Ether() /
- IPv6(dst=str(ip6_dst), src=str(ip6_src)) /
- UDP(sport=2152, dport=2152) /
- GTP_U_Header(gtp_type="g_pdu", teid=200) /
- IPv6(dst=d, src=s) /
- UDP(sport=1000, dport=23))
+ pkt = (
+ Ether()
+ / IPv6(dst=str(ip6_dst), src=str(ip6_src))
+ / UDP(sport=2152, dport=2152)
+ / GTP_U_Header(gtp_type="g_pdu", teid=200)
+ / IPv6(dst=d, src=s)
+ / UDP(sport=1000, dport=23)
+ )
self.logger.info(pkt.show2(dump=True))
pkts.append(pkt)
return pkts
def test_srv6_mobile(self):
- """ test_srv6_mobile """
+ """test_srv6_mobile"""
pkts = self.create_packets([("A::1", "B::1"), ("C::1", "D::1")])
self.vapi.cli("set sr encaps source addr A1::1")
self.vapi.cli("sr policy add bsid D4:: next D2:: next D3::")
self.vapi.cli(
- "sr localsid prefix 2001::/64 behavior end.m.gtp6.d " +
- "D4::/64 fib-table 0 drop-in")
+ "sr localsid prefix 2001::/64 behavior end.m.gtp6.d "
+ + "D4::/64 fib-table 0 drop-in"
+ )
self.vapi.cli("ip route add D2::/64 via {}".format(self.ip6_nhop))
self.logger.info(self.vapi.cli("show sr policies"))
@@ -337,11 +346,13 @@ class TestSRv6EndMGTP6D(VppTestCase):
for pkt in capture:
self.logger.info(pkt.show2(dump=True))
- self.logger.info("GTP6.D SID0={}".format(
- str(pkt[IPv6ExtHdrSegmentRouting].addresses[0])))
- self.logger.info("GTP6.D SID1={}".format(
- str(pkt[IPv6ExtHdrSegmentRouting].addresses[1])))
- self.assertEqual(
- str(pkt[IPv6ExtHdrSegmentRouting].addresses[0]), "2001::1")
+ self.logger.info(
+ "GTP6.D SID0={}".format(str(pkt[IPv6ExtHdrSegmentRouting].addresses[0]))
+ )
+ self.logger.info(
+ "GTP6.D SID1={}".format(str(pkt[IPv6ExtHdrSegmentRouting].addresses[1]))
+ )
+ self.assertEqual(str(pkt[IPv6ExtHdrSegmentRouting].addresses[0]), "2001::1")
self.assertEqual(
- str(pkt[IPv6ExtHdrSegmentRouting].addresses[1]), "d4::c800:0")
+ str(pkt[IPv6ExtHdrSegmentRouting].addresses[1]), "d4::c800:0"
+ )