summaryrefslogtreecommitdiffstats
path: root/test/test_map.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_map.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_map.py')
-rw-r--r--test/test_map.py824
1 files changed, 452 insertions, 372 deletions
diff --git a/test/test_map.py b/test/test_map.py
index 22fe1e1d20a..16e9607cedc 100644
--- a/test/test_map.py
+++ b/test/test_map.py
@@ -12,12 +12,17 @@ import scapy.compat
from scapy.layers.l2 import Ether
from scapy.packet import Raw
from scapy.layers.inet import IP, UDP, ICMP, TCP
-from scapy.layers.inet6 import IPv6, ICMPv6TimeExceeded, IPv6ExtHdrFragment, \
- ICMPv6EchoRequest, ICMPv6DestUnreach
+from scapy.layers.inet6 import (
+ IPv6,
+ ICMPv6TimeExceeded,
+ IPv6ExtHdrFragment,
+ ICMPv6EchoRequest,
+ ICMPv6DestUnreach,
+)
class TestMAP(VppTestCase):
- """ MAP Test Case """
+ """MAP Test Case"""
@classmethod
def setUpClass(cls):
@@ -69,27 +74,24 @@ class TestMAP(VppTestCase):
self.assertEqual(rx[IPv6].src, ip6_src)
self.assertEqual(rx[IPv6].dst, ip6_dst)
- def send_and_assert_encapped_one(self, packet, ip6_src, ip6_dst,
- dmac=None):
+ def send_and_assert_encapped_one(self, packet, ip6_src, ip6_dst, dmac=None):
return self.send_and_assert_encapped([packet], ip6_src, ip6_dst, dmac)
def test_api_map_domain_dump(self):
- map_dst = '2001::/64'
- map_src = '3000::1/128'
- client_pfx = '192.168.0.0/16'
- tag = 'MAP-E tag.'
- index = self.vapi.map_add_domain(ip4_prefix=client_pfx,
- ip6_prefix=map_dst,
- ip6_src=map_src,
- tag=tag).index
+ map_dst = "2001::/64"
+ map_src = "3000::1/128"
+ client_pfx = "192.168.0.0/16"
+ tag = "MAP-E tag."
+ index = self.vapi.map_add_domain(
+ ip4_prefix=client_pfx, ip6_prefix=map_dst, ip6_src=map_src, tag=tag
+ ).index
rv = self.vapi.map_domain_dump()
# restore the state early so as to not impact subsequent tests.
# If an assert fails, we will not get the chance to do it at the end.
self.vapi.map_del_domain(index=index)
- self.assertGreater(len(rv), 0,
- "Expected output from 'map_domain_dump'")
+ self.assertGreater(len(rv), 0, "Expected output from 'map_domain_dump'")
# typedefs are returned as ipaddress objects.
# wrap results in str() ugh! to avoid the need to call unicode.
@@ -97,8 +99,7 @@ class TestMAP(VppTestCase):
self.assertEqual(str(rv[0].ip6_prefix), map_dst)
self.assertEqual(str(rv[0].ip6_src), map_src)
- self.assertEqual(rv[0].tag, tag,
- "output produced incorrect tag value.")
+ self.assertEqual(rv[0].tag, tag, "output produced incorrect tag value.")
def create_domains(self, ip4_pfx_str, ip6_pfx_str, ip6_src_str):
ip4_pfx = ipaddress.ip_network(ip4_pfx_str)
@@ -106,24 +107,25 @@ class TestMAP(VppTestCase):
mod = ip4_pfx.num_addresses / 1024
indicies = []
for i in range(ip4_pfx.num_addresses):
- rv = self.vapi.map_add_domain(ip6_prefix=ip6_pfx_str,
- ip4_prefix=str(ip4_pfx[i]) + "/32",
- ip6_src=ip6_src_str)
+ rv = self.vapi.map_add_domain(
+ ip6_prefix=ip6_pfx_str,
+ ip4_prefix=str(ip4_pfx[i]) + "/32",
+ ip6_src=ip6_src_str,
+ )
indicies.append(rv.index)
return indicies
def test_api_map_domains_get(self):
# Create a bunch of domains
no_domains = 4096 # This must be large enough to ensure VPP suspends
- domains = self.create_domains('130.67.0.0/20', '2001::/32',
- '2001::1/128')
+ domains = self.create_domains("130.67.0.0/20", "2001::/32", "2001::1/128")
self.assertEqual(len(domains), no_domains)
d = []
cursor = 0
# Invalid cursor
- rv, details = self.vapi.map_domains_get(cursor=no_domains+10)
+ rv, details = self.vapi.map_domains_get(cursor=no_domains + 10)
self.assertEqual(rv.retval, -7)
# Delete a domain in the middle of walk
@@ -144,48 +146,53 @@ class TestMAP(VppTestCase):
self.vapi.map_del_domain(index=i)
def test_map_e_udp(self):
- """ MAP-E UDP"""
+ """MAP-E UDP"""
#
# Add a route to the MAP-BR
#
map_br_pfx = "2001::"
map_br_pfx_len = 32
- map_route = VppIpRoute(self,
- map_br_pfx,
- map_br_pfx_len,
- [VppRoutePath(self.pg1.remote_ip6,
- self.pg1.sw_if_index)])
+ map_route = VppIpRoute(
+ self,
+ map_br_pfx,
+ map_br_pfx_len,
+ [VppRoutePath(self.pg1.remote_ip6, self.pg1.sw_if_index)],
+ )
map_route.add_vpp_config()
#
# Add a domain that maps from pg0 to pg1
#
- map_dst = '2001::/32'
- map_src = '3000::1/128'
- client_pfx = '192.168.0.0/16'
- map_translated_addr = '2001:0:101:7000:0:c0a8:101:7'
- tag = 'MAP-E tag.'
- self.vapi.map_add_domain(ip4_prefix=client_pfx,
- ip6_prefix=map_dst,
- ip6_src=map_src,
- ea_bits_len=20,
- psid_offset=4,
- psid_length=4,
- tag=tag)
+ map_dst = "2001::/32"
+ map_src = "3000::1/128"
+ client_pfx = "192.168.0.0/16"
+ map_translated_addr = "2001:0:101:7000:0:c0a8:101:7"
+ tag = "MAP-E tag."
+ self.vapi.map_add_domain(
+ ip4_prefix=client_pfx,
+ ip6_prefix=map_dst,
+ ip6_src=map_src,
+ ea_bits_len=20,
+ psid_offset=4,
+ psid_length=4,
+ tag=tag,
+ )
self.vapi.map_param_set_security_check(enable=1, fragments=1)
# Enable MAP on interface.
- self.vapi.map_if_enable_disable(is_enable=1,
- sw_if_index=self.pg0.sw_if_index,
- is_translation=0)
+ self.vapi.map_if_enable_disable(
+ is_enable=1, sw_if_index=self.pg0.sw_if_index, is_translation=0
+ )
# Ensure MAP doesn't steal all packets!
- v4 = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
- IP(src=self.pg0.remote_ip4, dst=self.pg0.remote_ip4) /
- UDP(sport=20000, dport=10000) /
- Raw(b'\xa5' * 100))
+ v4 = (
+ Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
+ / IP(src=self.pg0.remote_ip4, dst=self.pg0.remote_ip4)
+ / UDP(sport=20000, dport=10000)
+ / Raw(b"\xa5" * 100)
+ )
rx = self.send_and_expect(self.pg0, v4 * 4, self.pg0)
v4_reply = v4[1]
v4_reply.ttl -= 1
@@ -195,20 +202,24 @@ class TestMAP(VppTestCase):
#
# Fire in a v4 packet that will be encapped to the BR
#
- v4 = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
- IP(src=self.pg0.remote_ip4, dst='192.168.1.1') /
- UDP(sport=20000, dport=10000) /
- Raw(b'\xa5' * 100))
+ v4 = (
+ Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
+ / IP(src=self.pg0.remote_ip4, dst="192.168.1.1")
+ / UDP(sport=20000, dport=10000)
+ / Raw(b"\xa5" * 100)
+ )
self.send_and_assert_encapped(v4 * 4, "3000::1", map_translated_addr)
#
# Verify reordered fragments are able to pass as well
#
- v4 = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
- IP(id=1, src=self.pg0.remote_ip4, dst='192.168.1.1') /
- UDP(sport=20000, dport=10000) /
- Raw(b'\xa5' * 1000))
+ v4 = (
+ Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
+ / IP(id=1, src=self.pg0.remote_ip4, dst="192.168.1.1")
+ / UDP(sport=20000, dport=10000)
+ / Raw(b"\xa5" * 1000)
+ )
frags = fragment_rfc791(v4, 400)
frags.reverse()
@@ -216,16 +227,18 @@ class TestMAP(VppTestCase):
self.send_and_assert_encapped(frags, "3000::1", map_translated_addr)
# Enable MAP on interface.
- self.vapi.map_if_enable_disable(is_enable=1,
- sw_if_index=self.pg1.sw_if_index,
- is_translation=0)
+ self.vapi.map_if_enable_disable(
+ is_enable=1, sw_if_index=self.pg1.sw_if_index, is_translation=0
+ )
# Ensure MAP doesn't steal all packets
- v6 = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
- IPv6(src=self.pg1.remote_ip6, dst=self.pg1.remote_ip6) /
- UDP(sport=20000, dport=10000) /
- Raw(b'\xa5' * 100))
- rx = self.send_and_expect(self.pg1, v6*1, self.pg1)
+ v6 = (
+ Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac)
+ / IPv6(src=self.pg1.remote_ip6, dst=self.pg1.remote_ip6)
+ / UDP(sport=20000, dport=10000)
+ / Raw(b"\xa5" * 100)
+ )
+ rx = self.send_and_expect(self.pg1, v6 * 1, self.pg1)
v6_reply = v6[1]
v6_reply.hlim -= 1
for p in rx:
@@ -235,11 +248,13 @@ class TestMAP(VppTestCase):
# Fire in a V6 encapped packet.
# expect a decapped packet on the inside ip4 link
#
- p = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
- IPv6(dst='3000::1', src=map_translated_addr) /
- IP(dst=self.pg0.remote_ip4, src='192.168.1.1') /
- UDP(sport=10000, dport=20000) /
- Raw(b'\xa5' * 100))
+ p = (
+ Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac)
+ / IPv6(dst="3000::1", src=map_translated_addr)
+ / IP(dst=self.pg0.remote_ip4, src="192.168.1.1")
+ / UDP(sport=10000, dport=20000)
+ / Raw(b"\xa5" * 100)
+ )
self.pg1.add_stream(p)
@@ -256,15 +271,20 @@ class TestMAP(VppTestCase):
#
# Verify encapped reordered fragments pass as well
#
- p = (IP(id=1, dst=self.pg0.remote_ip4, src='192.168.1.1') /
- UDP(sport=10000, dport=20000) /
- Raw(b'\xa5' * 1500))
+ p = (
+ IP(id=1, dst=self.pg0.remote_ip4, src="192.168.1.1")
+ / UDP(sport=10000, dport=20000)
+ / Raw(b"\xa5" * 1500)
+ )
frags = fragment_rfc791(p, 400)
frags.reverse()
- stream = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
- IPv6(dst='3000::1', src=map_translated_addr) /
- x for x in frags)
+ stream = (
+ Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac)
+ / IPv6(dst="3000::1", src=map_translated_addr)
+ / x
+ for x in frags
+ )
self.pg1.add_stream(stream)
@@ -279,15 +299,15 @@ class TestMAP(VppTestCase):
self.assertEqual(r[IP].dst, p[IP].dst)
# Verify that fragments pass even if ipv6 layer is fragmented
- stream = (IPv6(dst='3000::1', src=map_translated_addr) / x
- for x in frags)
+ stream = (IPv6(dst="3000::1", src=map_translated_addr) / x for x in frags)
v6_stream = [
Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) / x
for i in range(len(frags))
for x in fragment_rfc8200(
- IPv6(dst='3000::1', src=map_translated_addr) / frags[i],
- i, 200)]
+ IPv6(dst="3000::1", src=map_translated_addr) / frags[i], i, 200
+ )
+ ]
self.pg1.add_stream(v6_stream)
@@ -306,32 +326,35 @@ class TestMAP(VppTestCase):
#
self.vapi.ppcli("map params pre-resolve ip6-nh 4001::1")
- self.send_and_assert_no_replies(self.pg0, v4,
- "resolved via default route")
+ self.send_and_assert_no_replies(self.pg0, v4, "resolved via default route")
#
# Add a route to 4001::1. Expect the encapped traffic to be
# sent via that routes next-hop
#
- pre_res_route = VppIpRoute(self, "4001::1", 128,
- [VppRoutePath(self.pg1.remote_hosts[2].ip6,
- self.pg1.sw_if_index)])
+ pre_res_route = VppIpRoute(
+ self,
+ "4001::1",
+ 128,
+ [VppRoutePath(self.pg1.remote_hosts[2].ip6, self.pg1.sw_if_index)],
+ )
pre_res_route.add_vpp_config()
- self.send_and_assert_encapped_one(v4, "3000::1",
- map_translated_addr,
- dmac=self.pg1.remote_hosts[2].mac)
+ self.send_and_assert_encapped_one(
+ v4, "3000::1", map_translated_addr, dmac=self.pg1.remote_hosts[2].mac
+ )
#
# change the route to the pre-solved next-hop
#
- pre_res_route.modify([VppRoutePath(self.pg1.remote_hosts[3].ip6,
- self.pg1.sw_if_index)])
+ pre_res_route.modify(
+ [VppRoutePath(self.pg1.remote_hosts[3].ip6, self.pg1.sw_if_index)]
+ )
pre_res_route.add_vpp_config()
- self.send_and_assert_encapped_one(v4, "3000::1",
- map_translated_addr,
- dmac=self.pg1.remote_hosts[3].mac)
+ self.send_and_assert_encapped_one(
+ v4, "3000::1", map_translated_addr, dmac=self.pg1.remote_hosts[3].mac
+ )
#
# cleanup. The test infra's object registry will ensure
@@ -341,51 +364,56 @@ class TestMAP(VppTestCase):
self.vapi.ppcli("map params pre-resolve del ip6-nh 4001::1")
def test_map_e_inner_frag(self):
- """ MAP-E Inner fragmentation """
+ """MAP-E Inner fragmentation"""
#
# Add a route to the MAP-BR
#
map_br_pfx = "2001::"
map_br_pfx_len = 32
- map_route = VppIpRoute(self,
- map_br_pfx,
- map_br_pfx_len,
- [VppRoutePath(self.pg1.remote_ip6,
- self.pg1.sw_if_index)])
+ map_route = VppIpRoute(
+ self,
+ map_br_pfx,
+ map_br_pfx_len,
+ [VppRoutePath(self.pg1.remote_ip6, self.pg1.sw_if_index)],
+ )
map_route.add_vpp_config()
#
# Add a domain that maps from pg0 to pg1
#
- map_dst = '2001::/32'
- map_src = '3000::1/128'
- client_pfx = '192.168.0.0/16'
- map_translated_addr = '2001:0:101:7000:0:c0a8:101:7'
- tag = 'MAP-E tag.'
- self.vapi.map_add_domain(ip4_prefix=client_pfx,
- ip6_prefix=map_dst,
- ip6_src=map_src,
- ea_bits_len=20,
- psid_offset=4,
- psid_length=4,
- mtu=1000,
- tag=tag)
+ map_dst = "2001::/32"
+ map_src = "3000::1/128"
+ client_pfx = "192.168.0.0/16"
+ map_translated_addr = "2001:0:101:7000:0:c0a8:101:7"
+ tag = "MAP-E tag."
+ self.vapi.map_add_domain(
+ ip4_prefix=client_pfx,
+ ip6_prefix=map_dst,
+ ip6_src=map_src,
+ ea_bits_len=20,
+ psid_offset=4,
+ psid_length=4,
+ mtu=1000,
+ tag=tag,
+ )
# Enable MAP on interface.
- self.vapi.map_if_enable_disable(is_enable=1,
- sw_if_index=self.pg0.sw_if_index,
- is_translation=0)
+ self.vapi.map_if_enable_disable(
+ is_enable=1, sw_if_index=self.pg0.sw_if_index, is_translation=0
+ )
# Enable inner fragmentation
self.vapi.map_param_set_fragmentation(inner=1)
- v4 = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
- IP(src=self.pg0.remote_ip4, dst='192.168.1.1') /
- UDP(sport=20000, dport=10000) /
- Raw(b'\xa5' * 1300))
+ v4 = (
+ Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
+ / IP(src=self.pg0.remote_ip4, dst="192.168.1.1")
+ / UDP(sport=20000, dport=10000)
+ / Raw(b"\xa5" * 1300)
+ )
- self.pg_send(self.pg0, v4*1)
+ self.pg_send(self.pg0, v4 * 1)
rx = self.pg1.get_capture(2)
# 1000-sizeof(ip6_header_t) = 960.
@@ -397,10 +425,8 @@ class TestMAP(VppTestCase):
frags[0].chksum = 0
frags[1].chksum = 0
- v6_reply1 = (IPv6(src='3000::1', dst=map_translated_addr, hlim=63) /
- frags[0])
- v6_reply2 = (IPv6(src='3000::1', dst=map_translated_addr, hlim=63) /
- frags[1])
+ v6_reply1 = IPv6(src="3000::1", dst=map_translated_addr, hlim=63) / frags[0]
+ v6_reply2 = IPv6(src="3000::1", dst=map_translated_addr, hlim=63) / frags[1]
rx[0][1].fl = 0
rx[1][1].fl = 0
rx[0][1][IP].id = 0
@@ -412,45 +438,48 @@ class TestMAP(VppTestCase):
self.validate(rx[1][1], v6_reply2)
def test_map_e_tcp_mss(self):
- """ MAP-E TCP MSS"""
+ """MAP-E TCP MSS"""
#
# Add a route to the MAP-BR
#
map_br_pfx = "2001::"
map_br_pfx_len = 32
- map_route = VppIpRoute(self,
- map_br_pfx,
- map_br_pfx_len,
- [VppRoutePath(self.pg1.remote_ip6,
- self.pg1.sw_if_index)])
+ map_route = VppIpRoute(
+ self,
+ map_br_pfx,
+ map_br_pfx_len,
+ [VppRoutePath(self.pg1.remote_ip6, self.pg1.sw_if_index)],
+ )
map_route.add_vpp_config()
#
# Add a domain that maps from pg0 to pg1
#
- map_dst = '2001::/32'
- map_src = '3000::1/128'
- client_pfx = '192.168.0.0/16'
- map_translated_addr = '2001:0:101:5000:0:c0a8:101:5'
- tag = 'MAP-E TCP tag.'
- self.vapi.map_add_domain(ip4_prefix=client_pfx,
- ip6_prefix=map_dst,
- ip6_src=map_src,
- ea_bits_len=20,
- psid_offset=4,
- psid_length=4,
- tag=tag)
+ map_dst = "2001::/32"
+ map_src = "3000::1/128"
+ client_pfx = "192.168.0.0/16"
+ map_translated_addr = "2001:0:101:5000:0:c0a8:101:5"
+ tag = "MAP-E TCP tag."
+ self.vapi.map_add_domain(
+ ip4_prefix=client_pfx,
+ ip6_prefix=map_dst,
+ ip6_src=map_src,
+ ea_bits_len=20,
+ psid_offset=4,
+ psid_length=4,
+ tag=tag,
+ )
# Enable MAP on pg0 interface.
- self.vapi.map_if_enable_disable(is_enable=1,
- sw_if_index=self.pg0.sw_if_index,
- is_translation=0)
+ self.vapi.map_if_enable_disable(
+ is_enable=1, sw_if_index=self.pg0.sw_if_index, is_translation=0
+ )
# Enable MAP on pg1 interface.
- self.vapi.map_if_enable_disable(is_enable=1,
- sw_if_index=self.pg1.sw_if_index,
- is_translation=0)
+ self.vapi.map_if_enable_disable(
+ is_enable=1, sw_if_index=self.pg1.sw_if_index, is_translation=0
+ )
# TCP MSS clamping
mss_clamp = 1300
@@ -460,9 +489,8 @@ class TestMAP(VppTestCase):
# Send a v4 packet that will be encapped.
#
p_ether = Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
- p_ip4 = IP(src=self.pg0.remote_ip4, dst='192.168.1.1')
- p_tcp = TCP(sport=20000, dport=30000, flags="S",
- options=[("MSS", 1455)])
+ p_ip4 = IP(src=self.pg0.remote_ip4, dst="192.168.1.1")
+ p_tcp = TCP(sport=20000, dport=30000, flags="S", options=[("MSS", 1455)])
p4 = p_ether / p_ip4 / p_tcp
self.pg1.add_stream(p4)
@@ -476,8 +504,7 @@ class TestMAP(VppTestCase):
self.assertEqual(rx[IP].src, p4[IP].src)
self.assertEqual(rx[IP].dst, p4[IP].dst)
self.assertEqual(rx[IPv6].src, "3000::1")
- self.assertEqual(rx[TCP].options,
- TCP(options=[('MSS', mss_clamp)]).options)
+ self.assertEqual(rx[TCP].options, TCP(options=[("MSS", mss_clamp)]).options)
def validate(self, rx, expected):
self.assertEqual(rx, expected.__class__(scapy.compat.raw(expected)))
@@ -520,173 +547,202 @@ class TestMAP(VppTestCase):
self.assertEqual(payload_total, payload_len_expected)
def payload(self, len):
- return 'x' * len
+ return "x" * len
def test_map_t(self):
- """ MAP-T """
+ """MAP-T"""
#
# Add a domain that maps from pg0 to pg1
#
- map_dst = '2001:db8::/32'
- map_src = '1234:5678:90ab:cdef::/64'
- ip4_pfx = '192.168.0.0/24'
- tag = 'MAP-T Tag.'
-
- self.vapi.map_add_domain(ip6_prefix=map_dst,
- ip4_prefix=ip4_pfx,
- ip6_src=map_src,
- ea_bits_len=16,
- psid_offset=6,
- psid_length=4,
- mtu=1500,
- tag=tag)
+ map_dst = "2001:db8::/32"
+ map_src = "1234:5678:90ab:cdef::/64"
+ ip4_pfx = "192.168.0.0/24"
+ tag = "MAP-T Tag."
+
+ self.vapi.map_add_domain(
+ ip6_prefix=map_dst,
+ ip4_prefix=ip4_pfx,
+ ip6_src=map_src,
+ ea_bits_len=16,
+ psid_offset=6,
+ psid_length=4,
+ mtu=1500,
+ tag=tag,
+ )
# Enable MAP-T on interfaces.
- self.vapi.map_if_enable_disable(is_enable=1,
- sw_if_index=self.pg0.sw_if_index,
- is_translation=1)
- self.vapi.map_if_enable_disable(is_enable=1,
- sw_if_index=self.pg1.sw_if_index,
- is_translation=1)
+ self.vapi.map_if_enable_disable(
+ is_enable=1, sw_if_index=self.pg0.sw_if_index, is_translation=1
+ )
+ self.vapi.map_if_enable_disable(
+ is_enable=1, sw_if_index=self.pg1.sw_if_index, is_translation=1
+ )
# Ensure MAP doesn't steal all packets!
- v4 = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
- IP(src=self.pg0.remote_ip4, dst=self.pg0.remote_ip4) /
- UDP(sport=20000, dport=10000) /
- Raw(b'\xa5' * 100))
- rx = self.send_and_expect(self.pg0, v4*1, self.pg0)
+ v4 = (
+ Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
+ / IP(src=self.pg0.remote_ip4, dst=self.pg0.remote_ip4)
+ / UDP(sport=20000, dport=10000)
+ / Raw(b"\xa5" * 100)
+ )
+ rx = self.send_and_expect(self.pg0, v4 * 1, self.pg0)
v4_reply = v4[1]
v4_reply.ttl -= 1
for p in rx:
self.validate(p[1], v4_reply)
# Ensure MAP doesn't steal all packets
- v6 = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
- IPv6(src=self.pg1.remote_ip6, dst=self.pg1.remote_ip6) /
- UDP(sport=20000, dport=10000) /
- Raw(b'\xa5' * 100))
- rx = self.send_and_expect(self.pg1, v6*1, self.pg1)
+ v6 = (
+ Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac)
+ / IPv6(src=self.pg1.remote_ip6, dst=self.pg1.remote_ip6)
+ / UDP(sport=20000, dport=10000)
+ / Raw(b"\xa5" * 100)
+ )
+ rx = self.send_and_expect(self.pg1, v6 * 1, self.pg1)
v6_reply = v6[1]
v6_reply.hlim -= 1
for p in rx:
self.validate(p[1], v6_reply)
- map_route = VppIpRoute(self,
- "2001:db8::",
- 32,
- [VppRoutePath(self.pg1.remote_ip6,
- self.pg1.sw_if_index,
- proto=DpoProto.DPO_PROTO_IP6)])
+ map_route = VppIpRoute(
+ self,
+ "2001:db8::",
+ 32,
+ [
+ VppRoutePath(
+ self.pg1.remote_ip6,
+ self.pg1.sw_if_index,
+ proto=DpoProto.DPO_PROTO_IP6,
+ )
+ ],
+ )
map_route.add_vpp_config()
#
# Send a v4 packet that will be translated
#
p_ether = Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
- p_ip4 = IP(src=self.pg0.remote_ip4, dst='192.168.0.1')
- payload = TCP(sport=0xabcd, dport=0xabcd)
-
- p4 = (p_ether / p_ip4 / payload)
- p6_translated = (IPv6(src="1234:5678:90ab:cdef:ac:1001:200:0",
- dst="2001:db8:1f0::c0a8:1:f") / payload)
+ p_ip4 = IP(src=self.pg0.remote_ip4, dst="192.168.0.1")
+ payload = TCP(sport=0xABCD, dport=0xABCD)
+
+ p4 = p_ether / p_ip4 / payload
+ p6_translated = (
+ IPv6(src="1234:5678:90ab:cdef:ac:1001:200:0", dst="2001:db8:1f0::c0a8:1:f")
+ / payload
+ )
p6_translated.hlim -= 1
- rx = self.send_and_expect(self.pg0, p4*1, self.pg1)
+ rx = self.send_and_expect(self.pg0, p4 * 1, self.pg1)
for p in rx:
self.validate(p[1], p6_translated)
# Send back an IPv6 packet that will be "untranslated"
p_ether6 = Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac)
- p_ip6 = IPv6(src='2001:db8:1f0::c0a8:1:f',
- dst='1234:5678:90ab:cdef:ac:1001:200:0')
- p6 = (p_ether6 / p_ip6 / payload)
- p4_translated = (IP(src='192.168.0.1',
- dst=self.pg0.remote_ip4) / payload)
+ p_ip6 = IPv6(
+ src="2001:db8:1f0::c0a8:1:f", dst="1234:5678:90ab:cdef:ac:1001:200:0"
+ )
+ p6 = p_ether6 / p_ip6 / payload
+ p4_translated = IP(src="192.168.0.1", dst=self.pg0.remote_ip4) / payload
p4_translated.id = 0
p4_translated.ttl -= 1
- rx = self.send_and_expect(self.pg1, p6*1, self.pg0)
+ rx = self.send_and_expect(self.pg1, p6 * 1, self.pg0)
for p in rx:
self.validate(p[1], p4_translated)
# IPv4 TTL=0
- ip4_ttl_expired = IP(src=self.pg0.remote_ip4, dst='192.168.0.1', ttl=0)
- p4 = (p_ether / ip4_ttl_expired / payload)
-
- icmp4_reply = (IP(id=0, ttl=254, src=self.pg0.local_ip4,
- dst=self.pg0.remote_ip4) /
- ICMP(type='time-exceeded',
- code='ttl-zero-during-transit') /
- IP(src=self.pg0.remote_ip4,
- dst='192.168.0.1', ttl=0) / payload)
- rx = self.send_and_expect(self.pg0, p4*1, self.pg0)
+ ip4_ttl_expired = IP(src=self.pg0.remote_ip4, dst="192.168.0.1", ttl=0)
+ p4 = p_ether / ip4_ttl_expired / payload
+
+ icmp4_reply = (
+ IP(id=0, ttl=254, src=self.pg0.local_ip4, dst=self.pg0.remote_ip4)
+ / ICMP(type="time-exceeded", code="ttl-zero-during-transit")
+ / IP(src=self.pg0.remote_ip4, dst="192.168.0.1", ttl=0)
+ / payload
+ )
+ rx = self.send_and_expect(self.pg0, p4 * 1, self.pg0)
for p in rx:
self.validate(p[1], icmp4_reply)
# IPv4 TTL=1
- ip4_ttl_expired = IP(src=self.pg0.remote_ip4, dst='192.168.0.1', ttl=1)
- p4 = (p_ether / ip4_ttl_expired / payload)
-
- icmp4_reply = (IP(id=0, ttl=254, src=self.pg0.local_ip4,
- dst=self.pg0.remote_ip4) /
- ICMP(type='time-exceeded',
- code='ttl-zero-during-transit') /
- IP(src=self.pg0.remote_ip4,
- dst='192.168.0.1', ttl=1) / payload)
- rx = self.send_and_expect(self.pg0, p4*1, self.pg0)
+ ip4_ttl_expired = IP(src=self.pg0.remote_ip4, dst="192.168.0.1", ttl=1)
+ p4 = p_ether / ip4_ttl_expired / payload
+
+ icmp4_reply = (
+ IP(id=0, ttl=254, src=self.pg0.local_ip4, dst=self.pg0.remote_ip4)
+ / ICMP(type="time-exceeded", code="ttl-zero-during-transit")
+ / IP(src=self.pg0.remote_ip4, dst="192.168.0.1", ttl=1)
+ / payload
+ )
+ rx = self.send_and_expect(self.pg0, p4 * 1, self.pg0)
for p in rx:
self.validate(p[1], icmp4_reply)
# IPv6 Hop limit at BR
- ip6_hlim_expired = IPv6(hlim=1, src='2001:db8:1ab::c0a8:1:ab',
- dst='1234:5678:90ab:cdef:ac:1001:200:0')
- p6 = (p_ether6 / ip6_hlim_expired / payload)
-
- icmp6_reply = (IPv6(hlim=255, src=self.pg1.local_ip6,
- dst="2001:db8:1ab::c0a8:1:ab") /
- ICMPv6TimeExceeded(code=0) /
- IPv6(src="2001:db8:1ab::c0a8:1:ab",
- dst='1234:5678:90ab:cdef:ac:1001:200:0',
- hlim=1) / payload)
- rx = self.send_and_expect(self.pg1, p6*1, self.pg1)
+ ip6_hlim_expired = IPv6(
+ hlim=1,
+ src="2001:db8:1ab::c0a8:1:ab",
+ dst="1234:5678:90ab:cdef:ac:1001:200:0",
+ )
+ p6 = p_ether6 / ip6_hlim_expired / payload
+
+ icmp6_reply = (
+ IPv6(hlim=255, src=self.pg1.local_ip6, dst="2001:db8:1ab::c0a8:1:ab")
+ / ICMPv6TimeExceeded(code=0)
+ / IPv6(
+ src="2001:db8:1ab::c0a8:1:ab",
+ dst="1234:5678:90ab:cdef:ac:1001:200:0",
+ hlim=1,
+ )
+ / payload
+ )
+ rx = self.send_and_expect(self.pg1, p6 * 1, self.pg1)
for p in rx:
self.validate(p[1], icmp6_reply)
# IPv6 Hop limit beyond BR
- ip6_hlim_expired = IPv6(hlim=0, src='2001:db8:1ab::c0a8:1:ab',
- dst='1234:5678:90ab:cdef:ac:1001:200:0')
- p6 = (p_ether6 / ip6_hlim_expired / payload)
-
- icmp6_reply = (IPv6(hlim=255, src=self.pg1.local_ip6,
- dst="2001:db8:1ab::c0a8:1:ab") /
- ICMPv6TimeExceeded(code=0) /
- IPv6(src="2001:db8:1ab::c0a8:1:ab",
- dst='1234:5678:90ab:cdef:ac:1001:200:0',
- hlim=0) / payload)
- rx = self.send_and_expect(self.pg1, p6*1, self.pg1)
+ ip6_hlim_expired = IPv6(
+ hlim=0,
+ src="2001:db8:1ab::c0a8:1:ab",
+ dst="1234:5678:90ab:cdef:ac:1001:200:0",
+ )
+ p6 = p_ether6 / ip6_hlim_expired / payload
+
+ icmp6_reply = (
+ IPv6(hlim=255, src=self.pg1.local_ip6, dst="2001:db8:1ab::c0a8:1:ab")
+ / ICMPv6TimeExceeded(code=0)
+ / IPv6(
+ src="2001:db8:1ab::c0a8:1:ab",
+ dst="1234:5678:90ab:cdef:ac:1001:200:0",
+ hlim=0,
+ )
+ / payload
+ )
+ rx = self.send_and_expect(self.pg1, p6 * 1, self.pg1)
for p in rx:
self.validate(p[1], icmp6_reply)
# IPv4 Well-known port
- p_ip4 = IP(src=self.pg0.remote_ip4, dst='192.168.0.1')
+ p_ip4 = IP(src=self.pg0.remote_ip4, dst="192.168.0.1")
payload = UDP(sport=200, dport=200)
- p4 = (p_ether / p_ip4 / payload)
- self.send_and_assert_no_replies(self.pg0, p4*1)
+ p4 = p_ether / p_ip4 / payload
+ self.send_and_assert_no_replies(self.pg0, p4 * 1)
# IPv6 Well-known port
payload = UDP(sport=200, dport=200)
- p6 = (p_ether6 / p_ip6 / payload)
- self.send_and_assert_no_replies(self.pg1, p6*1)
+ p6 = p_ether6 / p_ip6 / payload
+ self.send_and_assert_no_replies(self.pg1, p6 * 1)
# UDP packet fragmentation
payload_len = 1453
payload = UDP(sport=40000, dport=4000) / self.payload(payload_len)
- p4 = (p_ether / p_ip4 / payload)
+ p4 = p_ether / p_ip4 / payload
self.pg_enable_capture()
self.pg0.add_stream(p4)
self.pg_start()
rx = self.pg1.get_capture(2)
- p_ip6_translated = IPv6(src='1234:5678:90ab:cdef:ac:1001:200:0',
- dst='2001:db8:1e0::c0a8:1:e')
+ p_ip6_translated = IPv6(
+ src="1234:5678:90ab:cdef:ac:1001:200:0", dst="2001:db8:1e0::c0a8:1:e"
+ )
for p in rx:
self.validate_frag6(p, p_ip6_translated)
@@ -695,7 +751,7 @@ class TestMAP(VppTestCase):
# UDP packet fragmentation send fragments
payload_len = 1453
payload = UDP(sport=40000, dport=4000) / self.payload(payload_len)
- p4 = (p_ether / p_ip4 / payload)
+ p4 = p_ether / p_ip4 / payload
frags = fragment_rfc791(p4, fragsize=1000)
self.pg_enable_capture()
self.pg0.add_stream(frags)
@@ -710,13 +766,14 @@ class TestMAP(VppTestCase):
# Send back an fragmented IPv6 UDP packet that will be "untranslated"
payload = UDP(sport=4000, dport=40000) / self.payload(payload_len)
p_ether6 = Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac)
- p_ip6 = IPv6(src='2001:db8:1e0::c0a8:1:e',
- dst='1234:5678:90ab:cdef:ac:1001:200:0')
- p6 = (p_ether6 / p_ip6 / payload)
- frags6 = fragment_rfc8200(p6, identification=0xdcba, fragsize=1000)
-
- p_ip4_translated = IP(src='192.168.0.1', dst=self.pg0.remote_ip4)
- p4_translated = (p_ip4_translated / payload)
+ p_ip6 = IPv6(
+ src="2001:db8:1e0::c0a8:1:e", dst="1234:5678:90ab:cdef:ac:1001:200:0"
+ )
+ p6 = p_ether6 / p_ip6 / payload
+ frags6 = fragment_rfc8200(p6, identification=0xDCBA, fragsize=1000)
+
+ p_ip4_translated = IP(src="192.168.0.1", dst=self.pg0.remote_ip4)
+ p4_translated = p_ip4_translated / payload
p4_translated.id = 0
p4_translated.ttl -= 1
@@ -732,14 +789,15 @@ class TestMAP(VppTestCase):
# ICMP packet fragmentation
payload = ICMP(id=6529) / self.payload(payload_len)
- p4 = (p_ether / p_ip4 / payload)
+ p4 = p_ether / p_ip4 / payload
self.pg_enable_capture()
self.pg0.add_stream(p4)
self.pg_start()
rx = self.pg1.get_capture(2)
- p_ip6_translated = IPv6(src='1234:5678:90ab:cdef:ac:1001:200:0',
- dst='2001:db8:160::c0a8:1:6')
+ p_ip6_translated = IPv6(
+ src="1234:5678:90ab:cdef:ac:1001:200:0", dst="2001:db8:160::c0a8:1:6"
+ )
for p in rx:
self.validate_frag6(p, p_ip6_translated)
@@ -747,7 +805,7 @@ class TestMAP(VppTestCase):
# ICMP packet fragmentation send fragments
payload = ICMP(id=6529) / self.payload(payload_len)
- p4 = (p_ether / p_ip4 / payload)
+ p4 = p_ether / p_ip4 / payload
frags = fragment_rfc791(p4, fragsize=1000)
self.pg_enable_capture()
self.pg0.add_stream(frags)
@@ -766,30 +824,31 @@ class TestMAP(VppTestCase):
# Send a v4 TCP SYN packet that will be translated and MSS clamped
#
p_ether = Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
- p_ip4 = IP(src=self.pg0.remote_ip4, dst='192.168.0.1')
- payload = TCP(sport=0xabcd, dport=0xabcd, flags="S",
- options=[('MSS', 1460)])
-
- p4 = (p_ether / p_ip4 / payload)
- p6_translated = (IPv6(src="1234:5678:90ab:cdef:ac:1001:200:0",
- dst="2001:db8:1f0::c0a8:1:f") / payload)
+ p_ip4 = IP(src=self.pg0.remote_ip4, dst="192.168.0.1")
+ payload = TCP(sport=0xABCD, dport=0xABCD, flags="S", options=[("MSS", 1460)])
+
+ p4 = p_ether / p_ip4 / payload
+ p6_translated = (
+ IPv6(src="1234:5678:90ab:cdef:ac:1001:200:0", dst="2001:db8:1f0::c0a8:1:f")
+ / payload
+ )
p6_translated.hlim -= 1
- p6_translated[TCP].options = [('MSS', 1300)]
- rx = self.send_and_expect(self.pg0, p4*1, self.pg1)
+ p6_translated[TCP].options = [("MSS", 1300)]
+ rx = self.send_and_expect(self.pg0, p4 * 1, self.pg1)
for p in rx:
self.validate(p[1], p6_translated)
# Send back an IPv6 packet that will be "untranslated"
p_ether6 = Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac)
- p_ip6 = IPv6(src='2001:db8:1f0::c0a8:1:f',
- dst='1234:5678:90ab:cdef:ac:1001:200:0')
- p6 = (p_ether6 / p_ip6 / payload)
- p4_translated = (IP(src='192.168.0.1',
- dst=self.pg0.remote_ip4) / payload)
+ p_ip6 = IPv6(
+ src="2001:db8:1f0::c0a8:1:f", dst="1234:5678:90ab:cdef:ac:1001:200:0"
+ )
+ p6 = p_ether6 / p_ip6 / payload
+ p4_translated = IP(src="192.168.0.1", dst=self.pg0.remote_ip4) / payload
p4_translated.id = 0
p4_translated.ttl -= 1
- p4_translated[TCP].options = [('MSS', 1300)]
- rx = self.send_and_expect(self.pg1, p6*1, self.pg0)
+ p4_translated[TCP].options = [("MSS", 1300)]
+ rx = self.send_and_expect(self.pg1, p6 * 1, self.pg0)
for p in rx:
self.validate(p[1], p4_translated)
@@ -803,19 +862,22 @@ class TestMAP(VppTestCase):
# Send back an IPv6 packet that will be droppped due to security
# check fail
p_ether6 = Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac)
- p_ip6_sec_check_fail = IPv6(src='2001:db8:1fe::c0a8:1:f',
- dst='1234:5678:90ab:cdef:ac:1001:200:0')
- payload = TCP(sport=0xabcd, dport=0xabcd)
- p6 = (p_ether6 / p_ip6_sec_check_fail / payload)
+ p_ip6_sec_check_fail = IPv6(
+ src="2001:db8:1fe::c0a8:1:f", dst="1234:5678:90ab:cdef:ac:1001:200:0"
+ )
+ payload = TCP(sport=0xABCD, dport=0xABCD)
+ p6 = p_ether6 / p_ip6_sec_check_fail / payload
- self.pg_send(self.pg1, p6*1)
+ self.pg_send(self.pg1, p6 * 1)
self.pg0.get_capture(0, timeout=1)
rx = self.pg1.get_capture(1)
- icmp6_reply = (IPv6(hlim=255, src=self.pg1.local_ip6,
- dst='2001:db8:1fe::c0a8:1:f') /
- ICMPv6DestUnreach(code=5) /
- p_ip6_sec_check_fail / payload)
+ icmp6_reply = (
+ IPv6(hlim=255, src=self.pg1.local_ip6, dst="2001:db8:1fe::c0a8:1:f")
+ / ICMPv6DestUnreach(code=5)
+ / p_ip6_sec_check_fail
+ / payload
+ )
for p in rx:
self.validate(p[1], icmp6_reply)
@@ -824,142 +886,160 @@ class TestMAP(VppTestCase):
self.vapi.map_param_set_icmp6(enable_unreachable=0)
def test_map_t_ip6_psid(self):
- """ MAP-T v6->v4 PSID validation"""
+ """MAP-T v6->v4 PSID validation"""
#
# Add a domain that maps from pg0 to pg1
#
- map_dst = '2001:db8::/32'
- map_src = '1234:5678:90ab:cdef::/64'
- ip4_pfx = '192.168.0.0/24'
- tag = 'MAP-T Test Domain'
-
- self.vapi.map_add_domain(ip6_prefix=map_dst,
- ip4_prefix=ip4_pfx,
- ip6_src=map_src,
- ea_bits_len=16,
- psid_offset=6,
- psid_length=4,
- mtu=1500,
- tag=tag)
+ map_dst = "2001:db8::/32"
+ map_src = "1234:5678:90ab:cdef::/64"
+ ip4_pfx = "192.168.0.0/24"
+ tag = "MAP-T Test Domain"
+
+ self.vapi.map_add_domain(
+ ip6_prefix=map_dst,
+ ip4_prefix=ip4_pfx,
+ ip6_src=map_src,
+ ea_bits_len=16,
+ psid_offset=6,
+ psid_length=4,
+ mtu=1500,
+ tag=tag,
+ )
# Enable MAP-T on interfaces.
- self.vapi.map_if_enable_disable(is_enable=1,
- sw_if_index=self.pg0.sw_if_index,
- is_translation=1)
- self.vapi.map_if_enable_disable(is_enable=1,
- sw_if_index=self.pg1.sw_if_index,
- is_translation=1)
-
- map_route = VppIpRoute(self,
- "2001:db8::",
- 32,
- [VppRoutePath(self.pg1.remote_ip6,
- self.pg1.sw_if_index,
- proto=DpoProto.DPO_PROTO_IP6)])
+ self.vapi.map_if_enable_disable(
+ is_enable=1, sw_if_index=self.pg0.sw_if_index, is_translation=1
+ )
+ self.vapi.map_if_enable_disable(
+ is_enable=1, sw_if_index=self.pg1.sw_if_index, is_translation=1
+ )
+
+ map_route = VppIpRoute(
+ self,
+ "2001:db8::",
+ 32,
+ [
+ VppRoutePath(
+ self.pg1.remote_ip6,
+ self.pg1.sw_if_index,
+ proto=DpoProto.DPO_PROTO_IP6,
+ )
+ ],
+ )
map_route.add_vpp_config()
p_ether6 = Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac)
- p_ip6 = IPv6(src='2001:db8:1f0::c0a8:1:f',
- dst='1234:5678:90ab:cdef:ac:1001:200:0')
+ p_ip6 = IPv6(
+ src="2001:db8:1f0::c0a8:1:f", dst="1234:5678:90ab:cdef:ac:1001:200:0"
+ )
# Send good IPv6 source port, ensure translated IPv4 received
- payload = TCP(sport=0xabcd, dport=80)
- p6 = (p_ether6 / p_ip6 / payload)
- p4_translated = (IP(src='192.168.0.1',
- dst=self.pg0.remote_ip4) / payload)
+ payload = TCP(sport=0xABCD, dport=80)
+ p6 = p_ether6 / p_ip6 / payload
+ p4_translated = IP(src="192.168.0.1", dst=self.pg0.remote_ip4) / payload
p4_translated.id = 0
p4_translated.ttl -= 1
- rx = self.send_and_expect(self.pg1, p6*1, self.pg0)
+ rx = self.send_and_expect(self.pg1, p6 * 1, self.pg0)
for p in rx:
self.validate(p[1], p4_translated)
# Send bad IPv6 source port, ensure translated IPv4 not received
- payload = TCP(sport=0xdcba, dport=80)
- p6 = (p_ether6 / p_ip6 / payload)
- self.send_and_assert_no_replies(self.pg1, p6*1)
+ payload = TCP(sport=0xDCBA, dport=80)
+ p6 = p_ether6 / p_ip6 / payload
+ self.send_and_assert_no_replies(self.pg1, p6 * 1)
def test_map_t_pre_resolve(self):
- """ MAP-T pre-resolve"""
+ """MAP-T pre-resolve"""
# Add a domain that maps from pg0 to pg1
- map_dst = '2001:db8::/32'
- map_src = '1234:5678:90ab:cdef::/64'
- ip4_pfx = '192.168.0.0/24'
- tag = 'MAP-T Test Domain.'
-
- self.vapi.map_add_domain(ip6_prefix=map_dst,
- ip4_prefix=ip4_pfx,
- ip6_src=map_src,
- ea_bits_len=16,
- psid_offset=6,
- psid_length=4,
- mtu=1500,
- tag=tag)
+ map_dst = "2001:db8::/32"
+ map_src = "1234:5678:90ab:cdef::/64"
+ ip4_pfx = "192.168.0.0/24"
+ tag = "MAP-T Test Domain."
+
+ self.vapi.map_add_domain(
+ ip6_prefix=map_dst,
+ ip4_prefix=ip4_pfx,
+ ip6_src=map_src,
+ ea_bits_len=16,
+ psid_offset=6,
+ psid_length=4,
+ mtu=1500,
+ tag=tag,
+ )
# Enable MAP-T on interfaces.
- self.vapi.map_if_enable_disable(is_enable=1,
- sw_if_index=self.pg0.sw_if_index,
- is_translation=1)
- self.vapi.map_if_enable_disable(is_enable=1,
- sw_if_index=self.pg1.sw_if_index,
- is_translation=1)
+ self.vapi.map_if_enable_disable(
+ is_enable=1, sw_if_index=self.pg0.sw_if_index, is_translation=1
+ )
+ self.vapi.map_if_enable_disable(
+ is_enable=1, sw_if_index=self.pg1.sw_if_index, is_translation=1
+ )
# Enable pre-resolve option
- self.vapi.map_param_add_del_pre_resolve(ip4_nh_address="10.1.2.3",
- ip6_nh_address="4001::1",
- is_add=1)
+ self.vapi.map_param_add_del_pre_resolve(
+ ip4_nh_address="10.1.2.3", ip6_nh_address="4001::1", is_add=1
+ )
# Add a route to 4001::1 and expect the translated traffic to be
# sent via that route next-hop.
- pre_res_route6 = VppIpRoute(self, "4001::1", 128,
- [VppRoutePath(self.pg1.remote_hosts[2].ip6,
- self.pg1.sw_if_index)])
+ pre_res_route6 = VppIpRoute(
+ self,
+ "4001::1",
+ 128,
+ [VppRoutePath(self.pg1.remote_hosts[2].ip6, self.pg1.sw_if_index)],
+ )
pre_res_route6.add_vpp_config()
# Add a route to 10.1.2.3 and expect the "untranslated" traffic to be
# sent via that route next-hop.
- pre_res_route4 = VppIpRoute(self, "10.1.2.3", 32,
- [VppRoutePath(self.pg0.remote_hosts[1].ip4,
- self.pg0.sw_if_index)])
+ pre_res_route4 = VppIpRoute(
+ self,
+ "10.1.2.3",
+ 32,
+ [VppRoutePath(self.pg0.remote_hosts[1].ip4, self.pg0.sw_if_index)],
+ )
pre_res_route4.add_vpp_config()
# Send an IPv4 packet that will be translated
p_ether = Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
- p_ip4 = IP(src=self.pg0.remote_ip4, dst='192.168.0.1')
- payload = TCP(sport=0xabcd, dport=0xabcd)
- p4 = (p_ether / p_ip4 / payload)
-
- p6_translated = (IPv6(src="1234:5678:90ab:cdef:ac:1001:200:0",
- dst="2001:db8:1f0::c0a8:1:f") / payload)
+ p_ip4 = IP(src=self.pg0.remote_ip4, dst="192.168.0.1")
+ payload = TCP(sport=0xABCD, dport=0xABCD)
+ p4 = p_ether / p_ip4 / payload
+
+ p6_translated = (
+ IPv6(src="1234:5678:90ab:cdef:ac:1001:200:0", dst="2001:db8:1f0::c0a8:1:f")
+ / payload
+ )
p6_translated.hlim -= 1
- rx = self.send_and_expect(self.pg0, p4*1, self.pg1)
+ rx = self.send_and_expect(self.pg0, p4 * 1, self.pg1)
for p in rx:
self.assertEqual(p[Ether].dst, self.pg1.remote_hosts[2].mac)
self.validate(p[1], p6_translated)
# Send back an IPv6 packet that will be "untranslated"
p_ether6 = Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac)
- p_ip6 = IPv6(src='2001:db8:1f0::c0a8:1:f',
- dst='1234:5678:90ab:cdef:ac:1001:200:0')
- p6 = (p_ether6 / p_ip6 / payload)
+ p_ip6 = IPv6(
+ src="2001:db8:1f0::c0a8:1:f", dst="1234:5678:90ab:cdef:ac:1001:200:0"
+ )
+ p6 = p_ether6 / p_ip6 / payload
- p4_translated = (IP(src='192.168.0.1',
- dst=self.pg0.remote_ip4) / payload)
+ p4_translated = IP(src="192.168.0.1", dst=self.pg0.remote_ip4) / payload
p4_translated.id = 0
p4_translated.ttl -= 1
- rx = self.send_and_expect(self.pg1, p6*1, self.pg0)
+ rx = self.send_and_expect(self.pg1, p6 * 1, self.pg0)
for p in rx:
self.assertEqual(p[Ether].dst, self.pg0.remote_hosts[1].mac)
self.validate(p[1], p4_translated)
# Cleanup pre-resolve option
- self.vapi.map_param_add_del_pre_resolve(ip4_nh_address="10.1.2.3",
- ip6_nh_address="4001::1",
- is_add=0)
+ self.vapi.map_param_add_del_pre_resolve(
+ ip4_nh_address="10.1.2.3", ip6_nh_address="4001::1", is_add=0
+ )
-if __name__ == '__main__':
+if __name__ == "__main__":
unittest.main(testRunner=VppTestRunner)