x/vmware-nsxlib
Code Issues Proposed changes

Merge "Add IPPool static subnet policy support"

Browse Source
This commit is contained in:
Zuul 2019-04-01 18:19:41 +00:00 committed by Gerrit Code Review
commit 279ae87202
4 changed files with 203 additions and 12 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -3213,6 +3213,26 @@ class TestPolicyIpPool(NsxPolicyLibTestCase):
tenant=TEST_TENANT)
self.assert_called_with_def(delete_call, expected_def)
def test_list_block_subnets(self):
ip_pool_id = 'ip-pool-id'
api_results = {
'results': [{'id': 'static_subnet_1',
'resource_type': 'IpAddressPoolStaticSubnet'},
{'id': 'block_subnet_2',
'resource_type': 'IpAddressPoolBlockSubnet'}]
}
with mock.patch.object(
self.policy_api, "list", return_value=api_results) as api_call:
result = self.resourceApi.list_block_subnets(
ip_pool_id, tenant=TEST_TENANT)
expected_def = core_defs.IpPoolBlockSubnetDef(
ip_pool_id=ip_pool_id,
tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
expected_result = [{'id': 'block_subnet_2',
'resource_type': 'IpAddressPoolBlockSubnet'}]
self.assertEqual(result, expected_result)
def test_get_ip_subnet_realization_info(self):
ip_pool_id = '111'
ip_subnet_id = 'subnet-id'
@ -3277,6 +3297,94 @@ class TestPolicyIpPool(NsxPolicyLibTestCase):
self.assertEqual('5.5.0.8', ip)
api_get.assert_called_once()
def test_create_or_update_static_subnet(self):
ip_pool_id = 'ip-pool-id'
ip_subnet_id = 'static-subnet-id'
cidr = '10.10.10.0/24'
allocation_ranges = [{'start': '10.10.10.2', 'end': '10.10.10.250'}]
with mock.patch.object(self.policy_api,
"create_or_update") as api_call:
self.resourceApi.create_or_update_static_subnet(
ip_pool_id, cidr, allocation_ranges, ip_subnet_id,
tenant=TEST_TENANT)
expected_def = core_defs.IpPoolStaticSubnetDef(
ip_pool_id=ip_pool_id,
cidr=cidr,
allocation_ranges=allocation_ranges,
ip_subnet_id=ip_subnet_id,
tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
def test_release_static_subnet(self):
ip_pool_id = 'ip-pool-id'
ip_subnet_id = 'static-subnet-id'
with mock.patch.object(self.policy_api, "delete") as delete_call:
self.resourceApi.release_static_subnet(
ip_pool_id, ip_subnet_id, tenant=TEST_TENANT)
expected_def = core_defs.IpPoolStaticSubnetDef(
ip_pool_id=ip_pool_id,
ip_subnet_id=ip_subnet_id,
tenant=TEST_TENANT)
self.assert_called_with_def(delete_call, expected_def)
def test_list_static_subnet(self):
ip_pool_id = 'ip-pool-id'
api_results = {
'results': [{'id': 'static_subnet_1',
'resource_type': 'IpAddressPoolStaticSubnet'},
{'id': 'block_subnet_2',
'resource_type': 'IpAddressPoolBlockSubnet'}]
}
with mock.patch.object(
self.policy_api, "list", return_value=api_results) as api_call:
result = self.resourceApi.list_static_subnets(
ip_pool_id, tenant=TEST_TENANT)
expected_def = core_defs.IpPoolStaticSubnetDef(
ip_pool_id=ip_pool_id,
tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
expected_result = [{'id': 'static_subnet_1',
'resource_type': 'IpAddressPoolStaticSubnet'}]
self.assertEqual(result, expected_result)
def test_get_static_subnet(self):
ip_pool_id = 'ip-pool-id'
ip_subnet_id = 'static-subnet-id'
with mock.patch.object(self.policy_api, "get") as api_call:
self.resourceApi.get_static_subnet(
ip_pool_id, ip_subnet_id, tenant=TEST_TENANT)
expected_def = core_defs.IpPoolStaticSubnetDef(
ip_pool_id=ip_pool_id,
ip_subnet_id=ip_subnet_id,
tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
def test_get_static_subnet_realization_info(self):
ip_pool_id = 'ip-pool-id'
ip_subnet_id = 'static-subnet-id'
result = {'extended_attributes': [
{'values': '10.10.10.0/24', 'key': 'cidr'},
{'values': [{'value': '10.10.10.2', 'key': 'start'},
{'value': '10.10.10.250', 'key': 'end'}],
'key': 'allocation_ranges'}]}
with mock.patch.object(
self.resourceApi, "_get_realization_info",
return_value=result) as api_get:
self.resourceApi.get_ip_subnet_realization_info(
ip_pool_id, ip_subnet_id, tenant=TEST_TENANT,
subnet_type=constants.IPPOOL_STATIC_SUBNET)
api_get.assert_called_once()
# Test with wait set to True
with mock.patch.object(
self.resourceApi, "_wait_until_realized",
return_value=result) as api_get:
self.resourceApi.get_ip_subnet_realization_info(
ip_pool_id, ip_subnet_id, tenant=TEST_TENANT,
wait=True, subnet_type=constants.IPPOOL_STATIC_SUBNET)
api_get.assert_called_once()
class TestPolicySegmentPort(NsxPolicyLibTestCase):

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

@ -88,3 +88,7 @@ WAF_LOG_LEVEL_NOTICE = 'NOTICE'
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"

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

@ -949,8 +949,8 @@ class IpPoolAllocationDef(ResourceDef):
return body
class IpPoolBlockSubnetDef(ResourceDef):
'''Infra IpPoolSubnet belonging to IpBlock'''
class IpPoolSubnetDef(ResourceDef):
'''Infra IpPool Subnet'''
@property
def path_pattern(self):
@ -964,13 +964,17 @@ class IpPoolBlockSubnetDef(ResourceDef):
def resource_class(cls):
return 'IpAddressPoolSubnet'
def path_defs(self):
return (TenantDef, IpPoolDef)
class IpPoolBlockSubnetDef(IpPoolSubnetDef):
'''Infra IpPoolSubnet belonging to IpBlock'''
@staticmethod
def resource_type():
return 'IpAddressPoolBlockSubnet'
def path_defs(self):
return (TenantDef, IpPoolDef)
def get_obj_dict(self):
body = super(IpPoolBlockSubnetDef, self).get_obj_dict()
self._set_attrs_if_specified(body, ['auto_assign_gateway', 'size'])
@ -986,6 +990,21 @@ class IpPoolBlockSubnetDef(ResourceDef):
return body
class IpPoolStaticSubnetDef(IpPoolSubnetDef):
'''Infra IpPool static subnet'''
@staticmethod
def resource_type():
return 'IpAddressPoolStaticSubnet'
def get_obj_dict(self):
body = super(IpPoolStaticSubnetDef, self).get_obj_dict()
self._set_attrs_if_specified(body, ['cidr',
'allocation_ranges',
'gateway_ip'])
return body
class Condition(object):
def __init__(self, value, key=constants.CONDITION_KEY_TAG,
member_type=constants.CONDITION_MEMBER_PORT,

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

@ -2389,7 +2389,12 @@ class NsxPolicyIpPoolApi(NsxPolicyResourceBase):
ip_subnet_def = core_defs.IpPoolBlockSubnetDef(
ip_pool_id=ip_pool_id,
tenant=tenant)
return self._list(ip_subnet_def)
subnets = self._list(ip_subnet_def)
block_subnets = []
for subnet in subnets:
if subnet['resource_type'] == ip_subnet_def.resource_type():
block_subnets.append(subnet)
return block_subnets
def get_ip_block_subnet(self, ip_pool_id, ip_subnet_id,
tenant=constants.POLICY_INFRA_TENANT):
@ -2414,15 +2419,70 @@ class NsxPolicyIpPoolApi(NsxPolicyResourceBase):
return self._get_extended_attr_from_realized_info(
realized_info, requested_attr='cidr')
def get_ip_subnet_realization_info(self, ip_pool_id, ip_subnet_id,
entity_type=None,
tenant=constants.POLICY_INFRA_TENANT,
wait=False, sleep=None,
max_attempts=None):
ip_subnet_def = core_defs.IpPoolBlockSubnetDef(
def create_or_update_static_subnet(self, ip_pool_id, cidr,
allocation_ranges, ip_subnet_id=None,
name=IGNORE, description=IGNORE,
gateway_ip=IGNORE, tags=IGNORE,
tenant=constants.POLICY_INFRA_TENANT):
ip_subnet_id = self._init_obj_uuid(ip_subnet_id)
args = self._get_user_args(
ip_pool_id=ip_pool_id,
ip_subnet_id=ip_subnet_id,
cidr=cidr,
allocation_ranges=allocation_ranges,
name=name,
description=description,
tags=tags,
tenant=tenant)
ip_subnet_def = core_defs.IpPoolStaticSubnetDef(**args)
self._create_or_store(ip_subnet_def)
def release_static_subnet(self, ip_pool_id, ip_subnet_id,
tenant=constants.POLICY_INFRA_TENANT):
ip_subnet_def = core_defs.IpPoolStaticSubnetDef(
ip_subnet_id=ip_subnet_id,
ip_pool_id=ip_pool_id,
tenant=tenant)
self.policy_api.delete(ip_subnet_def)
def list_static_subnets(self, ip_pool_id,
tenant=constants.POLICY_INFRA_TENANT):
ip_subnet_def = core_defs.IpPoolStaticSubnetDef(
ip_pool_id=ip_pool_id,
tenant=tenant)
subnets = self._list(ip_subnet_def)
static_subnets = []
for subnet in subnets:
if subnet['resource_type'] == ip_subnet_def.resource_type():
static_subnets.append(subnet)
return static_subnets
def get_static_subnet(self, ip_pool_id, ip_subnet_id,
tenant=constants.POLICY_INFRA_TENANT):
ip_subnet_def = core_defs.IpPoolStaticSubnetDef(
ip_pool_id=ip_pool_id,
ip_subnet_id=ip_subnet_id,
tenant=tenant)
return self.policy_api.get(ip_subnet_def)
def get_ip_subnet_realization_info(
self, ip_pool_id, ip_subnet_id,
entity_type=None,
tenant=constants.POLICY_INFRA_TENANT,
wait=False, sleep=None,
max_attempts=None,
subnet_type=constants.IPPOOL_BLOCK_SUBNET):
if subnet_type == constants.IPPOOL_BLOCK_SUBNET:
ip_subnet_def = core_defs.IpPoolBlockSubnetDef(
ip_pool_id=ip_pool_id,
ip_subnet_id=ip_subnet_id,
tenant=tenant)
else:
ip_subnet_def = core_defs.IpPoolStaticSubnetDef(
ip_pool_id=ip_pool_id,
ip_subnet_id=ip_subnet_id,
tenant=tenant)
if wait:
return self._wait_until_realized(
ip_subnet_def, entity_type=entity_type,