test/vpp_vxlan_gbp_tunnel.py

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