"""GBP utilities library."""
from enum import IntEnum
+
from ipaddress import ip_address
from resources.libraries.python.IPUtil import IPUtil
from resources.libraries.python.L2Util import L2Util
from resources.libraries.python.PapiExecutor import PapiSocketExecutor
-from resources.libraries.python.topology import Topology
class GBPEndpointFlags(IntEnum):
class GBPSubnetType(IntEnum):
"""GBP Subnet Type."""
GBP_API_SUBNET_TRANSPORT = 1
- GBP_API_SUBNET_STITCHED_INTERNAL = 2
- GBP_API_SUBNET_STITCHED_EXTERNAL = 3
+ GBP_API_SUBNET_STITCHED_INTERNAL = 2 # pylint: disable=invalid-name
+ GBP_API_SUBNET_STITCHED_EXTERNAL = 3 # pylint: disable=invalid-name
GBP_API_SUBNET_L3_OUT = 4
GBP_API_SUBNET_ANON_L3_OUT = 5
GBP_API_RULE_REDIRECT = 3
-class GBP(object):
+class GBPHashMode(IntEnum):
+ """GBP Hash Mode."""
+ GBP_API_HASH_MODE_SRC_IP = 1
+ GBP_API_HASH_MODE_DST_IP = 2
+ GBP_API_HASH_MODE_SYMETRIC = 3
+
+
+class GBP:
"""GBP utilities."""
@staticmethod
:type ip4_uu_sw_if_index: int
:type ip6_uu_sw_if_index: int
"""
- cmd = 'gbp_route_domain_add'
- err_msg = 'Failed to add GBP route domain on {node}!'\
- .format(node=node['host'])
+ cmd = u"gbp_route_domain_add"
+ err_msg = f"Failed to add GBP route domain on {node[u'host']}!"
args_in = dict(
- rd = dict (
- rd_id = rd_id,
- ip4_table_id = ip4_table_id,
- ip6_table_id = ip6_table_id,
- ip4_uu_sw_if_index = ip4_uu_sw_if_index,
- ip6_uu_sw_if_index = ip6_uu_sw_if_index
+ rd=dict(
+ rd_id=rd_id,
+ ip4_table_id=ip4_table_id,
+ ip6_table_id=ip6_table_id,
+ ip4_uu_sw_if_index=ip4_uu_sw_if_index,
+ ip6_uu_sw_if_index=ip6_uu_sw_if_index
)
)
:type uu_fwd_sw_if_index: int
:type bm_flood_sw_if_index: int
"""
- cmd = 'gbp_bridge_domain_add'
- err_msg = 'Failed to add GBP bridge domain on {node}!'\
- .format(node=node['host'])
+ cmd = u"gbp_bridge_domain_add"
+ err_msg = f"Failed to add GBP bridge domain on {node[u'host']}!"
args_in = dict(
- bd = dict(
- flags = getattr(GBPBridgeDomainFlags,
- 'GBP_BD_API_FLAG_NONE').value,
- bvi_sw_if_index = bvi_sw_if_index,
- uu_fwd_sw_if_index = uu_fwd_sw_if_index,
- bm_flood_sw_if_index = bm_flood_sw_if_index,
- bd_id = bd_id,
- rd_id = rd_id
+ bd=dict(
+ flags=getattr(
+ GBPBridgeDomainFlags, u"GBP_BD_API_FLAG_NONE"
+ ).value,
+ bvi_sw_if_index=bvi_sw_if_index,
+ uu_fwd_sw_if_index=uu_fwd_sw_if_index,
+ bm_flood_sw_if_index=bm_flood_sw_if_index,
+ bd_id=bd_id,
+ rd_id=rd_id
)
)
:type uplink_sw_if_index: int
:type remote_ep_timeout: int
"""
- cmd = 'gbp_endpoint_group_add'
- err_msg = 'Failed to add GBP endpoint group on {node}!'\
- .format(node=node['host'])
+ cmd = u"gbp_endpoint_group_add"
+ err_msg = f"Failed to add GBP endpoint group on {node[u'host']}!"
args_in = dict(
- epg = dict (
- uplink_sw_if_index = uplink_sw_if_index,
- bd_id = bd_id,
- rd_id = rd_id,
- vnid = vnid,
- sclass = sclass,
- retention = dict (
- remote_ep_timeout = remote_ep_timeout
+ epg=dict(
+ uplink_sw_if_index=uplink_sw_if_index,
+ bd_id=bd_id,
+ rd_id=rd_id,
+ vnid=vnid,
+ sclass=sclass,
+ retention=dict(
+ remote_ep_timeout=remote_ep_timeout
)
)
)
:type mac_addr: str
:type sclass: int
"""
- cmd = 'gbp_endpoint_add'
- err_msg = 'Failed to add GBP endpoint on {node}!'\
- .format(node=node['host'])
+ cmd = u"gbp_endpoint_add"
+ err_msg = f"Failed to add GBP endpoint on {node[u'host']}!"
ips = list()
- ips.append(IPUtil.create_ip_address_object(
- ip_address(unicode(ip_addr))))
- tun_src = IPUtil.create_ip_address_object(
- ip_address(unicode('0.0.0.0')))
- tun_dst = IPUtil.create_ip_address_object(
- ip_address(unicode('0.0.0.0')))
+ 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"))
args_in = dict(
- endpoint = dict(
- sw_if_index = sw_if_index,
- ips = ips,
- n_ips = len(ips),
- mac = L2Util.mac_to_bin(mac_addr),
- sclass = sclass,
- flags = getattr(GBPEndpointFlags,
- 'GBP_API_ENDPOINT_FLAG_EXTERNAL').value,
- tun = dict(
- src = tun_src,
- dst = tun_dst
+ endpoint=dict(
+ sw_if_index=sw_if_index,
+ ips=ips,
+ n_ips=len(ips),
+ mac=L2Util.mac_to_bin(mac_addr),
+ sclass=sclass,
+ flags=getattr(
+ GBPEndpointFlags, u"GBP_API_ENDPOINT_FLAG_EXTERNAL"
+ ).value,
+ tun=dict(
+ src=tun_src,
+ dst=tun_dst
)
)
)
:type bd_id: int
:type rd_id: int
"""
- cmd = 'gbp_ext_itf_add_del'
- err_msg = 'Failed to add external GBP interface on {node}!'\
- .format(node=node['host'])
+ cmd = u"gbp_ext_itf_add_del"
+ err_msg = u"Failed to add external GBP interface on {node[u'host']}!"
args_in = dict(
- is_add = 1,
- ext_itf = dict(
- sw_if_index = sw_if_index,
- bd_id = bd_id,
- rd_id = rd_id,
- flags = getattr(GBPExtItfFlags,
- 'GBP_API_EXT_ITF_F_NONE').value
+ is_add=1,
+ ext_itf=dict(
+ sw_if_index=sw_if_index,
+ bd_id=bd_id,
+ rd_id=rd_id,
+ flags=getattr(GBPExtItfFlags, u"GBP_API_EXT_ITF_F_NONE").value
)
)
:type rd_id: int
:type sw_if_index: int
"""
- cmd = 'gbp_subnet_add_del'
- err_msg = 'Failed to add GBP subnet on {node}!'\
- .format(node=node['host'])
+ cmd = u"gbp_subnet_add_del"
+ err_msg = f"Failed to add GBP subnet on {node[u'host']}!"
args_in = dict(
- is_add = 1,
- subnet = dict(
- type = getattr(GBPSubnetType,
- 'GBP_API_SUBNET_L3_OUT').value,
- sw_if_index = sw_if_index,
- sclass = sclass,
- prefix = dict(
- address = IPUtil.create_ip_address_object(
- ip_address(unicode(address))),
- len = int(subnet_length)
+ is_add=1,
+ subnet=dict(
+ type=getattr(GBPSubnetType, u"GBP_API_SUBNET_L3_OUT").value,
+ sw_if_index=sw_if_index,
+ sclass=sclass,
+ prefix=dict(
+ address=IPUtil.create_ip_address_object(
+ ip_address(address)
+ ),
+ len=int(subnet_length)
),
- rd_id = rd_id
+ rd_id=rd_id
)
)
papi_exec.add(cmd, **args_in).get_reply(err_msg)
@staticmethod
- def gbp_contract_add_del(node, sclass, dclass, acl_index=0):
+ def gbp_contract_add_del(node, sclass, dclass, acl_index=0, hash_mode=None):
"""Add GBP contract.
:param node: Node to add GBP contract on.
:param sclass: Source CLASS.
:param dclass: Destination CLASS.
:param acl_index: Index of ACL rule.
+ :param hash_mode: GBP contract hash mode.
:type node: dict
:type sclass: int
:type dclass: int
:type acl_index: int
+ :type hash_mode: str
"""
- cmd = 'gbp_contract_add_del'
- err_msg = 'Failed to add GBP contract on {node}!'\
- .format(node=node['host'])
+ cmd = u"gbp_contract_add_del"
+ err_msg = f"Failed to add GBP contract on {node[u'host']}!"
+ hash_mode = u"GBP_API_HASH_MODE_SRC_IP" if hash_mode is None \
+ else hash_mode
rule_permit = dict(
- action = getattr(GBPRuleAction,
- 'GBP_API_RULE_PERMIT').value,
- nh_set = dict(
- hash_mode = list(),
- n_nhs = 8,
- nhs = [dict()]*8,
+ action=getattr(GBPRuleAction, u"GBP_API_RULE_PERMIT").value,
+ nh_set=dict(
+ hash_mode=getattr(GBPHashMode, hash_mode).value,
+ n_nhs=8,
+ nhs=[dict()]*8,
)
)
rules = [rule_permit, rule_permit]
args_in = dict(
- is_add = 1,
- contract = dict(
- acl_index = acl_index,
- sclass = sclass,
- dclass = dclass,
- n_rules = len(rules),
- rules = rules,
- n_ether_types = 16,
- allowed_ethertypes = [0x800, 0x86dd] + [0]*14
+ is_add=1,
+ contract=dict(
+ acl_index=acl_index,
+ sclass=sclass,
+ dclass=dclass,
+ n_rules=len(rules),
+ rules=rules,
+ n_ether_types=16,
+ allowed_ethertypes=[0x800, 0x86dd] + [0]*14
)
)
Code Review / csit.git / blobdiff