diff options
Diffstat (limited to 'test/test_vxlan_gpe.py')
| -rw-r--r-- | test/test_vxlan_gpe.py | 185 |
1 files changed, 103 insertions, 82 deletions
diff --git a/test/test_vxlan_gpe.py b/test/test_vxlan_gpe.py index 828b5fc30cd..f432becfce3 100644 --- a/test/test_vxlan_gpe.py +++ b/test/test_vxlan_gpe.py @@ -20,7 +20,7 @@ from vpp_ip import INVALID_INDEX @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) @@ -31,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): @@ -46,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. @@ -92,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): @@ -121,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() @@ -151,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: @@ -184,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)) @@ -198,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() @@ -216,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) @@ -257,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) @@ -269,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() @@ -293,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) @@ -316,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. @@ -339,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) |