STATS: Dynamically mapped shared memory segment

[vpp.git] / test / test_ipip.py

diff --git a/test/test_ipip.py b/test/test_ipip.py
index 00721ec..efaaf52 100644 (file)
--- a/test/test_ipip.py
+++ b/test/test_ipip.py
@@ -3,9 +3,12 @@
 
 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 vpp_ip import DpoProto
+from vpp_ip_route import VppIpRoute, VppRoutePath, VppIpTable
 from socket import AF_INET, AF_INET6, inet_pton
+import StringIO
 
 """ Testipip is a subclass of  VPPTestCase classes.
 
@@ -14,6 +17,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 +61,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)
@@ -116,6 +145,10 @@ class TestIPIP(VppTestCase):
         for p in rx:
             self.validate(p[1], p4_reply)
 
+        err = self.statistics.get_counter(
+            '/err/ipip4-input/packets decapsulated')
+        self.assertEqual(err, 10)
+
         # IPv4 tunnel to IPv6
         p_ip6 = IPv6(src="1:2:3::4", dst=self.pg0.remote_ip6)
         p6 = (p_ether / IP(src=self.pg1.remote_ip4,
@@ -126,6 +159,76 @@ class TestIPIP(VppTestCase):
         for p in rx:
             self.validate(p[1], p6_reply)
 
+        err = self.statistics.get_counter(
+            '/err/ipip4-input/packets decapsulated')
+        self.assertEqual(err, 20)
+
+        #
+        # 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)
+
+        err = self.statistics.get_counter(
+            '/err/ipip4-input/packets decapsulated')
+        self.assertEqual(err, 1020)
+
+        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 +316,21 @@ class TestIPIP(VppTestCase):
         sw_if_index = rv.sw_if_index
         self.vapi.ipip_del_tunnel(sw_if_index)
 
+    def test_ipip_vrf_create(self):
+        """ ipip create / delete interface VRF test """
+
+        t = VppIpTable(self, 20)
+        t.add_vpp_config()
+        rv = self.vapi.ipip_add_tunnel(
+            src_address=inet_pton(AF_INET, '1.2.3.4'),
+            dst_address=inet_pton(AF_INET, '2.3.4.5'), is_ipv6=0,
+            table_id=20)
+        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)