8 from ipaddress import ip_address
9 from vpp_object import VppObject
16 _log = logging.getLogger(__name__)
23 DPO_PROTO_ETHERNET = 3
28 INVALID_INDEX = 0xFFFFFFFF
31 def get_dpo_proto(addr):
32 if ip_address(addr).version == 6:
33 return DpoProto.DPO_PROTO_IP6
35 return DpoProto.DPO_PROTO_IP4
38 class VppIpAddressUnion:
39 def __init__(self, addr):
41 self.ip_addr = ip_address(text_type(self.addr))
45 return {"ip6": self.ip_addr}
47 return {"ip4": self.ip_addr}
51 return self.ip_addr.version
59 return self.ip_addr.max_prefixlen
63 return self.ip_addr.packed
65 def __eq__(self, other):
66 if isinstance(other, self.__class__):
67 return self.ip_addr == other.ip_addr
68 elif hasattr(other, "ip4") and hasattr(other, "ip6"):
69 # vl_api_address_union_t
71 return self.ip_addr == other.ip4
73 return self.ip_addr == other.ip6
76 "Comparing VppIpAddressUnions:%s with incomparable type: %s",
81 def __ne__(self, other):
82 return not (self == other)
85 return str(self.ip_addr)
89 def __init__(self, saddr, gaddr, glen):
93 if ip_address(self.saddr).version != ip_address(self.gaddr).version:
95 "Source and group addresses must be of the same address family."
100 "af": ip_address(self.gaddr).vapi_af,
101 "grp_address": {ip_address(self.gaddr).vapi_af_name: self.gaddr},
102 "src_address": {ip_address(self.saddr).vapi_af_name: self.saddr},
103 "grp_address_length": self.glen,
112 return ip_address(self.gaddr).version
115 return "(%s,%s)/%d" % (self.saddr, self.gaddr, self.glen)
117 def __eq__(self, other):
118 if isinstance(other, self.__class__):
120 self.glen == other.glen
121 and self.saddr == other.gaddr
122 and self.saddr == other.saddr
125 hasattr(other, "grp_address_length")
126 and hasattr(other, "grp_address")
127 and hasattr(other, "src_address")
130 if 4 == self.version:
132 self.glen == other.grp_address_length
133 and self.gaddr == str(other.grp_address.ip4)
134 and self.saddr == str(other.src_address.ip4)
138 self.glen == other.grp_address_length
139 and self.gaddr == str(other.grp_address.ip6)
140 and self.saddr == str(other.src_address.ip6)
142 return NotImplemented
145 class VppIpPuntPolicer(VppObject):
146 def __init__(self, test, policer_index, is_ip6=False):
148 self._policer_index = policer_index
149 self._is_ip6 = is_ip6
151 def add_vpp_config(self):
152 self._test.vapi.ip_punt_police(
153 policer_index=self._policer_index, is_ip6=self._is_ip6, is_add=True
156 def remove_vpp_config(self):
157 self._test.vapi.ip_punt_police(
158 policer_index=self._policer_index, is_ip6=self._is_ip6, is_add=False
161 def query_vpp_config(self):
165 class VppIpPuntRedirect(VppObject):
166 def __init__(self, test, rx_index, tx_index, nh_addr):
168 self._rx_index = rx_index
169 self._tx_index = tx_index
170 self._nh_addr = ip_address(nh_addr)
174 "rx_sw_if_index": self._rx_index,
175 "tx_sw_if_index": self._tx_index,
179 def add_vpp_config(self):
180 self._test.vapi.ip_punt_redirect(punt=self.encode(), is_add=True)
181 self._test.registry.register(self, self._test.logger)
184 def remove_vpp_config(self):
185 self._test.vapi.ip_punt_redirect(punt=self.encode(), is_add=False)
187 def get_vpp_config(self):
188 is_ipv6 = True if self._nh_addr.version == 6 else False
189 return self._test.vapi.ip_punt_redirect_dump(
190 sw_if_index=self._rx_index, is_ipv6=is_ipv6
193 def query_vpp_config(self):
194 if self.get_vpp_config():
199 class VppIpPathMtu(VppObject):
200 def __init__(self, test, nh, pmtu, table_id=0):
204 self.table_id = table_id
206 def add_vpp_config(self):
207 self._test.vapi.ip_path_mtu_update(
208 pmtu={"nh": self.nh, "table_id": self.table_id, "path_mtu": self.pmtu}
210 self._test.registry.register(self, self._test.logger)
213 def modify(self, pmtu):
215 self._test.vapi.ip_path_mtu_update(
216 pmtu={"nh": self.nh, "table_id": self.table_id, "path_mtu": self.pmtu}
220 def remove_vpp_config(self):
221 self._test.vapi.ip_path_mtu_update(
222 pmtu={"nh": self.nh, "table_id": self.table_id, "path_mtu": 0}
225 def query_vpp_config(self):
226 ds = list(self._test.vapi.vpp.details_iter(self._test.vapi.ip_path_mtu_get))
230 self.nh == str(d.pmtu.nh)
231 and self.table_id == d.pmtu.table_id
232 and self.pmtu == d.pmtu.path_mtu
238 return "ip-path-mtu-%d-%s-%d" % (self.table_id, self.nh, self.pmtu)
241 return self.object_id()