Browse Source

Support segment profiles mapping

Change-Id: I4cdcc93ac55838ed0410b4678f11d703d103d1b4
changes/22/734022/1
asarfaty 3 weeks ago
committed by Adit Sarfaty
parent
commit
f750ca06a0
4 changed files with 341 additions and 4 deletions
  1. +180
    -4
      vmware_nsxlib/tests/unit/v3/policy/test_resources.py
  2. +6
    -0
      vmware_nsxlib/v3/policy/__init__.py
  3. +69
    -0
      vmware_nsxlib/v3/policy/core_defs.py
  4. +86
    -0
      vmware_nsxlib/v3/policy/core_resources.py

+ 180
- 4
vmware_nsxlib/tests/unit/v3/policy/test_resources.py View File

@@ -4679,6 +4679,182 @@ class TestPolicySegmentSecProfilesBinding(NsxPolicyLibTestCase):
update_call, expected_def)


class TestPolicySegmentDiscoveryProfilesBinding(NsxPolicyLibTestCase):

def setUp(self, resource_api_name='segment_discovery_profile_maps',
resource_def=core_defs.SegmentDiscoveryProfilesBindingMapDef):
super(TestPolicySegmentDiscoveryProfilesBinding, self).setUp()
self.resourceApi = getattr(self.policy_lib, resource_api_name)
self.resourceDef = resource_def

def test_create(self):
name = 'test'
segment_id = 'seg1'
prf1 = '1'
prf2 = '2'
with mock.patch.object(self.policy_api,
"create_or_update") as api_call:
result = self.resourceApi.create_or_overwrite(
name, segment_id,
ip_discovery_profile_id=prf1,
mac_discovery_profile_id=prf2,
tenant=TEST_TENANT)

expected_def = self.resourceDef(
segment_id=segment_id,
map_id=core_resources.DEFAULT_MAP_ID,
name=name,
ip_discovery_profile_id=prf1,
mac_discovery_profile_id=prf2,
tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
self.assertIsNotNone(result)

def test_delete(self):
segment_id = 'seg1'
with mock.patch.object(self.policy_api, "delete") as api_call:
self.resourceApi.delete(segment_id, tenant=TEST_TENANT)
expected_def = self.resourceDef(
segment_id=segment_id,
map_id=core_resources.DEFAULT_MAP_ID,
tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)

def test_get(self):
segment_id = 'seg1'
with mock.patch.object(self.policy_api, "get",
return_value={'id': segment_id}) as api_call:
result = self.resourceApi.get(segment_id,
tenant=TEST_TENANT)
expected_def = self.resourceDef(
segment_id=segment_id,
map_id=core_resources.DEFAULT_MAP_ID,
tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
self.assertEqual(segment_id, result['id'])

def test_list(self):
segment_id = 'seg1'
with mock.patch.object(self.policy_api, "list",
return_value={'results': []}) as api_call:
result = self.resourceApi.list(segment_id,
tenant=TEST_TENANT)
expected_def = self.resourceDef(
segment_id=segment_id,
tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
self.assertEqual([], result)

def test_update(self):
name = 'new name'
segment_id = 'seg1'
prf1 = '1'
prf2 = '2'
with self.mock_get(segment_id, name), \
self.mock_create_update() as update_call:

self.resourceApi.update(
segment_id=segment_id,
name=name,
ip_discovery_profile_id=prf1,
mac_discovery_profile_id=prf2,
tenant=TEST_TENANT)
expected_def = self.resourceDef(
segment_id=segment_id,
map_id=core_resources.DEFAULT_MAP_ID,
name=name,
ip_discovery_profile_id=prf1,
mac_discovery_profile_id=prf2,
tenant=TEST_TENANT)
self.assert_called_with_def(
update_call, expected_def)


class TestPolicySegmentQosProfilesBinding(NsxPolicyLibTestCase):

def setUp(self, resource_api_name='segment_qos_profile_maps',
resource_def=core_defs.SegmentQosProfilesBindingMapDef):
super(TestPolicySegmentQosProfilesBinding, self).setUp()
self.resourceApi = getattr(self.policy_lib, resource_api_name)
self.resourceDef = resource_def

def test_create(self):
name = 'test'
segment_id = 'seg1'
prf1 = '1'
with mock.patch.object(self.policy_api,
"create_or_update") as api_call:
result = self.resourceApi.create_or_overwrite(
name, segment_id,
qos_profile_id=prf1,
tenant=TEST_TENANT)

expected_def = self.resourceDef(
segment_id=segment_id,
map_id=core_resources.DEFAULT_MAP_ID,
name=name,
qos_profile_id=prf1,
tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
self.assertIsNotNone(result)

def test_delete(self):
segment_id = 'seg1'
with mock.patch.object(self.policy_api, "delete") as api_call:
self.resourceApi.delete(segment_id, tenant=TEST_TENANT)
expected_def = self.resourceDef(
segment_id=segment_id,
map_id=core_resources.DEFAULT_MAP_ID,
tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)

def test_get(self):
segment_id = 'seg1'
with mock.patch.object(self.policy_api, "get",
return_value={'id': segment_id}) as api_call:
result = self.resourceApi.get(segment_id,
tenant=TEST_TENANT)
expected_def = self.resourceDef(
segment_id=segment_id,
map_id=core_resources.DEFAULT_MAP_ID,
tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
self.assertEqual(segment_id, result['id'])

def test_list(self):
segment_id = 'seg1'
with mock.patch.object(self.policy_api, "list",
return_value={'results': []}) as api_call:
result = self.resourceApi.list(segment_id,
tenant=TEST_TENANT)
expected_def = self.resourceDef(
segment_id=segment_id,
tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
self.assertEqual([], result)

def test_update(self):
name = 'new name'
segment_id = 'seg1'
prf1 = '1'
with self.mock_get(segment_id, name), \
self.mock_create_update() as update_call:

self.resourceApi.update(
segment_id=segment_id,
name=name,
qos_profile_id=prf1,
tenant=TEST_TENANT)
expected_def = self.resourceDef(
segment_id=segment_id,
map_id=core_resources.DEFAULT_MAP_ID,
name=name,
qos_profile_id=prf1,
tenant=TEST_TENANT)
self.assert_called_with_def(
update_call, expected_def)


class TestPolicySegmentPortSecProfilesBinding(NsxPolicyLibTestCase):

def setUp(self, resource_api_name='segment_port_security_profiles',
@@ -4781,13 +4957,13 @@ class TestPolicySegmentPortSecProfilesBinding(NsxPolicyLibTestCase):
update_call, expected_def)


class TestPolicySegmentDiscoveryProfilesBinding(NsxPolicyLibTestCase):
class TestPolicySegmentPortDiscoveryProfilesBinding(NsxPolicyLibTestCase):

def setUp(
self, resource_api_name='segment_port_discovery_profiles',
resource_def=core_defs.SegmentPortDiscoveryProfilesBindingMapDef):

super(TestPolicySegmentDiscoveryProfilesBinding, self).setUp()
super(TestPolicySegmentPortDiscoveryProfilesBinding, self).setUp()
self.resourceApi = getattr(self.policy_lib, resource_api_name)
self.resourceDef = resource_def

@@ -4884,13 +5060,13 @@ class TestPolicySegmentDiscoveryProfilesBinding(NsxPolicyLibTestCase):
update_call, expected_def)


class TestPolicySegmentQosProfilesBinding(NsxPolicyLibTestCase):
class TestPolicySegmentPortQosProfilesBinding(NsxPolicyLibTestCase):

def setUp(
self, resource_api_name='segment_port_qos_profiles',
resource_def=core_defs.SegmentPortQoSProfilesBindingMapDef):

super(TestPolicySegmentQosProfilesBinding, self).setUp()
super(TestPolicySegmentPortQosProfilesBinding, self).setUp()
self.resourceApi = getattr(self.policy_lib, resource_api_name)
self.resourceDef = resource_def



+ 6
- 0
vmware_nsxlib/v3/policy/__init__.py View File

@@ -107,6 +107,12 @@ class NsxPolicyLib(lib.NsxLibBase):
self.segment_security_profile_maps = (
core_resources.SegmentSecurityProfilesBindingMapApi(
*args))
self.segment_qos_profile_maps = (
core_resources.SegmentQosProfilesBindingMapApi(
*args))
self.segment_discovery_profile_maps = (
core_resources.SegmentDiscoveryProfilesBindingMapApi(
*args))
self.segment_port_security_profiles = (
core_resources.SegmentPortSecurityProfilesBindingMapApi(
*args))


+ 69
- 0
vmware_nsxlib/v3/policy/core_defs.py View File

@@ -1173,6 +1173,75 @@ class SegmentSecProfilesBindingMapDef(SegmentBindingMapDefBase):
return body


class SegmentQosProfilesBindingMapDef(SegmentBindingMapDefBase):
@property
def path_pattern(self):
return (SEGMENTS_PATH_PATTERN +
"%s/segment-qos-profile-binding-maps/")

@staticmethod
def resource_type():
return 'SegmentQoSProfileBindingMap'

def get_obj_dict(self):
body = super(SegmentQosProfilesBindingMapDef, self).get_obj_dict()

if self.has_attr('segment_qos_profile_id'):
path = ""
if self.get_attr('segment_qos_profile_id'):
profile = QosProfileDef(
profile_id=self.get_attr('segment_qos_profile_id'),
tenant=self.get_tenant())
path = profile.get_resource_full_path()
self._set_attr_if_specified(
body, 'segment_qos_profile_id',
body_attr='segment_qos_profile_path',
value=path)

return body


class SegmentDiscoveryProfilesBindingMapDef(SegmentBindingMapDefBase):
@property
def path_pattern(self):
return (SEGMENTS_PATH_PATTERN +
"%s/segment-discovery-profile-binding-maps/")

@staticmethod
def resource_type():
return 'SegmentDiscoveryProfileBindingMap'

def get_obj_dict(self):
body = super(SegmentDiscoveryProfilesBindingMapDef,
self).get_obj_dict()

if self.has_attr('mac_discovery_profile_id'):
path = ""
if self.get_attr('mac_discovery_profile_id'):
profile = MacDiscoveryProfileDef(
profile_id=self.get_attr('mac_discovery_profile_id'),
tenant=self.get_tenant())
path = profile.get_resource_full_path()
self._set_attr_if_specified(
body, 'mac_discovery_profile_id',
body_attr='mac_discovery_profile_path',
value=path)

if self.has_attr('ip_discovery_profile_id'):
path = ""
if self.get_attr('ip_discovery_profile_id'):
profile = IpDiscoveryProfileDef(
profile_id=self.get_attr('ip_discovery_profile_id'),
tenant=self.get_tenant())
path = profile.get_resource_full_path()
self._set_attr_if_specified(
body, 'ip_discovery_profile_id',
body_attr='ip_discovery_profile_path',
value=path)

return body


class SegmentPortBindingMapDefBase(ResourceDef):

@property


+ 86
- 0
vmware_nsxlib/v3/policy/core_resources.py View File

@@ -2317,6 +2317,92 @@ class SegmentSecurityProfilesBindingMapApi(SegmentProfilesBindingMapBaseApi):
tenant=tenant)


class SegmentDiscoveryProfilesBindingMapApi(SegmentProfilesBindingMapBaseApi):

@property
def entry_def(self):
return core_defs.SegmentDiscoveryProfilesBindingMapDef

def create_or_overwrite(self, name, segment_id,
map_id=DEFAULT_MAP_ID,
description=IGNORE,
ip_discovery_profile_id=IGNORE,
mac_discovery_profile_id=IGNORE,
tags=IGNORE,
tenant=constants.POLICY_INFRA_TENANT):
map_id = self._init_obj_uuid(map_id)
map_def = self._init_def(
segment_id=segment_id,
map_id=map_id,
name=name,
description=description,
ip_discovery_profile_id=ip_discovery_profile_id,
mac_discovery_profile_id=mac_discovery_profile_id,
tags=tags,
tenant=tenant)
self._create_or_store(map_def)
return map_id

def update(self, segment_id,
map_id=DEFAULT_MAP_ID,
name=IGNORE,
description=IGNORE,
ip_discovery_profile_id=IGNORE,
mac_discovery_profile_id=IGNORE,
tags=IGNORE,
tenant=constants.POLICY_INFRA_TENANT):
self._update(
segment_id=segment_id,
map_id=map_id,
name=name,
description=description,
ip_discovery_profile_id=ip_discovery_profile_id,
mac_discovery_profile_id=mac_discovery_profile_id,
tags=tags,
tenant=tenant)


class SegmentQosProfilesBindingMapApi(SegmentProfilesBindingMapBaseApi):

@property
def entry_def(self):
return core_defs.SegmentQosProfilesBindingMapDef

def create_or_overwrite(self, name, segment_id,
map_id=DEFAULT_MAP_ID,
description=IGNORE,
qos_profile_id=IGNORE,
tags=IGNORE,
tenant=constants.POLICY_INFRA_TENANT):
map_id = self._init_obj_uuid(map_id)
map_def = self._init_def(
segment_id=segment_id,
map_id=map_id,
name=name,
description=description,
qos_profile_id=qos_profile_id,
tags=tags,
tenant=tenant)
self._create_or_store(map_def)
return map_id

def update(self, segment_id,
map_id=DEFAULT_MAP_ID,
name=IGNORE,
description=IGNORE,
qos_profile_id=IGNORE,
tags=IGNORE,
tenant=constants.POLICY_INFRA_TENANT):
self._update(
segment_id=segment_id,
map_id=map_id,
name=name,
description=description,
qos_profile_id=qos_profile_id,
tags=tags,
tenant=tenant)


class SegmentPortProfilesBindingMapBaseApi(NsxPolicyResourceBase):

def delete(self, segment_id, port_id, map_id=DEFAULT_MAP_ID,


Loading…
Cancel
Save