X-Git-Url: https://gerrit.fd.io/r/gitweb?a=blobdiff_plain;f=test%2Fvpp_ip_route.py;h=cd70b38101aaa268b7be7df6e80cefdc4a345831;hb=7784140f2bd2d5ae44f2be1507ac25f102006155;hp=3ac0e84c58ab0529840912381bd84bdbf351065b;hpb=9efcee6e7babd49999af3e3dcd13ee33d0a4c02d;p=vpp.git diff --git a/test/vpp_ip_route.py b/test/vpp_ip_route.py index 3ac0e84c58a..cd70b38101a 100644 --- a/test/vpp_ip_route.py +++ b/test/vpp_ip_route.py @@ -8,7 +8,7 @@ from vpp_object import VppObject from socket import inet_pton, inet_ntop, AF_INET, AF_INET6 from vpp_ip import DpoProto, INVALID_INDEX, VppIpAddressUnion, \ VppIpMPrefix -from ipaddress import ip_address, IPv4Network, IPv6Network +from ipaddress import ip_network, ip_address, IPv4Network, IPv6Network # from vnet/vnet/mpls/mpls_types.h MPLS_IETF_MAX_LABEL = 0xfffff @@ -87,16 +87,16 @@ def ip_to_dpo_proto(addr): def address_proto(ip_addr): - if ip_addr.ip_addr.version is 4: + if ip_addr.ip_addr.version == 4: return FibPathProto.FIB_PATH_NH_PROTO_IP4 else: return FibPathProto.FIB_PATH_NH_PROTO_IP6 -def find_route(test, addr, len, table_id=0): +def find_route(test, addr, len, table_id=0, sw_if_index=None): prefix = mk_network(addr, len) - if 4 is prefix.version: + if 4 == prefix.version: routes = test.vapi.ip_route_dump(table_id, False) else: routes = test.vapi.ip_route_dump(table_id, True) @@ -104,7 +104,16 @@ def find_route(test, addr, len, table_id=0): for e in routes: if table_id == e.route.table_id \ and str(e.route.prefix) == str(prefix): - return True + if not sw_if_index: + return True + else: + # should be only one path if the user is looking + # for the interface the route is reachable through + if e.route.n_paths != 1: + return False + else: + return (e.route.paths[0].sw_if_index == sw_if_index) + return False @@ -131,7 +140,7 @@ def find_mroute(test, grp_addr, src_addr, grp_addr_len, text_type(grp_addr), grp_addr_len) - if 4 is ip_mprefix.version: + if 4 == ip_mprefix.version: routes = test.vapi.ip_mroute_dump(table_id, False) else: routes = test.vapi.ip_mroute_dump(table_id, True) @@ -234,12 +243,16 @@ class VppIpTable(VppObject): class VppIpInterfaceAddress(VppObject): - def __init__(self, test, intf, addr, len): + def __init__(self, test, intf, addr, len, bind=None): self._test = test self.intf = intf self.addr = addr self.len = len self.prefix = "%s/%d" % (addr, len) + self.host_len = ip_network(self.prefix, strict=False).max_prefixlen + self.table_id = 0 + if bind: + self.table_id = bind.table.table_id def add_vpp_config(self): self._test.vapi.sw_interface_add_del_address( @@ -254,13 +267,65 @@ class VppIpInterfaceAddress(VppObject): is_add=0) def query_vpp_config(self): - return fib_interface_ip_prefix(self._test, - self.addr, - self.len, - self.intf.sw_if_index) + # search for the IP address mapping and the two expected + # FIB entries + v = ip_address(self.addr).version + + if ((v == 4 and self.len < 31) or (v == 6 and self.len < 127)): + return (fib_interface_ip_prefix(self._test, + self.addr, + self.len, + self.intf.sw_if_index) & + find_route(self._test, + self.addr, + self.len, + table_id=self.table_id, + sw_if_index=self.intf.sw_if_index) & + find_route(self._test, + self.addr, + self.host_len, + table_id=self.table_id, + sw_if_index=self.intf.sw_if_index)) + else: + return (fib_interface_ip_prefix(self._test, + self.addr, + self.len, + self.intf.sw_if_index) & + find_route(self._test, + self.addr, + self.host_len, + table_id=self.table_id, + sw_if_index=self.intf.sw_if_index)) def object_id(self): - return "interface-ip-%s-%s" % (self.intf, self.prefix) + return "interface-ip-%s-%d-%s" % (self.intf, + self.table_id, + self.prefix) + + +class VppIp6LinkLocalAddress(VppObject): + + def __init__(self, test, intf, addr): + self._test = test + self.intf = intf + self.addr = addr + + def add_vpp_config(self): + self._test.vapi.sw_interface_ip6_set_link_local_address( + sw_if_index=self.intf.sw_if_index, ip=self.addr) + self._test.registry.register(self, self._test.logger) + return self + + def remove_vpp_config(self): + # link locals can't be removed, only changed + pass + + def query_vpp_config(self): + # no API to query + return False + + def object_id(self): + return "ip6-link-local-%s-%s" % (self.intf, self.addr) class VppIpInterfaceBind(VppObject): @@ -276,6 +341,7 @@ class VppIpInterfaceBind(VppObject): else: self.intf.set_table_ip4(self.table.table_id) self._test.registry.register(self, self._test.logger) + return self def remove_vpp_config(self): if 0 == self.table.table_id: @@ -573,57 +639,53 @@ class VppIpMRoute(VppObject): for path in self.paths: self.encoded_paths.append(path.encode()) + def encode(self, paths=None): + _paths = self.encoded_paths if paths is None else paths + return {'table_id': self.table_id, + 'entry_flags': self.e_flags, + 'rpf_id': self.rpf_id, + 'prefix': self.prefix.encode(), + 'n_paths': len(_paths), + 'paths': _paths, + } + def add_vpp_config(self): - r = self._test.vapi.ip_mroute_add_del(self.table_id, - self.prefix.encode(), - self.e_flags, - self.rpf_id, - self.encoded_paths, + r = self._test.vapi.ip_mroute_add_del(route=self.encode(), + is_multipath=1, is_add=1) self.stats_index = r.stats_index self._test.registry.register(self, self._test.logger) return self def remove_vpp_config(self): - self._test.vapi.ip_mroute_add_del(self.table_id, - self.prefix.encode(), - self.e_flags, - self.rpf_id, - self.encoded_paths, + self._test.vapi.ip_mroute_add_del(route=self.encode(), + is_multipath=1, is_add=0) def update_entry_flags(self, flags): self.e_flags = flags - self._test.vapi.ip_mroute_add_del(self.table_id, - self.prefix.encode(), - self.e_flags, - self.rpf_id, - [], + self._test.vapi.ip_mroute_add_del(route=self.encode(paths=[]), + is_multipath=1, is_add=1) def update_rpf_id(self, rpf_id): self.rpf_id = rpf_id - self._test.vapi.ip_mroute_add_del(self.table_id, - self.prefix.encode(), - self.e_flags, - self.rpf_id, - [], + self._test.vapi.ip_mroute_add_del(route=self.encode(paths=[]), + is_multipath=1, is_add=1) def update_path_flags(self, itf, flags): for p in range(len(self.paths)): if self.paths[p].nh_itf == itf: self.paths[p].nh_i_flags = flags - self.encoded_paths[p] = self.paths[p].encode() - break - - self._test.vapi.ip_mroute_add_del(self.table_id, - self.prefix.encode(), - self.e_flags, - self.rpf_id, - [self.encoded_paths[p]], - is_add=1, - is_multipath=0) + self.encoded_paths[p] = self.paths[p].encode() + break + + self._test.vapi.ip_mroute_add_del( + route=self.encode( + paths=[self.encoded_paths[p]]), + is_add=1, + is_multipath=0) def query_vpp_config(self): return find_mroute(self._test,