lb: remove api boilerplate
[vpp.git] / test / vpp_ip.py
1 """
2   IP Types
3
4 """
5 import logging
6
7 from ipaddress import ip_address
8 from socket import AF_INET, AF_INET6
9 from vpp_papi import VppEnum
10 try:
11     text_type = unicode
12 except NameError:
13     text_type = str
14
15 _log = logging.getLogger(__name__)
16
17
18 class DpoProto:
19     DPO_PROTO_IP4 = 0
20     DPO_PROTO_IP6 = 1
21     DPO_PROTO_MPLS = 2
22     DPO_PROTO_ETHERNET = 3
23     DPO_PROTO_BIER = 4
24     DPO_PROTO_NSH = 5
25
26
27 INVALID_INDEX = 0xffffffff
28
29
30 class VppIpAddressUnion():
31     def __init__(self, addr):
32         self.addr = addr
33         self.ip_addr = ip_address(text_type(self.addr))
34
35     def encode(self):
36         if self.version == 6:
37             return {'ip6': self.ip_addr}
38         else:
39             return {'ip4': self.ip_addr}
40
41     @property
42     def version(self):
43         return self.ip_addr.version
44
45     @property
46     def address(self):
47         return self.addr
48
49     @property
50     def length(self):
51         return self.ip_addr.max_prefixlen
52
53     @property
54     def bytes(self):
55         return self.ip_addr.packed
56
57     def __eq__(self, other):
58         if isinstance(other, self.__class__):
59             return self.ip_addr == other.ip_addr
60         elif hasattr(other, "ip4") and hasattr(other, "ip6"):
61             # vl_api_address_union_t
62             if 4 == self.version:
63                 return self.ip_addr.packed == other.ip4
64             else:
65                 return self.ip_addr.packed == other.ip6
66         else:
67             _log.error("Comparing VppIpAddressUnions:%s"
68                        " with incomparable type: %s",
69                        self, other)
70             return NotImplemented
71
72     def __str__(self):
73         return str(self.ip_addr)
74
75
76 class VppIpAddress():
77     def __init__(self, addr):
78         self.addr = VppIpAddressUnion(addr)
79
80     def encode(self):
81         if self.addr.version == 6:
82             return {
83                 'af': VppEnum.vl_api_address_family_t.ADDRESS_IP6,
84                 'un': self.addr.encode()
85             }
86         else:
87             return {
88                 'af': VppEnum.vl_api_address_family_t.ADDRESS_IP4,
89                 'un': self.addr.encode()
90             }
91
92     def __eq__(self, other):
93         if isinstance(other, self.__class__):
94             return self.addr == other.addr
95         elif hasattr(other, "af") and hasattr(other, "un"):
96             # a vp_api_address_t
97             if 4 == self.version:
98                 return other.af == \
99                     VppEnum.vl_api_address_family_t.ADDRESS_IP4 and \
100                     other.un == self.addr
101             else:
102                 return other.af == \
103                     VppEnum.vl_api_address_family_t.ADDRESS_IP6 and \
104                     other.un == self.addr
105         else:
106             _log.error(
107                 "Comparing VppIpAddress:<%s> %s with incomparable "
108                 "type: <%s> %s",
109                 self.__class__.__name__, self,
110                 other.__class__.__name__, other)
111             return NotImplemented
112
113     def __ne__(self, other):
114         return not (self == other)
115
116     def __str__(self):
117         return self.address
118
119     @property
120     def bytes(self):
121         return self.addr.bytes
122
123     @property
124     def address(self):
125         return self.addr.address
126
127     @property
128     def length(self):
129         return self.addr.length
130
131     @property
132     def version(self):
133         return self.addr.version
134
135     @property
136     def is_ip6(self):
137         return (self.version == 6)
138
139     @property
140     def af(self):
141         if self.version == 6:
142             return AF_INET6
143         else:
144             return AF_INET
145
146     @property
147     def dpo_proto(self):
148         if self.version == 6:
149             return DpoProto.DPO_PROTO_IP6
150         else:
151             return DpoProto.DPO_PROTO_IP4
152
153
154 class VppIpPrefix():
155     def __init__(self, addr, len):
156         self.addr = VppIpAddress(addr)
157         self.len = len
158
159     def __eq__(self, other):
160         if self.address == other.address and self.len == other.len:
161             return True
162         return False
163
164     def encode(self):
165         return {'address': self.addr.encode(),
166                 'len': self.len}
167
168     @property
169     def version(self):
170         return self.addr.version
171
172     @property
173     def address(self):
174         return self.addr.address
175
176     @property
177     def bytes(self):
178         return self.addr.bytes
179
180     @property
181     def length(self):
182         return self.len
183
184     @property
185     def is_ip6(self):
186         return self.addr.is_ip6
187
188     def __str__(self):
189         return "%s/%d" % (self.address, self.length)
190
191     def __eq__(self, other):
192         if isinstance(other, self.__class__):
193             return (self.len == other.len and self.addr == other.addr)
194         elif hasattr(other, "address") and hasattr(other, "len"):
195             # vl_api_prefix_t
196             return self.len == other.len and \
197                    self.addr == other.address
198         else:
199             _log.error(
200                 "Comparing VppIpPrefix:%s with incomparable type: %s" %
201                 (self, other))
202             return NotImplemented
203
204
205 class VppIpMPrefix():
206     def __init__(self, saddr, gaddr, glen):
207         self.saddr = saddr
208         self.gaddr = gaddr
209         self.glen = glen
210         self.ip_saddr = VppIpAddressUnion(text_type(self.saddr))
211         self.ip_gaddr = VppIpAddressUnion(text_type(self.gaddr))
212         if self.ip_saddr.version != self.ip_gaddr.version:
213             raise ValueError('Source and group addresses must be of the '
214                              'same address family.')
215
216     def encode(self):
217         if 6 == self.ip_saddr.version:
218             prefix = {
219                 'af': VppEnum.vl_api_address_family_t.ADDRESS_IP6,
220                 'grp_address': {
221                     'ip6': self.gaddr
222                 },
223                 'src_address': {
224                     'ip6': self.saddr
225                 },
226                 'grp_address_length': self.glen,
227             }
228         else:
229             prefix = {
230                 'af': VppEnum.vl_api_address_family_t.ADDRESS_IP4,
231                 'grp_address': {
232                     'ip4': self.gaddr
233                 },
234                 'src_address': {
235                     'ip4':  self.saddr
236                 },
237                 'grp_address_length': self.glen,
238             }
239         return prefix
240
241     @property
242     def length(self):
243         return self.glen
244
245     @property
246     def version(self):
247         return self.ip_gaddr.version
248
249     def __str__(self):
250         return "(%s,%s)/%d" % (self.saddr, self.gaddr, self.glen)
251
252     def __eq__(self, other):
253         if isinstance(other, self.__class__):
254             return (self.glen == other.glen and
255                     self.ip_saddr == other.ip_gaddr and
256                     self.ip_saddr == other.ip_saddr)
257         elif (hasattr(other, "grp_address_length") and
258               hasattr(other, "grp_address") and
259               hasattr(other, "src_address")):
260             # vl_api_mprefix_t
261             if 4 == self.ip_saddr.version:
262                 if self.glen == other.grp_address_length and \
263                    self.gaddr == str(other.grp_address.ip4) and \
264                    self.saddr == str(other.src_address.ip4):
265                     return True
266                 return False
267             else:
268                 return (self.glen == other.grp_address_length and
269                         self.gaddr == other.grp_address.ip6 and
270                         self.saddr == other.src_address.ip6)
271         else:
272             raise Exception("Comparing VppIpPrefix:%s with unknown type: %s" %
273                             (self, other))
274         return False