cdaad65004
1. create when missing For NSX Policy resources, modify "update" method behavior to create the resource if it does not exist. This will save roundtrips. 2. update the path to use policy/api/vi 3. use POST instead of PUT 4. Update apis do not have to use get before update 5. Add enforcement point thumbprint Co-Authored-by: Adit Sarfaty <asarfaty@vmware.com> Change-Id: Ic44308d1df8944dfbcfa2a613ed8e1847eee3715
1023 lines
45 KiB
Python
1023 lines
45 KiB
Python
# Copyright 2017 VMware, Inc.
|
|
# All Rights Reserved
|
|
#
|
|
# Licensed under the Apache License, Version 2.0 (the "License"); you may
|
|
# not use this file except in compliance with the License. You may obtain
|
|
# a copy of the License at
|
|
#
|
|
# http://www.apache.org/licenses/LICENSE-2.0
|
|
#
|
|
# Unless required by applicable law or agreed to in writing, software
|
|
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
|
|
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
|
|
# License for the specific language governing permissions and limitations
|
|
# under the License.
|
|
#
|
|
import mock
|
|
import unittest
|
|
|
|
from vmware_nsxlib.tests.unit.v3 import nsxlib_testcase
|
|
from vmware_nsxlib import v3
|
|
from vmware_nsxlib.v3 import policy_constants
|
|
from vmware_nsxlib.v3 import policy_defs
|
|
|
|
TEST_TENANT = 'test'
|
|
|
|
|
|
class NsxPolicyLibTestCase(unittest.TestCase):
|
|
|
|
def setUp(self, *args, **kwargs):
|
|
super(NsxPolicyLibTestCase, self).setUp()
|
|
|
|
mock.patch.object(v3.NsxPolicyLib, "get_version",
|
|
return_value='2.0.0').start()
|
|
nsxlib_config = nsxlib_testcase.get_default_nsxlib_config()
|
|
self.policy_lib = v3.NsxPolicyLib(nsxlib_config)
|
|
self.policy_api = self.policy_lib.policy_api
|
|
|
|
self.maxDiff = None
|
|
|
|
def _compare_def(self, expected_def, actual_def):
|
|
# verify the resource definition class
|
|
self.assertEqual(expected_def.__class__, actual_def.__class__)
|
|
# verify the resource definition tenant
|
|
self.assertEqual(expected_def.tenant, actual_def.tenant)
|
|
# verify the resource definition values
|
|
self.assertEqual(expected_def.get_obj_dict(),
|
|
actual_def.get_obj_dict())
|
|
|
|
def assert_called_with_def(self, mock_api, expected_def, call_num=0):
|
|
# verify the api was called
|
|
mock_api.assert_called()
|
|
actual_def = mock_api.call_args_list[call_num][0][0]
|
|
self._compare_def(expected_def, actual_def)
|
|
|
|
def assert_called_with_defs(self, mock_api, expected_defs, call_num=0):
|
|
# verify the api & first resource definition
|
|
self.assert_called_with_def(mock_api, expected_defs[0],
|
|
call_num=call_num)
|
|
# compare the 2nd resource definition class & values
|
|
actual_def = mock_api.call_args_list[call_num][0][1]
|
|
expected_def = expected_defs[1]
|
|
self._compare_def(expected_def, actual_def)
|
|
|
|
def assert_called_with_def_and_dict(self, mock_api,
|
|
expected_def, expected_dict,
|
|
call_num=0):
|
|
# verify the api & resource definition
|
|
self.assert_called_with_def(mock_api, expected_def,
|
|
call_num=call_num)
|
|
# compare the 2nd api parameter which is a dictionary
|
|
actual_dict = mock_api.call_args_list[call_num][0][0].body
|
|
self.assertEqual(expected_dict, actual_dict)
|
|
|
|
|
|
class TestPolicyDomain(NsxPolicyLibTestCase):
|
|
|
|
def setUp(self, *args, **kwargs):
|
|
super(TestPolicyDomain, self).setUp()
|
|
self.resourceApi = self.policy_lib.domain
|
|
|
|
def test_create_with_id(self):
|
|
name = 'd1'
|
|
description = 'desc'
|
|
id = '111'
|
|
with mock.patch.object(self.policy_api,
|
|
"create_or_update") as api_call:
|
|
self.resourceApi.create_or_overwrite(name,
|
|
domain_id=id,
|
|
description=description,
|
|
tenant=TEST_TENANT)
|
|
expected_def = policy_defs.DomainDef(domain_id=id,
|
|
name=name,
|
|
description=description,
|
|
tenant=TEST_TENANT)
|
|
self.assert_called_with_def(api_call, expected_def)
|
|
|
|
def test_create_without_id(self):
|
|
name = 'd1'
|
|
description = 'desc'
|
|
with mock.patch.object(self.policy_api,
|
|
"create_or_update") as api_call:
|
|
self.resourceApi.create_or_overwrite(name, description=description,
|
|
tenant=TEST_TENANT)
|
|
expected_def = policy_defs.DomainDef(domain_id=mock.ANY,
|
|
name=name,
|
|
description=description,
|
|
tenant=TEST_TENANT)
|
|
self.assert_called_with_def(api_call, expected_def)
|
|
|
|
def test_delete(self):
|
|
id = '111'
|
|
with mock.patch.object(self.policy_api, "delete") as api_call:
|
|
self.resourceApi.delete(id, tenant=TEST_TENANT)
|
|
expected_def = policy_defs.DomainDef(domain_id=id,
|
|
tenant=TEST_TENANT)
|
|
self.assert_called_with_def(api_call, expected_def)
|
|
|
|
def test_get(self):
|
|
id = '111'
|
|
with mock.patch.object(self.policy_api, "get") as api_call:
|
|
self.resourceApi.get(id, tenant=TEST_TENANT)
|
|
expected_def = policy_defs.DomainDef(domain_id=id,
|
|
tenant=TEST_TENANT)
|
|
self.assert_called_with_def(api_call, expected_def)
|
|
|
|
def test_get_by_name(self):
|
|
name = 'd1'
|
|
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 = policy_defs.DomainDef(tenant=TEST_TENANT)
|
|
self.assert_called_with_def(api_call, expected_def)
|
|
|
|
def test_list(self):
|
|
with mock.patch.object(self.policy_api, "list") as api_call:
|
|
self.resourceApi.list(tenant=TEST_TENANT)
|
|
expected_def = policy_defs.DomainDef(tenant=TEST_TENANT)
|
|
self.assert_called_with_def(api_call, expected_def)
|
|
|
|
def test_update(self):
|
|
id = '111'
|
|
name = 'new name'
|
|
description = 'new desc'
|
|
with mock.patch.object(self.policy_api,
|
|
"create_or_update") as update_call:
|
|
self.resourceApi.update(id,
|
|
name=name,
|
|
description=description,
|
|
tenant=TEST_TENANT)
|
|
expected_def = policy_defs.DomainDef(domain_id=id,
|
|
tenant=TEST_TENANT)
|
|
expected_dict = {'display_name': name,
|
|
'description': description}
|
|
self.assert_called_with_def_and_dict(
|
|
update_call, expected_def, expected_dict)
|
|
|
|
|
|
class TestPolicyGroup(NsxPolicyLibTestCase):
|
|
|
|
def setUp(self, *args, **kwargs):
|
|
super(TestPolicyGroup, self).setUp()
|
|
self.resourceApi = self.policy_lib.group
|
|
|
|
def test_create_with_id(self):
|
|
domain_id = '111'
|
|
name = 'g1'
|
|
description = 'desc'
|
|
id = '222'
|
|
with mock.patch.object(self.policy_api,
|
|
"create_or_update") as api_call:
|
|
self.resourceApi.create_or_overwrite(name,
|
|
domain_id,
|
|
group_id=id,
|
|
description=description,
|
|
tenant=TEST_TENANT)
|
|
expected_def = policy_defs.GroupDef(domain_id=domain_id,
|
|
group_id=id,
|
|
name=name,
|
|
description=description,
|
|
conditions=[],
|
|
tenant=TEST_TENANT)
|
|
self.assert_called_with_def(api_call, expected_def)
|
|
|
|
def test_create_without_id(self):
|
|
domain_id = '111'
|
|
name = 'g1'
|
|
description = 'desc'
|
|
with mock.patch.object(self.policy_api,
|
|
"create_or_update") as api_call:
|
|
self.resourceApi.create_or_overwrite(name, domain_id,
|
|
description=description,
|
|
tenant=TEST_TENANT)
|
|
expected_def = policy_defs.GroupDef(domain_id=domain_id,
|
|
group_id=mock.ANY,
|
|
name=name,
|
|
description=description,
|
|
conditions=[],
|
|
tenant=TEST_TENANT)
|
|
self.assert_called_with_def(api_call, expected_def)
|
|
|
|
def test_create_with_condition(self):
|
|
domain_id = '111'
|
|
name = 'g1'
|
|
description = 'desc'
|
|
cond_val = '123'
|
|
cond_op = policy_constants.CONDITION_OP_EQUALS
|
|
cond_member_type = policy_constants.CONDITION_MEMBER_NET
|
|
cond_key = policy_constants.CONDITION_KEY_TAG
|
|
with mock.patch.object(self.policy_api,
|
|
"create_or_update") as api_call:
|
|
self.resourceApi.create_or_overwrite(
|
|
name, domain_id, description=description,
|
|
cond_val=cond_val,
|
|
cond_op=cond_op,
|
|
cond_member_type=cond_member_type,
|
|
cond_key=cond_key,
|
|
tenant=TEST_TENANT)
|
|
exp_cond = policy_defs.Condition(value=cond_val,
|
|
key=cond_key,
|
|
operator=cond_op,
|
|
member_type=cond_member_type)
|
|
expected_def = policy_defs.GroupDef(domain_id=domain_id,
|
|
group_id=mock.ANY,
|
|
name=name,
|
|
description=description,
|
|
conditions=[exp_cond],
|
|
tenant=TEST_TENANT)
|
|
self.assert_called_with_def(api_call, expected_def)
|
|
|
|
def test_delete(self):
|
|
domain_id = '111'
|
|
id = '222'
|
|
with mock.patch.object(self.policy_api, "delete") as api_call:
|
|
self.resourceApi.delete(domain_id, id, tenant=TEST_TENANT)
|
|
expected_def = policy_defs.GroupDef(domain_id=domain_id,
|
|
group_id=id,
|
|
tenant=TEST_TENANT)
|
|
self.assert_called_with_def(api_call, expected_def)
|
|
|
|
def test_get(self):
|
|
domain_id = '111'
|
|
id = '222'
|
|
with mock.patch.object(self.policy_api, "get") as api_call:
|
|
self.resourceApi.get(domain_id, id, tenant=TEST_TENANT)
|
|
expected_def = policy_defs.GroupDef(domain_id=domain_id,
|
|
group_id=id,
|
|
tenant=TEST_TENANT)
|
|
self.assert_called_with_def(api_call, expected_def)
|
|
|
|
def test_get_by_name(self):
|
|
domain_id = '111'
|
|
name = 'g1'
|
|
with mock.patch.object(
|
|
self.policy_api, "list",
|
|
return_value={'results': [{'display_name': name}]}) as api_call:
|
|
obj = self.resourceApi.get_by_name(domain_id, name,
|
|
tenant=TEST_TENANT)
|
|
self.assertIsNotNone(obj)
|
|
expected_def = policy_defs.GroupDef(domain_id, tenant=TEST_TENANT)
|
|
self.assert_called_with_def(api_call, expected_def)
|
|
|
|
def test_list(self):
|
|
domain_id = '111'
|
|
with mock.patch.object(self.policy_api, "list") as api_call:
|
|
self.resourceApi.list(domain_id, tenant=TEST_TENANT)
|
|
expected_def = policy_defs.GroupDef(domain_id=domain_id,
|
|
tenant=TEST_TENANT)
|
|
self.assert_called_with_def(api_call, expected_def)
|
|
|
|
def test_update(self):
|
|
domain_id = '111'
|
|
id = '222'
|
|
name = 'new name'
|
|
description = 'new desc'
|
|
with mock.patch.object(self.policy_api,
|
|
"create_or_update") as update_call:
|
|
self.resourceApi.update(domain_id, id,
|
|
name=name,
|
|
description=description,
|
|
tenant=TEST_TENANT)
|
|
expected_def = policy_defs.GroupDef(domain_id=domain_id,
|
|
group_id=id,
|
|
tenant=TEST_TENANT)
|
|
expected_dict = {'display_name': name,
|
|
'description': description}
|
|
self.assert_called_with_def_and_dict(
|
|
update_call, expected_def, expected_dict)
|
|
|
|
def test_update_condition(self):
|
|
domain_id = '111'
|
|
id = '222'
|
|
cond_val = '123'
|
|
with mock.patch.object(self.policy_api, "get",
|
|
return_value={}) as get_call,\
|
|
mock.patch.object(self.policy_api,
|
|
"create_or_update") as update_call:
|
|
self.resourceApi.update_condition(domain_id, id,
|
|
cond_val=cond_val,
|
|
tenant=TEST_TENANT)
|
|
expected_def = policy_defs.GroupDef(domain_id=domain_id,
|
|
group_id=id,
|
|
tenant=TEST_TENANT)
|
|
exp_cond = {'resource_type': 'Condition',
|
|
'member_type': policy_constants.CONDITION_MEMBER_PORT,
|
|
'key': policy_constants.CONDITION_KEY_TAG,
|
|
'value': cond_val,
|
|
'operator': policy_constants.CONDITION_OP_EQUALS}
|
|
expected_dict = {'expression': [exp_cond]}
|
|
self.assert_called_with_def(get_call, expected_def)
|
|
self.assert_called_with_def_and_dict(
|
|
update_call, expected_def, expected_dict)
|
|
|
|
def test_remove_condition(self):
|
|
domain_id = '111'
|
|
id = '222'
|
|
old_cond = {'resource_type': 'Condition',
|
|
'member_type': policy_constants.CONDITION_MEMBER_PORT,
|
|
'key': policy_constants.CONDITION_KEY_TAG,
|
|
'value': 'abc',
|
|
'operator': policy_constants.CONDITION_OP_EQUALS}
|
|
with mock.patch.object(self.policy_api, "get",
|
|
return_value={'expression': [old_cond]}) as get_call,\
|
|
mock.patch.object(self.policy_api,
|
|
"create_or_update") as update_call:
|
|
self.resourceApi.update_condition(domain_id, id,
|
|
cond_val=None,
|
|
tenant=TEST_TENANT)
|
|
expected_def = policy_defs.GroupDef(domain_id=domain_id,
|
|
group_id=id,
|
|
tenant=TEST_TENANT)
|
|
expected_dict = {'expression': []}
|
|
self.assert_called_with_def(get_call, expected_def)
|
|
self.assert_called_with_def_and_dict(
|
|
update_call, expected_def, expected_dict)
|
|
|
|
|
|
class TestPolicyService(NsxPolicyLibTestCase):
|
|
|
|
def setUp(self, *args, **kwargs):
|
|
super(TestPolicyService, self).setUp()
|
|
self.resourceApi = self.policy_lib.service
|
|
|
|
def test_create(self):
|
|
name = 's1'
|
|
description = 'desc'
|
|
protocol = policy_constants.TCP
|
|
dest_ports = [81, 82]
|
|
with mock.patch.object(self.policy_api,
|
|
"create_with_parent") as api_call:
|
|
self.resourceApi.create_or_overwrite(name,
|
|
description=description,
|
|
protocol=protocol,
|
|
dest_ports=dest_ports,
|
|
tenant=TEST_TENANT)
|
|
exp_srv_def = policy_defs.ServiceDef(service_id=mock.ANY,
|
|
name=name,
|
|
description=description,
|
|
tenant=TEST_TENANT)
|
|
exp_entry_def = policy_defs.L4ServiceEntryDef(
|
|
service_id=mock.ANY,
|
|
name=name,
|
|
description=description,
|
|
protocol=protocol,
|
|
dest_ports=dest_ports,
|
|
tenant=TEST_TENANT)
|
|
self.assert_called_with_defs(
|
|
api_call, [exp_srv_def, exp_entry_def])
|
|
|
|
def test_delete(self):
|
|
id = '111'
|
|
with mock.patch.object(self.policy_api, "delete") as api_call:
|
|
self.resourceApi.delete(id, tenant=TEST_TENANT)
|
|
expected_def = policy_defs.ServiceDef(service_id=id,
|
|
tenant=TEST_TENANT)
|
|
self.assert_called_with_def(api_call, expected_def)
|
|
|
|
def test_get(self):
|
|
id = '111'
|
|
with mock.patch.object(self.policy_api, "get") as api_call:
|
|
self.resourceApi.get(id, tenant=TEST_TENANT)
|
|
expected_def = policy_defs.ServiceDef(service_id=id,
|
|
tenant=TEST_TENANT)
|
|
self.assert_called_with_def(api_call, expected_def)
|
|
|
|
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 = policy_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") as api_call:
|
|
self.resourceApi.list(tenant=TEST_TENANT)
|
|
expected_def = policy_defs.ServiceDef(tenant=TEST_TENANT)
|
|
self.assert_called_with_def(api_call, expected_def)
|
|
|
|
def test_update(self):
|
|
id = '111'
|
|
name = 'new name'
|
|
description = 'new desc'
|
|
with mock.patch.object(self.policy_api, "get",
|
|
return_value={}) as get_call,\
|
|
mock.patch.object(self.policy_api,
|
|
"create_or_update") as update_call:
|
|
self.resourceApi.update(id,
|
|
name=name,
|
|
description=description,
|
|
tenant=TEST_TENANT)
|
|
expected_def = policy_defs.ServiceDef(service_id=id,
|
|
tenant=TEST_TENANT)
|
|
expected_dict = {'display_name': name,
|
|
'description': description,
|
|
'service_entries': []}
|
|
self.assert_called_with_def(get_call, expected_def)
|
|
self.assert_called_with_def_and_dict(
|
|
update_call, expected_def, expected_dict)
|
|
|
|
def test_update_entry(self):
|
|
id = '111'
|
|
protocol = 'udp'
|
|
dest_ports = [555]
|
|
service_entry_id = '222'
|
|
service_entry = {'id': service_entry_id}
|
|
|
|
with mock.patch.object(
|
|
self.policy_api, "get",
|
|
return_value={'service_entries': [service_entry]}) as get_call,\
|
|
mock.patch.object(self.policy_api,
|
|
"create_or_update") as update_call:
|
|
self.resourceApi.update(id,
|
|
protocol=protocol,
|
|
dest_ports=dest_ports,
|
|
tenant=TEST_TENANT)
|
|
# get will be called for the entire service
|
|
expected_def = policy_defs.ServiceDef(service_id=id,
|
|
tenant=TEST_TENANT)
|
|
self.assert_called_with_def(get_call, expected_def)
|
|
|
|
# update will be called for the service entry only
|
|
expected_entry_def = policy_defs.L4ServiceEntryDef(
|
|
service_id=id,
|
|
service_entry_id=service_entry_id,
|
|
tenant=TEST_TENANT)
|
|
expected_entry_dict = {'id': service_entry_id,
|
|
'l4_protocol': protocol.upper(),
|
|
'destination_ports': dest_ports}
|
|
self.assert_called_with_def_and_dict(
|
|
update_call, expected_entry_def, expected_entry_dict)
|
|
|
|
def test_update_all(self):
|
|
id = '111'
|
|
name = 'new name'
|
|
description = 'new desc'
|
|
protocol = 'udp'
|
|
dest_ports = [555]
|
|
service_entry_id = '222'
|
|
service_entry = {'id': service_entry_id}
|
|
|
|
with mock.patch.object(
|
|
self.policy_api, "get",
|
|
return_value={'service_entries': [service_entry]}) as get_call,\
|
|
mock.patch.object(self.policy_api,
|
|
"create_or_update") as update_call,\
|
|
mock.patch.object(self.policy_api, "list",
|
|
return_value={'results': []}):
|
|
self.resourceApi.update(id,
|
|
name=name,
|
|
description=description,
|
|
protocol=protocol,
|
|
dest_ports=dest_ports,
|
|
tenant=TEST_TENANT)
|
|
# get will be called for the entire service
|
|
expected_def = policy_defs.ServiceDef(service_id=id,
|
|
tenant=TEST_TENANT)
|
|
self.assert_called_with_def(get_call, expected_def)
|
|
|
|
# update will be called for the service and entry (2 calls)
|
|
expected_dict = {'display_name': name,
|
|
'description': description,
|
|
'service_entries': []}
|
|
self.assert_called_with_def_and_dict(
|
|
update_call, expected_def, expected_dict)
|
|
|
|
expected_entry_def = policy_defs.L4ServiceEntryDef(
|
|
service_id=id,
|
|
service_entry_id=service_entry_id,
|
|
tenant=TEST_TENANT)
|
|
expected_entry_dict = {'id': service_entry_id,
|
|
'display_name': name,
|
|
'description': description,
|
|
'l4_protocol': protocol.upper(),
|
|
'destination_ports': dest_ports}
|
|
self.assert_called_with_def_and_dict(
|
|
update_call, expected_entry_def, expected_entry_dict,
|
|
call_num=1)
|
|
|
|
|
|
class TestPolicyCommunicationProfile(NsxPolicyLibTestCase):
|
|
|
|
def setUp(self, *args, **kwargs):
|
|
super(TestPolicyCommunicationProfile, self).setUp()
|
|
self.resourceApi = self.policy_lib.comm_profile
|
|
|
|
def test_create(self):
|
|
name = 'c1'
|
|
description = 'desc'
|
|
service_id = '333'
|
|
action = 'DENY'
|
|
with mock.patch.object(self.policy_api,
|
|
"create_with_parent") as api_call:
|
|
self.resourceApi.create_or_overwrite(name, description=description,
|
|
services=[service_id],
|
|
action=action,
|
|
tenant=TEST_TENANT)
|
|
exp_srv_def = policy_defs.CommunicationProfileDef(
|
|
profile_id=mock.ANY,
|
|
name=name,
|
|
description=description,
|
|
tenant=TEST_TENANT)
|
|
exp_entry_def = policy_defs.CommunicationProfileEntryDef(
|
|
profile_id=mock.ANY,
|
|
name=name,
|
|
description=description,
|
|
services=[service_id],
|
|
action=action,
|
|
tenant=TEST_TENANT)
|
|
self.assert_called_with_defs(
|
|
api_call, [exp_srv_def, exp_entry_def])
|
|
|
|
def test_delete(self):
|
|
id = '111'
|
|
with mock.patch.object(self.policy_api, "delete") as api_call:
|
|
self.resourceApi.delete(id, tenant=TEST_TENANT)
|
|
expected_def = policy_defs.CommunicationProfileDef(
|
|
profile_id=id, tenant=TEST_TENANT)
|
|
self.assert_called_with_def(api_call, expected_def)
|
|
|
|
def test_get(self):
|
|
id = '111'
|
|
with mock.patch.object(self.policy_api, "get") as api_call:
|
|
self.resourceApi.get(id, tenant=TEST_TENANT)
|
|
expected_def = policy_defs.CommunicationProfileDef(
|
|
profile_id=id, tenant=TEST_TENANT)
|
|
self.assert_called_with_def(api_call, expected_def)
|
|
|
|
def test_get_by_name(self):
|
|
name = 'c1'
|
|
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 = policy_defs.CommunicationProfileDef(
|
|
tenant=TEST_TENANT)
|
|
self.assert_called_with_def(api_call, expected_def)
|
|
|
|
def test_list(self):
|
|
with mock.patch.object(self.policy_api, "list") as api_call:
|
|
self.resourceApi.list(tenant=TEST_TENANT)
|
|
expected_def = policy_defs.CommunicationProfileDef(
|
|
tenant=TEST_TENANT)
|
|
self.assert_called_with_def(api_call, expected_def)
|
|
|
|
def test_update(self):
|
|
id = '111'
|
|
name = 'new name'
|
|
description = 'new desc'
|
|
with mock.patch.object(self.policy_api, "get",
|
|
return_value={}) as get_call,\
|
|
mock.patch.object(self.policy_api,
|
|
"create_or_update") as update_call:
|
|
self.resourceApi.update(id,
|
|
name=name,
|
|
description=description,
|
|
tenant=TEST_TENANT)
|
|
expected_def = policy_defs.CommunicationProfileDef(
|
|
profile_id=id, tenant=TEST_TENANT)
|
|
expected_dict = {'display_name': name,
|
|
'description': description,
|
|
'communication_profile_entries': []}
|
|
self.assert_called_with_def(get_call, expected_def)
|
|
self.assert_called_with_def_and_dict(
|
|
update_call, expected_def, expected_dict)
|
|
|
|
def test_update_entry(self):
|
|
id = '111'
|
|
service_id = '333'
|
|
action = 'deny'
|
|
entry_id = '222'
|
|
profile_entry = {'id': entry_id}
|
|
entries_dict = {'communication_profile_entries': [profile_entry]}
|
|
|
|
with mock.patch.object(
|
|
self.policy_api, "get", return_value=entries_dict) as get_call,\
|
|
mock.patch.object(self.policy_api,
|
|
"create_or_update") as update_call:
|
|
self.resourceApi.update(id,
|
|
services=[service_id],
|
|
action=action,
|
|
tenant=TEST_TENANT)
|
|
# get will be called for the entire service
|
|
expected_def = policy_defs.CommunicationProfileDef(
|
|
profile_id=id, tenant=TEST_TENANT)
|
|
self.assert_called_with_def(get_call, expected_def)
|
|
|
|
# update will be called for the service entry only
|
|
expected_entry_def = policy_defs.CommunicationProfileEntryDef(
|
|
profile_id=id,
|
|
profile_entry_id=entry_id,
|
|
tenant=TEST_TENANT)
|
|
expected_entry_dict = {'id': entry_id,
|
|
'action': action.upper(),
|
|
'services': [service_id]}
|
|
self.assert_called_with_def_and_dict(
|
|
update_call, expected_entry_def, expected_entry_dict)
|
|
|
|
def test_update_all(self):
|
|
id = '111'
|
|
name = 'new name'
|
|
description = 'new desc'
|
|
service_id = '333'
|
|
action = 'deny'
|
|
entry_id = '222'
|
|
profile_entry = {'id': entry_id}
|
|
entries_dict = {'communication_profile_entries': [profile_entry]}
|
|
|
|
with mock.patch.object(
|
|
self.policy_api, "get", return_value=entries_dict) as get_call,\
|
|
mock.patch.object(self.policy_api,
|
|
"create_or_update") as update_call:
|
|
self.resourceApi.update(id,
|
|
name=name,
|
|
description=description,
|
|
services=[service_id],
|
|
action=action,
|
|
tenant=TEST_TENANT)
|
|
# get will be called for the entire service
|
|
expected_def = policy_defs.CommunicationProfileDef(
|
|
profile_id=id, tenant=TEST_TENANT)
|
|
self.assert_called_with_def(get_call, expected_def)
|
|
|
|
# update will be called for the service and entry (2 calls)
|
|
expected_dict = {'display_name': name,
|
|
'description': description,
|
|
'communication_profile_entries': []}
|
|
self.assert_called_with_def_and_dict(
|
|
update_call, expected_def, expected_dict)
|
|
|
|
expected_entry_def = policy_defs.CommunicationProfileEntryDef(
|
|
profile_id=id,
|
|
profile_entry_id=entry_id,
|
|
tenant=TEST_TENANT)
|
|
expected_entry_dict = {'id': entry_id,
|
|
'display_name': name,
|
|
'description': description,
|
|
'action': action.upper(),
|
|
'services': [service_id]}
|
|
self.assert_called_with_def_and_dict(
|
|
update_call, expected_entry_def, expected_entry_dict,
|
|
call_num=1)
|
|
|
|
|
|
class TestPolicyCommunicationMap(NsxPolicyLibTestCase):
|
|
|
|
def setUp(self, *args, **kwargs):
|
|
super(TestPolicyCommunicationMap, self).setUp()
|
|
self.resourceApi = self.policy_lib.comm_map
|
|
|
|
def test_create(self):
|
|
domain_id = '111'
|
|
name = 'cm1'
|
|
description = 'desc'
|
|
source_group = 'g1'
|
|
dest_group = 'g2'
|
|
seq_num = 7
|
|
profile_id = 'c1'
|
|
list_return_value = {'results': [{'sequence_number': 1}]}
|
|
with mock.patch.object(self.policy_api,
|
|
"create_or_update") as api_call,\
|
|
mock.patch.object(self.policy_api, "list",
|
|
return_value=list_return_value):
|
|
self.resourceApi.create_or_overwrite(name, domain_id,
|
|
description=description,
|
|
sequence_number=seq_num,
|
|
profile_id=profile_id,
|
|
source_groups=[source_group],
|
|
dest_groups=[dest_group],
|
|
tenant=TEST_TENANT)
|
|
expected_def = policy_defs.CommunicationMapEntryDef(
|
|
domain_id=domain_id,
|
|
map_id=mock.ANY,
|
|
name=name,
|
|
description=description,
|
|
sequence_number=seq_num,
|
|
profile_id=profile_id,
|
|
source_groups=[source_group],
|
|
dest_groups=[dest_group],
|
|
tenant=TEST_TENANT)
|
|
self.assert_called_with_def(api_call, expected_def)
|
|
|
|
def test_create_first_seqnum(self):
|
|
domain_id = '111'
|
|
name = 'cm1'
|
|
description = 'desc'
|
|
source_group = 'g1'
|
|
dest_group = 'g2'
|
|
profile_id = 'c1'
|
|
with mock.patch.object(self.policy_api,
|
|
"create_or_update") as api_call, \
|
|
mock.patch.object(self.resourceApi, "list", return_value=[]):
|
|
self.resourceApi.create_or_overwrite(name, domain_id,
|
|
description=description,
|
|
profile_id=profile_id,
|
|
source_groups=[source_group],
|
|
dest_groups=[dest_group],
|
|
tenant=TEST_TENANT)
|
|
|
|
expected_def = policy_defs.CommunicationMapEntryDef(
|
|
domain_id=domain_id,
|
|
map_id=mock.ANY,
|
|
name=name,
|
|
description=description,
|
|
sequence_number=1,
|
|
profile_id=profile_id,
|
|
source_groups=[source_group],
|
|
dest_groups=[dest_group],
|
|
tenant=TEST_TENANT)
|
|
|
|
self.assert_called_with_def(api_call, expected_def)
|
|
|
|
def test_create_without_seqnum(self):
|
|
domain_id = '111'
|
|
name = 'cm1'
|
|
description = 'desc'
|
|
source_group = 'g1'
|
|
dest_group = 'g2'
|
|
profile_id = 'c1'
|
|
with mock.patch.object(self.policy_api,
|
|
"create_with_parent") as api_call, \
|
|
mock.patch.object(self.resourceApi, "_get_last_seq_num",
|
|
return_value=-1):
|
|
self.resourceApi.create_or_overwrite(name, domain_id,
|
|
description=description,
|
|
profile_id=profile_id,
|
|
source_groups=[source_group],
|
|
dest_groups=[dest_group],
|
|
tenant=TEST_TENANT)
|
|
|
|
expected_map_def = policy_defs.CommunicationMapDef(
|
|
domain_id=domain_id,
|
|
tenant=TEST_TENANT)
|
|
|
|
expected_entry_def = policy_defs.CommunicationMapEntryDef(
|
|
domain_id=domain_id,
|
|
map_id=mock.ANY,
|
|
name=name,
|
|
description=description,
|
|
sequence_number=1,
|
|
profile_id=profile_id,
|
|
source_groups=[source_group],
|
|
dest_groups=[dest_group],
|
|
tenant=TEST_TENANT)
|
|
|
|
self.assert_called_with_defs(
|
|
api_call,
|
|
[expected_map_def, expected_entry_def])
|
|
|
|
def test_delete(self):
|
|
domain_id = '111'
|
|
id = '222'
|
|
with mock.patch.object(self.policy_api, "delete") as api_call:
|
|
self.resourceApi.delete(domain_id, id, tenant=TEST_TENANT)
|
|
expected_def = policy_defs.CommunicationMapEntryDef(
|
|
domain_id=domain_id,
|
|
map_id=id,
|
|
tenant=TEST_TENANT)
|
|
self.assert_called_with_def(api_call, expected_def)
|
|
|
|
def test_get(self):
|
|
domain_id = '111'
|
|
id = '222'
|
|
with mock.patch.object(self.policy_api, "get") as api_call:
|
|
self.resourceApi.get(domain_id, id, tenant=TEST_TENANT)
|
|
expected_def = policy_defs.CommunicationMapEntryDef(
|
|
domain_id=domain_id,
|
|
map_id=id,
|
|
tenant=TEST_TENANT)
|
|
self.assert_called_with_def(api_call, expected_def)
|
|
|
|
def test_get_by_name(self):
|
|
domain_id = '111'
|
|
name = 'cm1'
|
|
with mock.patch.object(
|
|
self.policy_api, "list",
|
|
return_value={'results': [{'display_name': name}]}) as api_call:
|
|
obj = self.resourceApi.get_by_name(domain_id, name,
|
|
tenant=TEST_TENANT)
|
|
self.assertIsNotNone(obj)
|
|
expected_def = policy_defs.CommunicationMapEntryDef(
|
|
domain_id,
|
|
tenant=TEST_TENANT)
|
|
self.assert_called_with_def(api_call, expected_def)
|
|
|
|
def test_list(self):
|
|
domain_id = '111'
|
|
with mock.patch.object(self.policy_api, "list") as api_call:
|
|
self.resourceApi.list(domain_id, tenant=TEST_TENANT)
|
|
expected_def = policy_defs.CommunicationMapEntryDef(
|
|
domain_id=domain_id,
|
|
tenant=TEST_TENANT)
|
|
self.assert_called_with_def(api_call, expected_def)
|
|
|
|
def test_update(self):
|
|
domain_id = '111'
|
|
id = '222'
|
|
name = 'new name'
|
|
description = 'new desc'
|
|
source_group = 'ng1'
|
|
dest_group = 'ng2'
|
|
profile_id = 'nc1'
|
|
with mock.patch.object(self.policy_api, "get",
|
|
return_value={}) as get_call,\
|
|
mock.patch.object(self.policy_api,
|
|
"create_or_update") as update_call:
|
|
self.resourceApi.update(domain_id, id,
|
|
name=name,
|
|
description=description,
|
|
profile_id=profile_id,
|
|
source_groups=[source_group],
|
|
dest_groups=[dest_group],
|
|
tenant=TEST_TENANT)
|
|
expected_def = policy_defs.CommunicationMapEntryDef(
|
|
domain_id=domain_id,
|
|
map_id=id,
|
|
tenant=TEST_TENANT)
|
|
sgroup_path = "/%s/domains/%s/groups/%s" % (
|
|
TEST_TENANT, domain_id, source_group)
|
|
dgroup_path = "/%s/domains/%s/groups/%s" % (
|
|
TEST_TENANT, domain_id, dest_group)
|
|
profile_path = "/%s/communication-profiles/%s" % (
|
|
TEST_TENANT, profile_id)
|
|
expected_dict = {'display_name': name,
|
|
'description': description,
|
|
'communication_profile_path': profile_path,
|
|
'source_groups': [sgroup_path],
|
|
'destination_groups': [dgroup_path]}
|
|
self.assert_called_with_def(get_call, expected_def)
|
|
self.assert_called_with_def_and_dict(
|
|
update_call, expected_def, expected_dict)
|
|
|
|
|
|
class TestPolicyEnforcementPoint(NsxPolicyLibTestCase):
|
|
|
|
def setUp(self, *args, **kwargs):
|
|
super(TestPolicyEnforcementPoint, self).setUp()
|
|
self.resourceApi = self.policy_lib.enforcement_point
|
|
|
|
def test_create(self):
|
|
name = 'ep'
|
|
description = 'desc'
|
|
ip_address = '1.1.1.1'
|
|
username = 'admin'
|
|
password = 'zzz'
|
|
with mock.patch.object(self.policy_api,
|
|
"create_or_update") as api_call:
|
|
self.resourceApi.create_or_overwrite(
|
|
name, description=description,
|
|
ip_address=ip_address,
|
|
username=username,
|
|
password=password,
|
|
tenant=TEST_TENANT)
|
|
|
|
expected_def = policy_defs.EnforcementPointDef(
|
|
ep_id=mock.ANY,
|
|
name=name,
|
|
description=description,
|
|
ip_address=ip_address,
|
|
username=username,
|
|
password=password,
|
|
tenant=TEST_TENANT)
|
|
self.assert_called_with_def(api_call, expected_def)
|
|
|
|
def test_delete(self):
|
|
id = '111'
|
|
with mock.patch.object(self.policy_api, "delete") as api_call:
|
|
self.resourceApi.delete(id, tenant=TEST_TENANT)
|
|
expected_def = policy_defs.EnforcementPointDef(ep_id=id,
|
|
tenant=TEST_TENANT)
|
|
self.assert_called_with_def(api_call, expected_def)
|
|
|
|
def test_get(self):
|
|
id = '111'
|
|
with mock.patch.object(self.policy_api, "get") as api_call:
|
|
self.resourceApi.get(id, tenant=TEST_TENANT)
|
|
expected_def = policy_defs.EnforcementPointDef(ep_id=id,
|
|
tenant=TEST_TENANT)
|
|
self.assert_called_with_def(api_call, expected_def)
|
|
|
|
def test_get_by_name(self):
|
|
name = 'ep1'
|
|
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 = policy_defs.EnforcementPointDef(tenant=TEST_TENANT)
|
|
self.assert_called_with_def(api_call, expected_def)
|
|
|
|
def test_list(self):
|
|
with mock.patch.object(self.policy_api, "list") as api_call:
|
|
self.resourceApi.list(tenant=TEST_TENANT)
|
|
expected_def = policy_defs.EnforcementPointDef(tenant=TEST_TENANT)
|
|
self.assert_called_with_def(api_call, expected_def)
|
|
|
|
def test_update(self):
|
|
id = '111'
|
|
name = 'new name'
|
|
username = 'admin'
|
|
password = 'zzz'
|
|
with mock.patch.object(self.policy_api,
|
|
"create_or_update") as update_call:
|
|
self.resourceApi.update(id,
|
|
name=name,
|
|
username=username,
|
|
password=password,
|
|
tenant=TEST_TENANT)
|
|
expected_def = policy_defs.EnforcementPointDef(ep_id=id,
|
|
tenant=TEST_TENANT)
|
|
expected_dict = {'display_name': name,
|
|
'username': username,
|
|
'password': password}
|
|
self.assert_called_with_def_and_dict(
|
|
update_call, expected_def, expected_dict)
|
|
|
|
|
|
class TestPolicyDeploymentMap(NsxPolicyLibTestCase):
|
|
|
|
def setUp(self, *args, **kwargs):
|
|
super(TestPolicyDeploymentMap, self).setUp()
|
|
self.resourceApi = self.policy_lib.deployment_map
|
|
|
|
def test_create(self):
|
|
name = 'map1'
|
|
description = 'desc'
|
|
domain_id = 'domain1'
|
|
ep_id = 'ep1'
|
|
with mock.patch.object(self.policy_api,
|
|
"create_or_update") as api_call:
|
|
self.resourceApi.create_or_overwrite(name,
|
|
description=description,
|
|
ep_id=ep_id,
|
|
domain_id=domain_id,
|
|
tenant=TEST_TENANT)
|
|
expected_def = policy_defs.DeploymentMapDef(
|
|
map_id=mock.ANY,
|
|
name=name,
|
|
description=description,
|
|
ep_id=ep_id,
|
|
domain_id=domain_id,
|
|
tenant=TEST_TENANT)
|
|
self.assert_called_with_def(api_call, expected_def)
|
|
|
|
def test_delete(self):
|
|
id = '111'
|
|
with mock.patch.object(self.policy_api, "delete") as api_call:
|
|
self.resourceApi.delete(id, tenant=TEST_TENANT)
|
|
expected_def = policy_defs.DeploymentMapDef(map_id=id,
|
|
tenant=TEST_TENANT)
|
|
self.assert_called_with_def(api_call, expected_def)
|
|
|
|
def test_get(self):
|
|
id = '111'
|
|
with mock.patch.object(self.policy_api, "get") as api_call:
|
|
self.resourceApi.get(id, tenant=TEST_TENANT)
|
|
expected_def = policy_defs.DeploymentMapDef(map_id=id,
|
|
tenant=TEST_TENANT)
|
|
self.assert_called_with_def(api_call, expected_def)
|
|
|
|
def test_get_by_name(self):
|
|
name = 'ep1'
|
|
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 = policy_defs.DeploymentMapDef(tenant=TEST_TENANT)
|
|
self.assert_called_with_def(api_call, expected_def)
|
|
|
|
def test_list(self):
|
|
with mock.patch.object(self.policy_api, "list") as api_call:
|
|
self.resourceApi.list(tenant=TEST_TENANT)
|
|
expected_def = policy_defs.DeploymentMapDef(tenant=TEST_TENANT)
|
|
self.assert_called_with_def(api_call, expected_def)
|
|
|
|
def test_update(self):
|
|
id = '111'
|
|
name = 'new name'
|
|
domain_id = 'domain2'
|
|
ep_id = 'ep2'
|
|
with mock.patch.object(self.policy_api,
|
|
"create_or_update") as update_call:
|
|
self.resourceApi.update(id,
|
|
name=name,
|
|
ep_id=ep_id,
|
|
domain_id=domain_id,
|
|
tenant=TEST_TENANT)
|
|
expected_def = policy_defs.DeploymentMapDef(map_id=id,
|
|
tenant=TEST_TENANT)
|
|
domain_path = "/%s/domains/%s" % (TEST_TENANT, domain_id)
|
|
ep_path = ("/%s/deploymentzones/default-deployment-zone/"
|
|
"enforcementpoints/%s" % (TEST_TENANT, ep_id))
|
|
expected_dict = {'display_name': name,
|
|
'enforcement_point_paths': [ep_path],
|
|
'domain_path': domain_path}
|
|
self.assert_called_with_def_and_dict(
|
|
update_call, expected_def, expected_dict)
|