from framework import VppTestCase, VppTestRunner
from vpp_ip import DpoProto
from vpp_ip_route import VppIpRoute, VppRoutePath
-from util import fragment_rfc791
+from util import fragment_rfc791, fragment_rfc8200
import scapy.compat
from scapy.layers.l2 import Ether, Raw
for p in rx:
self.validate(p[1], v4_reply)
- self.logger.debug("show trace")
-
#
# Fire in a v4 packet that will be encapped to the BR
#
self.send_and_assert_encapped_one(v4, "3000::1", map_translated_addr)
- self.logger.debug("show trace")
#
# Verify reordered fragments are able to pass as well
#
self.send_and_assert_encapped(frags, "3000::1", map_translated_addr)
- self.logger.debug("show trace")
-
# Enable MAP on interface.
self.vapi.map_if_enable_disable(is_enable=1,
sw_if_index=self.pg1.sw_if_index,
self.assertEqual(r[IP].src, p[IP].src)
self.assertEqual(r[IP].dst, p[IP].dst)
- return
+ # Verify that fragments pass even if ipv6 layer is fragmented
+ 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)]
+
+ self.pg1.add_stream(v6_stream)
+
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+
+ rx = self.pg0.get_capture(len(frags))
+
+ for r in rx:
+ self.assertFalse(r.haslayer(IPv6))
+ self.assertEqual(r[IP].src, p[IP].src)
+ self.assertEqual(r[IP].dst, p[IP].dst)
#
# Pre-resolve. No API for this!!
pre_res_route.add_vpp_config()
self.send_and_assert_encapped_one(v4, "3000::1",
- "2001::c0a8:0:0",
+ map_translated_addr,
dmac=self.pg1.remote_hosts[2].mac)
#
pre_res_route.add_vpp_config()
self.send_and_assert_encapped_one(v4, "3000::1",
- "2001::c0a8:0:0",
+ map_translated_addr,
dmac=self.pg1.remote_hosts[3].mac)
#
for p in rx:
pass
# p.show2()
+
# reass_pkt = reassemble(rx)
# p4_reply.ttl -= 1
# p4_reply.id = 256