| from vpp_interface import VppInterface |
| |
| |
| class VppMPLSTunnelInterface(VppInterface): |
| """ |
| VPP MPLS Tunnel interface |
| """ |
| |
| def __init__(self, test, paths, is_multicast=0, is_l2=0): |
| """Create MPLS Tunnel interface""" |
| super(VppMPLSTunnelInterface, self).__init__(test) |
| self.t_paths = paths |
| self.is_multicast = is_multicast |
| self.is_l2 = is_l2 |
| self.encoded_paths = [] |
| for path in self.t_paths: |
| self.encoded_paths.append(path.encode()) |
| |
| def add_vpp_config(self): |
| reply = self.test.vapi.mpls_tunnel_add_del( |
| 0xFFFFFFFF, |
| self.encoded_paths, |
| is_multicast=self.is_multicast, |
| l2_only=self.is_l2, |
| ) |
| self.set_sw_if_index(reply.sw_if_index) |
| self.tunnel_index = reply.tunnel_index |
| self._test.registry.register(self, self._test.logger) |
| |
| def remove_vpp_config(self): |
| reply = self.test.vapi.mpls_tunnel_add_del( |
| self.sw_if_index, self.encoded_paths, is_add=0 |
| ) |
| |
| def query_vpp_config(self): |
| dump = self._test.vapi.mpls_tunnel_dump() |
| for t in dump: |
| if ( |
| self.sw_if_index == t.mt_tunnel.mt_sw_if_index |
| and self.tunnel_index == t.mt_tunnel.mt_tunnel_index |
| ): |
| return True |
| return False |
| |
| def object_id(self): |
| return "mpls-tunnel%d-%d" % (self.tunnel_index, self.sw_if_index) |