x/vmware-nsxlib
Code Issues Proposed changes
vmware-nsxlib/vmware_nsxlib/tests/unit/v3/policy/test_resources.py
Abhishek Raut a0731e2d69 Add API to retrieve realized info for IpAllocations 
Add API to retrieve realized info for IpAllocations
ability to retrieve the allocated IpAddress from
IpPool from the realized information.

In addition to the new APIs, make IpAddress as an
optional param while allocating IP from IpPool.

Change-Id: If12f292a6af60cd0e6abd8441e18f5e2b9efccb9
2019-01-22 19:39:52 -08:00

3897 lines
160 KiB
Python
Raw Blame History

# 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
from vmware_nsxlib.tests.unit.v3 import nsxlib_testcase
from vmware_nsxlib.tests.unit.v3.policy import policy_testcase
from vmware_nsxlib.v3 import exceptions as nsxlib_exc
from vmware_nsxlib.v3 import nsx_constants
from vmware_nsxlib.v3 import policy
from vmware_nsxlib.v3.policy import constants
from vmware_nsxlib.v3.policy import core_defs
from vmware_nsxlib.v3.policy import core_resources
from vmware_nsxlib.v3.policy import lb_defs
TEST_TENANT = 'test'
class NsxPolicyLibTestCase(policy_testcase.TestPolicyApi):
def setUp(self, *args, **kwargs):
super(NsxPolicyLibTestCase, self).setUp()
nsxlib_config = nsxlib_testcase.get_default_nsxlib_config(
allow_passthrough=kwargs.get('allow_passthrough', True))
# Mock the nsx-lib for the passthrough api
with mock.patch('vmware_nsxlib.v3.NsxLib'):
self.policy_lib = policy.NsxPolicyLib(nsxlib_config)
self.policy_api = self.policy_lib.policy_api
self.policy_api.client = self.client
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.get_tenant(), actual_def.get_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
def_list = mock_api.call_args_list[call_num][0][1]
if not isinstance(def_list, list):
def_list = [def_list]
for i in range(1, len(expected_defs)):
actual_def = def_list[i - 1]
expected_def = expected_defs[i]
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 = core_defs.DomainDef(domain_id=id,
name=name,
description=description,
tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
def test_minimalistic_create(self):
name = 'test'
with mock.patch.object(self.policy_api,
"create_or_update") as api_call:
self.resourceApi.create_or_overwrite(name,
tenant=TEST_TENANT)
expected_def = core_defs.DomainDef(domain_id=mock.ANY,
name=name,
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 = core_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 = core_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 = core_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 = core_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 = core_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 = core_defs.DomainDef(domain_id=id,
name=name,
description=description,
tenant=TEST_TENANT)
self.assert_called_with_def(update_call, expected_def)
def test_unset(self):
domain_id = '111'
self.resourceApi.update(domain_id,
description=None,
tags=None,
tenant=TEST_TENANT)
expected_body = {'id': domain_id,
'resource_type': 'Domain',
'description': None,
'tags': None}
self.assert_json_call('PATCH', self.client,
'%s/domains/%s' % (TEST_TENANT, domain_id),
data=expected_body)
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 = core_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 = core_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 = constants.CONDITION_OP_EQUALS
cond_member_type = constants.CONDITION_MEMBER_VM
cond_key = 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 = core_defs.Condition(value=cond_val,
key=cond_key,
operator=cond_op,
member_type=cond_member_type)
expected_def = core_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_create_with_simple_condition(self):
domain_id = '111'
name = 'g1'
description = 'desc'
cond_val = '123'
cond_op = constants.CONDITION_OP_EQUALS
cond_member_type = constants.CONDITION_MEMBER_VM
cond_key = constants.CONDITION_KEY_TAG
cond = self.resourceApi.build_condition(
cond_val=cond_val,
cond_op=cond_op,
cond_member_type=cond_member_type,
cond_key=cond_key)
with mock.patch.object(self.policy_api,
"create_or_update") as api_call:
self.resourceApi.create_or_overwrite_with_conditions(
name, domain_id, description=description,
conditions=[cond],
tenant=TEST_TENANT)
exp_cond = core_defs.Condition(value=cond_val,
key=cond_key,
operator=cond_op,
member_type=cond_member_type)
expected_def = core_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_create_with_nested_condition(self):
domain_id = '111'
name = 'g1'
description = 'desc'
cond_val1 = '123'
cond_val2 = '456'
cond_op = constants.CONDITION_OP_EQUALS
cond_member_type = constants.CONDITION_MEMBER_VM
cond_key = constants.CONDITION_KEY_TAG
cond1 = self.resourceApi.build_condition(
cond_val=cond_val1,
cond_op=cond_op,
cond_member_type=cond_member_type,
cond_key=cond_key)
cond2 = self.resourceApi.build_condition(
cond_val=cond_val2,
cond_op=cond_op,
cond_member_type=cond_member_type,
cond_key=cond_key)
nested = self.resourceApi.build_nested_condition(
conditions=[cond1, cond2])
with mock.patch.object(self.policy_api,
"create_or_update") as api_call:
self.resourceApi.create_or_overwrite_with_conditions(
name, domain_id, description=description,
conditions=[nested],
tenant=TEST_TENANT)
exp_cond1 = core_defs.Condition(value=cond_val1,
key=cond_key,
operator=cond_op,
member_type=cond_member_type)
exp_cond2 = core_defs.Condition(value=cond_val2,
key=cond_key,
operator=cond_op,
member_type=cond_member_type)
and_cond = core_defs.ConjunctionOperator()
nested_cond = core_defs.NestedExpression(
expressions=[exp_cond1, and_cond, exp_cond2])
expected_def = core_defs.GroupDef(domain_id=domain_id,
group_id=mock.ANY,
name=name,
description=description,
conditions=[nested_cond],
tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
def test_create_with_ip_expression(self):
domain_id = '111'
name = 'g1'
description = 'desc'
cidr = '1.1.1.0/24'
cond = self.resourceApi.build_ip_address_expression([cidr])
with mock.patch.object(self.policy_api,
"create_or_update") as api_call:
self.resourceApi.create_or_overwrite_with_conditions(
name, domain_id, description=description,
conditions=[cond],
tenant=TEST_TENANT)
exp_cond = core_defs.IPAddressExpression([cidr])
expected_def = core_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 = core_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 = core_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 = core_defs.GroupDef(domain_id=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 = core_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 = core_defs.GroupDef(domain_id=domain_id,
group_id=id,
name=name,
description=description,
tenant=TEST_TENANT)
self.assert_called_with_def(
update_call, expected_def)
def test_unset(self):
domain_id = '111'
group_id = '222'
description = 'new'
self.resourceApi.update(domain_id,
group_id,
name=None,
description=description,
tenant=TEST_TENANT)
expected_body = {'id': group_id,
'resource_type': 'Group',
'display_name': None,
'description': description}
self.assert_json_call('PATCH', self.client,
'%s/domains/%s/groups/%s' % (TEST_TENANT,
domain_id,
group_id),
data=expected_body)
def test_get_realized(self):
domain_id = 'd1'
group_id = 'g1'
result = [{'state': constants.STATE_REALIZED,
'entity_type': 'RealizedGroup'}]
with mock.patch.object(
self.policy_api, "get_realized_entities",
return_value=result) as api_get:
state = self.resourceApi.get_realized_state(
domain_id, group_id, tenant=TEST_TENANT)
self.assertEqual(constants.STATE_REALIZED, state)
path = "/%s/domains/%s/groups/%s" % (
TEST_TENANT, domain_id, group_id)
api_get.assert_called_once_with(path)
def test_get_realized_multiple_results_get_default(self):
domain_id = 'd1'
group_id = 'g1'
result = [{'state': constants.STATE_UNREALIZED,
'entity_type': 'NotRealizedGroup'},
{'state': constants.STATE_REALIZED,
'entity_type': 'RealizedGroup'}]
with mock.patch.object(
self.policy_api, "get_realized_entities",
return_value=result) as api_get:
state = self.resourceApi.get_realized_state(
domain_id, group_id, tenant=TEST_TENANT)
self.assertEqual(constants.STATE_UNREALIZED, state)
path = "/%s/domains/%s/groups/%s" % (
TEST_TENANT, domain_id, group_id)
api_get.assert_called_once_with(path)
def test_get_realized_multiple_results_get_specific(self):
domain_id = 'd1'
group_id = 'g1'
result = [{'state': constants.STATE_UNREALIZED,
'entity_type': 'NotRealizedGroup'},
{'state': constants.STATE_REALIZED,
'entity_type': 'RealizedGroup'}]
with mock.patch.object(
self.policy_api, "get_realized_entities",
return_value=result) as api_get:
state = self.resourceApi.get_realized_state(
domain_id, group_id, entity_type='RealizedGroup',
tenant=TEST_TENANT)
self.assertEqual(constants.STATE_REALIZED, state)
path = "/%s/domains/%s/groups/%s" % (
TEST_TENANT, domain_id, group_id)
api_get.assert_called_once_with(path)
def test_get_realized_id(self):
domain_id = 'd1'
group_id = 'g1'
realized_id = 'realized_111'
result = [{'state': constants.STATE_REALIZED,
'entity_type': 'RealizedGroup',
'realization_specific_identifier': realized_id}]
with mock.patch.object(
self.policy_api, "get_realized_entities",
return_value=result) as api_get:
result_id = self.resourceApi.get_realized_id(
domain_id, group_id, tenant=TEST_TENANT)
self.assertEqual(realized_id, result_id)
path = "/%s/domains/%s/groups/%s" % (
TEST_TENANT, domain_id, group_id)
api_get.assert_called_once_with(path)
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 = constants.TCP
dest_ports = [81, 82]
tags = [{'scope': 'a', 'tag': 'b'}]
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,
tags=tags,
tenant=TEST_TENANT)
exp_srv_def = core_defs.ServiceDef(service_id=mock.ANY,
name=name,
description=description,
tags=tags,
tenant=TEST_TENANT)
exp_entry_def = core_defs.L4ServiceEntryDef(
service_id=mock.ANY,
entry_id='entry',
name='entry',
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 = core_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 = core_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 = 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") as api_call:
self.resourceApi.list(tenant=TEST_TENANT)
expected_def = core_defs.ServiceDef(tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
def test_update(self):
id = '111'
name = 'newName'
description = 'new desc'
protocol = 'tcp'
tags = [{'scope': 'a', 'tag': 'b'}]
entry_body = {'id': 'entry',
'l4_protocol': protocol}
with mock.patch.object(self.policy_api,
"get",
return_value=entry_body),\
mock.patch.object(self.policy_api,
"create_with_parent") as update_call:
self.resourceApi.update(id,
name=name,
description=description,
tags=tags,
tenant=TEST_TENANT)
service_def = core_defs.ServiceDef(service_id=id,
name=name,
description=description,
tags=tags,
tenant=TEST_TENANT)
entry_def = core_defs.L4ServiceEntryDef(
service_id=id,
entry_id='entry',
protocol=protocol,
tenant=TEST_TENANT)
self.assert_called_with_defs(update_call, [service_def, entry_def])
def test_update_all(self):
id = '111'
name = 'newName'
description = 'new desc'
protocol = 'udp'
dest_ports = [555]
entry_body = {'id': 'entry',
'l4_protocol': 'tcp'}
with mock.patch.object(self.policy_api,
"get",
return_value=entry_body),\
mock.patch.object(self.policy_api,
"create_with_parent") as update_call:
self.resourceApi.update(id,
name=name,
description=description,
protocol=protocol,
dest_ports=dest_ports,
tenant=TEST_TENANT)
service_def = core_defs.ServiceDef(service_id=id,
name=name,
description=description,
tenant=TEST_TENANT)
entry_def = core_defs.L4ServiceEntryDef(
service_id=id,
entry_id=mock.ANY,
protocol=protocol,
dest_ports=dest_ports,
tenant=TEST_TENANT)
self.assert_called_with_defs(
update_call, [service_def, entry_def])
def test_unset(self):
name = 'hello'
service_id = '111'
# Until policy PATCH is fixed to accept partial update, we
# call get on child entry
with mock.patch.object(
self.policy_api, "get",
return_value={'display_name': name}):
self.resourceApi.update(service_id,
description=None,
dest_ports=None,
tenant=TEST_TENANT)
expected_body = {'id': service_id,
'description': None,
'resource_type': 'Service',
'service_entries': [{
'display_name': name,
'id': 'entry',
'resource_type': 'L4PortSetServiceEntry',
'destination_ports': None}]
}
self.assert_json_call('PATCH', self.client,
'%s/services/%s' % (TEST_TENANT, service_id),
data=expected_body)
class TestPolicyIcmpService(NsxPolicyLibTestCase):
def setUp(self, *args, **kwargs):
super(TestPolicyIcmpService, self).setUp()
self.resourceApi = self.policy_lib.icmp_service
def test_create(self):
name = 's1'
description = 'desc'
icmp_type = 2
with mock.patch.object(self.policy_api,
"create_with_parent") as api_call:
self.resourceApi.create_or_overwrite(name,
description=description,
icmp_type=icmp_type,
tenant=TEST_TENANT)
exp_srv_def = core_defs.ServiceDef(service_id=mock.ANY,
name=name,
description=description,
tenant=TEST_TENANT)
exp_entry_def = core_defs.IcmpServiceEntryDef(
service_id=mock.ANY,
entry_id='entry',
name='entry',
version=4,
icmp_type=icmp_type,
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 = core_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 = core_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 = 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") as api_call:
self.resourceApi.list(tenant=TEST_TENANT)
expected_def = core_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={'id': 'entry',
'protocol': 'ICMPv4'}),\
mock.patch.object(self.policy_api,
"create_with_parent") as update_call:
self.resourceApi.update(id,
name=name,
description=description,
tenant=TEST_TENANT)
service_def = core_defs.ServiceDef(service_id=id,
name=name,
description=description,
tenant=TEST_TENANT)
entry_def = core_defs.IcmpServiceEntryDef(
service_id=id,
entry_id='entry',
version=4,
tenant=TEST_TENANT)
self.assert_called_with_defs(update_call, [service_def, entry_def])
def test_update_all(self):
id = '111'
name = 'newName'
description = 'new desc'
version = 6
icmp_type = 3
icmp_code = 3
with mock.patch.object(self.policy_api, "get",
return_value={'id': 'entry'}),\
mock.patch.object(self.policy_api,
"create_with_parent") as update_call:
self.resourceApi.update(id,
name=name,
description=description,
version=version,
icmp_type=icmp_type,
icmp_code=icmp_code,
tenant=TEST_TENANT)
# get will be called for the entire service
service_def = core_defs.ServiceDef(service_id=id,
name=name,
description=description,
tenant=TEST_TENANT)
entry_def = core_defs.IcmpServiceEntryDef(
service_id=id,
entry_id=mock.ANY,
version=version,
icmp_type=icmp_type,
icmp_code=icmp_code,
tenant=TEST_TENANT)
self.assert_called_with_defs(
update_call, [service_def, entry_def])
class TestPolicyIPProtocolService(NsxPolicyLibTestCase):
def setUp(self, *args, **kwargs):
super(TestPolicyIPProtocolService, self).setUp()
self.resourceApi = self.policy_lib.ip_protocol_service
def test_create(self):
name = 's1'
description = 'desc'
protocol_number = 2
with mock.patch.object(self.policy_api,
"create_with_parent") as api_call:
self.resourceApi.create_or_overwrite(
name,
description=description,
protocol_number=protocol_number,
tenant=TEST_TENANT)
exp_srv_def = core_defs.ServiceDef(service_id=mock.ANY,
name=name,
description=description,
tenant=TEST_TENANT)
exp_entry_def = core_defs.IPProtocolServiceEntryDef(
service_id=mock.ANY,
entry_id='entry',
name='entry',
protocol_number=protocol_number,
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 = core_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 = core_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 = 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") as api_call:
self.resourceApi.list(tenant=TEST_TENANT)
expected_def = core_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={'id': 'entry'}),\
mock.patch.object(self.policy_api,
"create_with_parent") as update_call:
self.resourceApi.update(id,
name=name,
description=description,
tenant=TEST_TENANT)
service_def = core_defs.ServiceDef(service_id=id,
name=name,
description=description,
tenant=TEST_TENANT)
entry_def = core_defs.IPProtocolServiceEntryDef(
service_id=id,
entry_id='entry',
tenant=TEST_TENANT)
self.assert_called_with_defs(update_call, [service_def, entry_def])
def test_update_all(self):
id = '111'
name = 'newName'
description = 'new desc'
protocol_number = 3
with mock.patch.object(self.policy_api, "get",
return_value={'id': 'entry'}),\
mock.patch.object(self.policy_api,
"create_with_parent") as service_update_call:
self.resourceApi.update(id,
name=name,
description=description,
protocol_number=protocol_number,
tenant=TEST_TENANT)
service_def = core_defs.ServiceDef(service_id=id,
name=name,
description=description,
tenant=TEST_TENANT)
entry_def = core_defs.IPProtocolServiceEntryDef(
service_id=id,
entry_id='entry',
protocol_number=protocol_number,
tenant=TEST_TENANT)
self.assert_called_with_defs(service_update_call,
[service_def, entry_def])
class TestPolicyCommunicationMap(NsxPolicyLibTestCase):
def setUp(self, *args, **kwargs):
super(TestPolicyCommunicationMap, self).setUp()
self.resourceApi = self.policy_lib.comm_map
def test_create_another(self):
domain_id = '111'
map_id = '222'
name = 'cm1'
description = 'desc'
source_group = 'g1'
dest_group = 'g2'
seq_num = 7
service_id = 'c1'
direction = nsx_constants.IN_OUT
get_return_value = {'rules': [{'sequence_number': 1}]}
with mock.patch.object(self.policy_api,
"create_with_parent") as api_call,\
mock.patch.object(self.policy_api, "get",
return_value=get_return_value):
self.resourceApi.create_or_overwrite(name, domain_id,
map_id=map_id,
description=description,
sequence_number=seq_num,
service_ids=[service_id],
source_groups=[source_group],
dest_groups=[dest_group],
direction=direction,
logged=True,
tenant=TEST_TENANT)
map_def = core_defs.CommunicationMapDef(
domain_id=domain_id,
map_id=map_id,
name=name,
description=description,
category=constants.CATEGORY_APPLICATION,
tenant=TEST_TENANT)
entry_def = core_defs.CommunicationMapEntryDef(
domain_id=domain_id,
map_id=map_id,
entry_id='entry',
name=name,
action=constants.ACTION_ALLOW,
description=description,
sequence_number=seq_num,
service_ids=[service_id],
source_groups=[source_group],
dest_groups=[dest_group],
direction=direction,
logged=True,
tenant=TEST_TENANT)
self.assert_called_with_defs(api_call, [map_def, entry_def])
def test_create_first_seqnum(self):
domain_id = '111'
map_id = '222'
name = 'cm1'
description = 'desc'
source_group = 'g1'
dest_group = 'g2'
service_id = 'c1'
category = 'Emergency'
get_return_value = {'rules': []}
with mock.patch.object(self.policy_api,
"create_with_parent") as api_call, \
mock.patch.object(self.resourceApi, "get",
return_value=get_return_value):
self.resourceApi.create_or_overwrite(name, domain_id,
map_id=map_id,
description=description,
service_ids=[service_id],
source_groups=[source_group],
dest_groups=[dest_group],
category=category,
logged=False,
tenant=TEST_TENANT)
map_def = core_defs.CommunicationMapDef(
domain_id=domain_id,
map_id=map_id,
name=name,
description=description,
category=category,
tenant=TEST_TENANT)
entry_def = core_defs.CommunicationMapEntryDef(
domain_id=domain_id,
map_id=map_id,
entry_id='entry',
name=name,
action=constants.ACTION_ALLOW,
direction=nsx_constants.IN_OUT,
description=description,
sequence_number=1,
service_ids=[service_id],
source_groups=[source_group],
dest_groups=[dest_group],
logged=False,
tenant=TEST_TENANT)
self.assert_called_with_defs(api_call, [map_def, entry_def])
def test_create_without_seqnum(self):
domain_id = '111'
name = 'cm1'
description = 'desc'
source_group = 'g1'
dest_group = 'g2'
service1_id = 'c1'
service2_id = 'c2'
with mock.patch.object(self.policy_api,
"create_with_parent") as api_call:
self.resourceApi.create_or_overwrite(name, domain_id,
description=description,
service_ids=[service1_id,
service2_id],
source_groups=[source_group],
dest_groups=[dest_group],
tenant=TEST_TENANT)
expected_map_def = core_defs.CommunicationMapDef(
domain_id=domain_id,
map_id=mock.ANY,
name=name,
description=description,
category=constants.CATEGORY_APPLICATION,
tenant=TEST_TENANT)
expected_entry_def = core_defs.CommunicationMapEntryDef(
domain_id=domain_id,
map_id=mock.ANY,
entry_id=mock.ANY,
action=constants.ACTION_ALLOW,
direction=nsx_constants.IN_OUT,
name=name,
description=description,
sequence_number=1,
service_ids=[service1_id, service2_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_create_map_only(self):
domain_id = '111'
name = 'cm1'
description = 'desc'
with mock.patch.object(self.policy_api,
"create_or_update") as api_call:
self.resourceApi.create_or_overwrite_map_only(
name, domain_id, description=description,
tenant=TEST_TENANT)
expected_map_def = core_defs.CommunicationMapDef(
domain_id=domain_id,
map_id=mock.ANY,
name=name,
description=description,
category=constants.CATEGORY_APPLICATION,
tenant=TEST_TENANT)
self.assert_called_with_def(
api_call, expected_map_def)
def test_create_entry(self):
domain_id = '111'
map_id = '333'
name = 'cm1'
description = 'desc'
source_group = 'g1'
dest_group = 'g2'
service1_id = 'c1'
service2_id = 'c2'
with mock.patch.object(self.policy_api,
"create_or_update") as api_call:
self.resourceApi.create_entry(name=name,
domain_id=domain_id,
map_id=map_id,
description=description,
service_ids=[service1_id,
service2_id],
source_groups=[source_group],
dest_groups=[dest_group],
sequence_number=1,
direction=nsx_constants.IN,
tenant=TEST_TENANT)
expected_entry_def = core_defs.CommunicationMapEntryDef(
domain_id=domain_id,
map_id=map_id,
entry_id=mock.ANY,
name=name,
action=constants.ACTION_ALLOW,
description=description,
sequence_number=1,
service_ids=[service1_id, service2_id],
source_groups=[source_group],
dest_groups=[dest_group],
direction=nsx_constants.IN,
logged=False,
tenant=TEST_TENANT)
self.assert_called_with_def(
api_call, expected_entry_def)
def test_create_entry_no_service(self):
domain_id = '111'
map_id = '333'
name = 'cm1'
description = 'desc'
source_group = 'g1'
dest_group = 'g2'
with mock.patch.object(self.policy_api,
"create_or_update") as api_call:
self.resourceApi.create_entry(name, domain_id, map_id,
description=description,
source_groups=[source_group],
dest_groups=[dest_group],
sequence_number=1,
tenant=TEST_TENANT)
expected_entry_def = core_defs.CommunicationMapEntryDef(
domain_id=domain_id,
map_id=map_id,
entry_id=mock.ANY,
name=name,
action=constants.ACTION_ALLOW,
direction=nsx_constants.IN_OUT,
description=description,
sequence_number=1,
service_ids=None,
source_groups=[source_group],
dest_groups=[dest_group],
logged=False,
tenant=TEST_TENANT)
self.assert_called_with_def(
api_call, expected_entry_def)
def test_create_entry_no_seq_num(self):
domain_id = '111'
map_id = '333'
name = 'cm1'
description = 'desc'
source_group = 'g1'
dest_group = 'g2'
service1_id = 'c1'
service2_id = 'c2'
seq_num = 1
ret_comm = {'rules': [{'sequence_number': seq_num}]}
with mock.patch.object(self.policy_api,
"create_or_update") as api_call,\
mock.patch.object(self.policy_api,
"get", return_value=ret_comm):
self.resourceApi.create_entry(name, domain_id, map_id,
description=description,
service_ids=[service1_id,
service2_id],
source_groups=[source_group],
dest_groups=[dest_group],
logged=False,
tenant=TEST_TENANT)
expected_entry_def = core_defs.CommunicationMapEntryDef(
domain_id=domain_id,
map_id=map_id,
entry_id=mock.ANY,
name=name,
action=constants.ACTION_ALLOW,
direction=nsx_constants.IN_OUT,
description=description,
service_ids=[service1_id, service2_id],
source_groups=[source_group],
dest_groups=[dest_group],
sequence_number=seq_num + 1,
logged=False,
tenant=TEST_TENANT)
self.assert_called_with_def(
api_call, expected_entry_def)
def test_create_with_entries(self):
domain_id = '111'
map_id = '222'
name = 'cm1'
description = 'desc'
source_group = 'g1'
dest_group = 'g2'
service_id = 'c1'
category = 'Emergency'
rule_id = 1
entry1 = self.resourceApi.build_entry(
'DHCP Reply', domain_id, map_id,
rule_id, sequence_number=rule_id, service_ids=[service_id],
action=constants.ACTION_DENY,
source_groups=None,
dest_groups=[dest_group],
direction=nsx_constants.IN)
rule_id += 1
entry2 = self.resourceApi.build_entry(
'DHCP Request', domain_id, map_id,
rule_id, sequence_number=rule_id, service_ids=None,
action=constants.ACTION_DENY,
source_groups=[source_group],
dest_groups=None,
direction=nsx_constants.OUT)
with mock.patch.object(self.policy_api,
"create_with_parent") as api_call:
self.resourceApi.create_with_entries(name, domain_id,
map_id=map_id,
description=description,
entries=[entry1, entry2],
category=category,
tenant=TEST_TENANT)
expected_def = core_defs.CommunicationMapDef(
domain_id=domain_id,
map_id=map_id,
name=name,
description=description,
category=category,
tenant=TEST_TENANT)
self.assert_called_with_defs(api_call,
[expected_def, entry1, entry2])
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 = core_defs.CommunicationMapDef(
domain_id=domain_id,
map_id=id,
tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
def test_entry(self):
domain_id = '111'
map_id = '222'
entry_id = '333'
with mock.patch.object(self.policy_api, "delete") as api_call:
self.resourceApi.delete_entry(domain_id, map_id, entry_id,
tenant=TEST_TENANT)
expected_def = core_defs.CommunicationMapEntryDef(
domain_id=domain_id,
map_id=map_id,
entry_id=entry_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 = core_defs.CommunicationMapDef(
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 = core_defs.CommunicationMapDef(
domain_id=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 = core_defs.CommunicationMapDef(
domain_id=domain_id,
tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
def test_update(self):
domain_id = '111'
map_id = '222'
name = 'new name'
description = 'new desc'
source_group = 'ng1'
dest_group = 'ng2'
service1_id = 'nc1'
service2_id = 'nc2'
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(domain_id, map_id,
name=name,
description=description,
service_ids=[service1_id, service2_id],
source_groups=[source_group],
dest_groups=[dest_group],
tenant=TEST_TENANT)
map_def = core_defs.CommunicationMapDef(
domain_id=domain_id,
map_id=map_id,
name=name,
description=description,
tenant=TEST_TENANT)
entry_def = core_defs.CommunicationMapEntryDef(
domain_id=domain_id,
map_id=map_id,
entry_id='entry',
service_ids=[service1_id, service2_id],
source_groups=[source_group],
dest_groups=[dest_group],
tenant=TEST_TENANT)
self.assert_called_with_defs(update_call, [map_def, entry_def])
def test_unset(self):
name = 'hello'
domain_id = 'test'
map_id = '111'
dest_groups = ['/infra/stuff']
# Until policy PATCH is fixed to accept partial update, we
# call get on child entry
with mock.patch.object(
self.policy_api, "get",
return_value={'display_name': name,
'source_groups': ['/infra/other/stuff'],
'destination_groups': dest_groups}):
self.resourceApi.update(domain_id, map_id,
description=None,
source_groups=None,
service_ids=None,
tenant=TEST_TENANT)
expected_body = {'id': map_id,
'description': None,
'resource_type': 'SecurityPolicy',
'rules': [{
'display_name': name,
'id': 'entry',
'resource_type': 'Rule',
'services': ["ANY"],
'source_groups': ["ANY"],
'destination_groups': dest_groups}]
}
url = '%s/domains/%s/security-policies/%s' % (TEST_TENANT,
domain_id,
map_id)
self.assert_json_call('PATCH', self.client, url, data=expected_body)
def test_update_entries_logged(self):
domain_id = '111'
map_id = '222'
dummy_map = {'rules': [{'logged': False}]}
updated_map = {'rules': [{'logged': True}]}
map_def = core_defs.CommunicationMapDef(
domain_id=domain_id,
map_id=map_id,
tenant=TEST_TENANT)
with mock.patch.object(self.policy_api, "get",
return_value=dummy_map),\
mock.patch.object(self.policy_api.client,
"update") as update_call:
self.resourceApi.update_entries_logged(
domain_id, map_id,
logged=True,
tenant=TEST_TENANT)
update_call.assert_called_once_with(
map_def.get_resource_path(), updated_map)
def test_get_realized(self):
domain_id = 'd1'
map_id = '111'
result = [{'state': constants.STATE_REALIZED,
'entity_type': 'RealizedFirewallSection'}]
with mock.patch.object(
self.policy_api, "get_realized_entities",
return_value=result) as api_get:
state = self.resourceApi.get_realized_state(
domain_id, map_id, tenant=TEST_TENANT)
self.assertEqual(constants.STATE_REALIZED, state)
path = "/%s/domains/%s/security-policies/%s" % (
TEST_TENANT, domain_id, map_id)
api_get.assert_called_once_with(path)
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'
thumbprint = 'abc'
edge_cluster_id = 'ec1'
transport_zone_id = 'tz1'
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,
thumbprint=thumbprint,
username=username,
password=password,
edge_cluster_id=edge_cluster_id,
transport_zone_id=transport_zone_id,
tenant=TEST_TENANT)
expected_def = core_defs.EnforcementPointDef(
ep_id=mock.ANY,
name=name,
description=description,
ip_address=ip_address,
username=username,
thumbprint=thumbprint,
password=password,
edge_cluster_id=edge_cluster_id,
transport_zone_id=transport_zone_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 = core_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 = core_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 = core_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 = core_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'
ip_address = '1.1.1.1'
thumbprint = 'abc'
edge_cluster_id = 'ec1'
transport_zone_id = 'tz1'
entry = {'id': id,
'connection_info': {'thumbprint': thumbprint,
'resource_type': 'NSXTConnectionInfo'}}
with mock.patch.object(self.policy_api,
"create_or_update") as update_call,\
mock.patch.object(self.policy_api, "get",
return_value=entry):
self.resourceApi.update(id,
name=name,
username=username,
password=password,
ip_address=ip_address,
edge_cluster_id=edge_cluster_id,
transport_zone_id=transport_zone_id,
tenant=TEST_TENANT)
expected_def = core_defs.EnforcementPointDef(
ep_id=id,
name=name,
username=username,
password=password,
ip_address=ip_address,
thumbprint=thumbprint,
edge_cluster_id=edge_cluster_id,
transport_zone_id=transport_zone_id,
tenant=TEST_TENANT)
self.assert_called_with_def(update_call, expected_def)
def test_get_realized(self):
ep_id = 'ef1'
result = [{'state': constants.STATE_REALIZED}]
with mock.patch.object(
self.policy_api, "get_realized_entities",
return_value=result) as api_get:
state = self.resourceApi.get_realized_state(
ep_id, tenant=TEST_TENANT)
self.assertEqual(constants.STATE_REALIZED, state)
path = "/%s/sites/default/enforcement-points/%s" % (
TEST_TENANT, ep_id)
api_get.assert_called_once_with(path)
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 = core_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'
domain_id = 'domain1'
with mock.patch.object(self.policy_api, "delete") as api_call:
self.resourceApi.delete(id, domain_id=domain_id,
tenant=TEST_TENANT)
expected_def = core_defs.DeploymentMapDef(map_id=id,
domain_id=domain_id,
tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
def test_get(self):
id = '111'
domain_id = 'domain1'
with mock.patch.object(self.policy_api, "get") as api_call:
self.resourceApi.get(id, domain_id=domain_id, tenant=TEST_TENANT)
expected_def = core_defs.DeploymentMapDef(map_id=id,
domain_id=domain_id,
tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
def test_get_by_name(self):
name = 'ep1'
domain_id = 'domain1'
with mock.patch.object(
self.policy_api, "list",
return_value={'results': [{'display_name': name}]}) as api_call:
obj = self.resourceApi.get_by_name(name, domain_id=domain_id,
tenant=TEST_TENANT)
self.assertIsNotNone(obj)
expected_def = core_defs.DeploymentMapDef(domain_id=domain_id,
tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
def test_list(self):
domain_id = 'domain1'
with mock.patch.object(self.policy_api, "list") as api_call:
self.resourceApi.list(domain_id=domain_id, tenant=TEST_TENANT)
expected_def = core_defs.DeploymentMapDef(domain_id=domain_id,
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 = core_defs.DeploymentMapDef(
map_id=id,
name=name,
ep_id=ep_id,
domain_id=domain_id,
tenant=TEST_TENANT)
self.assert_called_with_def(
update_call, expected_def)
class TestPolicyTransportZone(NsxPolicyLibTestCase):
def setUp(self, *args, **kwargs):
super(TestPolicyTransportZone, self).setUp()
self.resourceApi = self.policy_lib.transport_zone
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 = core_defs.TransportZoneDef(tz_id=id,
tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
def test_get_with_cache(self):
id = '111'
with mock.patch.object(self.policy_api.client, "get") as client_get:
self.resourceApi.get(id, tenant=TEST_TENANT)
self.resourceApi.get(id, tenant=TEST_TENANT)
self.assertEqual(1, client_get.call_count)
def test_get_by_name(self):
name = 'tz1'
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.TransportZoneDef(tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
def test_get_tz_type(self):
id = '111'
tz_type = self.resourceApi.TZ_TYPE_OVERLAY
with mock.patch.object(self.policy_api, "get",
return_value={'tz_type': tz_type}) as api_call:
actual_tz_type = self.resourceApi.get_tz_type(
id, tenant=TEST_TENANT)
expected_def = core_defs.TransportZoneDef(tz_id=id,
tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
self.assertEqual(tz_type, actual_tz_type)
def test_get_transport_type(self):
id = '111'
tz_type = self.resourceApi.TZ_TYPE_OVERLAY
with mock.patch.object(self.policy_api, "get",
return_value={'tz_type': tz_type}) as api_call:
actual_tz_type = self.resourceApi.get_transport_type(
id, tenant=TEST_TENANT)
expected_def = core_defs.TransportZoneDef(tz_id=id,
tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
self.assertEqual(nsx_constants.TRANSPORT_TYPE_OVERLAY,
actual_tz_type)
def test_get_switch_mode(self):
id = '111'
tz_type = self.resourceApi.TZ_TYPE_OVERLAY
with mock.patch.object(self.policy_api, "get",
return_value={'tz_type': tz_type}) as api_call:
actual_sm = self.resourceApi.get_host_switch_mode(
id, tenant=TEST_TENANT)
expected_def = core_defs.TransportZoneDef(tz_id=id,
tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
self.assertEqual(nsx_constants.HOST_SWITCH_MODE_STANDARD,
actual_sm)
def test_list(self):
with mock.patch.object(self.policy_api, "list") as api_call:
self.resourceApi.list(tenant=TEST_TENANT)
expected_def = core_defs.TransportZoneDef(tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
class TestPolicyEdgeCluster(NsxPolicyLibTestCase):
def setUp(self, *args, **kwargs):
super(TestPolicyEdgeCluster, self).setUp()
self.resourceApi = self.policy_lib.edge_cluster
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 = core_defs.EdgeClusterDef(ec_id=id,
tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
def test_get_by_name(self):
name = 'tz1'
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.EdgeClusterDef(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 = core_defs.EdgeClusterDef(tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
class TestPolicyTier1(NsxPolicyLibTestCase):
def setUp(self, *args, **kwargs):
super(TestPolicyTier1, self).setUp(*args, **kwargs)
self.resourceApi = self.policy_lib.tier1
def test_create(self):
name = 'test'
description = 'desc'
tier0_id = '111'
route_adv = self.resourceApi.build_route_advertisement(
lb_vip=True,
lb_snat=True)
with mock.patch.object(self.policy_api,
"create_or_update") as api_call:
self.resourceApi.create_or_overwrite(
name, description=description,
tier0=tier0_id,
force_whitelisting=True,
route_advertisement=route_adv,
tenant=TEST_TENANT)
expected_def = core_defs.Tier1Def(
tier1_id=mock.ANY,
name=name,
description=description,
tier0=tier0_id,
force_whitelisting=True,
failover_mode=constants.NON_PREEMPTIVE,
route_advertisement=route_adv,
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 = core_defs.Tier1Def(tier1_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 = core_defs.Tier1Def(tier1_id=id,
tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
def test_get_with_no_cache(self):
id = '111'
with mock.patch.object(self.policy_api.client, "get") as client_get:
self.resourceApi.get(id, tenant=TEST_TENANT)
self.resourceApi.get(id, tenant=TEST_TENANT)
self.assertEqual(2, client_get.call_count)
def test_get_by_name(self):
name = 'test'
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.Tier1Def(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 = core_defs.Tier1Def(tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
def test_update(self):
id = '111'
name = 'new name'
tier0 = 'tier0'
with mock.patch.object(self.policy_api,
"create_or_update") as update_call:
self.resourceApi.update(id,
name=name, tier0=tier0,
tenant=TEST_TENANT)
expected_def = core_defs.Tier1Def(tier1_id=id,
name=name,
tier0=tier0,
tenant=TEST_TENANT)
self.assert_called_with_def(
update_call, expected_def)
def test_update_ignore_tier0(self):
id = '111'
name = 'new name'
with mock.patch.object(self.policy_api,
"create_or_update") as update_call:
self.resourceApi.update(id,
name=name,
tenant=TEST_TENANT)
expected_def = core_defs.Tier1Def(tier1_id=id,
name=name,
tenant=TEST_TENANT)
self.assert_called_with_def(update_call, expected_def)
# make sure tier0 is not in the body
actual_def = update_call.call_args_list[0][0][0]
self.assertNotIn('tier0_path', actual_def.get_obj_dict())
def test_update_unset_tier0(self):
id = '111'
name = 'new name'
with mock.patch.object(self.policy_api,
"create_or_update") as update_call:
self.resourceApi.update(id,
name=name,
tier0=None,
tenant=TEST_TENANT)
expected_def = core_defs.Tier1Def(tier1_id=id,
name=name,
tier0=None,
tenant=TEST_TENANT)
self.assert_called_with_def(update_call, expected_def)
# make sure tier0 is in the body with value None
actual_def = update_call.call_args_list[0][0][0]
self.assertIn('tier0_path', actual_def.get_obj_dict())
self.assertIsNone(actual_def.get_obj_dict()['tier0_path'])
def test_update_route_adv(self):
id = '111'
rtr_name = 'rtr111'
get_result = {'id': '111',
'display_name': rtr_name,
'route_advertisement_types': ['TIER1_NAT',
'TIER1_LB_VIP']}
with mock.patch.object(self.policy_api, "get",
return_value=get_result),\
mock.patch.object(self.policy_api,
"create_or_update") as update_call:
self.resourceApi.update_route_advertisement(
id,
static_routes=True,
lb_vip=False,
lb_snat=True,
tenant=TEST_TENANT)
new_adv = self.resourceApi.build_route_advertisement(
nat=True, static_routes=True, lb_snat=True)
expected_def = core_defs.Tier1Def(tier1_id=id,
name=rtr_name,
route_adv=new_adv,
tenant=TEST_TENANT)
self.assert_called_with_def(
update_call, expected_def)
def test_wait_until_realized_fail(self):
tier1_id = '111'
logical_router_id = 'realized_111'
info = {'state': constants.STATE_UNREALIZED,
'realization_specific_identifier': logical_router_id,
'entity_type': 'RealizedLogicalRouter'}
with mock.patch.object(self.resourceApi, "_get_realization_info",
return_value=info):
self.assertRaises(nsxlib_exc.ManagerError,
self.resourceApi.wait_until_realized,
tier1_id, max_attempts=5, sleep=0.1,
tenant=TEST_TENANT)
def test_wait_until_realized_succeed(self):
tier1_id = '111'
logical_router_id = 'realized_111'
info = {'state': constants.STATE_REALIZED,
'realization_specific_identifier': logical_router_id,
'entity_type': 'RealizedLogicalRouter'}
with mock.patch.object(self.resourceApi, "_get_realization_info",
return_value=info):
actual_info = self.resourceApi.wait_until_realized(
tier1_id, max_attempts=5, sleep=0.1, tenant=TEST_TENANT)
self.assertEqual(info, actual_info)
def test_update_transport_zone(self):
# Test the passthrough api
tier1_id = '111'
logical_router_id = 'realized_111'
tz_uuid = 'dummy_tz'
info = {'state': constants.STATE_REALIZED,
'entity_type': 'RealizedLogicalRouter',
'realization_specific_identifier': logical_router_id}
passthrough_mock = self.resourceApi.nsx_api.logical_router.update
with mock.patch.object(self.resourceApi, "_get_realization_info",
return_value=info) as realization:
self.resourceApi.update_transport_zone(tier1_id, tz_uuid,
tenant=TEST_TENANT)
realization.assert_called_once()
passthrough_mock.assert_called_once_with(
logical_router_id, transport_zone_id=tz_uuid)
def test_wait_until_realized(self):
tier1_id = '111'
logical_router_id = 'realized_111'
info = {'state': constants.STATE_UNREALIZED,
'realization_specific_identifier': logical_router_id}
with mock.patch.object(self.resourceApi, "_get_realization_info",
return_value=info):
self.assertRaises(nsxlib_exc.ManagerError,
self.resourceApi.wait_until_realized,
tier1_id, tenant=TEST_TENANT,
max_attempts=5, sleep=0.1)
def test_get_realized_downlink_port(self):
tier1_id = '111'
segment_id = '222'
lrp_id = '333'
info = {'state': constants.STATE_REALIZED,
'realization_specific_identifier': lrp_id,
'entity_type': 'RealizedLogicalRouterPort'}
dummy_port = {'resource_type': nsx_constants.LROUTERPORT_DOWNLINK,
'id': lrp_id,
'display_name': 'test_%s' % segment_id}
with mock.patch.object(self.resourceApi.policy_api,
"get_realized_entities",
return_value=[info]),\
mock.patch.object(self.resourceApi.nsx_api.logical_router_port,
"get", return_value=dummy_port):
actual_id = self.resourceApi._get_realized_downlink_port(
tier1_id, segment_id)
self.assertEqual(lrp_id, actual_id)
def test_set_dhcp_relay(self):
tier1_id = '111'
segment_id = '222'
lrp_id = '333'
relay_id = '444'
info = {'state': constants.STATE_REALIZED,
'realization_specific_identifier': lrp_id,
'entity_type': 'RealizedLogicalRouterPort'}
dummy_port = {'resource_type': nsx_constants.LROUTERPORT_DOWNLINK,
'id': lrp_id,
'display_name': 'test_%s' % segment_id}
with mock.patch.object(self.resourceApi.policy_api,
"get_realized_entities",
return_value=[info]),\
mock.patch.object(self.resourceApi.nsx_api.logical_router_port,
"get", return_value=dummy_port),\
mock.patch.object(self.resourceApi.nsx_api.logical_router_port,
"update") as nsx_lrp_update:
self.resourceApi.set_dhcp_relay(tier1_id, segment_id, relay_id)
nsx_lrp_update.assert_called_once_with(
lrp_id, relay_service_uuid=relay_id)
class TestPolicyTier1NoPassthrough(TestPolicyTier1):
def setUp(self, *args, **kwargs):
super(TestPolicyTier1NoPassthrough, self).setUp(
allow_passthrough=False)
def test_update_transport_zone(self):
# Will not work without passthrough api
tier1_id = '111'
tz_uuid = 'dummy_tz'
with mock.patch.object(self.resourceApi,
"_get_realization_info") as realization:
self.resourceApi.update_transport_zone(tier1_id, tz_uuid,
tenant=TEST_TENANT)
realization.assert_not_called()
def test_get_realized_downlink_port(self):
# Will not work without passthrough api
tier1_id = '111'
segment_id = '222'
with mock.patch.object(self.resourceApi.policy_api,
"get_realized_entities") as realization:
actual_id = self.resourceApi._get_realized_downlink_port(
tier1_id, segment_id)
self.assertIsNone(actual_id)
realization.assert_not_called()
def test_set_dhcp_relay(self):
# Will not work without passthrough api
tier1_id = '111'
segment_id = '222'
relay_id = '444'
with mock.patch.object(self.resourceApi.policy_api,
"get_realized_entities") as realization:
self.resourceApi.set_dhcp_relay(tier1_id, segment_id, relay_id)
realization.assert_not_called()
class TestPolicyTier1NatRule(NsxPolicyLibTestCase):
def setUp(self, *args, **kwargs):
super(TestPolicyTier1NatRule, self).setUp()
self.resourceApi = self.policy_lib.tier1_nat_rule
def test_create(self):
name = 'test'
description = 'desc'
tier1_id = '111'
nat_rule_id = 'rule1'
action = constants.NAT_ACTION_SNAT
cidr1 = '1.1.1.1/32'
cidr2 = '2.2.2.0/24'
with mock.patch.object(self.policy_api,
"create_or_update") as api_call:
self.resourceApi.create_or_overwrite(
name, tier1_id,
nat_rule_id=nat_rule_id,
description=description,
action=action,
translated_network=cidr1,
source_network=cidr2,
tenant=TEST_TENANT)
expected_def = core_defs.Tier1NatRule(
tier1_id=tier1_id,
nat_rule_id=nat_rule_id,
nat_id=self.resourceApi.DEFAULT_NAT_ID,
name=name,
description=description,
action=action,
translated_network=cidr1,
source_network=cidr2,
tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
def test_delete(self):
tier1_id = '111'
nat_rule_id = 'rule1'
with mock.patch.object(self.policy_api, "delete") as api_call:
self.resourceApi.delete(
tier1_id,
nat_rule_id,
tenant=TEST_TENANT)
expected_def = core_defs.Tier1NatRule(
tier1_id=tier1_id,
nat_rule_id=nat_rule_id,
nat_id=self.resourceApi.DEFAULT_NAT_ID,
tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
class TestPolicyTier1StaticRoute(NsxPolicyLibTestCase):
def setUp(self, *args, **kwargs):
super(TestPolicyTier1StaticRoute, self).setUp()
self.resourceApi = self.policy_lib.tier1_static_route
def test_create(self):
name = 'test'
description = 'desc'
tier1_id = '111'
static_route_id = '222'
network = '1.1.1.1/24'
nexthop = '2.2.2.2'
with mock.patch.object(self.policy_api,
"create_or_update") as api_call:
self.resourceApi.create_or_overwrite(
name, tier1_id,
static_route_id=static_route_id,
description=description,
network=network,
next_hop=nexthop,
tenant=TEST_TENANT)
expected_def = core_defs.Tier1StaticRoute(
tier1_id=tier1_id,
static_route_id=static_route_id,
name=name,
description=description,
network=network,
next_hop=nexthop,
tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
def test_delete(self):
tier1_id = '111'
static_route_id = '222'
with mock.patch.object(self.policy_api, "delete") as api_call:
self.resourceApi.delete(
tier1_id,
static_route_id,
tenant=TEST_TENANT)
expected_def = core_defs.Tier1StaticRoute(
tier1_id=tier1_id,
static_route_id=static_route_id,
tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
class TestPolicyTier0(NsxPolicyLibTestCase):
def setUp(self, *args, **kwargs):
super(TestPolicyTier0, self).setUp()
self.resourceApi = self.policy_lib.tier0
def test_create(self):
name = 'test'
description = 'desc'
dhcp_config = '111'
subnets = ["2.2.2.0/24"]
with mock.patch.object(self.policy_api,
"create_or_update") as api_call:
self.resourceApi.create_or_overwrite(
name, description=description,
dhcp_config=dhcp_config,
force_whitelisting=True,
default_rule_logging=True,
transit_subnets=subnets,
tenant=TEST_TENANT)
expected_def = core_defs.Tier0Def(
tier0_id=mock.ANY,
name=name,
description=description,
dhcp_config=dhcp_config,
default_rule_logging=True,
force_whitelisting=True,
ha_mode=constants.ACTIVE_ACTIVE,
failover_mode=constants.NON_PREEMPTIVE,
transit_subnets=subnets,
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 = core_defs.Tier0Def(tier0_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 = core_defs.Tier0Def(tier0_id=id,
tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
def test_get_with_cache(self):
id = '111'
with mock.patch.object(self.policy_api.client, "get") as client_get:
self.resourceApi.get(id, tenant=TEST_TENANT)
self.resourceApi.get(id, tenant=TEST_TENANT)
self.assertEqual(1, client_get.call_count)
def test_get_by_name(self):
name = 'test'
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.Tier0Def(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 = core_defs.Tier0Def(tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
def test_update(self):
id = '111'
name = 'new name'
with mock.patch.object(self.policy_api,
"create_or_update") as update_call:
self.resourceApi.update(id,
name=name,
tenant=TEST_TENANT)
expected_def = core_defs.Tier0Def(tier0_id=id,
name=name,
tenant=TEST_TENANT)
self.assert_called_with_def(
update_call, expected_def)
def test_get_overlay_transport_zone(self):
# Test the passthrough api
tier0_id = '111'
logical_router_id = 'realized_111'
info = {'state': constants.STATE_REALIZED,
'entity_type': 'RealizedLogicalRouter',
'realization_specific_identifier': logical_router_id}
pt_mock = self.resourceApi.nsx_api.router.get_tier0_router_overlay_tz
with mock.patch.object(self.resourceApi, "_get_realization_info",
return_value=info) as realization:
self.resourceApi.get_overlay_transport_zone(
tier0_id, tenant=TEST_TENANT)
realization.assert_called_once()
pt_mock.assert_called_once_with(logical_router_id)
def test_wait_until_realized(self):
tier1_id = '111'
logical_router_id = 'realized_111'
info = {'state': constants.STATE_UNREALIZED,
'realization_specific_identifier': logical_router_id}
with mock.patch.object(self.resourceApi, "_get_realization_info",
return_value=info):
self.assertRaises(nsxlib_exc.ManagerError,
self.resourceApi.wait_until_realized,
tier1_id, max_attempts=5, sleep=0.1,
tenant=TEST_TENANT)
def test_get_uplink_ips(self):
tier0_id = '111'
ip_addr = '5.5.5.5'
interface = {'id': '222', 'type': 'EXTERNAL',
'subnets': [{'ip_addresses': [ip_addr]}]}
with mock.patch.object(self.resourceApi.policy_api, "list",
return_value={'results': [interface]}):
uplink_ips = self.resourceApi.get_uplink_ips(
tier0_id, tenant=TEST_TENANT)
self.assertEqual([ip_addr], uplink_ips)
def test_get_transport_zones(self):
# Test the passthrough api
tier0_id = '111'
logical_router_id = 'realized_111'
info = {'state': constants.STATE_REALIZED,
'entity_type': 'RealizedLogicalRouter',
'realization_specific_identifier': logical_router_id}
pt_mock = self.resourceApi.nsx_api.router.get_tier0_router_tz
with mock.patch.object(self.resourceApi, "_get_realization_info",
return_value=info) as realization:
self.resourceApi.get_transport_zones(
tier0_id, tenant=TEST_TENANT)
realization.assert_called_once()
pt_mock.assert_called_once_with(logical_router_id)
class TestPolicySegment(NsxPolicyLibTestCase):
def setUp(self, *args, **kwargs):
super(TestPolicySegment, self).setUp()
self.resourceApi = self.policy_lib.segment
def test_create(self):
name = 'test'
description = 'desc'
tier1_id = '111'
subnets = [core_defs.Subnet(gateway_address="2.2.2.0/24")]
with mock.patch.object(self.policy_api,
"create_or_update") as api_call:
self.resourceApi.create_or_overwrite(
name, description=description,
tier1_id=tier1_id,
subnets=subnets,
tenant=TEST_TENANT)
expected_def = core_defs.SegmentDef(
segment_id=mock.ANY,
name=name,
description=description,
tier1_id=tier1_id,
subnets=subnets,
tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
def test_delete(self):
segment_id = '111'
with mock.patch.object(self.policy_api, "delete") as api_call:
self.resourceApi.delete(segment_id, tenant=TEST_TENANT)
expected_def = core_defs.SegmentDef(segment_id=segment_id,
tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
def test_get(self):
segment_id = '111'
with mock.patch.object(self.policy_api, "get") as api_call:
self.resourceApi.get(segment_id, tenant=TEST_TENANT)
expected_def = core_defs.SegmentDef(segment_id=segment_id,
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 = core_defs.SegmentDef(tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
def test_update(self):
segment_id = '111'
name = 'new name'
with mock.patch.object(self.policy_api,
"create_or_update") as update_call:
self.resourceApi.update(segment_id,
name=name,
tenant=TEST_TENANT)
expected_def = core_defs.SegmentDef(segment_id=segment_id,
name=name,
tenant=TEST_TENANT)
self.assert_called_with_def(update_call, expected_def)
def test_remove_connectivity_and_subnets(self):
segment_id = '111'
with mock.patch.object(self.policy_api, "get",
return_value={'id': segment_id}) as api_get,\
mock.patch.object(self.policy_api.client, "update") as api_put:
self.resourceApi.remove_connectivity_and_subnets(
segment_id, tenant=TEST_TENANT)
api_get.assert_called_once()
api_put.assert_called_once_with(
'%s/segments/%s' % (TEST_TENANT, segment_id),
{'id': segment_id, 'connectivity_path': None, 'subnets': None})
class TestPolicyIpPool(NsxPolicyLibTestCase):
def setUp(self, *args, **kwargs):
super(TestPolicyIpPool, self).setUp()
self.resourceApi = self.policy_lib.ip_pool
def test_create(self):
name = 'test'
description = 'desc'
ip_pool_id = '111'
with mock.patch.object(self.policy_api,
"create_or_update") as api_call:
self.resourceApi.create_or_overwrite(
name, ip_pool_id, description=description,
tenant=TEST_TENANT)
expected_def = core_defs.IpPoolDef(
ip_pool_id=ip_pool_id,
name=name,
description=description,
tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
def test_delete(self):
ip_pool_id = '111'
with mock.patch.object(self.policy_api, "delete") as api_call:
self.resourceApi.delete(ip_pool_id, tenant=TEST_TENANT)
expected_def = core_defs.IpPoolDef(ip_pool_id=ip_pool_id,
tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
def test_get(self):
ip_pool_id = '111'
with mock.patch.object(self.policy_api, "get") as api_call:
self.resourceApi.get(ip_pool_id, tenant=TEST_TENANT)
expected_def = core_defs.IpPoolDef(ip_pool_id=ip_pool_id,
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 = core_defs.IpPoolDef(tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
def test_update(self):
ip_pool_id = '111'
name = 'new name'
with mock.patch.object(self.policy_api,
"create_or_update") as update_call:
self.resourceApi.update(ip_pool_id,
name=name,
tenant=TEST_TENANT)
expected_def = core_defs.IpPoolDef(ip_pool_id=ip_pool_id,
name=name,
tenant=TEST_TENANT)
self.assert_called_with_def(update_call, expected_def)
def test_allocate_ip(self):
ip_pool_id = '111'
ip_allocation_id = 'alloc-id'
with mock.patch.object(self.policy_api,
"create_or_update") as update_call:
self.resourceApi.allocate_ip(ip_pool_id,
ip_allocation_id,
tenant=TEST_TENANT)
expected_def = core_defs.IpPoolAllocationDef(
ip_pool_id=ip_pool_id,
ip_allocation_id=ip_allocation_id,
tenant=TEST_TENANT)
self.assert_called_with_def(update_call, expected_def)
def test_release_ip(self):
ip_pool_id = '111'
ip_allocation_id = 'alloc-id'
with mock.patch.object(self.policy_api, "delete") as delete_call:
self.resourceApi.release_ip(ip_pool_id,
ip_allocation_id,
tenant=TEST_TENANT)
expected_def = core_defs.IpPoolAllocationDef(
ip_pool_id=ip_pool_id,
ip_allocation_id=ip_allocation_id,
tenant=TEST_TENANT)
self.assert_called_with_def(delete_call, expected_def)
def test_allocate_block_subnet(self):
ip_pool_id = '111'
ip_block_id = 'block-id'
size = 256
ip_subnet_id = 'subnet-id'
with mock.patch.object(self.policy_api,
"create_or_update") as api_call:
self.resourceApi.allocate_block_subnet(
ip_pool_id, ip_block_id, size, ip_subnet_id,
tenant=TEST_TENANT)
expected_def = core_defs.IpPoolBlockSubnetDef(
ip_pool_id=ip_pool_id,
ip_block_id=ip_block_id,
ip_subnet_id=ip_subnet_id,
size=size,
tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
def test_release_block_subnet(self):
ip_pool_id = '111'
ip_subnet_id = 'subnet-id'
with mock.patch.object(self.policy_api, "delete") as delete_call:
self.resourceApi.release_block_subnet(ip_pool_id,
ip_subnet_id,
tenant=TEST_TENANT)
expected_def = core_defs.IpPoolBlockSubnetDef(
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_get_ip_alloc_realization_info(self):
ip_pool_id = '111'
ip_allocation_id = 'alloc-id'
result = {'extended_attributes': [{'values': ['5.5.0.8']}]}
with mock.patch.object(
self.resourceApi, "_get_realization_info",
return_value=result) as api_get:
self.resourceApi.get_ip_alloc_realization_info(
ip_pool_id, ip_allocation_id, tenant=TEST_TENANT)
api_get.assert_called_once()
def test_get_realized_allocated_ip(self):
ip_pool_id = '111'
ip_allocation_id = 'alloc-id'
result = {'extended_attributes': [{'values': ['5.5.0.8']}]}
with mock.patch.object(
self.resourceApi, "_get_realization_info",
return_value=result) as api_get:
ip = self.resourceApi.get_realized_allocated_ip(
ip_pool_id, ip_allocation_id, tenant=TEST_TENANT)
self.assertEqual('5.5.0.8', ip)
api_get.assert_called_once()
class TestPolicySegmentProfileBase(NsxPolicyLibTestCase):
def setUp(self, resource_api_name='segment_security_profile',
resource_def=core_defs.SegmentSecurityProfileDef):
super(TestPolicySegmentProfileBase, self).setUp()
self.resourceApi = getattr(self.policy_lib, resource_api_name)
self.resourceDef = resource_def
def test_create(self):
name = 'test'
with mock.patch.object(self.policy_api,
"create_or_update") as api_call:
self.resourceApi.create_or_overwrite(
name,
tenant=TEST_TENANT)
expected_def = self.resourceDef(
profile_id=mock.ANY,
name=name,
tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
def test_delete(self):
profile_id = '111'
with mock.patch.object(self.policy_api, "delete") as api_call:
self.resourceApi.delete(profile_id, tenant=TEST_TENANT)
expected_def = self.resourceDef(profile_id=profile_id,
tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
def test_get(self):
profile_id = '111'
with mock.patch.object(self.policy_api, "get") as api_call:
self.resourceApi.get(profile_id, tenant=TEST_TENANT)
expected_def = self.resourceDef(profile_id=profile_id,
tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
def test_get_by_name(self):
name = 'test'
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 = self.resourceDef(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 = self.resourceDef(tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
def test_update(self):
profile_id = '111'
name = 'new name'
with mock.patch.object(self.policy_api,
"create_or_update") as update_call:
self.resourceApi.update(profile_id,
name=name,
tenant=TEST_TENANT)
expected_def = self.resourceDef(profile_id=profile_id,
name=name,
tenant=TEST_TENANT)
self.assert_called_with_def(
update_call, expected_def)
class TestPolicyQosProfile(TestPolicySegmentProfileBase):
def setUp(self):
super(TestPolicyQosProfile, self).setUp(
resource_api_name='qos_profile',
resource_def=core_defs.QosProfileDef)
def test_create_with_params(self):
name = 'test'
description = 'desc'
dscp = self.resourceApi.build_dscp(trusted=False, priority=7)
limiter = self.resourceApi.build_ingress_rate_limiter(
average_bandwidth=700,
enabled=True)
with mock.patch.object(self.policy_api,
"create_or_update") as api_call:
self.resourceApi.create_or_overwrite(
name,
description=description,
dscp=dscp,
shaper_configurations=[limiter],
tenant=TEST_TENANT)
expected_def = self.resourceDef(
profile_id=mock.ANY,
name=name,
description=description,
dscp=dscp,
shaper_configurations=[limiter],
tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
class TestPolicySpoofguardProfile(TestPolicySegmentProfileBase):
def setUp(self):
super(TestPolicySpoofguardProfile, self).setUp(
resource_api_name='spoofguard_profile',
resource_def=core_defs.SpoofguardProfileDef)
class TestPolicyIpDiscoveryProfile(TestPolicySegmentProfileBase):
def setUp(self):
super(TestPolicyIpDiscoveryProfile, self).setUp(
resource_api_name='ip_discovery_profile',
resource_def=core_defs.IpDiscoveryProfileDef)
class TestPolicyMacDiscoveryProfile(TestPolicySegmentProfileBase):
def setUp(self):
super(TestPolicyMacDiscoveryProfile, self).setUp(
resource_api_name='mac_discovery_profile',
resource_def=core_defs.MacDiscoveryProfileDef)
class TestPolicySegmentSecurityProfile(TestPolicySegmentProfileBase):
def test_create_with_params(self):
name = 'test'
with mock.patch.object(self.policy_api,
"create_or_update") as api_call:
self.resourceApi.create_or_overwrite(
name,
bpdu_filter_enable=True,
dhcp_client_block_enabled=False,
dhcp_client_block_v6_enabled=True,
dhcp_server_block_enabled=False,
dhcp_server_block_v6_enabled=True,
non_ip_traffic_block_enabled=False,
ra_guard_enabled=True,
rate_limits_enabled=False,
tenant=TEST_TENANT)
expected_def = self.resourceDef(
profile_id=mock.ANY,
name=name,
bpdu_filter_enable=True,
dhcp_client_block_enabled=False,
dhcp_client_block_v6_enabled=True,
dhcp_server_block_enabled=False,
dhcp_server_block_v6_enabled=True,
non_ip_traffic_block_enabled=False,
ra_guard_enabled=True,
rate_limits_enabled=False,
tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
class TestPolicySegmentSecProfilesBinding(NsxPolicyLibTestCase):
def setUp(self, resource_api_name='segment_port_security_profiles',
resource_def=core_defs.SegmentPortSecProfilesBindingMapDef):
super(TestPolicySegmentSecProfilesBinding, self).setUp()
self.resourceApi = getattr(self.policy_lib, resource_api_name)
self.resourceDef = resource_def
def test_create(self):
name = 'test'
segment_id = 'seg1'
port_id = 'port1'
prf1 = '1'
prf2 = '2'
with mock.patch.object(self.policy_api,
"create_or_update") as api_call:
self.resourceApi.create_or_overwrite(
name, segment_id, port_id,
segment_security_profile_id=prf1,
spoofguard_profile_id=prf2,
tenant=TEST_TENANT)
expected_def = self.resourceDef(
segment_id=segment_id,
port_id=port_id,
map_id=core_resources.DEFAULT_MAP_ID,
name=name,
segment_security_profile_id=prf1,
spoofguard_profile_id=prf2,
tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
def test_delete(self):
segment_id = 'seg1'
port_id = 'port1'
with mock.patch.object(self.policy_api, "delete") as api_call:
self.resourceApi.delete(segment_id, port_id, tenant=TEST_TENANT)
expected_def = self.resourceDef(
segment_id=segment_id,
port_id=port_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'
port_id = 'port1'
with mock.patch.object(self.policy_api, "get") as api_call:
self.resourceApi.get(segment_id, port_id, tenant=TEST_TENANT)
expected_def = self.resourceDef(
segment_id=segment_id,
port_id=port_id,
map_id=core_resources.DEFAULT_MAP_ID,
tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
def test_list(self):
segment_id = 'seg1'
port_id = 'port1'
with mock.patch.object(self.policy_api, "list") as api_call:
self.resourceApi.list(segment_id, port_id, tenant=TEST_TENANT)
expected_def = self.resourceDef(
segment_id=segment_id,
port_id=port_id,
tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
def test_update(self):
name = 'new name'
segment_id = 'seg1'
port_id = 'port1'
prf1 = '1'
prf2 = '2'
with mock.patch.object(self.policy_api,
"create_or_update") as update_call:
self.resourceApi.update(
segment_id=segment_id,
port_id=port_id,
name=name,
segment_security_profile_id=prf1,
spoofguard_profile_id=prf2,
tenant=TEST_TENANT)
expected_def = self.resourceDef(
segment_id=segment_id,
port_id=port_id,
map_id=core_resources.DEFAULT_MAP_ID,
name=name,
segment_security_profile_id=prf1,
spoofguard_profile_id=prf2,
tenant=TEST_TENANT)
self.assert_called_with_def(
update_call, expected_def)
class TestPolicySegmentDiscoveryProfilesBinding(NsxPolicyLibTestCase):
def setUp(
self, resource_api_name='segment_port_discovery_profiles',
resource_def=core_defs.SegmentPortDiscoveryProfilesBindingMapDef):
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'
port_id = 'port1'
prf1 = '1'
prf2 = '2'
with mock.patch.object(self.policy_api,
"create_or_update") as api_call:
self.resourceApi.create_or_overwrite(
name, segment_id, port_id,
mac_discovery_profile_id=prf1,
ip_discovery_profile_id=prf2,
tenant=TEST_TENANT)
expected_def = self.resourceDef(
segment_id=segment_id,
port_id=port_id,
map_id=core_resources.DEFAULT_MAP_ID,
name=name,
mac_discovery_profile_id=prf1,
ip_discovery_profile_id=prf2,
tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
def test_delete(self):
segment_id = 'seg1'
port_id = 'port1'
with mock.patch.object(self.policy_api, "delete") as api_call:
self.resourceApi.delete(segment_id, port_id, tenant=TEST_TENANT)
expected_def = self.resourceDef(
segment_id=segment_id,
port_id=port_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'
port_id = 'port1'
with mock.patch.object(self.policy_api, "get") as api_call:
self.resourceApi.get(segment_id, port_id, tenant=TEST_TENANT)
expected_def = self.resourceDef(
segment_id=segment_id,
port_id=port_id,
map_id=core_resources.DEFAULT_MAP_ID,
tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
def test_list(self):
segment_id = 'seg1'
port_id = 'port1'
with mock.patch.object(self.policy_api, "list") as api_call:
self.resourceApi.list(segment_id, port_id, tenant=TEST_TENANT)
expected_def = self.resourceDef(
segment_id=segment_id,
port_id=port_id,
tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
def test_update(self):
name = 'new name'
segment_id = 'seg1'
port_id = 'port1'
prf1 = '1'
prf2 = '2'
with mock.patch.object(self.policy_api,
"create_or_update") as update_call:
self.resourceApi.update(
segment_id=segment_id,
port_id=port_id,
name=name,
mac_discovery_profile_id=prf1,
ip_discovery_profile_id=prf2,
tenant=TEST_TENANT)
expected_def = self.resourceDef(
segment_id=segment_id,
port_id=port_id,
map_id=core_resources.DEFAULT_MAP_ID,
name=name,
mac_discovery_profile_id=prf1,
ip_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_port_qos_profiles',
resource_def=core_defs.SegmentPortQoSProfilesBindingMapDef):
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'
port_id = 'port1'
prf1 = '1'
with mock.patch.object(self.policy_api,
"create_or_update") as api_call:
self.resourceApi.create_or_overwrite(
name, segment_id, port_id,
qos_profile_id=prf1,
tenant=TEST_TENANT)
expected_def = self.resourceDef(
segment_id=segment_id,
port_id=port_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)
def test_delete(self):
segment_id = 'seg1'
port_id = 'port1'
with mock.patch.object(self.policy_api, "delete") as api_call:
self.resourceApi.delete(segment_id, port_id, tenant=TEST_TENANT)
expected_def = self.resourceDef(
segment_id=segment_id,
port_id=port_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'
port_id = 'port1'
with mock.patch.object(self.policy_api, "get") as api_call:
self.resourceApi.get(segment_id, port_id, tenant=TEST_TENANT)
expected_def = self.resourceDef(
segment_id=segment_id,
port_id=port_id,
map_id=core_resources.DEFAULT_MAP_ID,
tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
def test_list(self):
segment_id = 'seg1'
port_id = 'port1'
with mock.patch.object(self.policy_api, "list") as api_call:
self.resourceApi.list(segment_id, port_id, tenant=TEST_TENANT)
expected_def = self.resourceDef(
segment_id=segment_id,
port_id=port_id,
tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
def test_update(self):
name = 'new name'
segment_id = 'seg1'
port_id = 'port1'
prf1 = '1'
with mock.patch.object(self.policy_api,
"create_or_update") as update_call:
self.resourceApi.update(
segment_id=segment_id,
port_id=port_id,
name=name,
qos_profile_id=prf1,
tenant=TEST_TENANT)
expected_def = self.resourceDef(
segment_id=segment_id,
port_id=port_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 TestPolicyDhcpRelayConfig(NsxPolicyLibTestCase):
def setUp(self, *args, **kwargs):
super(TestPolicyDhcpRelayConfig, self).setUp()
self.resourceApi = self.policy_lib.dhcp_relay_config
def test_create(self):
name = 'test'
description = 'desc'
server_addr = '1.1.1.1'
with mock.patch.object(self.policy_api,
"create_or_update") as api_call:
self.resourceApi.create_or_overwrite(
name, description=description,
server_addresses=[server_addr],
tenant=TEST_TENANT)
expected_def = core_defs.DhcpRelayConfigDef(
config_id=mock.ANY,
name=name,
description=description,
server_addresses=[server_addr],
tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
def test_delete(self):
config_id = '111'
with mock.patch.object(self.policy_api, "delete") as api_call:
self.resourceApi.delete(config_id, tenant=TEST_TENANT)
expected_def = core_defs.DhcpRelayConfigDef(config_id=config_id,
tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
def test_get(self):
config_id = '111'
with mock.patch.object(self.policy_api, "get") as api_call:
self.resourceApi.get(config_id, tenant=TEST_TENANT)
expected_def = core_defs.DhcpRelayConfigDef(config_id=config_id,
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 = core_defs.DhcpRelayConfigDef(tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
class TestPolicyLBClientSSLProfileApi(NsxPolicyLibTestCase):
def setUp(self, *args, **kwargs):
super(TestPolicyLBClientSSLProfileApi, self).setUp()
self.resourceApi = self.policy_lib.load_balancer.client_ssl_profile
def test_create_with_id(self):
name = 'd1'
description = 'desc'
id = '111'
protocols = ['TLS_V1_1']
with mock.patch.object(self.policy_api,
"create_or_update") as api_call:
self.resourceApi.create_or_overwrite(
name,
client_ssl_profile_id=id,
description=description,
protocols=protocols,
tenant=TEST_TENANT)
expected_def = lb_defs.LBClientSslProfileDef(
client_ssl_profile_id=id,
name=name,
description=description,
protocols=protocols,
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 = lb_defs.LBClientSslProfileDef(
client_ssl_profile_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 = lb_defs.LBClientSslProfileDef(
client_ssl_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 = lb_defs.LBClientSslProfileDef(
client_ssl_profile_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 = lb_defs.LBClientSslProfileDef(
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 = lb_defs.LBClientSslProfileDef(
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 = lb_defs.LBClientSslProfileDef(
client_ssl_profile_id=id,
name=name,
description=description,
tenant=TEST_TENANT)
self.assert_called_with_def(update_call, expected_def)
class TestPolicyLBCookiePersistenceProfile(NsxPolicyLibTestCase):
def setUp(self, *args, **kwargs):
super(TestPolicyLBCookiePersistenceProfile, self).setUp()
self.resourceApi = (
self.policy_lib.load_balancer.lb_cookie_persistence_profile)
def test_create_with_id(self):
name = 'd1'
description = 'desc'
id = '111'
cookie_garble = 'test_garble'
cookie_name = 'test_name'
cookie_mode = 'INSERT'
cookie_path = 'path'
cookie_time = 'time'
persistence_shared = False
with mock.patch.object(self.policy_api,
"create_or_update") as api_call:
self.resourceApi.create_or_overwrite(
name,
persistence_profile_id=id,
description=description,
cookie_name=cookie_name,
cookie_garble=cookie_garble,
cookie_mode=cookie_mode,
cookie_path=cookie_path,
cookie_time=cookie_time,
persistence_shared=persistence_shared,
tenant=TEST_TENANT)
expected_def = (
lb_defs.LBCookiePersistenceProfileDef(
persistence_profile_id=id,
name=name,
description=description,
cookie_name=cookie_name,
cookie_garble=cookie_garble,
cookie_mode=cookie_mode,
cookie_path=cookie_path,
cookie_time=cookie_time,
persistence_shared=persistence_shared,
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 = (
lb_defs.LBCookiePersistenceProfileDef(
persistence_profile_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 = (
lb_defs.LBCookiePersistenceProfileDef(
persistence_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 = (
lb_defs.LBCookiePersistenceProfileDef(
persistence_profile_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 = (
lb_defs.LBCookiePersistenceProfileDef(
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 = (
lb_defs.LBCookiePersistenceProfileDef(
tenant=TEST_TENANT))
self.assert_called_with_def(api_call, expected_def)
def test_update(self):
id = '111'
name = 'new name'
description = 'new desc'
cookie_garble = 'test_garble'
cookie_name = 'test_name'
cookie_mode = 'INSERT'
cookie_path = 'path'
cookie_time = 'time'
persistence_shared = False
with mock.patch.object(self.policy_api,
"create_or_update") as update_call:
self.resourceApi.update(id,
name=name,
description=description,
cookie_name=cookie_name,
cookie_garble=cookie_garble,
cookie_mode=cookie_mode,
cookie_path=cookie_path,
cookie_time=cookie_time,
persistence_shared=persistence_shared,
tenant=TEST_TENANT)
expected_def = (
lb_defs.LBCookiePersistenceProfileDef(
persistence_profile_id=id,
name=name,
description=description,
cookie_name=cookie_name,
cookie_garble=cookie_garble,
cookie_mode=cookie_mode,
cookie_path=cookie_path,
cookie_time=cookie_time,
persistence_shared=persistence_shared,
tenant=TEST_TENANT))
self.assert_called_with_def(update_call, expected_def)
class TestPolicyLBSourceIpProfileApi(NsxPolicyLibTestCase):
def setUp(self, *args, **kwargs):
super(TestPolicyLBSourceIpProfileApi, self).setUp()
self.resourceApi = (
self.policy_lib.load_balancer.lb_source_ip_persistence_profile)
def test_create_with_id(self):
name = 'd1'
description = 'desc'
id = '111'
ha = 'ha'
persistence_shared = True
purge = 'purge'
timeout = 100
with mock.patch.object(self.policy_api,
"create_or_update") as api_call:
self.resourceApi.create_or_overwrite(
name,
persistence_profile_id=id,
description=description,
ha_persistence_mirroring_enabled=ha,
persistence_shared=persistence_shared,
purge=purge,
timeout=timeout,
tenant=TEST_TENANT)
expected_def = (
lb_defs.LBSourceIpPersistenceProfileDef(
persistence_profile_id=id,
name=name,
description=description,
ha_persistence_mirroring_enabled=ha,
persistence_shared=persistence_shared,
purge=purge,
timeout=timeout,
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 = (
lb_defs.LBSourceIpPersistenceProfileDef(
persistence_profile_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 = (
lb_defs.LBSourceIpPersistenceProfileDef(
persistence_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 = (
lb_defs.LBSourceIpPersistenceProfileDef(
persistence_profile_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 = (
lb_defs.LBSourceIpPersistenceProfileDef(
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 = (
lb_defs.LBSourceIpPersistenceProfileDef(
tenant=TEST_TENANT))
self.assert_called_with_def(api_call, expected_def)
def test_update(self):
id = '111'
name = 'new name'
description = 'new desc'
ha = False
persistence_shared = False
purge = 'no purge'
timeout = 101
with mock.patch.object(self.policy_api,
"create_or_update") as update_call:
self.resourceApi.update(id,
name=name,
description=description,
ha_persistence_mirroring_enabled=ha,
persistence_shared=persistence_shared,
purge=purge,
timeout=timeout,
tenant=TEST_TENANT)
expected_def = (
lb_defs.LBSourceIpPersistenceProfileDef(
persistence_profile_id=id,
name=name,
description=description,
ha_persistence_mirroring_enabled=ha,
persistence_shared=persistence_shared,
purge=purge,
timeout=timeout,
tenant=TEST_TENANT))
self.assert_called_with_def(update_call, expected_def)
class TestPolicyLBApplicationProfile(NsxPolicyLibTestCase):
def setUp(self, *args, **kwargs):
super(TestPolicyLBApplicationProfile, self).setUp()
self.resourceApi = self.policy_lib.load_balancer.lb_http_profile
def test_create_with_id(self):
name = 'd1'
description = 'desc'
id = '111'
http_redirect_to_https = False
http_redirect_to = "sample-url"
idle_timeout = 100
ntlm = False
request_body_size = 1025
request_header_size = 10
response_header_size = 10
response_timeout = 10
x_forwarded_for = 'INSERT'
with mock.patch.object(self.policy_api,
"create_or_update") as api_call:
self.resourceApi.create_or_overwrite(
name,
lb_app_profile_id=id,
description=description,
http_redirect_to_https=http_redirect_to_https,
http_redirect_to=http_redirect_to,
idle_timeout=idle_timeout,
ntlm=ntlm,
request_body_size=request_body_size,
request_header_size=request_header_size,
response_header_size=response_header_size,
response_timeout=response_timeout,
x_forwarded_for=x_forwarded_for,
tenant=TEST_TENANT)
expected_def = (
lb_defs.LBHttpProfileDef(
lb_app_profile_id=id,
name=name,
description=description,
http_redirect_to_https=http_redirect_to_https,
http_redirect_to=http_redirect_to,
idle_timeout=idle_timeout,
ntlm=ntlm,
request_body_size=request_body_size,
request_header_size=request_header_size,
response_header_size=response_header_size,
response_timeout=response_timeout,
x_forwarded_for=x_forwarded_for,
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 = (
lb_defs.LBHttpProfileDef(
lb_app_profile_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 = (
lb_defs.LBHttpProfileDef(
lb_app_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 = (
lb_defs.LBHttpProfileDef(
lb_app_profile_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 = (
lb_defs.LBHttpProfileDef(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 = (
lb_defs.LBHttpProfileDef(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 = (
lb_defs.LBHttpProfileDef(
lb_app_profile_id=id,
name=name,
description=description,
tenant=TEST_TENANT))
self.assert_called_with_def(update_call, expected_def)
class TestPolicyLBService(NsxPolicyLibTestCase):
def setUp(self, *args, **kwargs):
super(TestPolicyLBService, self).setUp()
self.resourceApi = self.policy_lib.load_balancer.lb_service
def test_create_with_id(self):
name = 'd1'
description = 'desc'
id = '111'
size = 'SMALL'
with mock.patch.object(self.policy_api,
"create_or_update") as api_call:
self.resourceApi.create_or_overwrite(name,
lb_service_id=id,
description=description,
size=size,
tenant=TEST_TENANT)
expected_def = (
lb_defs.LBServiceDef(
lb_service_id=id,
name=name,
description=description,
size=size,
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 = (
lb_defs.LBServiceDef(lb_service_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 = lb_defs.LBServiceDef(
lb_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 = lb_defs.LBServiceDef(
lb_service_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 = lb_defs.LBServiceDef(
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 = lb_defs.LBServiceDef(
tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
def test_update(self):
id = '111'
name = 'new name'
description = 'new desc'
size = 'SMALL'
with mock.patch.object(self.policy_api,
"create_or_update") as update_call:
self.resourceApi.update(id,
name=name,
description=description,
tenant=TEST_TENANT,
size=size)
expected_def = lb_defs.LBServiceDef(
lb_service_id=id,
name=name,
description=description,
tenant=TEST_TENANT,
size=size)
self.assert_called_with_def(update_call, expected_def)
class TestPolicyLBVirtualServer(NsxPolicyLibTestCase):
def setUp(self, *args, **kwargs):
super(TestPolicyLBVirtualServer, self).setUp()
self.resourceApi = self.policy_lib.load_balancer.virtual_server
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,
virtual_server_id=id,
description=description,
tenant=TEST_TENANT)
expected_def = lb_defs.LBVirtualServerDef(
virtual_server_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 = lb_defs.LBVirtualServerDef(
virtual_server_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 = lb_defs.LBVirtualServerDef(
virtual_server_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 = lb_defs.LBVirtualServerDef(
virtual_server_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 = lb_defs.LBVirtualServerDef(
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 = lb_defs.LBVirtualServerDef(
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 = lb_defs.LBVirtualServerDef(
virtual_server_id=id, name=name, description=description,
tenant=TEST_TENANT)
self.assert_called_with_def(update_call, expected_def)
class TestPolicyLBPoolApi(NsxPolicyLibTestCase):
def setUp(self, *args, **kwargs):
super(TestPolicyLBPoolApi, self).setUp()
self.resourceApi = self.policy_lib.load_balancer.lb_pool
def test_create_with_id(self):
name = 'd1'
description = 'desc'
id = '111'
members = [
lb_defs.LBPoolMemberDef(ip_address='10.0.0.1')]
algorithm = 'algo'
active_monitor_paths = 'path1'
member_group = 'group1'
snat_translation = False
with mock.patch.object(self.policy_api,
"create_or_update") as api_call:
self.resourceApi.create_or_overwrite(
name,
lb_pool_id=id,
description=description,
members=members,
active_monitor_paths=active_monitor_paths,
algorithm=algorithm,
member_group=member_group,
snat_translation=snat_translation,
tenant=TEST_TENANT)
expected_def = lb_defs.LBPoolDef(
lb_pool_id=id,
name=name,
description=description,
members=members,
active_monitor_paths=active_monitor_paths,
algorithm=algorithm,
member_group=member_group,
snat_translation=snat_translation,
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 = lb_defs.LBPoolDef(
lb_pool_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 = lb_defs.LBPoolDef(
lb_pool_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 = lb_defs.LBPoolDef(
lb_pool_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 = lb_defs.LBPoolDef(
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 = lb_defs.LBPoolDef(
tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
def test_update(self):
id = '111'
name = 'new name'
description = 'new desc'
members = [{'ip_address': '10.0.0.1'}]
algorithm = 'algo'
active_monitor_paths = 'path1'
member_group = 'group1'
snat_translation = False
with mock.patch.object(self.policy_api,
"create_or_update") as update_call:
self.resourceApi.update(id,
name=name,
description=description,
members=members,
active_monitor_paths=active_monitor_paths,
algorithm=algorithm,
member_group=member_group,
snat_translation=snat_translation,
tenant=TEST_TENANT)
expected_def = lb_defs.LBPoolDef(
lb_pool_id=id,
name=name,
description=description,
members=members,
active_monitor_paths=active_monitor_paths,
algorithm=algorithm,
member_group=member_group,
snat_translation=snat_translation,
tenant=TEST_TENANT)
self.assert_called_with_def(update_call, expected_def)
class TestPolicyCertificate(NsxPolicyLibTestCase):
def setUp(self, *args, **kwargs):
super(TestPolicyCertificate, self).setUp()
self.resourceApi = self.policy_lib.certificate
def test_create_with_id(self):
name = 'd1'
description = 'desc'
id = '111'
pem_encoded = 'pem_encoded'
private_key = 'private_key'
passphrase = 'passphrase'
key_algo = 'algo'
with mock.patch.object(self.policy_api,
"create_or_update") as api_call:
self.resourceApi.create_or_overwrite(name,
certificate_id=id,
description=description,
pem_encoded=pem_encoded,
private_key=private_key,
passphrase=passphrase,
key_algo=key_algo,
tenant=TEST_TENANT)
expected_def = (
core_defs.CertificateDef(
certificate_id=id,
name=name,
description=description,
pem_encoded=pem_encoded,
private_key=private_key,
passphrase=passphrase,
key_algo=key_algo,
tenant=TEST_TENANT))
self.assert_called_with_def(api_call, expected_def)
def test_create_without_id(self):
name = 'd1'
description = 'desc'
pem_encoded = 'pem_encoded'
with mock.patch.object(self.policy_api,
"create_or_update") as api_call:
self.resourceApi.create_or_overwrite(name, description=description,
tenant=TEST_TENANT,
pem_encoded=pem_encoded)
expected_def = (
core_defs.CertificateDef(certificate_id=mock.ANY,
name=name,
description=description,
tenant=TEST_TENANT,
pem_encoded=pem_encoded))
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 = core_defs.CertificateDef(
certificate_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 = core_defs.CertificateDef(
certificate_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 = core_defs.CertificateDef(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 = core_defs.CertificateDef(tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
def test_update(self):
id = '111'
name = 'new name'
description = 'new desc'
pem_encoded = 'pem_encoded'
private_key = 'private_key'
passphrase = '12'
key_algo = 'new_algo'
with mock.patch.object(self.policy_api,
"create_or_update") as update_call:
self.resourceApi.update(id,
name=name,
description=description,
tenant=TEST_TENANT,
pem_encoded=pem_encoded,
private_key=private_key,
passphrase=passphrase,
key_algo=key_algo)
expected_def = core_defs.CertificateDef(
certificate_id=id,
name=name,
description=description,
tenant=TEST_TENANT,
pem_encoded=pem_encoded,
private_key=private_key,
passphrase=passphrase,
key_algo=key_algo
)
self.assert_called_with_def(update_call, expected_def)
View Git Blame Copy Permalink