x
/
vmware-nsxlib
Code Issues Proposed changes

Add Service class that allows multiple/mixed entries 

Change-Id: I3012f8db6a8393d3dee59795b8136d923f8cb2b6
This commit is contained in:
Shih-Hao Li 2019-03-27 11:20:50 -07:00
parent 207ce5eca1
commit 0a085a6062
4 changed files with 273 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

34
vmware_nsxlib/tests/unit/v3/policy/test_api.py
View File

@ -175,6 +175,40 @@ class TestPolicyService(policy_testcase.TestPolicyApi):
'infra/services/icmpservice',
data=expected_data)
def test_create_mixed_with_parent(self):
service_def = policy.ServiceDef(name='mixedservice',
service_id='mixedservice')
l4_entry_def = policy.L4ServiceEntryDef(service_id='mixedservice',
protocol='TCP',
entry_id='http',
name='http',
dest_ports=[80, 8080])
icmp_entry_def = policy.IcmpServiceEntryDef(service_id='mixedservice',
version=4,
entry_id='icmp',
name='icmpv4')
self.policy_api.create_with_parent(service_def,
[l4_entry_def, icmp_entry_def])
expected_l4_entry = {'id': 'http',
'resource_type': 'L4PortSetServiceEntry',
'display_name': 'http',
'l4_protocol': 'TCP',
'destination_ports': [80, 8080]}
expected_icmp_entry = {'id': 'icmp',
'resource_type': 'ICMPTypeServiceEntry',
'display_name': 'icmpv4',
'protocol': 'ICMPv4'}
expected_data = {'id': 'mixedservice',
'resource_type': 'Service',
'display_name': 'mixedservice',
'service_entries': [
expected_l4_entry, expected_icmp_entry]}
self.assert_json_call('PATCH', self.client,
'infra/services/mixedservice',
data=expected_data)
class TestPolicyCommunicationMap(policy_testcase.TestPolicyApi):

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

@ -1028,6 +1028,167 @@ class TestPolicyIPProtocolService(NsxPolicyLibTestCase):
[service_def, entry_def])
class TestPolicyMixedService(NsxPolicyLibTestCase):
def setUp(self, *args, **kwargs):
super(TestPolicyMixedService, self).setUp()
self.l4ServiceApi = self.policy_lib.service
self.icmpServiceApi = self.policy_lib.icmp_service
self.ipServiceApi = self.policy_lib.ip_protocol_service
self.resourceApi = self.policy_lib.mixed_service
def test_create_service_only(self):
name = 's1'
srv_id = '111'
description = 'desc'
tags = [{'scope': 'a', 'tag': 'b'}]
with mock.patch.object(self.policy_api,
"create_or_update") as api_call:
result = self.resourceApi.create_or_overwrite(
name,
srv_id,
description=description,
tags=tags,
tenant=TEST_TENANT)
exp_srv_def = core_defs.ServiceDef(
service_id=srv_id,
name=name,
description=description,
tags=tags,
tenant=TEST_TENANT)
self.assert_called_with_def(api_call, exp_srv_def)
self.assertIsNotNone(result)
def test_create_with_entries(self):
name = 's1'
srv_id = '111'
description = 'desc'
tags = [{'scope': 'a', 'tag': 'b'}]
protocol = constants.TCP
dest_ports = [81, 82]
source_ports = [83, 84]
icmp_type = 2
protocol_number = 2
l4_entry = self.l4ServiceApi.build_entry(
'l4_entry', srv_id, 'l4_entry', protocol=protocol,
dest_ports=dest_ports, source_ports=source_ports,
tenant=TEST_TENANT)
icmp_entry = self.icmpServiceApi.build_entry(
'icmp_entry', srv_id, 'icmp_entry', icmp_type=icmp_type,
tenant=TEST_TENANT)
ip_entry = self.ipServiceApi.build_entry(
'ip_entry', srv_id, 'ip_entry',
protocol_number=protocol_number, tenant=TEST_TENANT)
with mock.patch.object(self.policy_api,
"create_with_parent") as api_call:
result = self.resourceApi.create_or_overwrite(
name,
srv_id,
description=description,
entries=[l4_entry, icmp_entry, ip_entry],
tags=tags,
tenant=TEST_TENANT)
service_def = core_defs.ServiceDef(
service_id=srv_id,
name=name,
description=description,
tags=tags,
tenant=TEST_TENANT)
self.assert_called_with_defs(
api_call, [service_def, l4_entry, icmp_entry, ip_entry])
self.assertIsNotNone(result)
def test_delete(self):
srv_id = '111'
with mock.patch.object(self.policy_api, "delete") as api_call:
self.resourceApi.delete(srv_id, tenant=TEST_TENANT)
expected_def = core_defs.ServiceDef(service_id=srv_id,
tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
def test_get(self):
srv_id = '111'
with mock.patch.object(self.policy_api, "get",
return_value={'id': srv_id}) as api_call:
result = self.resourceApi.get(srv_id, tenant=TEST_TENANT)
expected_def = core_defs.ServiceDef(service_id=srv_id,
tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
self.assertEqual(srv_id, result['id'])
def test_get_by_name(self):
name = 's1'
with mock.patch.object(
self.policy_api, "list",
return_value={'results': [{'display_name': name}]}) as api_call:
obj = self.resourceApi.get_by_name(name, tenant=TEST_TENANT)
self.assertIsNotNone(obj)
expected_def = core_defs.ServiceDef(tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
def test_list(self):
with mock.patch.object(self.policy_api, "list",
return_value={'results': []}) as api_call:
result = self.resourceApi.list(tenant=TEST_TENANT)
expected_def = core_defs.ServiceDef(tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
self.assertEqual([], result)
def test_update(self):
name = 'newName'
srv_id = '111'
description = 'new desc'
tags = [{'scope': 'c', 'tag': 'd'}]
protocol = constants.UDP
dest_ports = [91, 92]
source_ports = [93, 94]
icmp_type = 3
protocol_number = 3
l4_entry = self.l4ServiceApi.build_entry(
'l4_entry', srv_id, 'l4_entry', protocol=protocol,
dest_ports=dest_ports, source_ports=source_ports,
tenant=TEST_TENANT)
icmp_entry = self.icmpServiceApi.build_entry(
'icmp_entry', srv_id, 'icmp_entry', icmp_type=icmp_type,
tenant=TEST_TENANT)
ip_entry = self.ipServiceApi.build_entry(
'ip_entry', srv_id, 'ip_entry',
protocol_number=protocol_number, tenant=TEST_TENANT)
with mock.patch.object(self.policy_api, "get",
return_value={}),\
mock.patch.object(self.policy_api,
"create_with_parent") as update_call:
self.resourceApi.update(
srv_id,
name=name,
description=description,
entries=[l4_entry, icmp_entry, ip_entry],
tags=tags,
tenant=TEST_TENANT)
service_def = core_defs.ServiceDef(
service_id=srv_id,
name=name,
description=description,
tags=tags,
tenant=TEST_TENANT)
self.assert_called_with_defs(
update_call, [service_def, l4_entry, icmp_entry, ip_entry])
class TestPolicyCommunicationMap(NsxPolicyLibTestCase):
def setUp(self, *args, **kwargs):

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

@ -58,6 +58,7 @@ class NsxPolicyLib(lib.NsxLibBase):
*args)
self.ip_protocol_service = (
core_resources.NsxPolicyIPProtocolServiceApi(*args))
self.mixed_service = core_resources.NsxPolicyMixedServiceApi(*args)
self.tier0 = core_resources.NsxPolicyTier0Api(*args)
self.tier0_nat_rule = core_resources.NsxPolicyTier0NatRuleApi(
*args)

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

@ -598,6 +598,20 @@ class NsxPolicyL4ServiceApi(NsxPolicyServiceBase):
self.policy_api.create_with_parent(parent_def, entry_def)
def build_entry(self, name, service_id, entry_id,
description=None, protocol=None,
dest_ports=None, source_ports=None,
tags=None, tenant=constants.POLICY_INFRA_TENANT):
return self._init_def(service_id=service_id,
entry_id=entry_id,
name=name,
description=description,
protocol=protocol,
dest_ports=dest_ports,
source_ports=source_ports,
tags=tags,
tenant=tenant)
class NsxPolicyIcmpServiceApi(NsxPolicyServiceBase):
"""NSX Policy Service with a single ICMP service entry.
@ -656,6 +670,20 @@ class NsxPolicyIcmpServiceApi(NsxPolicyServiceBase):
return self.policy_api.create_with_parent(parent_def, entry_def)
def build_entry(self, name, service_id, entry_id,
description=None, version=4,
icmp_type=None, icmp_code=None,
tags=None, tenant=constants.POLICY_INFRA_TENANT):
return self._init_def(service_id=service_id,
entry_id=entry_id,
name=name,
description=description,
version=version,
icmp_type=icmp_type,
icmp_code=icmp_code,
tags=tags,
tenant=tenant)
class NsxPolicyIPProtocolServiceApi(NsxPolicyServiceBase):
"""NSX Policy Service with a single IPProtocol service entry.
@ -708,6 +736,55 @@ class NsxPolicyIPProtocolServiceApi(NsxPolicyServiceBase):
return self.policy_api.create_with_parent(parent_def, entry_def)
def build_entry(self, name, service_id, entry_id,
description=None, protocol_number=None,
tags=None, tenant=constants.POLICY_INFRA_TENANT):
return self._init_def(service_id=service_id,
entry_id=entry_id,
name=name,
protocol_number=protocol_number,
tags=tags,
tenant=tenant)
class NsxPolicyMixedServiceApi(NsxPolicyServiceBase):
"""NSX Policy Service with mixed service entries."""
def create_or_overwrite(self, name, service_id,
description=IGNORE,
entries=IGNORE,
tags=IGNORE,
tenant=constants.POLICY_INFRA_TENANT):
service_def = self._init_parent_def(service_id=service_id,
name=name,
description=description,
entries=entries,
tags=tags,
tenant=tenant)
if entries != IGNORE:
self._create_or_store(service_def, entries)
else:
self._create_or_store(service_def)
return service_id
def update(self, service_id,
name=IGNORE, description=IGNORE,
entries=IGNORE, tags=IGNORE,
tenant=constants.POLICY_INFRA_TENANT):
parent_def = self._init_parent_def(
service_id=service_id,
name=name,
description=description,
tags=tags,
tenant=tenant)
if entries != IGNORE:
self.policy_api.create_with_parent(parent_def, entries)
else:
self.policy_api.create_or_update(parent_def)
class NsxPolicyTier1Api(NsxPolicyResourceBase):
"""NSX Tier1 API """