diff options
Diffstat (limited to 'test/test_vxlan_gpe.py')
| -rw-r--r-- | test/test_vxlan_gpe.py | 227 |
1 files changed, 152 insertions, 75 deletions
diff --git a/test/test_vxlan_gpe.py b/test/test_vxlan_gpe.py index c5d6bf07f7c..28c31ccd3bb 100644 --- a/test/test_vxlan_gpe.py +++ b/test/test_vxlan_gpe.py @@ -7,13 +7,13 @@ from framework import VppTestCase, VppTestRunner, running_extended_tests from template_bd import BridgeDomain from scapy.layers.l2 import Ether -from scapy.packet import Raw +from scapy.packet import Raw, bind_layers from scapy.layers.inet import IP, UDP from scapy.layers.vxlan import VXLAN import util from vpp_ip_route import VppIpRoute, VppRoutePath - +from vpp_vxlan_gpe_tunnel import VppVxlanGpeTunnel from vpp_ip import INVALID_INDEX @@ -79,14 +79,14 @@ class TestVxlanGpe(BridgeDomain, VppTestCase): self.assertEqual(pkt[IP].dst, type(self).mcast_ip4) # Verify UDP destination port is VXLAN-GPE 4790, source UDP port # could be arbitrary. - self.assertEqual(pkt[UDP].dport, type(self).dport) + self.assertEqual(pkt[UDP].dport, self.dport) # Verify UDP checksum self.assert_udp_checksum_valid(pkt) # Verify VNI self.assertEqual(pkt[VXLAN].vni, vni) @classmethod - def create_vxlan_gpe_flood_test_bd(cls, vni, n_ucast_tunnels): + def create_vxlan_gpe_flood_test_bd(cls, vni, n_ucast_tunnels, port): # Create 10 ucast vxlan tunnels under bd ip_range_start = 10 ip_range_end = ip_range_start + n_ucast_tunnels @@ -100,15 +100,18 @@ class TestVxlanGpe(BridgeDomain, VppTestCase): register=False) rip.add_vpp_config() - r = cls.vapi.vxlan_gpe_add_del_tunnel( - src_addr=cls.pg0.local_ip4, - dst_addr=dest_ip4, - vni=vni) + r = VppVxlanGpeTunnel(cls, + src_addr=cls.pg0.local_ip4, + dst_addr=dest_ip4, + src_port=port, + dst_port=port, + vni=vni) + r.add_vpp_config() cls.vapi.sw_interface_set_l2_bridge(rx_sw_if_index=r.sw_if_index, bd_id=vni) @classmethod - def add_del_shared_mcast_dst_load(cls, is_add): + def add_del_shared_mcast_dst_load(cls, port, is_add): """ add or del tunnels sharing the same mcast dst to test vxlan_gpe ref_count mechanism @@ -117,25 +120,30 @@ class TestVxlanGpe(BridgeDomain, VppTestCase): vni_start = 1000 vni_end = vni_start + n_shared_dst_tunnels for vni in range(vni_start, vni_end): - r = cls.vapi.vxlan_gpe_add_del_tunnel( - local=cls.pg0.local_ip4, - remote=cls.mcast_ip4, - mcast_sw_if_index=1, - vni=vni, - is_add=is_add) - if r.sw_if_index == 0xffffffff: - raise ValueError("bad sw_if_index: ~0") + r = VppVxlanGpeTunnel(cls, + src_addr=cls.pg0.local_ip4, + dst_addr=cls.mcast_ip4, + src_port=port, + dst_port=port, + mcast_sw_if_index=1, + vni=vni) + if is_add: + r.add_vpp_config() + if r.sw_if_index == 0xffffffff: + raise ValueError("bad sw_if_index: ~0") + else: + r.remove_vpp_config() @classmethod - def add_shared_mcast_dst_load(cls): - cls.add_del_shared_mcast_dst_load(is_add=1) + def add_shared_mcast_dst_load(cls, port): + cls.add_del_shared_mcast_dst_load(port=port, is_add=1) @classmethod - def del_shared_mcast_dst_load(cls): - cls.add_del_shared_mcast_dst_load(is_add=0) + def del_shared_mcast_dst_load(cls, port): + cls.add_del_shared_mcast_dst_load(port=port, is_add=0) @classmethod - def add_del_mcast_tunnels_load(cls, is_add): + def add_del_mcast_tunnels_load(cls, port, is_add): """ add or del tunnels to test vxlan_gpe stability """ @@ -145,20 +153,25 @@ class TestVxlanGpe(BridgeDomain, VppTestCase): for dest_ip4 in ip4_range(cls.mcast_ip4, ip_range_start, ip_range_end): vni = int(dest_ip4.split(".")[3]) - cls.vapi.vxlan_gpe_add_del_tunnel( - src_addr=cls.pg0.local_ip4, - dst_addr=dest_ip4, - mcast_sw_if_index=1, - vni=vni, - is_add=is_add) + r = VppVxlanGpeTunnel(cls, + src_addr=cls.pg0.local_ip4, + dst_addr=dest_ip4, + src_port=port, + dst_port=port, + mcast_sw_if_index=1, + vni=vni) + if is_add: + r.add_vpp_config() + else: + r.remove_vpp_config() @classmethod - def add_mcast_tunnels_load(cls): - cls.add_del_mcast_tunnels_load(is_add=1) + def add_mcast_tunnels_load(cls, port): + cls.add_del_mcast_tunnels_load(port=port, is_add=1) @classmethod - def del_mcast_tunnels_load(cls): - cls.add_del_mcast_tunnels_load(is_add=0) + def del_mcast_tunnels_load(cls, port): + cls.add_del_mcast_tunnels_load(port=port, is_add=0) # Class method to start the VXLAN-GPE test case. # Overrides setUpClass method in VppTestCase class. @@ -170,7 +183,6 @@ class TestVxlanGpe(BridgeDomain, VppTestCase): super(TestVxlanGpe, cls).setUpClass() try: - cls.dport = 4790 cls.flags = 0x0c # Create 2 pg interfaces. @@ -187,55 +199,120 @@ class TestVxlanGpe(BridgeDomain, VppTestCase): # Our Multicast address cls.mcast_ip4 = '239.1.1.1' cls.mcast_mac = util.mcast_ip_to_mac(cls.mcast_ip4) - - # Create VXLAN-GPE VTEP on VPP pg0, and put vxlan_gpe_tunnel0 - # and pg1 into BD. - cls.single_tunnel_vni = 0xabcde - cls.single_tunnel_bd = 11 - r = cls.vapi.vxlan_gpe_add_del_tunnel( - src_addr=cls.pg0.local_ip4, - dst_addr=cls.pg0.remote_ip4, - vni=cls.single_tunnel_vni) - cls.vapi.sw_interface_set_l2_bridge(rx_sw_if_index=r.sw_if_index, - bd_id=cls.single_tunnel_bd) - cls.vapi.sw_interface_set_l2_bridge( - rx_sw_if_index=cls.pg1.sw_if_index, bd_id=cls.single_tunnel_bd) - - # Setup vni 2 to test multicast flooding - cls.n_ucast_tunnels = 10 - cls.mcast_flood_bd = 12 - cls.create_vxlan_gpe_flood_test_bd(cls.mcast_flood_bd, - cls.n_ucast_tunnels) - r = cls.vapi.vxlan_gpe_add_del_tunnel( - src_addr=cls.pg0.local_ip4, - dst_addr=cls.mcast_ip4, - mcast_sw_if_index=1, - vni=cls.mcast_flood_bd) - cls.vapi.sw_interface_set_l2_bridge(rx_sw_if_index=r.sw_if_index, - bd_id=cls.mcast_flood_bd) - cls.vapi.sw_interface_set_l2_bridge( - rx_sw_if_index=cls.pg2.sw_if_index, bd_id=cls.mcast_flood_bd) - - # Add and delete mcast tunnels to check stability - cls.add_shared_mcast_dst_load() - cls.add_mcast_tunnels_load() - cls.del_shared_mcast_dst_load() - cls.del_mcast_tunnels_load() - - # Setup vni 3 to test unicast flooding - cls.ucast_flood_bd = 13 - cls.create_vxlan_gpe_flood_test_bd(cls.ucast_flood_bd, - cls.n_ucast_tunnels) - cls.vapi.sw_interface_set_l2_bridge( - rx_sw_if_index=cls.pg3.sw_if_index, bd_id=cls.ucast_flood_bd) except Exception: - super(TestVxlanGpe, cls).tearDownClass() + cls.tearDownClass() raise @classmethod def tearDownClass(cls): super(TestVxlanGpe, cls).tearDownClass() + def setUp(self): + super(TestVxlanGpe, self).setUp() + + def createVxLANInterfaces(self, port=4790): + # Create VXLAN-GPE VTEP on VPP pg0, and put vxlan_gpe_tunnel0 + # and pg1 into BD. + self.dport = port + + self.single_tunnel_vni = 0xabcde + self.single_tunnel_bd = 11 + r = VppVxlanGpeTunnel(self, + src_addr=self.pg0.local_ip4, + dst_addr=self.pg0.remote_ip4, + src_port=port, + dst_port=port, + 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) + + # Setup vni 2 to test multicast flooding + self.n_ucast_tunnels = 10 + self.mcast_flood_bd = 12 + self.create_vxlan_gpe_flood_test_bd(self.mcast_flood_bd, + self.n_ucast_tunnels, + self.dport) + r = VppVxlanGpeTunnel(self, + src_addr=self.pg0.local_ip4, + dst_addr=self.mcast_ip4, + src_port=port, + dst_port=port, + 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) + + # Add and delete mcast tunnels to check stability + self.add_shared_mcast_dst_load(self.dport) + self.add_mcast_tunnels_load(self.dport) + self.del_shared_mcast_dst_load(self.dport) + self.del_mcast_tunnels_load(self.dport) + + # Setup vni 3 to test unicast flooding + self.ucast_flood_bd = 13 + self.create_vxlan_gpe_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) + + # Set scapy listen custom port for VxLAN + bind_layers(UDP, VXLAN, dport=self.dport) + + """ + Tests with default port (4790) + """ + def test_decap(self): + """ Decapsulation test + from BridgeDoman + """ + self.createVxLANInterfaces() + super(TestVxlanGpe, self).test_decap() + + def test_encap(self): + """ Encapsulation test + from BridgeDoman + """ + self.createVxLANInterfaces() + super(TestVxlanGpe, self).test_encap() + + def test_ucast_flood(self): + """ Unicast flood test + from BridgeDoman + """ + self.createVxLANInterfaces() + super(TestVxlanGpe, self).test_ucast_flood() + + """ + Tests with custom port (1112) + """ + def test_decap_custom_port(self): + """ Decapsulation test custom port + from BridgeDoman + """ + self.createVxLANInterfaces(1112) + super(TestVxlanGpe, self).test_decap() + + def test_encap_custom_port(self): + """ Encapsulation test custom port + from BridgeDoman + """ + self.createVxLANInterfaces(1112) + super(TestVxlanGpe, self).test_encap() + + def test_ucast_flood_custom_port(self): + """ Unicast flood test custom port + from BridgeDoman + """ + self.createVxLANInterfaces(1112) + super(TestVxlanGpe, self).test_ucast_flood() + @unittest.skip("test disabled for vxlan-gpe") def test_mcast_flood(self): """ inherited from BridgeDomain """ |