test/vpp_vxlan_gpe_tunnel.py

   1 from vpp_interface import VppInterface
   2 from vpp_papi import VppEnum
   3
   4
   5 INDEX_INVALID = 0xffffffff
   6 DEFAULT_PORT = 4790
   7 UNDEFINED_PORT = 0
   8
   9
  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)
  12
  13     src_port = DEFAULT_PORT
  14     if s_port != UNDEFINED_PORT:
  15         src_port = s_port
  16
  17     dst_port = DEFAULT_PORT
  18     if d_port != UNDEFINED_PORT:
  19         dst_port = d_port
  20
  21     for t in ts:
  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 \
  26            t.vni == vni:
  27             return t.sw_if_index
  28     return INDEX_INVALID
  29
  30
  31 class VppVxlanGpeTunnel(VppInterface):
  32     """
  33     VPP VXLAN GPE interface
  34     """
  35
  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,
  39                  encap_vrf_id=None,
  40                  decap_vrf_id=None, protocol=3):
  41         """ Create VXLAN GPE Tunnel interface """
  42         super(VppVxlanGpeTunnel, self).__init__(test)
  43         self.src = src_addr
  44         self.dst = dst_addr
  45         self.vni = vni
  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
  51         self.protocol = 3
  52
  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)
  63
  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)
  72
  73     def query_vpp_config(self):
  74         return (INDEX_INVALID != find_vxlan_gpe_tunnel(self._test,
  75                                                        self.src,
  76                                                        self.dst,
  77                                                        self.src_port,
  78                                                        self.dst_port,
  79                                                        self.vni))
  80
  81     def object_id(self):
  82         return "vxlan-%d-%d-%s-%s" % (self.sw_if_index, self.vni,
  83                                       self.src, self.dst)