X-Git-Url: https://gerrit.fd.io/r/gitweb?a=blobdiff_plain;ds=sidebyside;f=test%2Ftest_vxlan_gpe.py;h=11f17f59f0e23bacfe6dd8c1f6f7bac4b1169923;hb=853cc9f2ad3ee52cbdd891fb09d51c25678baed0;hp=28c31ccd3bb131de831a5bf71dede458c6540b9a;hpb=ea96292985760b7fc3af54669e3eab35e7561a10;p=vpp.git diff --git a/test/test_vxlan_gpe.py b/test/test_vxlan_gpe.py index 28c31ccd3bb..11f17f59f0e 100644 --- a/test/test_vxlan_gpe.py +++ b/test/test_vxlan_gpe.py @@ -1,13 +1,14 @@ #!/usr/bin/env python3 -import socket from util import ip4_range import unittest -from framework import VppTestCase, VppTestRunner, running_extended_tests +from config import config +from framework import VppTestCase +from asfframework import VppTestRunner from template_bd import BridgeDomain from scapy.layers.l2 import Ether -from scapy.packet import Raw, bind_layers +from scapy.packet import bind_layers from scapy.layers.inet import IP, UDP from scapy.layers.vxlan import VXLAN @@ -17,9 +18,9 @@ from vpp_vxlan_gpe_tunnel import VppVxlanGpeTunnel from vpp_ip import INVALID_INDEX -@unittest.skipUnless(running_extended_tests, "part of extended tests") +@unittest.skipUnless(config.extended, "part of extended tests") class TestVxlanGpe(BridgeDomain, VppTestCase): - """ VXLAN-GPE Test Case """ + """VXLAN-GPE Test Case""" def __init__(self, *args): BridgeDomain.__init__(self) @@ -30,14 +31,16 @@ class TestVxlanGpe(BridgeDomain, VppTestCase): Encapsulate the original payload frame by adding VXLAN-GPE 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): @@ -45,18 +48,20 @@ class TestVxlanGpe(BridgeDomain, VppTestCase): Encapsulate the original payload frame by adding VXLAN-GPE 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-GPE header """ # check if is set I and P flag - self.assertEqual(pkt[VXLAN].flags, 0x0c) + self.assertEqual(pkt[VXLAN].flags, 0x0C) return pkt[VXLAN].payload # Method for checking VXLAN-GPE encapsulation. @@ -91,24 +96,27 @@ class TestVxlanGpe(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_ip4n 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 = VppVxlanGpeTunnel(cls, - src_addr=cls.pg0.local_ip4, - dst_addr=dest_ip4, - src_port=port, - dst_port=port, - 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) + 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, port, is_add): @@ -120,16 +128,18 @@ class TestVxlanGpe(BridgeDomain, VppTestCase): vni_start = 1000 vni_end = vni_start + n_shared_dst_tunnels for vni in range(vni_start, vni_end): - 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) + 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: + if r.sw_if_index == 0xFFFFFFFF: raise ValueError("bad sw_if_index: ~0") else: r.remove_vpp_config() @@ -150,16 +160,17 @@ class TestVxlanGpe(BridgeDomain, VppTestCase): n_distinct_dst_tunnels = 20 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 = int(dest_ip4.split(".")[3]) - 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) + 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: @@ -183,7 +194,7 @@ class TestVxlanGpe(BridgeDomain, VppTestCase): super(TestVxlanGpe, cls).setUpClass() try: - cls.flags = 0x0c + cls.flags = 0x0C # Create 2 pg interfaces. cls.create_pg_interfaces(range(4)) @@ -197,7 +208,7 @@ class TestVxlanGpe(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() @@ -215,38 +226,46 @@ class TestVxlanGpe(BridgeDomain, VppTestCase): # and pg1 into BD. self.dport = port - self.single_tunnel_vni = 0xabcde + 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 = 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) + 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) + 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) + 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) @@ -256,11 +275,12 @@ class TestVxlanGpe(BridgeDomain, VppTestCase): # 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.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) + 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) @@ -268,22 +288,23 @@ class TestVxlanGpe(BridgeDomain, VppTestCase): """ Tests with default port (4790) """ + def test_decap(self): - """ Decapsulation test + """Decapsulation test from BridgeDoman """ self.createVxLANInterfaces() super(TestVxlanGpe, self).test_decap() def test_encap(self): - """ Encapsulation test + """Encapsulation test from BridgeDoman """ self.createVxLANInterfaces() super(TestVxlanGpe, self).test_encap() def test_ucast_flood(self): - """ Unicast flood test + """Unicast flood test from BridgeDoman """ self.createVxLANInterfaces() @@ -292,22 +313,23 @@ class TestVxlanGpe(BridgeDomain, VppTestCase): """ Tests with custom port (1112) """ + def test_decap_custom_port(self): - """ Decapsulation test custom port + """Decapsulation test custom port from BridgeDoman """ self.createVxLANInterfaces(1112) super(TestVxlanGpe, self).test_decap() def test_encap_custom_port(self): - """ Encapsulation test custom port + """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 + """Unicast flood test custom port from BridgeDoman """ self.createVxLANInterfaces(1112) @@ -315,12 +337,12 @@ class TestVxlanGpe(BridgeDomain, VppTestCase): @unittest.skip("test disabled for vxlan-gpe") def test_mcast_flood(self): - """ inherited from BridgeDomain """ + """inherited from BridgeDomain""" pass @unittest.skip("test disabled for vxlan-gpe") def test_mcast_rcv(self): - """ inherited from BridgeDomain """ + """inherited from BridgeDomain""" pass # Method to define VPP actions before tear down of the test case. @@ -338,5 +360,5 @@ class TestVxlanGpe(BridgeDomain, VppTestCase): self.logger.info(self.vapi.cli("show trace")) -if __name__ == '__main__': +if __name__ == "__main__": unittest.main(testRunner=VppTestRunner)