aboutsummaryrefslogtreecommitdiffstats
path: root/test
diff options
context:
space:
mode:
Diffstat (limited to 'test')
-rw-r--r--test/test_ipip.py51
1 files changed, 51 insertions, 0 deletions
diff --git a/test/test_ipip.py b/test/test_ipip.py
index 00721ec90a0..582ab5be84c 100644
--- a/test/test_ipip.py
+++ b/test/test_ipip.py
@@ -3,9 +3,11 @@
import unittest
from scapy.layers.inet6 import IPv6, Ether, IP, UDP
+from scapy.all import fragment
from framework import VppTestCase, VppTestRunner
from vpp_ip_route import VppIpRoute, VppRoutePath, DpoProto
from socket import AF_INET, AF_INET6, inet_pton
+import StringIO
""" Testipip is a subclass of VPPTestCase classes.
@@ -14,6 +16,20 @@ IPIP tests.
"""
+def reassemble(listoffragments):
+ buffer = StringIO.StringIO()
+ first = listoffragments[0]
+ buffer.seek(20)
+ for pkt in listoffragments:
+ buffer.seek(pkt[IP].frag*8)
+ buffer.write(pkt[IP].payload)
+ first.len = len(buffer.getvalue()) + 20
+ first.flags = 0
+ del(first.chksum)
+ header = str(first[IP])[:20]
+ return first[IP].__class__(header + buffer.getvalue())
+
+
class TestIPIP(VppTestCase):
""" IPIP Test Case """
@@ -126,6 +142,38 @@ class TestIPIP(VppTestCase):
for p in rx:
self.validate(p[1], p6_reply)
+ # Fragmentation / Reassembly and Re-fragmentation
+ rv = self.vapi.ip_reassembly_enable_disable(
+ sw_if_index=self.pg1.sw_if_index,
+ enable_ip4=1)
+ # Decapsulation
+ p_ether = Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac)
+ p_payload = UDP(sport=1234, dport=1234) / self.payload(3123)
+ p_ip4 = IP(src="1.2.3.4", dst=self.pg0.remote_ip4)
+ outer_ip4 = (p_ether / IP(src=self.pg1.remote_ip4,
+ dst=self.pg0.local_ip4) / p_ip4 / p_payload)
+ frags = fragment(outer_ip4, 1400)
+ p4_reply = (p_ip4 / p_payload)
+ p4_reply.ttl -= 1
+
+ self.pg_enable_capture()
+ self.pg1.add_stream(frags)
+ self.pg_start()
+ rx = self.pg0.get_capture(1)
+ for p in rx:
+ self.validate(p[1], p4_reply)
+
+ # Now try with re-fragmentation
+ self.vapi.sw_interface_set_mtu(self.pg0.sw_if_index, [576, 0, 0, 0])
+ self.pg_enable_capture()
+ self.pg1.add_stream(frags)
+ self.pg_start()
+ rx = self.pg0.get_capture(6)
+ reass_pkt = reassemble(rx)
+ p4_reply.ttl -= 1
+ p4_reply.id = 256
+ self.validate(reass_pkt, p4_reply)
+
def test_ipip6(self):
""" ip{v4,v6} over ip6 test """
p_ether = Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
@@ -213,6 +261,9 @@ class TestIPIP(VppTestCase):
sw_if_index = rv.sw_if_index
self.vapi.ipip_del_tunnel(sw_if_index)
+ def payload(self, len):
+ return 'x' * len
+
if __name__ == '__main__':
unittest.main(testRunner=VppTestRunner)