X-Git-Url: https://gerrit.fd.io/r/gitweb?a=blobdiff_plain;f=test%2Ftest_ipip.py;h=8d99f83e6c21f3047514dd5044c2dd493d7b41f0;hb=7eb9d9608b5b47b95f4860b2c470411658f7cdac;hp=00721ec90a097ab070e54d39c2e5f29de9df0fd5;hpb=d57f63698f99fad0288ac040d83b3ecd380d4bfd;p=vpp.git

diff --git a/test/test_ipip.py b/test/test_ipip.py
index 00721ec90a0..8d99f83e6c2 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, RandShort
 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 """
 
@@ -44,6 +60,18 @@ class TestIPIP(VppTestCase):
     def validate(self, rx, expected):
         self.assertEqual(rx, expected.__class__(str(expected)))
 
+    def generate_frags(self, payload_length, fragment_size):
+        p_ether = Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac)
+        p_payload = UDP(sport=1234, dport=1234) / self.payload(payload_length)
+        p_ip4 = IP(src="1.2.3.4", dst=self.pg0.remote_ip4)
+        outer_ip4 = (p_ether / IP(src=self.pg1.remote_ip4,
+                                  id=RandShort(),
+                                  dst=self.pg0.local_ip4) / p_ip4 / p_payload)
+        frags = fragment(outer_ip4, fragment_size)
+        p4_reply = (p_ip4 / p_payload)
+        p4_reply.ttl -= 1
+        return frags, p4_reply
+
     def test_ipip4(self):
         """ ip{v4,v6} over ip4 test """
         p_ether = Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
@@ -126,6 +154,68 @@ 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)
+
+        # Send lots of fragments, verify reassembled packet
+        frags, p4_reply = self.generate_frags(3131, 1400)
+        f = []
+        for i in range(0, 1000):
+            f.extend(frags)
+        self.pg1.add_stream(f)
+        self.pg_enable_capture()
+        self.pg_start()
+        rx = self.pg0.get_capture(1000)
+
+        for p in rx:
+            self.validate(p[1], p4_reply)
+
+        f = []
+        r = []
+        for i in range(1, 90):
+            frags, p4_reply = self.generate_frags(i * 100, 1000)
+            f.extend(frags)
+            r.extend(p4_reply)
+        self.pg_enable_capture()
+        self.pg1.add_stream(f)
+        self.pg_start()
+        rx = self.pg0.get_capture(89)
+        i = 0
+        for p in rx:
+            self.validate(p[1], r[i])
+            i += 1
+
+        # Now try with re-fragmentation
+        #
+        # Send fragments to tunnel head-end, for the tunnel head end
+        # to reassemble and then refragment
+        #
+        self.vapi.sw_interface_set_mtu(self.pg0.sw_if_index, [576, 0, 0, 0])
+        frags, p4_reply = self.generate_frags(3123, 1200)
+        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)
+
+        self.vapi.sw_interface_set_mtu(self.pg0.sw_if_index, [1600, 0, 0, 0])
+        frags, p4_reply = self.generate_frags(3123, 1200)
+        self.pg_enable_capture()
+        self.pg1.add_stream(frags)
+        self.pg_start()
+        rx = self.pg0.get_capture(2)
+        reass_pkt = reassemble(rx)
+        p4_reply.ttl -= 1
+        p4_reply.id = 512
+        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 +303,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)