aboutsummaryrefslogtreecommitdiffstats
path: root/resources/libraries/python
diff options
context:
space:
mode:
Diffstat (limited to 'resources/libraries/python')
-rw-r--r--resources/libraries/python/GBP.py12
-rw-r--r--resources/libraries/python/IPAddress.py55
-rw-r--r--resources/libraries/python/IPUtil.py40
-rw-r--r--resources/libraries/python/IPsecUtil.py33
-rw-r--r--resources/libraries/python/TestConfig.py7
5 files changed, 85 insertions, 62 deletions
diff --git a/resources/libraries/python/GBP.py b/resources/libraries/python/GBP.py
index f0d07a12b7..9d56d4475b 100644
--- a/resources/libraries/python/GBP.py
+++ b/resources/libraries/python/GBP.py
@@ -1,4 +1,4 @@
-# Copyright (c) 2019 Cisco and/or its affiliates.
+# Copyright (c) 2020 Cisco and/or its affiliates.
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at:
@@ -17,7 +17,7 @@ from enum import IntEnum
from ipaddress import ip_address
-from resources.libraries.python.IPUtil import IPUtil
+from resources.libraries.python.IPAddress import IPAddress
from resources.libraries.python.L2Util import L2Util
from resources.libraries.python.PapiExecutor import PapiSocketExecutor
@@ -205,9 +205,9 @@ class GBP:
err_msg = f"Failed to add GBP endpoint on {node[u'host']}!"
ips = list()
- ips.append(IPUtil.create_ip_address_object(ip_address(ip_addr)))
- tun_src = IPUtil.create_ip_address_object(ip_address(u"0.0.0.0"))
- tun_dst = IPUtil.create_ip_address_object(ip_address(u"0.0.0.0"))
+ ips.append(IPAddress.create_ip_address_object(ip_address(ip_addr)))
+ tun_src = IPAddress.create_ip_address_object(ip_address(u"0.0.0.0"))
+ tun_dst = IPAddress.create_ip_address_object(ip_address(u"0.0.0.0"))
args_in = dict(
endpoint=dict(
@@ -287,7 +287,7 @@ class GBP:
sw_if_index=sw_if_index,
sclass=sclass,
prefix=dict(
- address=IPUtil.create_ip_address_object(
+ address=IPAddress.create_ip_address_object(
ip_address(address)
),
len=int(subnet_length)
diff --git a/resources/libraries/python/IPAddress.py b/resources/libraries/python/IPAddress.py
new file mode 100644
index 0000000000..b8a4d7443d
--- /dev/null
+++ b/resources/libraries/python/IPAddress.py
@@ -0,0 +1,55 @@
+# Copyright (c) 2020 Cisco and/or its affiliates.
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at:
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Common IP utilities library."""
+
+from enum import IntEnum
+
+
+class AddressFamily(IntEnum):
+ """IP address family."""
+ ADDRESS_IP4 = 0
+ ADDRESS_IP6 = 1
+
+
+class IPAddress:
+ """Common IP address utilities"""
+
+ @staticmethod
+ def union_addr(ip_addr):
+ """Creates union IP address.
+
+ :param ip_addr: IPv4 or IPv6 address.
+ :type ip_addr: IPv4Address or IPv6Address
+ :returns: Union IP address.
+ :rtype: dict
+ """
+ return dict(ip6=ip_addr.packed) if ip_addr.version == 6 \
+ else dict(ip4=ip_addr.packed)
+
+ @staticmethod
+ def create_ip_address_object(ip_addr):
+ """Create IP address object.
+
+ :param ip_addr: IPv4 or IPv6 address
+ :type ip_addr: IPv4Address or IPv6Address
+ :returns: IP address object.
+ :rtype: dict
+ """
+ return dict(
+ af=getattr(
+ AddressFamily, u"ADDRESS_IP6" if ip_addr.version == 6
+ else u"ADDRESS_IP4"
+ ).value,
+ un=IPAddress.union_addr(ip_addr)
+ )
diff --git a/resources/libraries/python/IPUtil.py b/resources/libraries/python/IPUtil.py
index 1e0d90f065..e91cd7df82 100644
--- a/resources/libraries/python/IPUtil.py
+++ b/resources/libraries/python/IPUtil.py
@@ -20,6 +20,7 @@ from ipaddress import ip_address
from resources.libraries.python.Constants import Constants
from resources.libraries.python.InterfaceUtil import InterfaceUtil
+from resources.libraries.python.IPAddress import IPAddress
from resources.libraries.python.PapiExecutor import PapiSocketExecutor
from resources.libraries.python.ssh import exec_cmd_no_error, exec_cmd
from resources.libraries.python.topology import Topology
@@ -32,12 +33,6 @@ MPLS_IETF_MAX_LABEL = 0xfffff
MPLS_LABEL_INVALID = MPLS_IETF_MAX_LABEL + 1
-class AddressFamily(IntEnum):
- """IP address family."""
- ADDRESS_IP4 = 0
- ADDRESS_IP6 = 1
-
-
class FibPathType(IntEnum):
"""FIB path types."""
FIB_PATH_TYPE_NORMAL = 0
@@ -513,35 +508,6 @@ class IPUtil:
papi_exec.add(cmd, **args).get_reply(err_msg)
@staticmethod
- def union_addr(ip_addr):
- """Creates union IP address.
-
- :param ip_addr: IPv4 or IPv6 address.
- :type ip_addr: IPv4Address or IPv6Address
- :returns: Union IP address.
- :rtype: dict
- """
- return dict(ip6=ip_addr.packed) if ip_addr.version == 6 \
- else dict(ip4=ip_addr.packed)
-
- @staticmethod
- def create_ip_address_object(ip_addr):
- """Create IP address object.
-
- :param ip_addr: IPv4 or IPv6 address
- :type ip_addr: IPv4Address or IPv6Address
- :returns: IP address object.
- :rtype: dict
- """
- return dict(
- af=getattr(
- AddressFamily, u"ADDRESS_IP6" if ip_addr.version == 6
- else u"ADDRESS_IP4"
- ).value,
- un=IPUtil.union_addr(ip_addr)
- )
-
- @staticmethod
def create_prefix_object(ip_addr, addr_len):
"""Create prefix object.
@@ -552,7 +518,7 @@ class IPUtil:
:returns: Prefix object.
:rtype: dict
"""
- addr = IPUtil.create_ip_address_object(ip_addr)
+ addr = IPAddress.create_ip_address_object(ip_addr)
return dict(
len=int(addr_len),
@@ -594,7 +560,7 @@ class IPUtil:
paths = list()
n_hop = dict(
- address=IPUtil.union_addr(ip_address(gateway)) if gateway else 0,
+ address=IPAddress.union_addr(ip_address(gateway)) if gateway else 0,
via_label=MPLS_LABEL_INVALID,
obj_id=Constants.BITWISE_NON_ZERO
)
diff --git a/resources/libraries/python/IPsecUtil.py b/resources/libraries/python/IPsecUtil.py
index 7cbfea1545..2e6574f40e 100644
--- a/resources/libraries/python/IPsecUtil.py
+++ b/resources/libraries/python/IPsecUtil.py
@@ -22,9 +22,10 @@ from string import ascii_letters
from ipaddress import ip_network, ip_address
-from resources.libraries.python.IPUtil import IPUtil
from resources.libraries.python.InterfaceUtil import InterfaceUtil, \
InterfaceStatusFlags
+from resources.libraries.python.IPAddress import IPAddress
+from resources.libraries.python.IPUtil import IPUtil
from resources.libraries.python.PapiExecutor import PapiSocketExecutor
from resources.libraries.python.ssh import scp_node
from resources.libraries.python.topology import Topology
@@ -632,16 +633,16 @@ class IPsecUtil:
sa_id=int(sa_id) if sa_id else 0,
policy=action.policy_int_repr,
protocol=int(proto) if proto else 0,
- remote_address_start=IPUtil.create_ip_address_object(
+ remote_address_start=IPAddress.create_ip_address_object(
ip_network(raddr_range, strict=False).network_address
),
- remote_address_stop=IPUtil.create_ip_address_object(
+ remote_address_stop=IPAddress.create_ip_address_object(
ip_network(raddr_range, strict=False).broadcast_address
),
- local_address_start=IPUtil.create_ip_address_object(
+ local_address_start=IPAddress.create_ip_address_object(
ip_network(laddr_range, strict=False).network_address
),
- local_address_stop=IPUtil.create_ip_address_object(
+ local_address_stop=IPAddress.create_ip_address_object(
ip_network(laddr_range, strict=False).broadcast_address
),
remote_port_start=int(rport_range.split(u"-")[0]) if rport_range
@@ -722,12 +723,12 @@ class IPsecUtil:
sa_id=int(sa_id) if sa_id else 0,
policy=IPsecUtil.policy_action_protect().policy_int_repr,
protocol=0,
- remote_address_start=IPUtil.create_ip_address_object(raddr_ip),
- remote_address_stop=IPUtil.create_ip_address_object(raddr_ip),
- local_address_start=IPUtil.create_ip_address_object(
+ remote_address_start=IPAddress.create_ip_address_object(raddr_ip),
+ remote_address_stop=IPAddress.create_ip_address_object(raddr_ip),
+ local_address_start=IPAddress.create_ip_address_object(
ip_network(laddr_range, strict=False).network_address
),
- local_address_stop=IPUtil.create_ip_address_object(
+ local_address_stop=IPAddress.create_ip_address_object(
ip_network(laddr_range, strict=False).broadcast_address
),
remote_port_start=0,
@@ -743,9 +744,9 @@ class IPsecUtil:
with PapiSocketExecutor(node) as papi_exec:
for i in range(n_entries):
args[u"entry"][u"remote_address_start"][u"un"] = \
- IPUtil.union_addr(raddr_ip + i)
+ IPAddress.union_addr(raddr_ip + i)
args[u"entry"][u"remote_address_stop"][u"un"] = \
- IPUtil.union_addr(raddr_ip + i)
+ IPAddress.union_addr(raddr_ip + i)
history = bool(not 1 < i < n_entries - 2)
papi_exec.add(cmd, history=history, **args)
papi_exec.get_replies(err_msg)
@@ -1001,10 +1002,10 @@ class IPsecUtil:
)
args2[u"local_spi"] = spi_1 + i
args2[u"remote_spi"] = spi_2 + i
- args2[u"local_ip"] = IPUtil.create_ip_address_object(
+ args2[u"local_ip"] = IPAddress.create_ip_address_object(
if1_ip + i * addr_incr
)
- args2[u"remote_ip"] = IPUtil.create_ip_address_object(if2_ip)
+ args2[u"remote_ip"] = IPAddress.create_ip_address_object(if2_ip)
args2[u"local_crypto_key_len"] = len(ckeys[i])
args2[u"local_crypto_key"] = ckeys[i]
args2[u"remote_crypto_key_len"] = len(ckeys[i])
@@ -1078,7 +1079,7 @@ class IPsecUtil:
cmd2 = u"ipsec_tunnel_if_add_del"
args2 = dict(
is_add=True,
- local_ip=IPUtil.create_ip_address_object(if2_ip),
+ local_ip=IPAddress.create_ip_address_object(if2_ip),
remote_ip=None,
local_spi=0,
remote_spi=0,
@@ -1100,8 +1101,8 @@ class IPsecUtil:
for i in range(existing_tunnels, n_tunnels):
args2[u"local_spi"] = spi_2 + i
args2[u"remote_spi"] = spi_1 + i
- args2[u"local_ip"] = IPUtil.create_ip_address_object(if2_ip)
- args2[u"remote_ip"] = IPUtil.create_ip_address_object(
+ args2[u"local_ip"] = IPAddress.create_ip_address_object(if2_ip)
+ args2[u"remote_ip"] = IPAddress.create_ip_address_object(
if1_ip + i * addr_incr)
args2[u"local_crypto_key_len"] = len(ckeys[i])
args2[u"local_crypto_key"] = ckeys[i]
diff --git a/resources/libraries/python/TestConfig.py b/resources/libraries/python/TestConfig.py
index 28778b2c31..6366f2e3fb 100644
--- a/resources/libraries/python/TestConfig.py
+++ b/resources/libraries/python/TestConfig.py
@@ -1,4 +1,4 @@
-# Copyright (c) 2019 Cisco and/or its affiliates.
+# Copyright (c) 2020 Cisco and/or its affiliates.
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at:
@@ -19,6 +19,7 @@ from robot.api import logger
from resources.libraries.python.Constants import Constants
from resources.libraries.python.InterfaceUtil import InterfaceUtil, \
InterfaceStatusFlags
+from resources.libraries.python.IPAddress import IPAddress
from resources.libraries.python.IPUtil import IPUtil
from resources.libraries.python.PapiExecutor import PapiSocketExecutor
from resources.libraries.python.topology import Topology
@@ -423,9 +424,9 @@ class TestConfig:
args1[u"neighbor"][u"ip_address"] = \
str(dst_ip_start + i * ip_step)
args2[u"route"][u"prefix"][u"address"][u"un"] = \
- IPUtil.union_addr(dst_ip_start + i * ip_step)
+ IPAddress.union_addr(dst_ip_start + i * ip_step)
args2[u"route"][u"paths"][0][u"nh"][u"address"] = \
- IPUtil.union_addr(dst_ip_start + i * ip_step)
+ IPAddress.union_addr(dst_ip_start + i * ip_step)
args3[u"rx_sw_if_index"] = Topology.get_interface_sw_index(
node, f"vxlan_tunnel{i+1}"
)