1 from vpp_interface import VppInterface
2 from vpp_papi import VppEnum
5 INDEX_INVALID = 0xffffffff
10 def find_vxlan_gpe_tunnel(test, src, dst, s_port, d_port, vni):
11 ts = test.vapi.vxlan_gpe_tunnel_v2_dump(INDEX_INVALID)
13 src_port = DEFAULT_PORT
14 if s_port != UNDEFINED_PORT:
17 dst_port = DEFAULT_PORT
18 if d_port != UNDEFINED_PORT:
22 if src == str(t.local) and \
23 dst == str(t.remote) and \
24 src_port == t.local_port and \
25 dst_port == t.remote_port and \
31 class VppVxlanGpeTunnel(VppInterface):
33 VPP VXLAN GPE interface
36 def __init__(self, test, src_addr, dst_addr, vni,
37 src_port=UNDEFINED_PORT, dst_port=UNDEFINED_PORT,
38 mcast_sw_if_index=INDEX_INVALID,
40 decap_vrf_id=None, protocol=3):
41 """ Create VXLAN GPE Tunnel interface """
42 super(VppVxlanGpeTunnel, self).__init__(test)
46 self.src_port = src_port
47 self.dst_port = dst_port
48 self.mcast_sw_if_index = mcast_sw_if_index
49 self.encap_vrf_id = encap_vrf_id
50 self.decap_vrf_id = decap_vrf_id
53 def add_vpp_config(self):
54 reply = self.test.vapi.vxlan_gpe_add_del_tunnel_v2(
55 is_add=1, local=self.src, remote=self.dst, vni=self.vni,
56 local_port=self.src_port, remote_port=self.dst_port,
57 mcast_sw_if_index=self.mcast_sw_if_index,
58 encap_vrf_id=self.encap_vrf_id,
59 decap_vrf_id=self.decap_vrf_id,
60 protocol=self.protocol)
61 self.set_sw_if_index(reply.sw_if_index)
62 self._test.registry.register(self, self._test.logger)
64 def remove_vpp_config(self):
65 self.test.vapi.vxlan_gpe_add_del_tunnel_v2(
66 is_add=0, local=self.src, remote=self.dst, vni=self.vni,
67 local_port=self.src_port, remote_port=self.dst_port,
68 mcast_sw_if_index=self.mcast_sw_if_index,
69 encap_vrf_id=self.encap_vrf_id,
70 decap_vrf_id=self.decap_vrf_id,
71 protocol=self.protocol)
73 def query_vpp_config(self):
74 return (INDEX_INVALID != find_vxlan_gpe_tunnel(self._test,
82 return "vxlan-%d-%d-%s-%s" % (self.sw_if_index, self.vni,