summaryrefslogtreecommitdiffstats
path: root/extras/deprecated/vnet
diff options
context:
space:
mode:
Diffstat (limited to 'extras/deprecated/vnet')
-rw-r--r--extras/deprecated/vnet/vxlan-gbp/test_vxlan_gbp.py143
-rw-r--r--extras/deprecated/vnet/vxlan-gbp/vpp_vxlan_gbp_tunnel.py64
2 files changed, 117 insertions, 90 deletions
diff --git a/extras/deprecated/vnet/vxlan-gbp/test_vxlan_gbp.py b/extras/deprecated/vnet/vxlan-gbp/test_vxlan_gbp.py
index f332aced7d8..1d64937b6dd 100644
--- a/extras/deprecated/vnet/vxlan-gbp/test_vxlan_gbp.py
+++ b/extras/deprecated/vnet/vxlan-gbp/test_vxlan_gbp.py
@@ -16,37 +16,43 @@ from vpp_ip import INVALID_INDEX
class TestVxlanGbp(VppTestCase):
- """ VXLAN GBP Test Case """
+ """VXLAN GBP Test Case"""
@property
def frame_request(self):
- """ Ethernet frame modeling a generic request """
- return (Ether(src='00:00:00:00:00:01', dst='00:00:00:00:00:02') /
- IP(src='1.2.3.4', dst='4.3.2.1') /
- UDP(sport=10000, dport=20000) /
- Raw(b'\xa5' * 100))
+ """Ethernet frame modeling a generic request"""
+ return (
+ Ether(src="00:00:00:00:00:01", dst="00:00:00:00:00:02")
+ / IP(src="1.2.3.4", dst="4.3.2.1")
+ / UDP(sport=10000, dport=20000)
+ / Raw(b"\xa5" * 100)
+ )
@property
def frame_reply(self):
- """ Ethernet frame modeling a generic reply """
- return (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' * 100))
+ """Ethernet frame modeling a generic reply"""
+ return (
+ 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" * 100)
+ )
def encapsulate(self, pkt, vni):
"""
Encapsulate the original payload frame by adding VXLAN GBP 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, gpflags=self.gpflags,
- gpid=self.sclass) / 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, gpflags=self.gpflags, gpid=self.sclass)
+ / 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 decapsulate(self, pkt):
@@ -54,7 +60,7 @@ class TestVxlanGbp(VppTestCase):
Decapsulate the original payload frame by removing VXLAN header
"""
# check if is set G and I flag
- self.assertEqual(pkt[VXLAN].flags, int('0x88', 16))
+ self.assertEqual(pkt[VXLAN].flags, int("0x88", 16))
return pkt[VXLAN].payload
# Method for checking VXLAN GBP encapsulation.
@@ -94,28 +100,28 @@ class TestVxlanGbp(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(cls.pg0.remote_ip4,
- ip_range_start,
- ip_range_end):
+ for dest_ip4 in ip4_range(cls.pg0.remote_ip4, 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 = cls.vapi.vxlan_gbp_tunnel_add_del(
tunnel={
- 'src': cls.pg0.local_ip4,
- 'dst': dest_ip4,
- 'vni': vni,
- 'instance': INVALID_INDEX,
- 'mcast_sw_if_index': INVALID_INDEX,
- 'mode': 1,
+ "src": cls.pg0.local_ip4,
+ "dst": dest_ip4,
+ "vni": vni,
+ "instance": INVALID_INDEX,
+ "mcast_sw_if_index": INVALID_INDEX,
+ "mode": 1,
},
- is_add=1
+ is_add=1,
)
- 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)
# Class method to start the VXLAN GBP test case.
# Overrides setUpClass method in VppTestCase class.
@@ -146,33 +152,33 @@ class TestVxlanGbp(VppTestCase):
# Create VXLAN GBP VTEP on VPP pg0, and put vxlan_gbp_tunnel0 and
# pg1 into BD.
cls.single_tunnel_bd = 1
- cls.single_tunnel_vni = 0xabcde
+ cls.single_tunnel_vni = 0xABCDE
r = cls.vapi.vxlan_gbp_tunnel_add_del(
tunnel={
- 'src': cls.pg0.local_ip4,
- 'dst': cls.pg0.remote_ip4,
- 'vni': cls.single_tunnel_vni,
- 'instance': INVALID_INDEX,
- 'mcast_sw_if_index': INVALID_INDEX,
- 'mode': 1,
+ "src": cls.pg0.local_ip4,
+ "dst": cls.pg0.remote_ip4,
+ "vni": cls.single_tunnel_vni,
+ "instance": INVALID_INDEX,
+ "mcast_sw_if_index": INVALID_INDEX,
+ "mode": 1,
},
- is_add=1
+ is_add=1,
+ )
+ 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=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)
+ 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 = 2
# Setup vni 3 to test unicast flooding
cls.ucast_flood_bd = 3
- cls.create_vxlan_gbp_flood_test_bd(cls.ucast_flood_bd,
- cls.n_ucast_tunnels)
+ cls.create_vxlan_gbp_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)
+ rx_sw_if_index=cls.pg3.sw_if_index, bd_id=cls.ucast_flood_bd
+ )
except Exception:
super(TestVxlanGbp, cls).tearDownClass()
raise
@@ -182,7 +188,7 @@ class TestVxlanGbp(VppTestCase):
super(TestVxlanGbp, cls).tearDownClass()
def assert_eq_pkts(self, pkt1, pkt2):
- """ Verify the Ether, IP, UDP, payload are equal in both
+ """Verify the Ether, IP, UDP, payload are equal in both
packets
"""
self.assertEqual(pkt1[Ether].src, pkt2[Ether].src)
@@ -194,14 +200,17 @@ class TestVxlanGbp(VppTestCase):
self.assertEqual(pkt1[Raw], pkt2[Raw])
def test_decap(self):
- """ Decapsulation test
+ """Decapsulation test
Send encapsulated frames from pg0
Verify receipt of decapsulated frames on pg1
"""
- encapsulated_pkt = self.encapsulate(self.frame_request,
- self.single_tunnel_vni)
+ encapsulated_pkt = self.encapsulate(self.frame_request, self.single_tunnel_vni)
- self.pg0.add_stream([encapsulated_pkt, ])
+ self.pg0.add_stream(
+ [
+ encapsulated_pkt,
+ ]
+ )
self.pg1.enable_capture()
@@ -214,7 +223,7 @@ class TestVxlanGbp(VppTestCase):
self.assert_eq_pkts(pkt, self.frame_request)
def test_encap(self):
- """ Encapsulation test
+ """Encapsulation test
Send frames from pg1
Verify receipt of encapsulated frames on pg0
"""
@@ -233,7 +242,7 @@ class TestVxlanGbp(VppTestCase):
self.assert_eq_pkts(payload, self.frame_reply)
def test_ucast_flood(self):
- """ Unicast flood test
+ """Unicast flood test
Send frames from pg3
Verify receipt of encapsulated frames on pg0
"""
@@ -251,16 +260,18 @@ class TestVxlanGbp(VppTestCase):
self.assert_eq_pkts(payload, self.frame_reply)
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.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])
@@ -276,9 +287,9 @@ class TestVxlanGbp(VppTestCase):
payload = self.decapsulate(pkt)
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.
+ # Method to define VPP actions before tear down of the test case.
+ # Overrides tearDown method in VppTestCase class.
+ # @param self The object pointer.
def tearDown(self):
super(TestVxlanGbp, self).tearDown()
@@ -289,5 +300,5 @@ class TestVxlanGbp(VppTestCase):
self.logger.info(self.vapi.cli("show error"))
-if __name__ == '__main__':
+if __name__ == "__main__":
unittest.main(testRunner=VppTestRunner)
diff --git a/extras/deprecated/vnet/vxlan-gbp/vpp_vxlan_gbp_tunnel.py b/extras/deprecated/vnet/vxlan-gbp/vpp_vxlan_gbp_tunnel.py
index 0898bd9f810..1b6b4e731ca 100644
--- a/extras/deprecated/vnet/vxlan-gbp/vpp_vxlan_gbp_tunnel.py
+++ b/extras/deprecated/vnet/vxlan-gbp/vpp_vxlan_gbp_tunnel.py
@@ -1,17 +1,18 @@
-
from vpp_interface import VppInterface
from vpp_papi import VppEnum
-INDEX_INVALID = 0xffffffff
+INDEX_INVALID = 0xFFFFFFFF
def find_vxlan_gbp_tunnel(test, src, dst, vni):
ts = test.vapi.vxlan_gbp_tunnel_dump(INDEX_INVALID)
for t in ts:
- if src == str(t.tunnel.src) and \
- dst == str(t.tunnel.dst) and \
- t.tunnel.vni == vni:
+ if (
+ src == str(t.tunnel.src)
+ and dst == str(t.tunnel.dst)
+ and t.tunnel.vni == vni
+ ):
return t.tunnel.sw_if_index
return INDEX_INVALID
@@ -21,9 +22,19 @@ class VppVxlanGbpTunnel(VppInterface):
VPP VXLAN GBP interface
"""
- def __init__(self, test, src, dst, vni, mcast_itf=None, mode=None,
- is_ipv6=None, encap_table_id=None, instance=0xffffffff):
- """ Create VXLAN-GBP Tunnel interface """
+ def __init__(
+ self,
+ test,
+ src,
+ dst,
+ vni,
+ mcast_itf=None,
+ mode=None,
+ is_ipv6=None,
+ encap_table_id=None,
+ instance=0xFFFFFFFF,
+ ):
+ """Create VXLAN-GBP Tunnel interface"""
super(VppVxlanGbpTunnel, self).__init__(test)
self.src = src
self.dst = dst
@@ -33,21 +44,23 @@ class VppVxlanGbpTunnel(VppInterface):
self.encap_table_id = encap_table_id
self.instance = instance
if not mode:
- self.mode = (VppEnum.vl_api_vxlan_gbp_api_tunnel_mode_t.
- VXLAN_GBP_API_TUNNEL_MODE_L2)
+ self.mode = (
+ VppEnum.vl_api_vxlan_gbp_api_tunnel_mode_t.VXLAN_GBP_API_TUNNEL_MODE_L2
+ )
else:
self.mode = mode
def encode(self):
return {
- 'src': self.src,
- 'dst': self.dst,
- 'mode': self.mode,
- 'vni': self.vni,
- 'mcast_sw_if_index': self.mcast_itf.sw_if_index
- if self.mcast_itf else INDEX_INVALID,
- 'encap_table_id': self.encap_table_id,
- 'instance': self.instance,
+ "src": self.src,
+ "dst": self.dst,
+ "mode": self.mode,
+ "vni": self.vni,
+ "mcast_sw_if_index": self.mcast_itf.sw_if_index
+ if self.mcast_itf
+ else INDEX_INVALID,
+ "encap_table_id": self.encap_table_id,
+ "instance": self.instance,
}
def add_vpp_config(self):
@@ -65,11 +78,14 @@ class VppVxlanGbpTunnel(VppInterface):
)
def query_vpp_config(self):
- return (INDEX_INVALID != find_vxlan_gbp_tunnel(self._test,
- self.src,
- self.dst,
- self.vni))
+ return INDEX_INVALID != find_vxlan_gbp_tunnel(
+ self._test, self.src, self.dst, self.vni
+ )
def object_id(self):
- return "vxlan-gbp-%d-%d-%s-%s" % (self.sw_if_index, self.vni,
- self.src, self.dst)
+ return "vxlan-gbp-%d-%d-%s-%s" % (
+ self.sw_if_index,
+ self.vni,
+ self.src,
+ self.dst,
+ )