7 from ipaddress import ip_address
8 from vpp_object import VppObject
15 _log = logging.getLogger(__name__)
22 DPO_PROTO_ETHERNET = 3
27 INVALID_INDEX = 0xFFFFFFFF
30 def get_dpo_proto(addr):
31 if ip_address(addr).version == 6:
32 return DpoProto.DPO_PROTO_IP6
34 return DpoProto.DPO_PROTO_IP4
37 class VppIpAddressUnion:
38 def __init__(self, addr):
40 self.ip_addr = ip_address(text_type(self.addr))
44 return {"ip6": self.ip_addr}
46 return {"ip4": self.ip_addr}
50 return self.ip_addr.version
58 return self.ip_addr.max_prefixlen
62 return self.ip_addr.packed
64 def __eq__(self, other):
65 if isinstance(other, self.__class__):
66 return self.ip_addr == other.ip_addr
67 elif hasattr(other, "ip4") and hasattr(other, "ip6"):
68 # vl_api_address_union_t
70 return self.ip_addr == other.ip4
72 return self.ip_addr == other.ip6
75 "Comparing VppIpAddressUnions:%s with incomparable type: %s",
80 def __ne__(self, other):
81 return not (self == other)
84 return str(self.ip_addr)
88 def __init__(self, saddr, gaddr, glen):
92 if ip_address(self.saddr).version != ip_address(self.gaddr).version:
94 "Source and group addresses must be of the same address family."
99 "af": ip_address(self.gaddr).vapi_af,
100 "grp_address": {ip_address(self.gaddr).vapi_af_name: self.gaddr},
101 "src_address": {ip_address(self.saddr).vapi_af_name: self.saddr},
102 "grp_address_length": self.glen,
111 return ip_address(self.gaddr).version
114 return "(%s,%s)/%d" % (self.saddr, self.gaddr, self.glen)
116 def __eq__(self, other):
117 if isinstance(other, self.__class__):
119 self.glen == other.glen
120 and self.saddr == other.gaddr
121 and self.saddr == other.saddr
124 hasattr(other, "grp_address_length")
125 and hasattr(other, "grp_address")
126 and hasattr(other, "src_address")
129 if 4 == self.version:
131 self.glen == other.grp_address_length
132 and self.gaddr == str(other.grp_address.ip4)
133 and self.saddr == str(other.src_address.ip4)
137 self.glen == other.grp_address_length
138 and self.gaddr == str(other.grp_address.ip6)
139 and self.saddr == str(other.src_address.ip6)
141 return NotImplemented
144 class VppIpPuntPolicer(VppObject):
145 def __init__(self, test, policer_index, is_ip6=False):
147 self._policer_index = policer_index
148 self._is_ip6 = is_ip6
150 def add_vpp_config(self):
151 self._test.vapi.ip_punt_police(
152 policer_index=self._policer_index, is_ip6=self._is_ip6, is_add=True
155 def remove_vpp_config(self):
156 self._test.vapi.ip_punt_police(
157 policer_index=self._policer_index, is_ip6=self._is_ip6, is_add=False
160 def query_vpp_config(self):
164 class VppIpPuntRedirect(VppObject):
165 def __init__(self, test, rx_index, tx_index, nh_addr):
167 self._rx_index = rx_index
168 self._tx_index = tx_index
169 self._nh_addr = ip_address(nh_addr)
173 "rx_sw_if_index": self._rx_index,
174 "tx_sw_if_index": self._tx_index,
178 def add_vpp_config(self):
179 self._test.vapi.ip_punt_redirect(punt=self.encode(), is_add=True)
180 self._test.registry.register(self, self._test.logger)
183 def remove_vpp_config(self):
184 self._test.vapi.ip_punt_redirect(punt=self.encode(), is_add=False)
186 def get_vpp_config(self):
187 is_ipv6 = True if self._nh_addr.version == 6 else False
188 return self._test.vapi.ip_punt_redirect_dump(
189 sw_if_index=self._rx_index, is_ipv6=is_ipv6
192 def query_vpp_config(self):
193 if self.get_vpp_config():
198 class VppIpPathMtu(VppObject):
199 def __init__(self, test, nh, pmtu, table_id=0):
203 self.table_id = table_id
205 def add_vpp_config(self):
206 self._test.vapi.ip_path_mtu_update(
207 pmtu={"nh": self.nh, "table_id": self.table_id, "path_mtu": self.pmtu}
209 self._test.registry.register(self, self._test.logger)
212 def modify(self, pmtu):
214 self._test.vapi.ip_path_mtu_update(
215 pmtu={"nh": self.nh, "table_id": self.table_id, "path_mtu": self.pmtu}
219 def remove_vpp_config(self):
220 self._test.vapi.ip_path_mtu_update(
221 pmtu={"nh": self.nh, "table_id": self.table_id, "path_mtu": 0}
224 def query_vpp_config(self):
225 ds = list(self._test.vapi.vpp.details_iter(self._test.vapi.ip_path_mtu_get))
229 self.nh == str(d.pmtu.nh)
230 and self.table_id == d.pmtu.table_id
231 and self.pmtu == d.pmtu.path_mtu
237 return "ip-path-mtu-%d-%s-%d" % (self.table_id, self.nh, self.pmtu)
240 return self.object_id()