x/vmware-nsxlib
Code Issues Proposed changes

NSXP: Support router advertisement rules

Browse Source 
Change-Id: Iad1011252fd21c87b1b5bd454fd6fb5a8064835d
This commit is contained in:
Kobi Samoray 2019-03-29 11:02:43 +03:00 committed by Anna Khmelnitsky
parent ad98fba5ec
commit 8f050a575b
4 changed files with 128 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

56
vmware_nsxlib/tests/unit/v3/policy/test_resources.py
View File

@ -2744,6 +2744,62 @@ class TestPolicyTier1(NsxPolicyLibTestCase):
self.assert_called_with_def(api_call, expected_def)
def test_add_advertisement_rule(self):
tier1_id = '111'
rule_name = 'rule_name'
rule_action = 'rule_action'
rule_pfx_operator = 'GE'
rule_adv_types = ['A']
rule_subnets = ['x', 'y', 'z']
with mock.patch.object(self.policy_api,
"get",
return_value={'id': tier1_id,
'display_name': 'tier1name'}),\
mock.patch.object(self.policy_api,
'create_or_update') as api_call:
self.resourceApi.add_advertisement_rule(
tier1_id, rule_name, action=rule_action,
prefix_operator=rule_pfx_operator,
route_advertisement_types=rule_adv_types, subnets=rule_subnets,
tenant=TEST_TENANT)
expected_def = core_defs.Tier1Def(
tier1_id=tier1_id,
name='tier1name',
route_advertisement_rules=[
core_defs.RouteAdvertisementRule(
rule_name,
action=rule_action,
prefix_operator=rule_pfx_operator,
route_advertisement_types=rule_adv_types,
subnets=rule_subnets)],
tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
def test_remove_advertisement_rule(self):
tier1_id = '111'
rule_name = 'rule_name'
get_retval = {
'id': tier1_id,
'display_name': 'tier1name',
'route_advertisement_rules': [{'name': rule_name}]}
with mock.patch.object(self.policy_api,
"get",
return_value=get_retval),\
mock.patch.object(self.policy_api,
'create_or_update') as api_call:
self.resourceApi.remove_advertisement_rule(
tier1_id, rule_name, tenant=TEST_TENANT)
expected_def = core_defs.Tier1Def(
tier1_id=tier1_id,
name='tier1name',
route_advertisement_rules=[],
tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
class TestPolicyTier1NoPassthrough(TestPolicyTier1):

15
vmware_nsxlib/v3/policy/constants.py
View File

@ -89,6 +89,21 @@ WAF_LOG_LEVEL_INFO = 'INFO'
WAF_LOG_LEVEL_DETAIL = 'DETAIL'
WAF_LOG_LEVEL_EVERYTHING = 'EVERYTHING'
# IpPool subnet type
IPPOOL_BLOCK_SUBNET = "IpAddressPoolBlockSubnet"
IPPOOL_STATIC_SUBNET = "IpAddressPoolStaticSubnet"
ADV_RULE_PERMIT = 'PERMIT'
ADV_RULE_DENY = 'DENY'
ADV_RULE_OPERATOR_GE = 'GE'
ADV_RULE_OPERATOR_EQ = 'EQ'
ADV_RULE_TYPE_TIER1_STATIC_ROUTES = 'TIER1_STATIC_ROUTES'
ADV_RULE_TIER1_CONNECTED = 'TIER1_CONNECTED'
ADV_RULE_TIER1_NAT = 'TIER1_NAT'
ADV_RULE_TIER1_LB_VIP = 'TIER1_LB_VIP'
ADV_RULE_TIER1_LB_SNAT = 'TIER1_LB_SNAT'
ADV_RULE_TIER1_DNS_FORWARDER_IP = 'TIER1_DNS_FORWARDER_IP'
ADV_RULE_TIER1_IPSEC_LOCAL_ENDPOINT = 'TIER1_IPSEC_LOCAL_ENDPOINT'

25
vmware_nsxlib/v3/policy/core_defs.py
View File

@ -387,6 +387,11 @@ class Tier1Def(RouterDef):
'route_advertisement').get_obj_dict()
self._set_attrs_if_specified(body, ['enable_standby_relocation'])
if self.has_attr('route_advertisement_rules'):
body['route_advertisement_rules'] = [
a.get_obj_dict()
if isinstance(a, RouteAdvertisementRule) else a
for a in self.get_attr('route_advertisement_rules')]
return body
@ -528,6 +533,26 @@ class Tier1NatRule(RouterNatRule):
return (TenantDef, Tier1Def)
class RouteAdvertisementRule(object):
def __init__(self, name, action=constants.ADV_RULE_PERMIT,
prefix_operator=constants.ADV_RULE_OPERATOR_GE,
route_advertisement_types=None,
subnets=None):
self.name = name
self.action = action
self.prefix_operator = prefix_operator
self.route_advertisement_types = route_advertisement_types
self.subnets = subnets
def get_obj_dict(self):
return {'name': self.name,
'action': self.action,
'prefix_operator': self.prefix_operator,
'route_advertisement_types': self.route_advertisement_types,
'subnets': self.subnets}
class RouterStaticRoute(ResourceDef):
@staticmethod

32
vmware_nsxlib/v3/policy/core_resources.py
View File

@ -961,6 +961,38 @@ class NsxPolicyTier1Api(NsxPolicyResourceBase):
tenant=tenant)
self.policy_api.create_or_update(tier1_def)
def add_advertisement_rule(
self, tier1_id, name, action=None, prefix_operator=None,
route_advertisement_types=None, subnets=None,
tenant=constants.POLICY_INFRA_TENANT):
tier1_dict = self.get(tier1_id, tenant)
adv_rules = tier1_dict.get('route_advertisement_rules', [])
adv_rules = [r for r in adv_rules if r.get('name') != name]
adv_rule = core_defs.RouteAdvertisementRule(
name=name, action=action, prefix_operator=prefix_operator,
route_advertisement_types=route_advertisement_types,
subnets=subnets)
adv_rules.append(adv_rule)
tier1_def = self.entry_def(tier1_id=tier1_id,
name=tier1_dict.get('display_name'),
route_advertisement_rules=adv_rules,
tenant=tenant)
self.policy_api.create_or_update(tier1_def)
def remove_advertisement_rule(self, tier1_id, name,
tenant=constants.POLICY_INFRA_TENANT):
tier1_dict = self.get(tier1_id, tenant)
adv_rules = tier1_dict.get('route_advertisement_rules', [])
updated_adv_rules = [r for r in adv_rules if r.get('name') != name]
if updated_adv_rules != adv_rules:
tier1_def = self.entry_def(
tier1_id=tier1_id,
name=tier1_dict.get('display_name'),
route_advertisement_rules=updated_adv_rules,
tenant=tenant)
self.policy_api.create_or_update(tier1_def)
def _locale_service_id(self, tier1_id):
# Supporting only a single locale-service per router for now
# with the same id as the router id with a constant suffix