from vpp_object import *
from socket import inet_pton, inet_ntop, AF_INET, AF_INET6
+from vpp_ip import *
# from vnet/vnet/mpls/mpls_types.h
MPLS_IETF_MAX_LABEL = 0xfffff
MFIB_ENTRY_FLAG_INHERIT_ACCEPT = 8
-class DpoProto:
- DPO_PROTO_IP4 = 0
- DPO_PROTO_IP6 = 1
- DPO_PROTO_MPLS = 2
- DPO_PROTO_ETHERNET = 3
- DPO_PROTO_BIER = 4
- DPO_PROTO_NSH = 5
-
-
class MplsLspMode:
PIPE = 0
UNIFORM = 1
return False
+def find_mroute(test, grp_addr, src_addr, grp_addr_len,
+ table_id=0, inet=AF_INET):
+ if inet == AF_INET:
+ s = 4
+ routes = test.vapi.ip_mfib_dump()
+ else:
+ s = 16
+ routes = test.vapi.ip6_mfib_dump()
+ gaddr = inet_pton(inet, grp_addr)
+ saddr = inet_pton(inet, src_addr)
+ for e in routes:
+ if gaddr == e.grp_address[:s] \
+ and grp_addr_len == e.address_length \
+ and saddr == e.src_address[:s] \
+ and table_id == e.table_id:
+ return True
+ return False
+
+
class VppIpTable(VppObject):
def __init__(self,
'n_labels': len(self.nh_labels),
'label_stack': self.encode_labels()}
+ def __eq__(self, other):
+ return self.nh_addr == other.nh_addr
+
class VppMRoutePath(VppRoutePath):
def __init__(self, test, dest_addr,
dest_addr_len, paths, table_id=0, is_ip6=0, is_local=0,
- is_unreach=0, is_prohibit=0):
+ is_unreach=0, is_prohibit=0, is_drop=0):
self._test = test
self.paths = paths
self.dest_addr_len = dest_addr_len
self.is_local = is_local
self.is_unreach = is_unreach
self.is_prohibit = is_prohibit
+ self.is_drop = is_drop
self.dest_addr_p = dest_addr
if is_ip6:
self.dest_addr = inet_pton(AF_INET6, dest_addr)
self.is_prohibit = is_prohibit
def add_vpp_config(self):
- if self.is_local or self.is_unreach or self.is_prohibit:
- self._test.vapi.ip_add_del_route(
+ if self.is_local or self.is_unreach or \
+ self.is_prohibit or self.is_drop:
+ r = self._test.vapi.ip_add_del_route(
self.dest_addr,
self.dest_addr_len,
inet_pton(AF_INET6, "::"),
is_local=self.is_local,
is_unreach=self.is_unreach,
is_prohibit=self.is_prohibit,
+ is_drop=self.is_drop,
table_id=self.table_id,
is_ipv6=self.is_ip6)
else:
for path in self.paths:
lstack = path.encode_labels()
- self._test.vapi.ip_add_del_route(
+ r = self._test.vapi.ip_add_del_route(
self.dest_addr,
self.dest_addr_len,
path.nh_addr,
is_source_lookup=path.is_source_lookup,
is_udp_encap=path.is_udp_encap,
is_multipath=1 if len(self.paths) > 1 else 0)
+ self.stats_index = r.stats_index
self._test.registry.register(self, self._test.logger)
def remove_vpp_config(self):
- if self.is_local or self.is_unreach or self.is_prohibit:
+ if self.is_local or self.is_unreach or \
+ self.is_prohibit or self.is_drop:
self._test.vapi.ip_add_del_route(
self.dest_addr,
self.dest_addr_len,
self.dest_addr_p,
self.dest_addr_len))
+ def get_stats_to(self):
+ c = self._test.statistics.get_counter("/net/route/to")
+ return c[0][self.stats_index]
+
+ def get_stats_via(self):
+ c = self._test.statistics.get_counter("/net/route/via")
+ return c[0][self.stats_index]
+
class VppIpMRoute(VppObject):
"""
self.is_ip6 = is_ip6
self.rpf_id = rpf_id
+ self.grp_addr_p = grp_addr
+ self.src_addr_p = src_addr
if is_ip6:
self.grp_addr = inet_pton(AF_INET6, grp_addr)
self.src_addr = inet_pton(AF_INET6, src_addr)
def add_vpp_config(self):
for path in self.paths:
- self._test.vapi.ip_mroute_add_del(self.src_addr,
- self.grp_addr,
- self.grp_addr_len,
- self.e_flags,
- path.proto,
- path.nh_itf,
- path.nh_addr,
- path.nh_i_flags,
- bier_imp=path.bier_imp,
- rpf_id=self.rpf_id,
- table_id=self.table_id,
- is_ipv6=self.is_ip6)
+ r = self._test.vapi.ip_mroute_add_del(self.src_addr,
+ self.grp_addr,
+ self.grp_addr_len,
+ self.e_flags,
+ path.proto,
+ path.nh_itf,
+ path.nh_addr,
+ path.nh_i_flags,
+ bier_imp=path.bier_imp,
+ rpf_id=self.rpf_id,
+ table_id=self.table_id,
+ is_ipv6=self.is_ip6)
+ self.stats_index = r.stats_index
self._test.registry.register(self, self._test.logger)
def remove_vpp_config(self):
is_ipv6=self.is_ip6)
def query_vpp_config(self):
- if self.is_ip6:
- dump = self._test.vapi.ip6_mfib_dump()
- else:
- dump = self._test.vapi.ip_mfib_dump()
- for e in dump:
- if self.grp_addr == e.grp_address \
- and self.grp_addr_len == e.address_length \
- and self.src_addr == e.src_address \
- and self.table_id == e.table_id:
- return True
- return False
+ return find_mroute(self._test,
+ self.grp_addr_p,
+ self.src_addr_p,
+ self.grp_addr_len,
+ self.table_id,
+ inet=AF_INET6 if self.is_ip6 == 1 else AF_INET)
def __str__(self):
return self.object_id()
inet_ntop(AF_INET, self.grp_addr),
self.grp_addr_len))
+ def get_stats(self):
+ c = self._test.statistics.get_counter("/net/mroute")
+ return c[0][self.stats_index]
+
class VppMFibSignal(object):
def __init__(self, test, route, interface, packet):
for path in self.paths:
lstack = path.encode_labels()
- self._test.vapi.mpls_route_add_del(
+ r = self._test.vapi.mpls_route_add_del(
self.local_label,
self.eos_bit,
path.proto,
next_hop_n_out_labels=len(lstack),
next_hop_via_label=path.nh_via_label,
next_hop_table_id=path.nh_table_id)
+ self.stats_index = r.stats_index
self._test.registry.register(self, self._test.logger)
def remove_vpp_config(self):
% (self.table_id,
self.local_label,
20+self.eos_bit))
+
+ def get_stats_to(self):
+ c = self._test.statistics.get_counter("/net/route/to")
+ return c[0][self.stats_index]
+
+ def get_stats_via(self):
+ c = self._test.statistics.get_counter("/net/route/via")
+ return c[0][self.stats_index]
Code Review / vpp.git / blobdiff