#!/usr/bin/env python
import socket
-from util import ip4n_range
+from util import ip4n_range, ip4_range
import unittest
from framework import VppTestCase, VppTestRunner
from template_bd import BridgeDomain
-from scapy.layers.l2 import Ether
+from scapy.layers.l2 import Ether, Raw
from scapy.layers.inet import IP, UDP
from scapy.layers.vxlan import VXLAN
from scapy.utils import atol
+import StringIO
+
+
+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 TestVxlan(BridgeDomain, VppTestCase):
""" VXLAN Test Case """
VXLAN(vni=vni, flags=self.flags) /
pkt)
+ def ip_range(self, start, end):
+ """ range of remote ip's """
+ return ip4_range(self.pg0.remote_ip4, start, end)
+
def encap_mcast(self, pkt, src_ip, src_mac, vni):
"""
Encapsulate the original payload frame by adding VXLAN header with its
# Method for checking VXLAN encapsulation.
#
- def check_encapsulation(self, pkt, vni, local_only=False):
+ def check_encapsulation(self, pkt, vni, local_only=False, mcast_pkt=False):
# TODO: add error messages
# Verify source MAC is VPP_MAC and destination MAC is MY_MAC resolved
# by VPP using ARP.
self.assertEqual(pkt[Ether].src, self.pg0.local_mac)
if not local_only:
- self.assertEqual(pkt[Ether].dst, self.pg0.remote_mac)
+ if not mcast_pkt:
+ self.assertEqual(pkt[Ether].dst, self.pg0.remote_mac)
+ else:
+ self.assertEqual(pkt[Ether].dst, type(self).mcast_mac)
# Verify VXLAN tunnel source IP is VPP_IP and destination IP is MY_IP.
self.assertEqual(pkt[IP].src, self.pg0.local_ip4)
if not local_only:
- self.assertEqual(pkt[IP].dst, self.pg0.remote_ip4)
+ if not mcast_pkt:
+ self.assertEqual(pkt[IP].dst, self.pg0.remote_ip4)
+ else:
+ self.assertEqual(pkt[IP].dst, type(self).mcast_ip4)
# Verify UDP destination port is VXLAN 4789, source UDP port could be
# arbitrary.
self.assertEqual(pkt[UDP].dport, type(self).dport)
add or del tunnels sharing the same mcast dst
to test vxlan ref_count mechanism
"""
- n_shared_dst_tunnels = 2000
+ n_shared_dst_tunnels = 20
vni_start = 10000
vni_end = vni_start + n_shared_dst_tunnels
for vni in range(vni_start, vni_end):
- cls.vapi.vxlan_add_del_tunnel(
+ r = cls.vapi.vxlan_add_del_tunnel(
src_addr=cls.pg0.local_ip4n,
dst_addr=cls.mcast_ip4n,
mcast_sw_if_index=1,
vni=vni,
is_add=is_add)
+ if r.sw_if_index == 0xffffffff:
+ raise "bad sw_if_index"
@classmethod
def add_shared_mcast_dst_load(cls):
super(TestVxlan, cls).tearDownClass()
raise
+ def test_encap_big_packet(self):
+ """ Encapsulation test send big frame from pg1
+ Verify receipt of encapsulated frames on pg0
+ """
+
+ self.vapi.sw_interface_set_mtu(self.pg0.sw_if_index, [1500, 0, 0, 0])
+
+ frame = (Ether(src='00:00:00:00:00:02', dst='00:00:00:00:00:01') /
+ IP(src='4.3.2.1', dst='1.2.3.4') /
+ UDP(sport=20000, dport=10000) /
+ Raw('\xa5' * 1450))
+
+ self.pg1.add_stream([frame])
+
+ self.pg0.enable_capture()
+
+ self.pg_start()
+
+ # Pick first received frame and check if it's correctly encapsulated.
+ out = self.pg0.get_capture(2)
+ ether = out[0]
+ pkt = reassemble(out)
+ pkt = ether / pkt
+ self.check_encapsulation(pkt, self.single_tunnel_bd)
+
+ payload = self.decapsulate(pkt)
+ # TODO: Scapy bug?
+ # self.assert_eq_pkts(payload, frame)
+
# Method to define VPP actions before tear down of the test case.
# Overrides tearDown method in VppTestCase class.
# @param self The object pointer.