-# 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:
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
class SRv6PolicySteeringTypes(IntEnum):
- """SRv6 LocalSID supported functions."""
+ """SRv6 steering types."""
SR_STEER_L2 = 2
SR_STEER_IPV4 = 4
SR_STEER_IPV6 = 6
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.
: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
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']}"
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(
: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"):
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} " \
: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(
: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.
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
)
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
: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
)
: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):
Code Review / csit.git / blobdiff