+ 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_ip4_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_ip4_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_ip4_frags(3123, 1200)
+ self.pg_enable_capture()
+ self.pg1.add_stream(frags)
+ self.pg_start()
+ rx = self.pg0.get_capture(6)
+ reass_pkt = reassemble4(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_ip4_frags(3123, 1200)
+ self.pg_enable_capture()
+ self.pg1.add_stream(frags)
+ self.pg_start()
+ rx = self.pg0.get_capture(2)
+ reass_pkt = reassemble4(rx)
+ p4_reply.ttl -= 1
+ p4_reply.id = 512
+ self.validate(reass_pkt, p4_reply)
+
+ def test_ipip_create(self):
+ """ ipip create / delete interface test """
+ 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)
+ 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 """