Code Review
/
csit.git
/ blobdiff
commit
grep
author
committer
pickaxe
?
search:
re
summary
|
shortlog
|
log
|
commit
|
commitdiff
|
review
|
tree
raw
|
inline
| side by side
FIX: Mellanox jumbo frames
[csit.git]
/
resources
/
libraries
/
python
/
SRv6.py
diff --git
a/resources/libraries/python/SRv6.py
b/resources/libraries/python/SRv6.py
index
f0a2eaf
..
86dd002
100644
(file)
--- a/
resources/libraries/python/SRv6.py
+++ b/
resources/libraries/python/SRv6.py
@@
-1,4
+1,4
@@
-# Copyright (c) 20
19
Cisco and/or its affiliates.
+# Copyright (c) 20
20
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:
# 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.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
from resources.libraries.python.PapiExecutor import PapiSocketExecutor
@@
-47,7
+49,7
@@
class SRv6Behavior(IntEnum):
class SRv6PolicySteeringTypes(IntEnum):
class SRv6PolicySteeringTypes(IntEnum):
- """SRv6
LocalSID supported function
s."""
+ """SRv6
steering type
s."""
SR_STEER_L2 = 2
SR_STEER_IPV4 = 4
SR_STEER_IPV6 = 6
SR_STEER_L2 = 2
SR_STEER_IPV4 = 4
SR_STEER_IPV6 = 6
@@
-56,19
+58,6
@@
class SRv6PolicySteeringTypes(IntEnum):
class SRv6:
"""SRv6 class."""
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.
@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.
:param sids: SID IPv6 addresses.
:type sids: list
:returns: SRv6 SID list object.
- :rtype:
lis
t
+ :rtype:
dic
t
"""
"""
- 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(
return dict(
- num_sids=len(sid
s
),
+ num_sids=len(sid
_list
),
weight=1,
weight=1,
- sids=sid_list
+ sids=sid_list
+ (16 - len(sid_list)) * [IPv6Address(0).packed]
)
@staticmethod
)
@staticmethod
@@
-167,15
+153,14
@@
class SRv6:
cmd = u"sr_localsid_add_del"
args = dict(
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,
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']}"
)
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
)
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(
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"
: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"):
@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,
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} " \
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"
: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(
@staticmethod
def _get_sr_steer_policy_args(
@@
-287,8
+265,7
@@
class SRv6:
:type interface: str
:type ip_addr: str
:type prefix: int
: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.
: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)
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
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)
)
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(
+ prefix = IPUtil.create_prefix_object(ip_addr, int(prefix))
+ traffic_type = getattr(
SRv6PolicySteeringTypes, u"SR_STEER_IPV4"
SRv6PolicySteeringTypes, u"SR_STEER_IPV4"
- ).value
- else:
- traffic_type = getattr(
+ ).value if ip_addr.version == 4 else getattr(
SRv6PolicySteeringTypes, u"SR_STEER_IPV6"
).value
else:
raise ValueError(f"Unsupported mode: {mode}")
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
# 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.
"""
: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(
+ 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(
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,
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
)
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"
: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):
@staticmethod
def set_sr_encaps_source_address(node, ip6_addr):