src/vnet/vxlan/test/vpp_vxlan_tunnel.py

   1 from vpp_interface import VppInterface
   2 from vpp_papi import VppEnum
   3
   4
   5 INDEX_INVALID = 0xffffffff
   6 DEFAULT_PORT = 4789
   7 UNDEFINED_PORT = 0
   8
   9
  10 def find_vxlan_tunnel(test, src, dst, s_port, d_port, vni):
  11     ts = test.vapi.vxlan_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.src_address) and \
  23            dst == str(t.dst_address) and \
  24            src_port == t.src_port and \
  25            dst_port == t.dst_port and \
  26            t.vni == vni:
  27             return t.sw_if_index
  28     return INDEX_INVALID
  29
  30
  31 class VppVxlanTunnel(VppInterface):
  32     """
  33     VPP VXLAN interface
  34     """
  35
  36     def __init__(self, test, src, dst, vni,
  37                  src_port=UNDEFINED_PORT, dst_port=UNDEFINED_PORT,
  38                  mcast_itf=None,
  39                  mcast_sw_if_index=INDEX_INVALID,
  40                  decap_next_index=INDEX_INVALID,
  41                  encap_vrf_id=None, instance=0xffffffff, is_l3=False):
  42         """ Create VXLAN Tunnel interface """
  43         super(VppVxlanTunnel, self).__init__(test)
  44         self.src = src
  45         self.dst = dst
  46         self.vni = vni
  47         self.src_port = src_port
  48         self.dst_port = dst_port
  49         self.mcast_itf = mcast_itf
  50         self.mcast_sw_if_index = mcast_sw_if_index
  51         self.encap_vrf_id = encap_vrf_id
  52         self.decap_next_index = decap_next_index
  53         self.instance = instance
  54         self.is_l3 = is_l3
  55
  56         if (self.mcast_itf):
  57             self.mcast_sw_if_index = self.mcast_itf.sw_if_index
  58
  59     def add_vpp_config(self):
  60         reply = self.test.vapi.vxlan_add_del_tunnel_v3(
  61             is_add=1, src_address=self.src, dst_address=self.dst, vni=self.vni,
  62             src_port=self.src_port, dst_port=self.dst_port,
  63             mcast_sw_if_index=self.mcast_sw_if_index,
  64             encap_vrf_id=self.encap_vrf_id, is_l3=self.is_l3,
  65             instance=self.instance, decap_next_index=self.decap_next_index)
  66         self.set_sw_if_index(reply.sw_if_index)
  67         self._test.registry.register(self, self._test.logger)
  68
  69     def remove_vpp_config(self):
  70         self.test.vapi.vxlan_add_del_tunnel_v2(
  71             is_add=0, src_address=self.src, dst_address=self.dst, vni=self.vni,
  72             src_port=self.src_port, dst_port=self.dst_port,
  73             mcast_sw_if_index=self.mcast_sw_if_index,
  74             encap_vrf_id=self.encap_vrf_id, instance=self.instance,
  75             decap_next_index=self.decap_next_index)
  76
  77     def query_vpp_config(self):
  78         return (INDEX_INVALID != find_vxlan_tunnel(self._test,
  79                                                    self.src,
  80                                                    self.dst,
  81                                                    self.src_port,
  82                                                    self.dst_port,
  83                                                    self.vni))
  84
  85     def object_id(self):
  86         return "vxlan-%d-%d-%s-%s" % (self.sw_if_index, self.vni,
  87                                       self.src, self.dst)