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
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
75 raise Exception("Comparing VppIpAddressUnions:%s"
76 " with incomparable type: %s",
79 def __ne__(self, other):
80 return not (self == other)
83 return str(self.ip_addr)
87 def __init__(self, saddr, gaddr, glen):
91 if ip_address(self.saddr).version != \
92 ip_address(self.gaddr).version:
93 raise ValueError('Source and group addresses must be of the '
94 'same address family.')
98 'af': ip_address(self.gaddr).vapi_af,
100 ip_address(self.gaddr).vapi_af_name: self.gaddr
103 ip_address(self.saddr).vapi_af_name: self.saddr
105 'grp_address_length': self.glen,
114 return ip_address(self.gaddr).version
117 return "(%s,%s)/%d" % (self.saddr, self.gaddr, self.glen)
119 def __eq__(self, other):
120 if isinstance(other, self.__class__):
121 return (self.glen == other.glen and
122 self.saddr == other.gaddr and
123 self.saddr == other.saddr)
124 elif (hasattr(other, "grp_address_length") and
125 hasattr(other, "grp_address") and
126 hasattr(other, "src_address")):
128 if 4 == self.version:
129 return (self.glen == other.grp_address_length and
130 self.gaddr == str(other.grp_address.ip4) and
131 self.saddr == str(other.src_address.ip4))
133 return (self.glen == other.grp_address_length and
134 self.gaddr == str(other.grp_address.ip6) and
135 self.saddr == str(other.src_address.ip6))
136 return NotImplemented
139 class VppIpPuntPolicer(VppObject):
140 def __init__(self, test, policer_index, is_ip6=False):
142 self._policer_index = policer_index
143 self._is_ip6 = is_ip6
145 def add_vpp_config(self):
146 self._test.vapi.ip_punt_police(policer_index=self._policer_index,
147 is_ip6=self._is_ip6, is_add=True)
149 def remove_vpp_config(self):
150 self._test.vapi.ip_punt_police(policer_index=self._policer_index,
151 is_ip6=self._is_ip6, is_add=False)
153 def query_vpp_config(self):
157 class VppIpPuntRedirect(VppObject):
158 def __init__(self, test, rx_index, tx_index, nh_addr):
160 self._rx_index = rx_index
161 self._tx_index = tx_index
162 self._nh_addr = ip_address(nh_addr)
165 return {"rx_sw_if_index": self._rx_index,
166 "tx_sw_if_index": self._tx_index, "nh": self._nh_addr}
168 def add_vpp_config(self):
169 self._test.vapi.ip_punt_redirect(punt=self.encode(), is_add=True)
170 self._test.registry.register(self, self._test.logger)
172 def remove_vpp_config(self):
173 self._test.vapi.ip_punt_redirect(punt=self.encode(), is_add=False)
175 def get_vpp_config(self):
176 is_ipv6 = True if self._nh_addr.version == 6 else False
177 return self._test.vapi.ip_punt_redirect_dump(
178 sw_if_index=self._rx_index, is_ipv6=is_ipv6)
180 def query_vpp_config(self):
181 if self.get_vpp_config():
186 class VppIpPathMtu(VppObject):
187 def __init__(self, test, nh, pmtu, table_id=0):
191 self.table_id = table_id
193 def add_vpp_config(self):
194 self._test.vapi.ip_path_mtu_update(pmtu={'nh': self.nh,
195 'table_id': self.table_id,
196 'path_mtu': self.pmtu})
197 self._test.registry.register(self, self._test.logger)
200 def modify(self, pmtu):
202 self._test.vapi.ip_path_mtu_update(pmtu={'nh': self.nh,
203 'table_id': self.table_id,
204 'path_mtu': self.pmtu})
207 def remove_vpp_config(self):
208 self._test.vapi.ip_path_mtu_update(pmtu={'nh': self.nh,
209 'table_id': self.table_id,
212 def query_vpp_config(self):
213 ds = list(self._test.vapi.vpp.details_iter(
214 self._test.vapi.ip_path_mtu_get))
217 if self.nh == str(d.pmtu.nh) \
218 and self.table_id == d.pmtu.table_id \
219 and self.pmtu == d.pmtu.path_mtu:
224 return ("ip-path-mtu-%d-%s-%d" % (self.table_id,
229 return self.object_id()