x/vmware-nsxlib
Code Issues Proposed changes

Add VPC IP address allocation to policy API

Browse Source 
Change-Id: I79e2026f972bc2a52aa2c2691aef3feeb3605db0
This commit is contained in:
Qian Sun 2024-06-14 10:20:34 +00:00
parent a797301b8e
commit 84eda65ba1
6 changed files with 206 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -7680,3 +7680,100 @@ class TestPolicySharedResource(NsxPolicyLibTestCase):
description=description,
tenant=TEST_TENANT)
self.assert_called_with_def(update_call, expected_def)
class TestNsxVpcIpAddressAllocation(NsxPolicyLibTestCase):
def setUp(self, *args, **kwargs):
super(TestNsxVpcIpAddressAllocation, self).setUp()
self.resourceApi = self.policy_lib.vpc_ip_address_allocation
def test_create(self):
name = "vpc_ip_address_allocation_test"
description = "desc"
ip_address_allocation_id = "allocation_id"
ip_address_block_visibility = "EXTERNAL"
allocation_ips = ["1.2.3.4"]
tenant = "orgs/org1/projects/project1/vpcs/vpc1"
with mock.patch.object(self.policy_api,
"create_or_update") as api_call:
result = self.resourceApi.create_or_overwrite(
name=name,
description=description,
ip_address_allocation_id=ip_address_allocation_id,
ip_address_block_visibility=ip_address_block_visibility,
allocation_ips=allocation_ips,
tenant=tenant)
expected_def = core_defs.VpcIpAddressAllocationDef(
name=name,
description=description,
ip_address_allocation_id=ip_address_allocation_id,
ip_address_block_visibility=ip_address_block_visibility,
allocation_ips=allocation_ips,
tenant=tenant)
self.assert_called_with_def(api_call, expected_def)
self.assertIsNotNone(result)
def test_delete(self):
ip_address_allocation_id = "allocation_id"
tenant = "orgs/org1/projects/project1/vpcs/vpc1"
with mock.patch.object(self.policy_api, "delete") as api_call:
self.resourceApi.delete(
tenant=tenant,
ip_address_allocation_id=ip_address_allocation_id)
expected_def = core_defs.VpcIpAddressAllocationDef(
ip_address_allocation_id=ip_address_allocation_id,
tenant=tenant)
self.assert_called_with_def(api_call, expected_def)
def test_list(self):
tenant = "orgs/org1/projects/project1/vpcs/vpc1"
with mock.patch.object(self.policy_api, "list",
return_value={'results': []}) as api_call:
result = self.resourceApi.list(tenant=tenant)
expected_def = core_defs.VpcIpAddressAllocationDef(tenant=tenant)
self.assert_called_with_def(api_call, expected_def)
self.assertEqual([], result)
def test_get(self):
ip_address_allocation_id = "allocation_id"
tenant = "orgs/org1/projects/project1/vpcs/vpc1"
with mock.patch.object(self.policy_api, "get") as api_call:
mock_t0_nat_rule = mock.Mock()
api_call.return_value = mock_t0_nat_rule
result = self.resourceApi.get(
tenant=tenant,
ip_address_allocation_id=ip_address_allocation_id)
expected_def = core_defs.VpcIpAddressAllocationDef(
ip_address_allocation_id=ip_address_allocation_id,
tenant=tenant)
self.assert_called_with_def(api_call, expected_def)
self.assertEqual(mock_t0_nat_rule, result)
def test_update(self):
name = "vpc_ip_address_allocation_test2"
description = "desc2"
ip_address_allocation_id = "allocation_id"
ip_address_block_visibility = "EXTERNAL"
allocation_ips = ["1.2.3.5"]
tenant = "orgs/org1/projects/project1/vpcs/vpc1"
with mock.patch.object(self.policy_api,
"create_or_update") as api_call:
self.resourceApi.update(
name=name,
description=description,
ip_address_allocation_id=ip_address_allocation_id,
ip_address_block_visibility=ip_address_block_visibility,
allocation_ips=allocation_ips,
tenant=tenant)
expected_def = core_defs.VpcIpAddressAllocationDef(
name=name,
description=description,
ip_address_allocation_id=ip_address_allocation_id,
ip_address_block_visibility=ip_address_block_visibility,
allocation_ips=allocation_ips,
tenant=tenant)
self.assert_called_with_def(api_call, expected_def)

3
vmware_nsxlib/v3/client.py
View File

@ -113,7 +113,8 @@ def http_error_to_exception(status_code, error_code, related_error_codes=None):
'503040': exceptions.NsxSegemntWithVM,
'100148': exceptions.StaleRevision,
'500012': exceptions.NsxInvalidPath,
'612504': exceptions.VifRestoreAlreadyCompleted},
'612504': exceptions.VifRestoreAlreadyCompleted,
'500073': exceptions.EntityAlreadyExists},
requests.codes.CONFLICT: exceptions.StaleRevision,
requests.codes.PRECONDITION_FAILED: exceptions.StaleRevision,
requests.codes.INTERNAL_SERVER_ERROR:

4
vmware_nsxlib/v3/exceptions.py
View File

@ -257,3 +257,7 @@ class ResourceInUse(ManagerError):
class VifRestoreAlreadyCompleted(ManagerError):
message = _("Vif Restore failed: %(details)s")
class EntityAlreadyExists(ManagerError):
message = _("Failed to create or update entity: %(msg)s")

2
vmware_nsxlib/v3/policy/__init__.py
View File

@ -158,6 +158,8 @@ class NsxPolicyLib(lib.NsxLibBase):
self.shared_resource = (
core_resources.NsxPolicySharedResourceApi(
*args))
self.vpc_ip_address_allocation = (
core_resources.NsxVpcIpAddressAllocationApi(*args))
def get_nsxlib_passthrough(self):
return self.nsx_api

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

@ -80,6 +80,9 @@ TIER1_LOCALE_SERVICES_PATH_PATTERN = (TIER1S_PATH_PATTERN +
OBJECT_PERMISSIONS_PATH_PATTERN = (TENANTS_PATH_PATTERN +
"object-permissions")
VPC_IP_ADDRESS_ALLOCATION_PATH_PATTERN = (TENANTS_PATH_PATTERN +
"/ip-address-allocations/")
class ResourceDef(object, metaclass=abc.ABCMeta):
def __init__(self, nsx_version=None, **kwargs):
@ -2947,3 +2950,24 @@ class ObjectRolePermissionGroupDef(ResourceDef):
if obj_id:
body.update({"id": obj_id})
return body
class VpcIpAddressAllocationDef(ResourceDef):
@staticmethod
def resource_type():
return "VpcIpAddressAllocationDef"
def get_obj_dict(self):
body = super(VpcIpAddressAllocationDef, self).get_obj_dict()
self._set_attrs_if_specified(body, ["ip_address_block_visibility",
"allocation_ip"])
return body
@property
def path_ids(self):
return ("tenant", "ip_address_allocation_id")
@property
def path_pattern(self):
return VPC_IP_ADDRESS_ALLOCATION_PATH_PATTERN

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

@ -5413,3 +5413,80 @@ class NsxPolicySharedResourceApi(NsxPolicyResourceBase):
description=description,
tags=tags,
tenant=tenant)
class NsxVpcIpAddressAllocationApi(NsxPolicyResourceBase):
@property
def entry_def(self):
return core_defs.VpcIpAddressAllocationDef
def create_or_overwrite(self,
tenant,
ip_address_allocation_id,
name=IGNORE,
allocation_ip=IGNORE,
allocation_ips=IGNORE,
allocation_size=IGNORE,
description=IGNORE,
ip_address_block_visibility="EXTERNAL",
ip_address_type="IPV4",
ip_block=IGNORE,
tags=IGNORE):
ip_address_allocation_id = self._init_obj_uuid(
ip_address_allocation_id)
ip_address_allocation_def = self._init_def(
name=name,
tenant=tenant,
ip_address_allocation_id=ip_address_allocation_id,
allocation_ip=allocation_ip,
allocation_ips=allocation_ips,
allocation_size=allocation_size,
description=description,
ip_address_block_visibility=ip_address_block_visibility,
ip_address_type=ip_address_type,
ip_block=ip_block,
tags=tags,
patch=True)
self._create_or_store(ip_address_allocation_def)
return ip_address_allocation_id
def delete(self, tenant, ip_address_allocation_id):
ip_address_allocation_def = self.entry_def(
tenant=tenant,
ip_address_allocation_id=ip_address_allocation_id)
self._delete_with_retry(ip_address_allocation_def)
def get(self, tenant, ip_address_allocation_id, silent=False):
ip_address_allocation_def = self.entry_def(
tenant=tenant,
ip_address_allocation_id=ip_address_allocation_id)
return self.policy_api.get(ip_address_allocation_def, silent=silent)
def list(self, tenant):
ip_address_allocation_def = self.entry_def(tenant=tenant)
return self._list(ip_address_allocation_def)
def update(self,
tenant,
ip_address_allocation_id,
name=IGNORE,
allocation_ip=IGNORE,
allocation_ips=IGNORE, allocation_size=IGNORE,
description=IGNORE,
ip_address_block_visibility=IGNORE,
ip_address_type=IGNORE,
ip_block=IGNORE,
tags=IGNORE):
self._update(tenant=tenant,
ip_address_allocation_id=ip_address_allocation_id,
name=name,
allocation_ip=allocation_ip,
allocation_ips=allocation_ips,
allocation_size=allocation_size,
description=description,
ip_address_block_visibility=ip_address_block_visibility,
ip_address_type=ip_address_type,
ip_block=ip_block,
tags=tags,
patch=True)