from vpp_ip import DpoProto
from vpp_ip_route import VppIpRoute, VppRoutePath, VppIpTable
from socket import AF_INET, AF_INET6, inet_pton
-import StringIO
+from util import reassemble4
+
""" Testipip is a subclass of VPPTestCase classes.
"""
-# Replace by deframent from scapy.
-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 """
cls.create_pg_interfaces(range(2))
cls.interfaces = list(cls.pg_interfaces)
- def setUp(cls):
- super(TestIPIP, cls).setUp()
- for i in cls.interfaces:
+ def setUp(self):
+ super(TestIPIP, self).setUp()
+ for i in self.interfaces:
i.admin_up()
i.config_ip4()
i.config_ip6()
i.admin_down()
def validate(self, rx, expected):
- self.assertEqual(rx, expected.__class__(str(expected)))
+ self.assertEqual(rx, expected.__class__(expected))
def generate_ip4_frags(self, payload_length, fragment_size):
p_ether = Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac)
self.pg1.add_stream(frags)
self.pg_start()
rx = self.pg0.get_capture(6)
- reass_pkt = reassemble(rx)
+ reass_pkt = reassemble4(rx)
p4_reply.ttl -= 1
p4_reply.id = 256
self.validate(reass_pkt, p4_reply)
self.pg1.add_stream(frags)
self.pg_start()
rx = self.pg0.get_capture(2)
- reass_pkt = reassemble(rx)
+ reass_pkt = reassemble4(rx)
p4_reply.ttl -= 1
p4_reply.id = 512
self.validate(reass_pkt, p4_reply)
rv = self.vapi.ipip_del_tunnel(sw_if_index=self.tunnel_if_index)
def validate(self, rx, expected):
- self.assertEqual(rx, expected.__class__(str(expected)))
+ self.assertEqual(rx, expected.__class__(expected))
def generate_ip6_frags(self, payload_length, fragment_size):
p_ether = Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac)