diff options
| -rw-r--r-- | resources/libraries/python/GBP.py | 12 | ||||
| -rw-r--r-- | resources/libraries/python/IPAddress.py | 55 | ||||
| -rw-r--r-- | resources/libraries/python/IPUtil.py | 40 | ||||
| -rw-r--r-- | resources/libraries/python/IPsecUtil.py | 33 | ||||
| -rw-r--r-- | resources/libraries/python/TestConfig.py | 7 |
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}" ) |