diff options
Diffstat (limited to 'test/test_vxlan.py')
| -rw-r--r-- | test/test_vxlan.py | 341 |
1 files changed, 254 insertions, 87 deletions
diff --git a/test/test_vxlan.py b/test/test_vxlan.py index 028275ccedf..6128d1070d3 100644 --- a/test/test_vxlan.py +++ b/test/test_vxlan.py @@ -3,22 +3,27 @@ import socket from util import ip4_range, reassemble4 import unittest -from framework import VppTestCase, VppTestRunner +from framework import VppTestCase +from asfframework import VppTestRunner from template_bd import BridgeDomain from scapy.layers.l2 import Ether +from scapy.layers.l2 import ARP from scapy.packet import Raw, bind_layers from scapy.layers.inet import IP, UDP +from scapy.layers.inet6 import IPv6 from scapy.layers.vxlan import VXLAN +from scapy.contrib.mpls import MPLS import util from vpp_ip_route import VppIpRoute, VppRoutePath from vpp_vxlan_tunnel import VppVxlanTunnel from vpp_ip import INVALID_INDEX +from vpp_neighbor import VppNeighbor class TestVxlan(BridgeDomain, VppTestCase): - """ VXLAN Test Case """ + """VXLAN Test Case""" def __init__(self, *args): BridgeDomain.__init__(self) @@ -29,14 +34,16 @@ class TestVxlan(BridgeDomain, VppTestCase): Encapsulate the original payload frame by adding VXLAN header with its UDP, IP and Ethernet fields """ - return (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) / - IP(src=self.pg0.remote_ip4, dst=self.pg0.local_ip4) / - UDP(sport=self.dport, dport=self.dport, chksum=0) / - VXLAN(vni=vni, flags=self.flags) / - pkt) + return ( + Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) + / IP(src=self.pg0.remote_ip4, dst=self.pg0.local_ip4) + / UDP(sport=self.dport, dport=self.dport, chksum=0) + / VXLAN(vni=vni, flags=self.flags) + / pkt + ) def ip_range(self, start, end): - """ range of remote ip's """ + """range of remote ip's""" return ip4_range(self.pg0.remote_ip4, start, end) def encap_mcast(self, pkt, src_ip, src_mac, vni): @@ -44,18 +51,20 @@ class TestVxlan(BridgeDomain, VppTestCase): Encapsulate the original payload frame by adding VXLAN header with its UDP, IP and Ethernet fields """ - return (Ether(src=src_mac, dst=self.mcast_mac) / - IP(src=src_ip, dst=self.mcast_ip4) / - UDP(sport=self.dport, dport=self.dport, chksum=0) / - VXLAN(vni=vni, flags=self.flags) / - pkt) + return ( + Ether(src=src_mac, dst=self.mcast_mac) + / IP(src=src_ip, dst=self.mcast_ip4) + / UDP(sport=self.dport, dport=self.dport, chksum=0) + / VXLAN(vni=vni, flags=self.flags) + / pkt + ) def decapsulate(self, pkt): """ Decapsulate the original payload frame by removing VXLAN header """ # check if is set I flag - self.assertEqual(pkt[VXLAN].flags, int('0x8', 16)) + self.assertEqual(pkt[VXLAN].flags, int("0x8", 16)) return pkt[VXLAN].payload # Method for checking VXLAN encapsulation. @@ -91,18 +100,25 @@ class TestVxlan(BridgeDomain, VppTestCase): ip_range_start = 10 ip_range_end = ip_range_start + n_ucast_tunnels next_hop_address = cls.pg0.remote_ip4 - for dest_ip4 in ip4_range(next_hop_address, ip_range_start, - ip_range_end): + for dest_ip4 in ip4_range(next_hop_address, ip_range_start, ip_range_end): # add host route so dest_ip4 will not be resolved - rip = VppIpRoute(cls, dest_ip4, 32, - [VppRoutePath(next_hop_address, - INVALID_INDEX)], - register=False) + rip = VppIpRoute( + cls, + dest_ip4, + 32, + [VppRoutePath(next_hop_address, INVALID_INDEX)], + register=False, + ) rip.add_vpp_config() - r = VppVxlanTunnel(cls, src=cls.pg0.local_ip4, - src_port=port, dst_port=port, - dst=dest_ip4, vni=vni) + r = VppVxlanTunnel( + cls, + src=cls.pg0.local_ip4, + src_port=port, + dst_port=port, + dst=dest_ip4, + vni=vni, + ) r.add_vpp_config() cls.vapi.sw_interface_set_l2_bridge(r.sw_if_index, bd_id=vni) @@ -116,12 +132,18 @@ class TestVxlan(BridgeDomain, VppTestCase): vni_start = 10000 vni_end = vni_start + n_shared_dst_tunnels for vni in range(vni_start, vni_end): - r = VppVxlanTunnel(cls, src=cls.pg0.local_ip4, - src_port=port, dst_port=port, - dst=cls.mcast_ip4, mcast_sw_if_index=1, vni=vni) + r = VppVxlanTunnel( + cls, + src=cls.pg0.local_ip4, + src_port=port, + dst_port=port, + dst=cls.mcast_ip4, + mcast_sw_if_index=1, + vni=vni, + ) if is_add: r.add_vpp_config() - if r.sw_if_index == 0xffffffff: + if r.sw_if_index == 0xFFFFFFFF: raise ValueError("bad sw_if_index: ~0") else: r.remove_vpp_config() @@ -142,12 +164,17 @@ class TestVxlan(BridgeDomain, VppTestCase): n_distinct_dst_tunnels = 200 ip_range_start = 10 ip_range_end = ip_range_start + n_distinct_dst_tunnels - for dest_ip4 in ip4_range(cls.mcast_ip4, ip_range_start, - ip_range_end): + for dest_ip4 in ip4_range(cls.mcast_ip4, ip_range_start, ip_range_end): vni = bytearray(socket.inet_pton(socket.AF_INET, dest_ip4))[3] - r = VppVxlanTunnel(cls, src=cls.pg0.local_ip4, - src_port=port, dst_port=port, - dst=dest_ip4, mcast_sw_if_index=1, vni=vni) + r = VppVxlanTunnel( + cls, + src=cls.pg0.local_ip4, + src_port=port, + dst_port=port, + dst=dest_ip4, + mcast_sw_if_index=1, + vni=vni, + ) if is_add: r.add_vpp_config() else: @@ -185,7 +212,7 @@ class TestVxlan(BridgeDomain, VppTestCase): cls.pg0.resolve_arp() # Our Multicast address - cls.mcast_ip4 = '239.1.1.1' + cls.mcast_ip4 = "239.1.1.1" cls.mcast_mac = util.mcast_ip_to_mac(cls.mcast_ip4) except Exception: cls.tearDownClass() @@ -205,30 +232,44 @@ class TestVxlan(BridgeDomain, VppTestCase): self.single_tunnel_vni = 0x12345 self.single_tunnel_bd = 1 - r = VppVxlanTunnel(self, src=self.pg0.local_ip4, - dst=self.pg0.remote_ip4, - src_port=self.dport, dst_port=self.dport, - vni=self.single_tunnel_vni) + r = VppVxlanTunnel( + self, + src=self.pg0.local_ip4, + dst=self.pg0.remote_ip4, + src_port=self.dport, + dst_port=self.dport, + vni=self.single_tunnel_vni, + ) r.add_vpp_config() - self.vapi.sw_interface_set_l2_bridge(rx_sw_if_index=r.sw_if_index, - bd_id=self.single_tunnel_bd) self.vapi.sw_interface_set_l2_bridge( - rx_sw_if_index=self.pg1.sw_if_index, bd_id=self.single_tunnel_bd) + rx_sw_if_index=r.sw_if_index, bd_id=self.single_tunnel_bd + ) + self.vapi.sw_interface_set_l2_bridge( + rx_sw_if_index=self.pg1.sw_if_index, bd_id=self.single_tunnel_bd + ) # Setup vni 2 to test multicast flooding self.n_ucast_tunnels = 10 self.mcast_flood_bd = 2 - self.create_vxlan_flood_test_bd(self.mcast_flood_bd, - self.n_ucast_tunnels, - self.dport) - r = VppVxlanTunnel(self, src=self.pg0.local_ip4, dst=self.mcast_ip4, - src_port=self.dport, dst_port=self.dport, - mcast_sw_if_index=1, vni=self.mcast_flood_bd) + self.create_vxlan_flood_test_bd( + self.mcast_flood_bd, self.n_ucast_tunnels, self.dport + ) + r = VppVxlanTunnel( + self, + src=self.pg0.local_ip4, + dst=self.mcast_ip4, + src_port=self.dport, + dst_port=self.dport, + mcast_sw_if_index=1, + vni=self.mcast_flood_bd, + ) r.add_vpp_config() - self.vapi.sw_interface_set_l2_bridge(rx_sw_if_index=r.sw_if_index, - bd_id=self.mcast_flood_bd) self.vapi.sw_interface_set_l2_bridge( - rx_sw_if_index=self.pg2.sw_if_index, bd_id=self.mcast_flood_bd) + rx_sw_if_index=r.sw_if_index, bd_id=self.mcast_flood_bd + ) + self.vapi.sw_interface_set_l2_bridge( + rx_sw_if_index=self.pg2.sw_if_index, bd_id=self.mcast_flood_bd + ) # Add and delete mcast tunnels to check stability self.add_shared_mcast_dst_load(self.dport) @@ -238,22 +279,90 @@ class TestVxlan(BridgeDomain, VppTestCase): # Setup vni 3 to test unicast flooding self.ucast_flood_bd = 3 - self.create_vxlan_flood_test_bd(self.ucast_flood_bd, - self.n_ucast_tunnels, - self.dport) + self.create_vxlan_flood_test_bd( + self.ucast_flood_bd, self.n_ucast_tunnels, self.dport + ) self.vapi.sw_interface_set_l2_bridge( - rx_sw_if_index=self.pg3.sw_if_index, bd_id=self.ucast_flood_bd) + rx_sw_if_index=self.pg3.sw_if_index, bd_id=self.ucast_flood_bd + ) # Set scapy listen custom port for VxLAN bind_layers(UDP, VXLAN, dport=self.dport) + def encap_packets(self): + def encap_frames(frame, n=10): + frames = [] + + # Provide IP flow hash difference. + for i in range(n): + p = frame.copy() + p[UDP].dport += i + frames.append(p) + + self.pg1.add_stream(frames) + + self.pg0.enable_capture() + self.pg_start() + + # Pick received frames and check if they're correctly encapsulated. + out = self.pg0.get_capture(n) + sports = set() + for i in range(n): + pkt = out[i] + self.check_encapsulation(pkt, self.single_tunnel_vni) + + payload = self.decapsulate(pkt) + self.assert_eq_pkts(payload, frames[i]) + + sports.add(pkt[UDP].sport) + + # Check src port randomization presence, not concerned with the + # src ports split ratio, just as long as there are more then one. + self.assertGreaterEqual(len(sports), min(n, 2)) + + frame_ip4 = ( + 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" * 100) + ) + encap_frames(frame_ip4) + + frame_ip6 = ( + Ether(src="00:00:00:00:00:02", dst="00:00:00:00:00:01") + / IPv6(src="2001:db8::4321", dst="2001:db8::1234") + / UDP(sport=20000, dport=10000) + / Raw("\xa5" * 100) + ) + encap_frames(frame_ip6) + + frame_mpls4 = ( + Ether(src="00:00:00:00:00:02", dst="00:00:00:00:00:01") + / MPLS(label=44, ttl=64) + / IP(src="4.3.2.1", dst="1.2.3.4") + / UDP(sport=20000, dport=10000) + / Raw("\xa5" * 100) + ) + encap_frames(frame_mpls4) + + frame_mpls6 = ( + Ether(src="00:00:00:00:00:02", dst="00:00:00:00:00:01") + / MPLS(label=44, ttl=64) + / IPv6(src="2001:db8::4321", dst="2001:db8::1234") + / UDP(sport=20000, dport=10000) + / Raw("\xa5" * 100) + ) + encap_frames(frame_mpls6) + def encap_big_packet(self): 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(b'\xa5' * 1450)) + 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(b"\xa5" * 1450) + ) self.pg1.add_stream([frame]) @@ -275,43 +384,44 @@ class TestVxlan(BridgeDomain, VppTestCase): """ Tests with default port (4789) """ + def test_decap(self): - """ Decapsulation test + """Decapsulation test from BridgeDoman """ self.createVxLANInterfaces() super(TestVxlan, self).test_decap() def test_encap(self): - """ Encapsulation test + """Encapsulation test from BridgeDoman """ self.createVxLANInterfaces() - super(TestVxlan, self).test_encap() + self.encap_packets() def test_encap_big_packet(self): - """ Encapsulation test send big frame from pg1 + """Encapsulation test send big frame from pg1 Verify receipt of encapsulated frames on pg0 """ self.createVxLANInterfaces() self.encap_big_packet() def test_ucast_flood(self): - """ Unicast flood test + """Unicast flood test from BridgeDoman """ self.createVxLANInterfaces() super(TestVxlan, self).test_ucast_flood() def test_mcast_flood(self): - """ Multicast flood test + """Multicast flood test from BridgeDoman """ self.createVxLANInterfaces() super(TestVxlan, self).test_mcast_flood() def test_mcast_rcv(self): - """ Multicast receive test + """Multicast receive test from BridgeDoman """ self.createVxLANInterfaces() @@ -320,36 +430,37 @@ class TestVxlan(BridgeDomain, VppTestCase): """ Tests with custom port """ + def test_decap_custom_port(self): - """ Decapsulation test custom port + """Decapsulation test custom port from BridgeDoman """ self.createVxLANInterfaces(1111) super(TestVxlan, self).test_decap() def test_encap_custom_port(self): - """ Encapsulation test custom port + """Encapsulation test custom port from BridgeDoman """ self.createVxLANInterfaces(1111) super(TestVxlan, self).test_encap() def test_ucast_flood_custom_port(self): - """ Unicast flood test custom port + """Unicast flood test custom port from BridgeDoman """ self.createVxLANInterfaces(1111) super(TestVxlan, self).test_ucast_flood() def test_mcast_flood_custom_port(self): - """ Multicast flood test custom port + """Multicast flood test custom port from BridgeDoman """ self.createVxLANInterfaces(1111) super(TestVxlan, self).test_mcast_flood() def test_mcast_rcv_custom_port(self): - """ Multicast receive test custom port + """Multicast receive test custom port from BridgeDoman """ self.createVxLANInterfaces(1111) @@ -370,7 +481,8 @@ class TestVxlan(BridgeDomain, VppTestCase): class TestVxlan2(VppTestCase): - """ VXLAN Test Case """ + """VXLAN Test Case""" + def setUp(self): super(TestVxlan2, self).setUp() @@ -387,35 +499,90 @@ class TestVxlan2(VppTestCase): super(TestVxlan2, self).tearDown() def test_xconnect(self): - """ VXLAN source address not local """ + """VXLAN source address not local""" # # test the broken configuration of a VXLAN tunnel whose # source address is not local ot the box. packets sent # through the tunnel should be dropped # - t = VppVxlanTunnel(self, - src="10.0.0.5", - dst=self.pg0.local_ip4, - vni=1000) + t = VppVxlanTunnel(self, src="10.0.0.5", dst=self.pg0.local_ip4, vni=1000) t.add_vpp_config() t.admin_up() - self.vapi.sw_interface_set_l2_xconnect(t.sw_if_index, - self.pg1.sw_if_index, - enable=1) - self.vapi.sw_interface_set_l2_xconnect(self.pg1.sw_if_index, - t.sw_if_index, - enable=1) + self.vapi.sw_interface_set_l2_xconnect( + t.sw_if_index, self.pg1.sw_if_index, enable=1 + ) + self.vapi.sw_interface_set_l2_xconnect( + self.pg1.sw_if_index, t.sw_if_index, enable=1 + ) - p = (Ether(src="00:11:22:33:44:55", - dst="00:00:00:11:22:33") / - IP(src="4.3.2.1", dst="1.2.3.4") / - UDP(sport=20000, dport=10000) / - Raw(b'\xa5' * 1450)) + p = ( + Ether(src="00:11:22:33:44:55", dst="00:00:00:11:22:33") + / IP(src="4.3.2.1", dst="1.2.3.4") + / UDP(sport=20000, dport=10000) + / Raw(b"\xa5" * 1450) + ) rx = self.send_and_assert_no_replies(self.pg1, [p]) -if __name__ == '__main__': +class TestVxlanL2Mode(VppTestCase): + """VXLAN Test Case""" + + def setUp(self): + super(TestVxlanL2Mode, self).setUp() + + # Create 2 pg interfaces. + self.create_pg_interfaces(range(2)) + for pg in self.pg_interfaces: + pg.admin_up() + + # Configure IPv4 addresses on VPP pg0. + self.pg0.config_ip4() + self.pg0.resolve_arp() + + # Configure IPv4 addresses on VPP pg1. + self.pg1.config_ip4() + + def tearDown(self): + super(TestVxlanL2Mode, self).tearDown() + + def test_l2_mode(self): + """VXLAN L2 mode""" + t = VppVxlanTunnel( + self, src=self.pg0.local_ip4, dst=self.pg0.remote_ip4, vni=1000, is_l3=False + ) + t.add_vpp_config() + t.config_ip4() + t.admin_up() + + dstIP = t.local_ip4[:-1] + "2" + + # Create a packet to send + p = ( + Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) + / IP(src=self.pg1.local_ip4, dst=dstIP) + / UDP(sport=555, dport=556) + / Raw(b"\x00" * 80) + ) + + # Expect ARP request + rx = self.send_and_expect(self.pg1, [p], self.pg0) + for p in rx: + self.assertEqual(p[Ether].dst, self.pg0.remote_mac) + self.assertEqual(p[Ether].src, self.pg0.local_mac) + self.assertEqual(p[ARP].op, 1) + self.assertEqual(p[ARP].pdst, dstIP) + + # Resolve ARP + VppNeighbor(self, t.sw_if_index, self.pg1.remote_mac, dstIP).add_vpp_config() + + # Send packets + NUM_PKTS = 128 + rx = self.send_and_expect(self.pg1, p * NUM_PKTS, self.pg0) + self.assertEqual(NUM_PKTS, len(rx)) + + +if __name__ == "__main__": unittest.main(testRunner=VppTestRunner) |