test/vpp_mpls_tunnel_interface.py

   1
   2 from vpp_interface import VppInterface
   3
   4
   5 class VppMPLSTunnelInterface(VppInterface):
   6     """
   7     VPP MPLS Tunnel interface
   8     """
   9
  10     def __init__(self, test, paths, is_multicast=0, is_l2=0):
  11         """ Create MPLS Tunnel interface """
  12         super(VppMPLSTunnelInterface, self).__init__(test)
  13         self.t_paths = paths
  14         self.is_multicast = is_multicast
  15         self.is_l2 = is_l2
  16
  17     def add_vpp_config(self):
  18         sw_if_index = 0xffffffff
  19         for path in self.t_paths:
  20             lstack = path.encode_labels()
  21
  22             reply = self.test.vapi.mpls_tunnel_add_del(
  23                 sw_if_index,
  24                 1,  # IPv4 next-hop
  25                 path.nh_addr,
  26                 path.nh_itf,
  27                 path.nh_table_id,
  28                 path.weight,
  29                 next_hop_via_label=path.nh_via_label,
  30                 next_hop_out_label_stack=lstack,
  31                 next_hop_n_out_labels=len(lstack),
  32                 is_multicast=self.is_multicast,
  33                 l2_only=self.is_l2)
  34             sw_if_index = reply.sw_if_index
  35             self.tunnel_index = reply.tunnel_index
  36         self.set_sw_if_index(sw_if_index)
  37         self._test.registry.register(self, self._test.logger)
  38
  39     def remove_vpp_config(self):
  40         for path in self.t_paths:
  41             lstack = path.encode_labels()
  42
  43             self.test.vapi.mpls_tunnel_add_del(
  44                 self.sw_if_index,
  45                 1,  # IPv4 next-hop
  46                 path.nh_addr,
  47                 path.nh_itf,
  48                 path.nh_table_id,
  49                 path.weight,
  50                 next_hop_via_label=path.nh_via_label,
  51                 next_hop_out_label_stack=lstack,
  52                 next_hop_n_out_labels=len(lstack),
  53                 is_add=0)
  54
  55     def query_vpp_config(self):
  56         dump = self._test.vapi.mpls_tunnel_dump()
  57         for t in dump:
  58             if self.sw_if_index == t.mt_sw_if_index and \
  59                self.tunnel_index == t.mt_tunnel_index:
  60                 return True
  61         return False
  62
  63     def __str__(self):
  64         return self.object_id()
  65
  66     def object_id(self):
  67         return ("mpls-tunnel%d-%d" % (self.tunnel_index,
  68                                       self.sw_if_index))