X-Git-Url: https://gerrit.fd.io/r/gitweb?p=csit.git;a=blobdiff_plain;f=resources%2Flibraries%2Fpython%2FSRv6.py;h=eca22a2b1744457bd9b4258a9996f38ab3d1cc31;hp=f0a2eaf7d498c9a14833f05f7dc799abc8015759;hb=b6606e7625e308a66bdfb9d5a9c065b58e429a99;hpb=f6eacaa97848720c5fbb7949ca35c0d2ba49160b diff --git a/resources/libraries/python/SRv6.py b/resources/libraries/python/SRv6.py index f0a2eaf7d4..eca22a2b17 100644 --- a/resources/libraries/python/SRv6.py +++ b/resources/libraries/python/SRv6.py @@ -1,4 +1,4 @@ -# Copyright (c) 2019 Cisco and/or its affiliates. +# Copyright (c) 2021 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,8 @@ from ipaddress import ip_address, IPv6Address 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.IPUtil import IPUtil from resources.libraries.python.PapiExecutor import PapiSocketExecutor @@ -47,7 +49,7 @@ class SRv6Behavior(IntEnum): class SRv6PolicySteeringTypes(IntEnum): - """SRv6 LocalSID supported functions.""" + """SRv6 steering types.""" SR_STEER_L2 = 2 SR_STEER_IPV4 = 4 SR_STEER_IPV6 = 6 @@ -56,19 +58,6 @@ class SRv6PolicySteeringTypes(IntEnum): class SRv6: """SRv6 class.""" - @staticmethod - def create_srv6_sid_object(ip_addr): - """Create SRv6 SID object. - - :param ip_addr: IPv6 address. - :type ip_addr: str - :returns: SRv6 SID object. - :rtype: dict - """ - return dict( - addr=IPv6Address(ip_addr).packed - ) - @staticmethod def create_srv6_sid_list(sids): """Create SRv6 SID list object. @@ -76,17 +65,14 @@ class SRv6: :param sids: SID IPv6 addresses. :type sids: list :returns: SRv6 SID list object. - :rtype: list + :rtype: dict """ - sid_list = [0] * 16 - - for index, item in enumerate(sids): - sid_list[index] = SRv6.create_srv6_sid_object(item) + sid_list = [IPv6Address(sid).packed for sid in sids] return dict( - num_sids=len(sids), + num_sids=len(sid_list), weight=1, - sids=sid_list + sids=sid_list + (16 - len(sid_list)) * [IPv6Address(0).packed] ) @staticmethod @@ -167,15 +153,14 @@ class SRv6: cmd = u"sr_localsid_add_del" args = dict( - is_del=0, - localsid=SRv6.create_srv6_sid_object(local_sid), - end_psp=0, + is_del=False, + localsid=IPv6Address(local_sid).packed, + end_psp=False, behavior=getattr(SRv6Behavior, beh).value, sw_if_index=Constants.BITWISE_NON_ZERO, vlan_index=0, fib_table=0, - nh_addr6=0, - nh_addr4=0 + nh_addr=0 ) err_msg = f"Failed to add SR localSID {local_sid} " \ f"host {node[u'host']}" @@ -191,11 +176,9 @@ class SRv6: args[u"sw_if_index"] = InterfaceUtil.get_interface_index( node, interface ) - next_hop = ip_address(next_hop) - if next_hop.version == 6: - args[u"nh_addr6"] = next_hop.packed - else: - args[u"nh_addr4"] = next_hop.packed + args[u"nh_addr"] = IPAddress.create_ip_address_object( + ip_address(next_hop) + ) elif beh == getattr(SRv6Behavior, u"END_DX2").name: if interface is None: raise ValueError( @@ -224,10 +207,7 @@ class SRv6: :type node: dict """ cmd = u"sr_localsids_dump" - err_msg = f"Failed to get SR localSID dump on host {node[u'host']}" - - with PapiSocketExecutor(node) as papi_exec: - papi_exec.add(cmd).get_details(err_msg) + PapiSocketExecutor.dump_and_log(node, (cmd,)) @staticmethod def configure_sr_policy(node, bsid, sid_list, mode=u"encap"): @@ -246,7 +226,8 @@ class SRv6: args = dict( bsid_addr=IPv6Address(bsid).packed, weight=1, - is_encap=1 if mode == u"encap" else 0, + is_encap=bool(mode == u"encap"), + is_spray=False, sids=SRv6.create_srv6_sid_list(sid_list) ) err_msg = f"Failed to add SR policy for BindingSID {bsid} " \ @@ -263,10 +244,7 @@ class SRv6: :type node: dict """ cmd = u"sr_policies_dump" - err_msg = f"Failed to get SR policies dump on host {node[u'host']}" - - with PapiSocketExecutor(node) as papi_exec: - papi_exec.add(cmd).get_details(err_msg) + PapiSocketExecutor.dump_and_log(node, (cmd,)) @staticmethod def _get_sr_steer_policy_args( @@ -287,8 +265,7 @@ class SRv6: :type interface: str :type ip_addr: str :type prefix: int - :returns: Values for sw_if_index, mask_width, prefix_addr and - traffic_type + :returns: Values for sw_if_index, prefix and traffic_type :rtype: tuple :raises ValueError: If unsupported mode used or required parameter is missing. @@ -300,8 +277,7 @@ class SRv6: f"interface: {interface}" ) sw_if_index = InterfaceUtil.get_interface_index(node, interface) - mask_width = 0 - prefix_addr = 16 * b"\0" + prefix = 0 traffic_type = getattr( SRv6PolicySteeringTypes, u"SR_STEER_L2" ).value @@ -314,21 +290,16 @@ class SRv6: ) sw_if_index = Constants.BITWISE_NON_ZERO ip_addr = ip_address(ip_addr) - prefix_addr = ip_addr.packed - mask_width = int(prefix) - if ip_addr.version == 4: - prefix_addr += 12 * b"\0" - traffic_type = getattr( - SRv6PolicySteeringTypes, u"SR_STEER_IPV4" - ).value - else: - traffic_type = getattr( - SRv6PolicySteeringTypes, u"SR_STEER_IPV6" - ).value + prefix = IPUtil.create_prefix_object(ip_addr, int(prefix)) + traffic_type = getattr( + SRv6PolicySteeringTypes, u"SR_STEER_IPV4" + ).value if ip_addr.version == 4 else getattr( + SRv6PolicySteeringTypes, u"SR_STEER_IPV6" + ).value else: raise ValueError(f"Unsupported mode: {mode}") - return sw_if_index, mask_width, prefix_addr, traffic_type + return sw_if_index, prefix, traffic_type # TODO: Bring L1 names, arguments and defaults closer to PAPI ones. @staticmethod @@ -354,19 +325,17 @@ class SRv6: :raises ValueError: If unsupported mode used or required parameter is missing. """ - sw_if_index, mask_width, prefix_addr, traffic_type = \ - SRv6._get_sr_steer_policy_args( - node, mode, interface, ip_addr, prefix - ) + sw_if_index, prefix, traffic_type = SRv6._get_sr_steer_policy_args( + node, mode, interface, ip_addr, prefix + ) cmd = u"sr_steering_add_del" args = dict( - is_del=0, + is_del=False, bsid_addr=IPv6Address(str(bsid)).packed, sr_policy_index=0, table_id=0, - prefix_addr=prefix_addr, - mask_width=mask_width, + prefix=prefix, sw_if_index=sw_if_index, traffic_type=traffic_type ) @@ -384,10 +353,7 @@ class SRv6: :type node: dict """ cmd = u"sr_steering_pol_dump" - err_msg = f"Failed to get SR localSID dump on host {node[u'host']}" - - with PapiSocketExecutor(node) as papi_exec: - papi_exec.add(cmd).get_details(err_msg) + PapiSocketExecutor.dump_and_log(node, (cmd,)) @staticmethod def set_sr_encaps_source_address(node, ip6_addr):