test/vpp_vxlan_gbp_tunnel.py

   1
   2 from vpp_interface import VppInterface
   3 from vpp_ip import VppIpAddress
   4 from vpp_papi import VppEnum
   5
   6
   7 INDEX_INVALID = 0xffffffff
   8
   9
  10 def find_vxlan_gbp_tunnel(test, src, dst, vni):
  11     vsrc = VppIpAddress(src)
  12     vdst = VppIpAddress(dst)
  13
  14     ts = test.vapi.vxlan_gbp_tunnel_dump(INDEX_INVALID)
  15     for t in ts:
  16         if vsrc == t.tunnel.src and \
  17            vdst == t.tunnel.dst and \
  18            t.tunnel.vni == vni:
  19             return t.tunnel.sw_if_index
  20     return INDEX_INVALID
  21
  22
  23 class VppVxlanGbpTunnel(VppInterface):
  24     """
  25     VPP VXLAN GBP interface
  26     """
  27
  28     def __init__(self, test, src, dst, vni, mcast_itf=None, mode=None):
  29         """ Create VXLAN-GBP Tunnel interface """
  30         super(VppVxlanGbpTunnel, self).__init__(test)
  31         self.src = VppIpAddress(src)
  32         self.dst = VppIpAddress(dst)
  33         self.vni = vni
  34         self.mcast_itf = mcast_itf
  35         if not mode:
  36             self.mode = (VppEnum.vl_api_vxlan_gbp_api_tunnel_mode_t.
  37                          VXLAN_GBP_API_TUNNEL_MODE_L2)
  38         else:
  39             self.mode = mode
  40
  41     def add_vpp_config(self):
  42         mcast_sw_if_index = INDEX_INVALID
  43         if (self.mcast_itf):
  44             mcast_sw_if_index = self.mcast_itf.sw_if_index
  45         reply = self.test.vapi.vxlan_gbp_tunnel_add_del(
  46             self.src.encode(),
  47             self.dst.encode(),
  48             mode=self.mode,
  49             vni=self.vni,
  50             mcast_sw_if_index=mcast_sw_if_index)
  51         self.set_sw_if_index(reply.sw_if_index)
  52         self._test.registry.register(self, self._test.logger)
  53
  54     def remove_vpp_config(self):
  55         mcast_sw_if_index = INDEX_INVALID
  56         if (self.mcast_itf):
  57             mcast_sw_if_index = self.mcast_itf.sw_if_index
  58         self.test.vapi.vxlan_gbp_tunnel_add_del(
  59             self.src.encode(),
  60             self.dst.encode(),
  61             vni=self.vni,
  62             is_add=0,
  63             mcast_sw_if_index=mcast_sw_if_index)
  64
  65     def query_vpp_config(self):
  66         return (INDEX_INVALID != find_vxlan_gbp_tunnel(self._test,
  67                                                        self.src,
  68                                                        self.dst,
  69                                                        self.vni))
  70
  71     def __str__(self):
  72         return self.object_id()
  73
  74     def object_id(self):
  75         return "vxlan-gbp-%d-%d-%s-%s" % (self.sw_if_index, self.vni,
  76                                           self.src, self.dst)