tests: Remove the unrequired VPP IP address/prefix class wrappers
Type: refactor
Change-Id: I204f3f8eebc5f5d5a377e91262f91c615fd00168
Signed-off-by: Neale Ranns <nranns@cisco.com>
diff --git a/test/bfd.py b/test/bfd.py
index e096284..9d44425 100644
--- a/test/bfd.py
+++ b/test/bfd.py
@@ -9,7 +9,6 @@
ConditionalField, StrField
from vpp_object import VppObject
from util import NumericConstant
-from vpp_ip import VppIpAddress
from vpp_papi import VppEnum
@@ -236,10 +235,10 @@
self._interface = interface
self._af = af
if local_addr:
- self._local_addr = VppIpAddress(local_addr)
+ self._local_addr = local_addr
else:
self._local_addr = None
- self._peer_addr = VppIpAddress(peer_addr)
+ self._peer_addr = peer_addr
self._desired_min_tx = desired_min_tx
self._required_min_rx = required_min_rx
self._detect_mult = detect_mult
@@ -275,12 +274,12 @@
return self._interface.local_ip6
else:
raise Exception("Unexpected af '%s'" % self.af)
- return self._local_addr.address
+ return self._local_addr
@property
def peer_addr(self):
""" BFD session peer address """
- return self._peer_addr.address
+ return self._peer_addr
def get_bfd_udp_session_dump_entry(self):
""" get the namedtuple entry from bfd udp session dump """
@@ -343,8 +342,8 @@
is_delayed = 1 if delayed else 0
self.test.vapi.bfd_udp_auth_activate(
sw_if_index=self._interface.sw_if_index,
- local_addr=self.local_addr.encode(),
- peer_addr=self.peer_addr.encode(),
+ local_addr=self.local_addr,
+ peer_addr=self.peer_addr,
bfd_key_id=self._bfd_key_id,
conf_key_id=conf_key_id,
is_delayed=is_delayed)
@@ -356,8 +355,8 @@
is_delayed = 1 if delayed else 0
self.test.vapi.bfd_udp_auth_deactivate(
sw_if_index=self._interface.sw_if_index,
- local_addr=self.local_addr.encode(),
- peer_addr=self.peer_addr.encode(),
+ local_addr=self.local_addr,
+ peer_addr=self.peer_addr,
is_delayed=is_delayed)
def modify_parameters(self,
@@ -375,8 +374,8 @@
desired_min_tx=self.desired_min_tx,
required_min_rx=self.required_min_rx,
detect_mult=self.detect_mult,
- local_addr=self.local_addr.encode(),
- peer_addr=self.peer_addr.encode())
+ local_addr=self.local_addr,
+ peer_addr=self.peer_addr)
def add_vpp_config(self):
bfd_key_id = self._bfd_key_id if self._sha1_key else None
@@ -386,8 +385,8 @@
desired_min_tx=self.desired_min_tx,
required_min_rx=self.required_min_rx,
detect_mult=self.detect_mult,
- local_addr=self.local_addr.encode(),
- peer_addr=self.peer_addr.encode(),
+ local_addr=self.local_addr,
+ peer_addr=self.peer_addr,
bfd_key_id=bfd_key_id,
conf_key_id=conf_key_id,
is_authenticated=is_authenticated)
@@ -399,8 +398,8 @@
def remove_vpp_config(self):
self.test.vapi.bfd_udp_del(self._interface.sw_if_index,
- local_addr=self.local_addr.encode(),
- peer_addr=self.peer_addr.encode())
+ local_addr=self.local_addr,
+ peer_addr=self.peer_addr)
def object_id(self):
return "bfd-udp-%s-%s-%s-%s" % (self._interface.sw_if_index,
@@ -413,12 +412,12 @@
self.test.vapi.bfd_udp_session_set_flags(
flags=VppEnum.vl_api_if_status_flags_t.IF_STATUS_API_FLAG_ADMIN_UP,
sw_if_index=self._interface.sw_if_index,
- local_addr=self.local_addr.encode(),
- peer_addr=self.peer_addr.encode())
+ local_addr=self.local_addr,
+ peer_addr=self.peer_addr)
def admin_down(self):
""" set bfd session admin-down """
self.test.vapi.bfd_udp_session_set_flags(
flags=0, sw_if_index=self._interface.sw_if_index,
- local_addr=self.local_addr.encode(),
- peer_addr=self.peer_addr.encode())
+ local_addr=self.local_addr,
+ peer_addr=self.peer_addr)
diff --git a/test/test_bond.py b/test/test_bond.py
index 1b6c10b..29ad33d 100644
--- a/test/test_bond.py
+++ b/test/test_bond.py
@@ -9,7 +9,6 @@
from scapy.layers.inet import IP, UDP
from vpp_bond_interface import VppBondInterface
from vpp_papi import MACAddress
-from vpp_ip import VppIpPrefix
class TestBondInterface(VppTestCase):
@@ -73,7 +72,7 @@
bond0.admin_up()
self.vapi.sw_interface_add_del_address(
sw_if_index=bond0.sw_if_index,
- prefix=VppIpPrefix("10.10.10.1", 24).encode())
+ prefix="10.10.10.1/24")
self.pg2.config_ip4()
self.pg2.resolve_arp()
diff --git a/test/test_interface_crud.py b/test/test_interface_crud.py
index f701e27..3ab83c9 100644
--- a/test/test_interface_crud.py
+++ b/test/test_interface_crud.py
@@ -85,7 +85,7 @@
# create
loopbacks = self.create_loopback_interfaces(20)
for i in loopbacks:
- i.local_ip4_prefix.len = 32
+ i.local_ip4_prefix_len = 32
i.config_ip4()
i.admin_up()
@@ -132,7 +132,7 @@
# create
loopbacks = self.create_loopback_interfaces(20)
for i in loopbacks:
- i.local_ip4_prefix.len = 32
+ i.local_ip4_prefix_len = 32
i.config_ip4()
i.admin_up()
diff --git a/test/test_ip4.py b/test/test_ip4.py
index 0f37e73..5c268a8 100644
--- a/test/test_ip4.py
+++ b/test/test_ip4.py
@@ -13,12 +13,10 @@
from framework import VppTestCase, VppTestRunner
from util import ppp
-from vpp_ip import VppIpPrefix
from vpp_ip_route import VppIpRoute, VppRoutePath, VppIpMRoute, \
VppMRoutePath, MRouteItfFlags, MRouteEntryFlags, VppMplsIpBind, \
VppMplsTable, VppIpTable, FibPathType, find_route, \
VppIpInterfaceAddress
-from vpp_ip import VppIpAddress
from vpp_sub_interface import VppSubInterface, VppDot1QSubint, VppDot1ADSubint
from vpp_papi import VppEnum
from vpp_neighbor import VppNeighbor
@@ -258,11 +256,9 @@
"""
# create two addresses, verify route not present
- if_addr1 = VppIpInterfaceAddress(self, self.pg0,
- VppIpAddress("10.10.10.10"), 24)
- if_addr2 = VppIpInterfaceAddress(self, self.pg0,
- VppIpAddress("10.10.10.20"), 24)
- self.assertFalse(if_addr1.query_vpp_config()) # 10.10.10.0/24
+ if_addr1 = VppIpInterfaceAddress(self, self.pg0, "10.10.10.10", 24)
+ if_addr2 = VppIpInterfaceAddress(self, self.pg0, "10.10.10.20", 24)
+ self.assertFalse(if_addr1.query_vpp_config()) # 10.10.10.10/24
self.assertFalse(find_route(self, "10.10.10.10", 32))
self.assertFalse(find_route(self, "10.10.10.20", 32))
self.assertFalse(find_route(self, "10.10.10.255", 32))
@@ -270,7 +266,7 @@
# configure first address, verify route present
if_addr1.add_vpp_config()
- self.assertTrue(if_addr1.query_vpp_config()) # 10.10.10.0/24
+ self.assertTrue(if_addr1.query_vpp_config()) # 10.10.10.10/24
self.assertTrue(find_route(self, "10.10.10.10", 32))
self.assertFalse(find_route(self, "10.10.10.20", 32))
self.assertTrue(find_route(self, "10.10.10.255", 32))
@@ -279,7 +275,8 @@
# configure second address, delete first, verify route not removed
if_addr2.add_vpp_config()
if_addr1.remove_vpp_config()
- self.assertTrue(if_addr1.query_vpp_config()) # 10.10.10.0/24
+ self.assertFalse(if_addr1.query_vpp_config()) # 10.10.10.10/24
+ self.assertTrue(if_addr2.query_vpp_config()) # 10.10.10.20/24
self.assertFalse(find_route(self, "10.10.10.10", 32))
self.assertTrue(find_route(self, "10.10.10.20", 32))
self.assertTrue(find_route(self, "10.10.10.255", 32))
@@ -287,7 +284,7 @@
# delete second address, verify route removed
if_addr2.remove_vpp_config()
- self.assertFalse(if_addr1.query_vpp_config()) # 10.10.10.0/24
+ self.assertFalse(if_addr2.query_vpp_config()) # 10.10.10.20/24
self.assertFalse(find_route(self, "10.10.10.10", 32))
self.assertFalse(find_route(self, "10.10.10.20", 32))
self.assertFalse(find_route(self, "10.10.10.255", 32))
@@ -451,11 +448,11 @@
pkts = []
for _ in range(count):
- dst_addr = random.choice(routes).prefix.address
+ dst_addr = random.choice(routes).prefix.network_address
info = self.create_packet_info(src_if, dst_if)
payload = self.info_to_payload(info)
p = (Ether(dst=src_if.local_mac, src=src_if.remote_mac) /
- IP(src=src_if.remote_ip4, dst=dst_addr) /
+ IP(src=src_if.remote_ip4, dst=str(dst_addr)) /
UDP(sport=1234, dport=1234) /
Raw(payload))
info.data = p.copy()
@@ -491,11 +488,15 @@
def verify_route_dump(self, routes):
for r in routes:
- self.assertTrue(find_route(self, r.prefix.address, r.prefix.len))
+ self.assertTrue(find_route(self,
+ r.prefix.network_address,
+ r.prefix.prefixlen))
def verify_not_in_route_dump(self, routes):
for r in routes:
- self.assertFalse(find_route(self, r.prefix.address, r.prefix.len))
+ self.assertFalse(find_route(self,
+ r.prefix.network_address,
+ r.prefix.prefixlen))
@classmethod
def setUpClass(cls):
@@ -923,7 +924,7 @@
self.vapi.sw_interface_add_del_address(
sw_if_index=self.pg0.sw_if_index,
- prefix=VppIpPrefix("10.10.10.10", 16).encode())
+ prefix="10.10.10.10/16")
pn = (Ether(src=self.pg1.remote_mac,
dst=self.pg1.local_mac) /
@@ -942,7 +943,8 @@
# remove the sub-net and we are forwarding via the cover again
self.vapi.sw_interface_add_del_address(
sw_if_index=self.pg0.sw_if_index,
- prefix=VppIpPrefix("10.10.10.10", 16).encode(), is_add=0)
+ prefix="10.10.10.10/16",
+ is_add=0)
self.pg1.add_stream(pn)
self.pg_enable_capture(self.pg_interfaces)
@@ -961,7 +963,7 @@
self.vapi.sw_interface_add_del_address(
sw_if_index=self.pg0.sw_if_index,
- prefix=VppIpPrefix("10.10.10.10", 31).encode())
+ prefix="10.10.10.10/31")
pn = (Ether(src=self.pg1.remote_mac,
dst=self.pg1.local_mac) /
@@ -978,7 +980,7 @@
# remove the sub-net and we are forwarding via the cover again
self.vapi.sw_interface_add_del_address(
sw_if_index=self.pg0.sw_if_index,
- prefix=VppIpPrefix("10.10.10.10", 31).encode(), is_add=0)
+ prefix="10.10.10.10/31", is_add=0)
self.pg1.add_stream(pn)
self.pg_enable_capture(self.pg_interfaces)
diff --git a/test/test_ip6.py b/test/test_ip6.py
index 4cd0828..36532ce 100644
--- a/test/test_ip6.py
+++ b/test/test_ip6.py
@@ -20,7 +20,7 @@
from framework import VppTestCase, VppTestRunner
from util import ppp, ip6_normalize, mk_ll_addr
-from vpp_ip import DpoProto, VppIpAddress
+from vpp_ip import DpoProto
from vpp_ip_route import VppIpRoute, VppRoutePath, find_route, VppIpMRoute, \
VppMRoutePath, MRouteItfFlags, MRouteEntryFlags, VppMplsIpBind, \
VppMplsRoute, VppMplsTable, VppIpTable, FibPathType, \
@@ -28,7 +28,7 @@
from vpp_neighbor import find_nbr, VppNeighbor
from vpp_pg_interface import is_ipv6_misc
from vpp_sub_interface import VppSubInterface, VppDot1QSubint
-from ipaddress import IPv6Network, IPv4Network, IPv6Address
+from ipaddress import IPv6Network, IPv6Address
AF_INET6 = socket.AF_INET6
@@ -982,30 +982,29 @@
addr1 = "2001:10::10"
addr2 = "2001:10::20"
- if_addr1 = VppIpInterfaceAddress(self, self.pg0,
- VppIpAddress(addr1), 64)
- if_addr2 = VppIpInterfaceAddress(self, self.pg0,
- VppIpAddress(addr2), 64)
- self.assertFalse(if_addr1.query_vpp_config()) # 2001:10::/64
+ if_addr1 = VppIpInterfaceAddress(self, self.pg0, addr1, 64)
+ if_addr2 = VppIpInterfaceAddress(self, self.pg0, addr2, 64)
+ self.assertFalse(if_addr1.query_vpp_config())
self.assertFalse(find_route(self, addr1, 128))
self.assertFalse(find_route(self, addr2, 128))
# configure first address, verify route present
if_addr1.add_vpp_config()
- self.assertTrue(if_addr1.query_vpp_config()) # 2001:10::/64
+ self.assertTrue(if_addr1.query_vpp_config())
self.assertTrue(find_route(self, addr1, 128))
self.assertFalse(find_route(self, addr2, 128))
# configure second address, delete first, verify route not removed
if_addr2.add_vpp_config()
if_addr1.remove_vpp_config()
- self.assertTrue(if_addr1.query_vpp_config()) # 2001:10::/64
+ self.assertFalse(if_addr1.query_vpp_config())
+ self.assertTrue(if_addr2.query_vpp_config())
self.assertFalse(find_route(self, addr1, 128))
self.assertTrue(find_route(self, addr2, 128))
# delete second address, verify route removed
if_addr2.remove_vpp_config()
- self.assertFalse(if_addr1.query_vpp_config()) # 2001:10::/64
+ self.assertFalse(if_addr1.query_vpp_config())
self.assertFalse(find_route(self, addr1, 128))
self.assertFalse(find_route(self, addr2, 128))
diff --git a/test/test_vxlan_gbp.py b/test/test_vxlan_gbp.py
index 64e51b0..17ee662 100644
--- a/test/test_vxlan_gbp.py
+++ b/test/test_vxlan_gbp.py
@@ -5,7 +5,6 @@
import unittest
from framework import VppTestCase, VppTestRunner
from template_bd import BridgeDomain
-from vpp_ip import VppIpAddress
from scapy.layers.l2 import Ether, Raw
from scapy.layers.inet import IP, UDP
@@ -104,8 +103,8 @@
rip.add_vpp_config()
r = cls.vapi.vxlan_gbp_tunnel_add_del(
tunnel={
- 'src': VppIpAddress(cls.pg0.local_ip4).encode(),
- 'dst': VppIpAddress(dest_ip4).encode(),
+ 'src': cls.pg0.local_ip4,
+ 'dst': dest_ip4,
'vni': vni,
'instance': INVALID_INDEX,
'mcast_sw_if_index': INVALID_INDEX,
@@ -147,8 +146,8 @@
cls.single_tunnel_bd = 1
r = cls.vapi.vxlan_gbp_tunnel_add_del(
tunnel={
- 'src': VppIpAddress(cls.pg0.local_ip4).encode(),
- 'dst': VppIpAddress(cls.pg0.remote_ip4).encode(),
+ 'src': cls.pg0.local_ip4,
+ 'dst': cls.pg0.remote_ip4,
'vni': cls.single_tunnel_bd,
'instance': INVALID_INDEX,
'mcast_sw_if_index': INVALID_INDEX,
diff --git a/test/vpp_interface.py b/test/vpp_interface.py
index a5f6f45..9bfa880 100644
--- a/test/vpp_interface.py
+++ b/test/vpp_interface.py
@@ -7,8 +7,7 @@
from util import Host, mk_ll_addr
from vpp_papi import mac_ntop, VppEnum
-from vpp_ip import VppIpAddress, VppIpPrefix
-from ipaddress import IPv4Network
+from ipaddress import IPv4Network, IPv6Network
try:
text_type = unicode
@@ -46,31 +45,31 @@
@property
def local_ip4(self):
"""Local IPv4 address on VPP interface (string)."""
- return self._local_ip4.address
+ return self._local_ip4
@local_ip4.setter
def local_ip4(self, value):
- self._local_ip4.address = value
+ self._local_ip4 = value
@property
def local_ip4_prefix_len(self):
"""Local IPv4 prefix length """
- return self._local_ip4.len
+ return self._local_ip4_len
@local_ip4_prefix_len.setter
def local_ip4_prefix_len(self, value):
- self._local_ip4.len = value
+ self._local_ip4_len = value
@property
def local_ip4_prefix(self):
"""Local IPv4 prefix """
- return self._local_ip4
+ return ("%s/%d" % (self._local_ip4, self._local_ip4_len))
@property
def local_ip4n(self):
"""DEPRECATED """
"""Local IPv4 address - raw, suitable as API parameter."""
- return socket.inet_pton(socket.AF_INET, self._local_ip4.address)
+ return socket.inet_pton(socket.AF_INET, self._local_ip4)
@property
def remote_ip4(self):
@@ -86,31 +85,31 @@
@property
def local_ip6(self):
"""Local IPv6 address on VPP interface (string)."""
- return self._local_ip6.address
+ return self._local_ip6
@local_ip6.setter
def local_ip6(self, value):
- self._local_ip6.address = value
+ self._local_ip6
@property
def local_ip6_prefix_len(self):
"""Local IPv6 prefix length """
- return self._local_ip6.len
+ return self._local_ip6_len
@local_ip6_prefix_len.setter
def local_ip6_prefix_len(self, value):
- self._local_ip6.len = value
+ self._local_ip6_len = value
@property
def local_ip6_prefix(self):
- """Local IPv6 prefix """
- return self._local_ip6
+ """Local IPv4 prefix """
+ return ("%s/%d" % (self._local_ip6, self._local_ip6_len))
@property
def local_ip6n(self):
"""DEPRECATED """
"""Local IPv6 address - raw, suitable as API parameter."""
- return socket.inet_pton(socket.AF_INET6, self._local_ip6.address)
+ return socket.inet_pton(socket.AF_INET6, self._local_ip6)
@property
def remote_ip6(self):
@@ -126,7 +125,7 @@
@property
def local_ip6_ll(self):
"""Local IPv6 link-local address on VPP interface (string)."""
- return self._local_ip6_ll.address
+ return self._local_ip6_ll
@property
def local_ip6n_ll(self):
@@ -233,7 +232,7 @@
def set_mac(self, mac):
self._local_mac = str(mac)
- self._local_ip6_ll = VppIpAddress(mk_ll_addr(self._local_mac))
+ self._local_ip6_ll = mk_ll_addr(self._local_mac)
self.test.vapi.sw_interface_set_mac_address(
self.sw_if_index, mac.packed)
@@ -242,13 +241,15 @@
self.generate_remote_hosts()
- self._local_ip4 = VppIpPrefix("172.16.%u.1" % self.sw_if_index, 24)
+ self._local_ip4 = "172.16.%u.1" % self.sw_if_index
+ self._local_ip4_len = 24
self._local_ip4_subnet = "172.16.%u.0" % self.sw_if_index
self._local_ip4_bcast = "172.16.%u.255" % self.sw_if_index
self.has_ip4_config = False
self.ip4_table_id = 0
- self._local_ip6 = VppIpPrefix("fd01:%x::1" % self.sw_if_index, 64)
+ self._local_ip6 = "fd01:%x::1" % self.sw_if_index
+ self._local_ip6_len = 64
self.has_ip6_config = False
self.ip6_table_id = 0
@@ -269,13 +270,13 @@
"Could not find interface with sw_if_index %d "
"in interface dump %s" %
(self.sw_if_index, moves.reprlib.repr(r)))
- self._local_ip6_ll = VppIpAddress(mk_ll_addr(self.local_mac))
+ self._local_ip6_ll = mk_ll_addr(self.local_mac)
self._remote_ip6_ll = mk_ll_addr(self.remote_mac)
def config_ip4(self):
"""Configure IPv4 address on the VPP interface."""
self.test.vapi.sw_interface_add_del_address(
- sw_if_index=self.sw_if_index, prefix=self._local_ip4.encode())
+ sw_if_index=self.sw_if_index, prefix=self.local_ip4_prefix)
self.has_ip4_config = True
def unconfig_ip4(self):
@@ -284,7 +285,7 @@
if self.has_ip4_config:
self.test.vapi.sw_interface_add_del_address(
sw_if_index=self.sw_if_index,
- prefix=self._local_ip4.encode(), is_add=0)
+ prefix=self.local_ip4_prefix, is_add=0)
except AttributeError:
self.has_ip4_config = False
self.has_ip4_config = False
@@ -302,7 +303,7 @@
def config_ip6(self):
"""Configure IPv6 address on the VPP interface."""
self.test.vapi.sw_interface_add_del_address(
- sw_if_index=self.sw_if_index, prefix=self._local_ip6.encode())
+ sw_if_index=self.sw_if_index, prefix=self.local_ip6_prefix)
self.has_ip6_config = True
def unconfig_ip6(self):
@@ -311,7 +312,7 @@
if self.has_ip6_config:
self.test.vapi.sw_interface_add_del_address(
sw_if_index=self.sw_if_index,
- prefix=self._local_ip6.encode(), is_add=0)
+ prefix=self.local_ip6_prefix, is_add=0)
except AttributeError:
self.has_ip6_config = False
self.has_ip6_config = False
diff --git a/test/vpp_ip.py b/test/vpp_ip.py
index 3bdfa62..8c3bbba 100644
--- a/test/vpp_ip.py
+++ b/test/vpp_ip.py
@@ -27,6 +27,13 @@
INVALID_INDEX = 0xffffffff
+def get_dpo_proto(addr):
+ if ip_address(addr).version == 6:
+ return DpoProto.DPO_PROTO_IP6
+ else:
+ return DpoProto.DPO_PROTO_IP4
+
+
class VppIpAddressUnion():
def __init__(self, addr):
self.addr = addr
@@ -73,148 +80,18 @@
return str(self.ip_addr)
-class VppIpAddress():
- def __init__(self, addr):
- self.addr = VppIpAddressUnion(addr)
-
- def encode(self):
- if self.addr.version == 6:
- return {
- 'af': VppEnum.vl_api_address_family_t.ADDRESS_IP6,
- 'un': self.addr.encode()
- }
- else:
- return {
- 'af': VppEnum.vl_api_address_family_t.ADDRESS_IP4,
- 'un': self.addr.encode()
- }
-
- def __eq__(self, other):
- if isinstance(other, self.__class__):
- return self.addr == other.addr
- elif hasattr(other, "af") and hasattr(other, "un"):
- # a vp_api_address_t
- if 4 == self.version:
- return other.af == \
- VppEnum.vl_api_address_family_t.ADDRESS_IP4 and \
- other.un == self.addr
- else:
- return other.af == \
- VppEnum.vl_api_address_family_t.ADDRESS_IP6 and \
- other.un == self.addr
- else:
- _log.error(
- "Comparing VppIpAddress:<%s> %s with incomparable "
- "type: <%s> %s",
- self.__class__.__name__, self,
- other.__class__.__name__, other)
- return NotImplemented
-
- def __ne__(self, other):
- return not (self == other)
-
- def __str__(self):
- return self.address
-
- @property
- def bytes(self):
- return self.addr.bytes
-
- @property
- def address(self):
- return self.addr.address
-
- @property
- def length(self):
- return self.addr.length
-
- @property
- def version(self):
- return self.addr.version
-
- @property
- def is_ip6(self):
- return (self.version == 6)
-
- @property
- def af(self):
- if self.version == 6:
- return AF_INET6
- else:
- return AF_INET
-
- @property
- def dpo_proto(self):
- if self.version == 6:
- return DpoProto.DPO_PROTO_IP6
- else:
- return DpoProto.DPO_PROTO_IP4
-
-
-class VppIpPrefix():
- def __init__(self, addr, len):
- self.addr = VppIpAddress(addr)
- self.len = len
-
- def __eq__(self, other):
- if self.address == other.address and self.len == other.len:
- return True
- return False
-
- def encode(self):
- return {'address': self.addr.encode(),
- 'len': self.len}
-
- @property
- def version(self):
- return self.addr.version
-
- @property
- def address(self):
- return self.addr.address
-
- @property
- def bytes(self):
- return self.addr.bytes
-
- @property
- def length(self):
- return self.len
-
- @property
- def is_ip6(self):
- return self.addr.is_ip6
-
- def __str__(self):
- return "%s/%d" % (self.address, self.length)
-
- def __eq__(self, other):
- if isinstance(other, self.__class__):
- return (self.len == other.len and self.addr == other.addr)
- elif hasattr(other, "address") and hasattr(other, "len"):
- # vl_api_prefix_t
- return self.len == other.len and \
- self.addr == other.address
- else:
- _log.error(
- "Comparing VppIpPrefix:%s with incomparable type: %s" %
- (self, other))
- return NotImplemented
-
-
class VppIpMPrefix():
def __init__(self, saddr, gaddr, glen):
self.saddr = saddr
self.gaddr = gaddr
self.glen = glen
- self.ip_saddr = VppIpAddressUnion(text_type(self.saddr))
- self.ip_gaddr = VppIpAddressUnion(text_type(self.gaddr))
- if self.ip_saddr.version != self.ip_gaddr.version:
+ if ip_address(self.saddr).version != \
+ ip_address(self.gaddr).version:
raise ValueError('Source and group addresses must be of the '
'same address family.')
def encode(self):
- if 6 == self.ip_saddr.version:
+ if 6 == self.version:
prefix = {
'af': VppEnum.vl_api_address_family_t.ADDRESS_IP6,
'grp_address': {
@@ -244,7 +121,7 @@
@property
def version(self):
- return self.ip_gaddr.version
+ return ip_address(self.gaddr).version
def __str__(self):
return "(%s,%s)/%d" % (self.saddr, self.gaddr, self.glen)
@@ -252,22 +129,20 @@
def __eq__(self, other):
if isinstance(other, self.__class__):
return (self.glen == other.glen and
- self.ip_saddr == other.ip_gaddr and
- self.ip_saddr == other.ip_saddr)
+ self.saddr == other.gaddr and
+ self.saddr == other.saddr)
elif (hasattr(other, "grp_address_length") and
hasattr(other, "grp_address") and
hasattr(other, "src_address")):
# vl_api_mprefix_t
- if 4 == self.ip_saddr.version:
- if self.glen == other.grp_address_length and \
- self.gaddr == str(other.grp_address.ip4) and \
- self.saddr == str(other.src_address.ip4):
- return True
- return False
+ if 4 == self.version:
+ return (self.glen == other.grp_address_length and
+ self.gaddr == str(other.grp_address.ip4) and
+ self.saddr == str(other.src_address.ip4))
else:
return (self.glen == other.grp_address_length and
- self.gaddr == other.grp_address.ip6 and
- self.saddr == other.src_address.ip6)
+ self.gaddr == str(other.grp_address.ip6) and
+ self.saddr == str(other.src_address.ip6))
else:
raise Exception("Comparing VppIpPrefix:%s with unknown type: %s" %
(self, other))
diff --git a/test/vpp_ip_route.py b/test/vpp_ip_route.py
index 8f9d51a..d600475 100644
--- a/test/vpp_ip_route.py
+++ b/test/vpp_ip_route.py
@@ -6,7 +6,7 @@
from vpp_object import VppObject
from socket import inet_pton, inet_ntop, AF_INET, AF_INET6
-from vpp_ip import DpoProto, VppIpPrefix, INVALID_INDEX, VppIpAddressUnion, \
+from vpp_ip import DpoProto, INVALID_INDEX, VppIpAddressUnion, \
VppIpMPrefix
from ipaddress import ip_address, IPv4Network, IPv6Network
@@ -72,6 +72,13 @@
UNIFORM = 1
+def mk_network(addr, len):
+ if ip_address(text_type(addr)).version == 4:
+ return IPv4Network("%s/%d" % (addr, len), strict=False)
+ else:
+ return IPv6Network("%s/%d" % (addr, len), strict=False)
+
+
def ip_to_dpo_proto(addr):
if addr.version == 6:
return DpoProto.DPO_PROTO_IP6
@@ -87,18 +94,16 @@
def find_route(test, addr, len, table_id=0):
- ip_addr = ip_address(text_type(addr))
+ prefix = mk_network(addr, len)
- if 4 is ip_addr.version:
+ if 4 is prefix.version:
routes = test.vapi.ip_route_dump(table_id, False)
- prefix = IPv4Network("%s/%d" % (text_type(addr), len), strict=False)
else:
routes = test.vapi.ip_route_dump(table_id, True)
- prefix = IPv6Network("%s/%d" % (text_type(addr), len), strict=False)
for e in routes:
if table_id == e.route.table_id \
- and prefix == e.route.prefix:
+ and str(e.route.prefix) == str(prefix):
return True
return False
@@ -138,22 +143,16 @@
return False
-def fib_interface_ip_prefix(test, address, length, sw_if_index):
- ip_addr = ip_address(text_type(address))
+def fib_interface_ip_prefix(test, addr, len, sw_if_index):
+ # can't use python net here since we need the host bits in the prefix
+ prefix = "%s/%d" % (addr, len)
+ addrs = test.vapi.ip_address_dump(
+ sw_if_index,
+ is_ipv6=(6 == ip_address(addr).version))
- if 4 is ip_addr.version:
- addrs = test.vapi.ip_address_dump(sw_if_index)
- prefix = IPv4Network("%s/%d" % (text_type(address), length),
- strict=False)
- else:
- addrs = test.vapi.ip_address_dump(sw_if_index, is_ipv6=1)
- prefix = IPv6Network("%s/%d" % (text_type(address), length),
- strict=False)
-
- # TODO: refactor this to VppIpPrefix.__eq__
for a in addrs:
if a.sw_if_index == sw_if_index and \
- a.prefix.network == prefix:
+ str(a.prefix) == prefix:
return True
return False
@@ -198,23 +197,25 @@
def __init__(self, test, intf, addr, len):
self._test = test
self.intf = intf
- self.prefix = VppIpPrefix(addr, len)
+ self.addr = addr
+ self.len = len
+ self.prefix = "%s/%d" % (addr, len)
def add_vpp_config(self):
self._test.vapi.sw_interface_add_del_address(
- sw_if_index=self.intf.sw_if_index, prefix=self.prefix.encode(),
+ sw_if_index=self.intf.sw_if_index, prefix=self.prefix,
is_add=1)
self._test.registry.register(self, self._test.logger)
def remove_vpp_config(self):
self._test.vapi.sw_interface_add_del_address(
- sw_if_index=self.intf.sw_if_index, prefix=self.prefix.encode(),
+ sw_if_index=self.intf.sw_if_index, prefix=self.prefix,
is_add=0)
def query_vpp_config(self):
return fib_interface_ip_prefix(self._test,
- self.prefix.address,
- self.prefix.length,
+ self.addr,
+ self.len,
self.intf.sw_if_index)
def object_id(self):
@@ -424,7 +425,7 @@
self._test = test
self.paths = paths
self.table_id = table_id
- self.prefix = VppIpPrefix(dest_addr, dest_addr_len)
+ self.prefix = mk_network(dest_addr, dest_addr_len)
self.register = register
self.stats_index = None
self.modified = False
@@ -447,7 +448,7 @@
self.modified = True
self._test.vapi.ip_route_add_del(route={'table_id': self.table_id,
- 'prefix': self.prefix.encode(),
+ 'prefix': self.prefix,
'n_paths': len(
self.encoded_paths),
'paths': self.encoded_paths,
@@ -458,7 +459,7 @@
def add_vpp_config(self):
r = self._test.vapi.ip_route_add_del(
route={'table_id': self.table_id,
- 'prefix': self.prefix.encode(),
+ 'prefix': self.prefix,
'n_paths': len(self.encoded_paths),
'paths': self.encoded_paths,
},
@@ -476,7 +477,7 @@
if self.modified:
self._test.vapi.ip_route_add_del(
route={'table_id': self.table_id,
- 'prefix': self.prefix.encode(),
+ 'prefix': self.prefix,
'n_paths': len(
self.encoded_paths),
'paths': self.encoded_paths},
@@ -485,23 +486,22 @@
else:
self._test.vapi.ip_route_add_del(
route={'table_id': self.table_id,
- 'prefix': self.prefix.encode(),
+ 'prefix': self.prefix,
'n_paths': 0},
is_add=0,
is_multipath=0)
def query_vpp_config(self):
return find_route(self._test,
- self.prefix.address,
- self.prefix.len,
+ self.prefix.network_address,
+ self.prefix.prefixlen,
self.table_id)
def object_id(self):
- return ("%s:table-%d-%s/%d" % (
- 'ip6-route' if self.prefix.addr.version == 6 else 'ip-route',
+ return ("%s:table-%d-%s" % (
+ 'ip6-route' if self.prefix.version == 6 else 'ip-route',
self.table_id,
- self.prefix.address,
- self.prefix.len))
+ self.prefix))
def get_stats_to(self):
c = self._test.statistics.get_counter("/net/route/to")
@@ -627,18 +627,18 @@
self.local_label = local_label
self.table_id = table_id
self.ip_table_id = ip_table_id
- self.prefix = VppIpPrefix(dest_addr, dest_addr_len)
+ self.prefix = mk_network(dest_addr, dest_addr_len)
def add_vpp_config(self):
self._test.vapi.mpls_ip_bind_unbind(self.local_label,
- self.prefix.encode(),
+ self.prefix,
table_id=self.table_id,
ip_table_id=self.ip_table_id)
self._test.registry.register(self, self._test.logger)
def remove_vpp_config(self):
self._test.vapi.mpls_ip_bind_unbind(self.local_label,
- self.prefix.encode(),
+ self.prefix,
table_id=self.table_id,
ip_table_id=self.ip_table_id,
is_bind=0)
diff --git a/test/vpp_l2.py b/test/vpp_l2.py
index 3ee0d35..114b1c7 100644
--- a/test/vpp_l2.py
+++ b/test/vpp_l2.py
@@ -4,7 +4,6 @@
"""
from vpp_object import VppObject
-from vpp_ip import VppIpAddress
from vpp_lo_interface import VppLoInterface
from vpp_papi import MACAddress
from vpp_sub_interface import L2_VTR_OP
diff --git a/test/vpp_papi_provider.py b/test/vpp_papi_provider.py
index cc7771e..a440312 100644
--- a/test/vpp_papi_provider.py
+++ b/test/vpp_papi_provider.py
@@ -1092,8 +1092,7 @@
def vxlan_gbp_tunnel_dump(self, sw_if_index=0xffffffff):
return self.api(self.papi.vxlan_gbp_tunnel_dump,
- {'sw_if_index': sw_if_index,
- '_no_type_conversion': True})
+ {'sw_if_index': sw_if_index})
def pppoe_add_del_session(
self,
@@ -1802,8 +1801,7 @@
def gbp_endpoint_dump(self):
""" GBP endpoint Dump """
- return self.api(self.papi.gbp_endpoint_dump,
- {'_no_type_conversion': True})
+ return self.api(self.papi.gbp_endpoint_dump, {})
def gbp_endpoint_group_add(self, vnid, sclass, bd,
rd, uplink_sw_if_index,
@@ -1912,8 +1910,7 @@
def gbp_subnet_dump(self):
""" GBP Subnet Dump """
- return self.api(self.papi.gbp_subnet_dump,
- {'_no_type_conversion': True})
+ return self.api(self.papi.gbp_subnet_dump, {})
def gbp_contract_dump(self):
""" GBP contract Dump """
diff --git a/test/vpp_vxlan_gbp_tunnel.py b/test/vpp_vxlan_gbp_tunnel.py
index 5352d6e..2255585 100644
--- a/test/vpp_vxlan_gbp_tunnel.py
+++ b/test/vpp_vxlan_gbp_tunnel.py
@@ -1,6 +1,5 @@
from vpp_interface import VppInterface
-from vpp_ip import VppIpAddress
from vpp_papi import VppEnum
@@ -8,13 +7,10 @@
def find_vxlan_gbp_tunnel(test, src, dst, vni):
- vsrc = VppIpAddress(src)
- vdst = VppIpAddress(dst)
-
ts = test.vapi.vxlan_gbp_tunnel_dump(INDEX_INVALID)
for t in ts:
- if vsrc == t.tunnel.src and \
- vdst == t.tunnel.dst and \
+ if src == str(t.tunnel.src) and \
+ dst == str(t.tunnel.dst) and \
t.tunnel.vni == vni:
return t.tunnel.sw_if_index
return INDEX_INVALID
@@ -29,8 +25,8 @@
is_ipv6=None, encap_table_id=None, instance=0xffffffff):
""" Create VXLAN-GBP Tunnel interface """
super(VppVxlanGbpTunnel, self).__init__(test)
- self.src = VppIpAddress(src)
- self.dst = VppIpAddress(dst)
+ self.src = src
+ self.dst = dst
self.vni = vni
self.mcast_itf = mcast_itf
self.ipv6 = is_ipv6
@@ -49,8 +45,8 @@
reply = self.test.vapi.vxlan_gbp_tunnel_add_del(
is_add=1,
tunnel={
- 'src': self.src.encode(),
- 'dst': self.dst.encode(),
+ 'src': self.src,
+ 'dst': self.dst,
'mode': self.mode,
'vni': self.vni,
'mcast_sw_if_index': mcast_sw_if_index,
@@ -67,8 +63,8 @@
self.test.vapi.vxlan_gbp_tunnel_add_del(
is_add=0,
tunnel={
- 'src': self.src.encode(),
- 'dst': self.dst.encode(),
+ 'src': self.src,
+ 'dst': self.dst,
'mode': self.mode,
'vni': self.vni,
'mcast_sw_if_index': mcast_sw_if_index,