7 from ipaddress import ip_address
8 from socket import AF_INET, AF_INET6
9 from vpp_papi import VppEnum
10 from vpp_object import VppObject
17 _log = logging.getLogger(__name__)
24 DPO_PROTO_ETHERNET = 3
29 INVALID_INDEX = 0xFFFFFFFF
32 def get_dpo_proto(addr):
33 if ip_address(addr).version == 6:
34 return DpoProto.DPO_PROTO_IP6
36 return DpoProto.DPO_PROTO_IP4
39 class VppIpAddressUnion:
40 def __init__(self, addr):
42 self.ip_addr = ip_address(text_type(self.addr))
46 return {"ip6": self.ip_addr}
48 return {"ip4": self.ip_addr}
52 return self.ip_addr.version
60 return self.ip_addr.max_prefixlen
64 return self.ip_addr.packed
66 def __eq__(self, other):
67 if isinstance(other, self.__class__):
68 return self.ip_addr == other.ip_addr
69 elif hasattr(other, "ip4") and hasattr(other, "ip6"):
70 # vl_api_address_union_t
72 return self.ip_addr == other.ip4
74 return self.ip_addr == other.ip6
77 "Comparing VppIpAddressUnions:%s with incomparable type: %s",
82 def __ne__(self, other):
83 return not (self == other)
86 return str(self.ip_addr)
90 def __init__(self, saddr, gaddr, glen):
94 if ip_address(self.saddr).version != ip_address(self.gaddr).version:
96 "Source and group addresses must be of the same address family."
101 "af": ip_address(self.gaddr).vapi_af,
102 "grp_address": {ip_address(self.gaddr).vapi_af_name: self.gaddr},
103 "src_address": {ip_address(self.saddr).vapi_af_name: self.saddr},
104 "grp_address_length": self.glen,
113 return ip_address(self.gaddr).version
116 return "(%s,%s)/%d" % (self.saddr, self.gaddr, self.glen)
118 def __eq__(self, other):
119 if isinstance(other, self.__class__):
121 self.glen == other.glen
122 and self.saddr == other.gaddr
123 and self.saddr == other.saddr
126 hasattr(other, "grp_address_length")
127 and hasattr(other, "grp_address")
128 and hasattr(other, "src_address")
131 if 4 == self.version:
133 self.glen == other.grp_address_length
134 and self.gaddr == str(other.grp_address.ip4)
135 and self.saddr == str(other.src_address.ip4)
139 self.glen == other.grp_address_length
140 and self.gaddr == str(other.grp_address.ip6)
141 and self.saddr == str(other.src_address.ip6)
143 return NotImplemented
146 class VppIpPuntPolicer(VppObject):
147 def __init__(self, test, policer_index, is_ip6=False):
149 self._policer_index = policer_index
150 self._is_ip6 = is_ip6
152 def add_vpp_config(self):
153 self._test.vapi.ip_punt_police(
154 policer_index=self._policer_index, is_ip6=self._is_ip6, is_add=True
157 def remove_vpp_config(self):
158 self._test.vapi.ip_punt_police(
159 policer_index=self._policer_index, is_ip6=self._is_ip6, is_add=False
162 def query_vpp_config(self):
166 class VppIpPuntRedirect(VppObject):
167 def __init__(self, test, rx_index, tx_index, nh_addr):
169 self._rx_index = rx_index
170 self._tx_index = tx_index
171 self._nh_addr = ip_address(nh_addr)
175 "rx_sw_if_index": self._rx_index,
176 "tx_sw_if_index": self._tx_index,
180 def add_vpp_config(self):
181 self._test.vapi.ip_punt_redirect(punt=self.encode(), is_add=True)
182 self._test.registry.register(self, self._test.logger)
185 def remove_vpp_config(self):
186 self._test.vapi.ip_punt_redirect(punt=self.encode(), is_add=False)
188 def get_vpp_config(self):
189 is_ipv6 = True if self._nh_addr.version == 6 else False
190 return self._test.vapi.ip_punt_redirect_dump(
191 sw_if_index=self._rx_index, is_ipv6=is_ipv6
194 def query_vpp_config(self):
195 if self.get_vpp_config():
200 class VppIpPathMtu(VppObject):
201 def __init__(self, test, nh, pmtu, table_id=0):
205 self.table_id = table_id
207 def add_vpp_config(self):
208 self._test.vapi.ip_path_mtu_update(
209 pmtu={"nh": self.nh, "table_id": self.table_id, "path_mtu": self.pmtu}
211 self._test.registry.register(self, self._test.logger)
214 def modify(self, pmtu):
216 self._test.vapi.ip_path_mtu_update(
217 pmtu={"nh": self.nh, "table_id": self.table_id, "path_mtu": self.pmtu}
221 def remove_vpp_config(self):
222 self._test.vapi.ip_path_mtu_update(
223 pmtu={"nh": self.nh, "table_id": self.table_id, "path_mtu": 0}
226 def query_vpp_config(self):
227 ds = list(self._test.vapi.vpp.details_iter(self._test.vapi.ip_path_mtu_get))
231 self.nh == str(d.pmtu.nh)
232 and self.table_id == d.pmtu.table_id
233 and self.pmtu == d.pmtu.path_mtu
239 return "ip-path-mtu-%d-%s-%d" % (self.table_id, self.nh, self.pmtu)
242 return self.object_id()