A common library that interfaces with VMware NSX.
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 
 

7066 lines
286 KiB

# 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.
#
from unittest import mock
from vmware_nsxlib.tests.unit.v3 import nsxlib_testcase
from vmware_nsxlib.tests.unit.v3.policy import policy_testcase
from vmware_nsxlib.tests.unit.v3 import test_client
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
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") as mock_lib:
mock_lib.return_value.get_version.return_value = (
nsxlib_testcase.LATEST_VERSION)
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)
def assert_called_with_def_and_kwargs(self, mock_api, expected_def,
call_num=0, **kwargs):
# verify the api & resource definition
self.assert_called_with_def(mock_api, expected_def, call_num=call_num)
actual_kwargs = mock_api.call_args_list[call_num][1]
self.assertDictEqual(actual_kwargs, kwargs)
def mock_get(self, obj_id, obj_name, **kwargs):
obj_dict = {
'id': obj_id,
'display_name': obj_name,
'resource_type': self.resourceApi.entry_def.resource_type()}
if kwargs:
obj_dict.update(kwargs)
return mock.patch.object(self.policy_api, "get",
return_value=obj_dict)
def mock_create_update(self):
return mock.patch.object(self.policy_api, "create_or_update")
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'
domain_id = '111'
with self.mock_create_update() as api_call:
result = self.resourceApi.create_or_overwrite(
name,
domain_id=domain_id,
description=description,
tenant=TEST_TENANT)
expected_def = core_defs.DomainDef(domain_id=domain_id,
name=name,
description=description,
tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
self.assertEqual(domain_id, result)
def test_minimalistic_create(self):
name = 'test'
with self.mock_create_update() as api_call:
result = 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)
self.assertIsNotNone(result)
def test_create_without_id(self):
name = 'd1'
description = 'desc'
with mock.patch.object(self.policy_api,
"create_or_update") as api_call:
result = 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)
self.assertIsNotNone(result)
def test_delete(self):
domain_id = '111'
with mock.patch.object(self.policy_api, "delete") as api_call:
self.resourceApi.delete(domain_id, tenant=TEST_TENANT)
expected_def = core_defs.DomainDef(domain_id=domain_id,
tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
def test_get(self):
domain_id = '111'
with mock.patch.object(self.policy_api, "get",
return_value={'id': domain_id}) as api_call:
result = self.resourceApi.get(domain_id, tenant=TEST_TENANT)
expected_def = core_defs.DomainDef(domain_id=domain_id,
tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
self.assertEqual(domain_id, result['id'])
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",
return_value={'results': []}) as api_call:
result = self.resourceApi.list(tenant=TEST_TENANT)
expected_def = core_defs.DomainDef(tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
self.assertEqual([], result)
def test_update(self):
domain_id = '111'
name = 'new name'
description = 'new desc'
with self.mock_get(domain_id, name), \
self.mock_create_update() as update_call:
self.resourceApi.update(domain_id,
name=name,
description=description,
tenant=TEST_TENANT)
expected_def = core_defs.DomainDef(domain_id=domain_id,
name=name,
description=description,
tenant=TEST_TENANT)
self.assert_called_with_def(update_call, expected_def)
def test_unset(self):
domain_id = '111'
with mock.patch.object(self.policy_api, "get",
return_value={'id': domain_id}):
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,
headers=test_client.PARTIAL_UPDATE_HEADERS)
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'
group_id = '222'
with mock.patch.object(self.policy_api,
"create_or_update") as api_call:
result = self.resourceApi.create_or_overwrite(
name, domain_id,
group_id=group_id,
description=description,
tenant=TEST_TENANT)
expected_def = core_defs.GroupDef(domain_id=domain_id,
group_id=group_id,
name=name,
description=description,
conditions=[],
tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
self.assertEqual(group_id, result)
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:
result = 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)
self.assertIsNotNone(result)
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:
result = 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)
self.assertIsNotNone(result)
def test_create_with_empty_condition(self):
domain_id = '111'
name = 'g1'
description = 'desc'
with mock.patch.object(self.policy_api,
"create_or_update") as api_call:
result = 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)
self.assertIsNotNone(result)
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:
result = 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)
self.assertIsNotNone(result)
def _test_create_with_condition(self, condition, exp_condition):
domain_id = '111'
name = 'g1'
description = 'desc'
with mock.patch.object(self.policy_api,
"create_or_update") as api_call:
result = self.resourceApi.create_or_overwrite_with_conditions(
name, domain_id, description=description,
conditions=condition, tenant=TEST_TENANT)
expected_def = core_defs.GroupDef(domain_id=domain_id,
group_id=mock.ANY,
name=name,
description=description,
conditions=exp_condition,
tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
self.assertIsNotNone(result)
def test_create_with_union_condition(self):
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)
cond1_dup = self.resourceApi.build_condition(
cond_val=cond_val1,
cond_op=cond_op,
cond_member_type=cond_member_type,
cond_key=cond_key)
union_cond_no_dup = self.resourceApi.build_union_condition(
conditions=[cond1, cond2])
union_cond_dup = self.resourceApi.build_union_condition(
conditions=[cond1, cond1_dup])
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)
or_cond = core_defs.ConjunctionOperator(
operator=constants.CONDITION_OP_OR)
exp_cond = list(set([exp_cond1, exp_cond2]))
exp_cond.insert(1, or_cond)
self._test_create_with_condition(union_cond_no_dup, exp_cond)
self._test_create_with_condition(union_cond_dup, [exp_cond1])
def test_create_with_nested_condition(self):
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])
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()
expressions = list(set([exp_cond1, exp_cond2]))
expressions.insert(1, and_cond)
exp_cond = core_defs.NestedExpression(expressions=expressions)
self._test_create_with_condition(nested, exp_cond)
def test_create_with_dup_nested_condition(self):
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])
cond1_dup = self.resourceApi.build_condition(
cond_val=cond_val1,
cond_op=cond_op,
cond_member_type=cond_member_type,
cond_key=cond_key)
cond2_dup = self.resourceApi.build_condition(
cond_val=cond_val2,
cond_op=cond_op,
cond_member_type=cond_member_type,
cond_key=cond_key)
nested_dup = self.resourceApi.build_nested_condition(
conditions=[cond1_dup, cond2_dup])
double_nested = self.resourceApi.build_nested_condition(
conditions=[nested, nested_dup])
exp_cond = core_defs.NestedExpression(expressions=[nested])
self._test_create_with_condition(double_nested, exp_cond)
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:
result = 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)
self.assertIsNotNone(result)
def test_create_with_path_expression(self):
domain_id = '111'
name = 'g1'
description = 'desc'
path = '/test/path1'
cond = self.resourceApi.build_path_expression([path])
with mock.patch.object(self.policy_api,
"create_or_update") as api_call:
result = self.resourceApi.create_or_overwrite_with_conditions(
name, domain_id, description=description,
conditions=[cond],
tenant=TEST_TENANT)
exp_cond = core_defs.PathExpression([path])
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)
self.assertIsNotNone(result)
def test_delete(self):
domain_id = '111'
group_id = '222'
with mock.patch.object(self.policy_api, "delete") as api_call:
self.resourceApi.delete(domain_id, group_id, tenant=TEST_TENANT)
expected_def = core_defs.GroupDef(domain_id=domain_id,
group_id=group_id,
tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
def test_get(self):
domain_id = '111'
group_id = '222'
with mock.patch.object(self.policy_api, "get",
return_value={'id': group_id}) as api_call:
result = self.resourceApi.get(domain_id, group_id,
tenant=TEST_TENANT)
expected_def = core_defs.GroupDef(domain_id=domain_id,
group_id=group_id,
tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
self.assertEqual(group_id, result['id'])
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",
return_value={'results': []}) as api_call:
result = 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)
self.assertEqual([], result)
def test_update(self):
domain_id = '111'
group_id = '222'
name = 'new name'
description = 'new desc'
with self.mock_get(group_id, name), \
self.mock_create_update() as update_call:
self.resourceApi.update(domain_id, group_id,
name=name,
description=description,
tenant=TEST_TENANT)
expected_def = core_defs.GroupDef(domain_id=domain_id,
group_id=group_id,
name=name,
description=description,
tenant=TEST_TENANT)
self.assert_called_with_def(
update_call, expected_def)
def test_update_with_conditions(self):
domain_id = '111'
group_id = '222'
name = 'name'
new_name = 'new name'
description = 'desc'
new_description = 'new 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_def = core_defs.Condition(value=cond_val1,
key=cond_key,
operator=cond_op,
member_type=cond_member_type)
cond2_def = core_defs.Condition(value=cond_val2,
key=cond_key,
operator=cond_op,
member_type=cond_member_type)
original_group = {
'id': group_id,
'resource_type': 'Group',
'display_name': name,
'description': description,
'expression': [cond1_def.get_obj_dict()]}
updated_group = {
'id': group_id,
'resource_type': 'Group',
'display_name': new_name,
'description': new_description,
'expression': [cond2_def.get_obj_dict()]}
group_def = core_defs.GroupDef(
domain_id=domain_id,
group_id=group_id,
tenant=TEST_TENANT)
with mock.patch.object(self.policy_api, "get",
return_value=original_group),\
mock.patch.object(self.policy_api.client,
"update") as update_call:
self.resourceApi.update_with_conditions(
domain_id, group_id, name=new_name,
description=new_description,
conditions=[cond2_def], tenant=TEST_TENANT)
update_call.assert_called_once_with(
group_def.get_resource_path(), updated_group)
def test_update_with_conditions_callback(self):
def update_payload_cbk(revised_payload, payload):
revised_ips = revised_payload["expression"][0]["ip_addresses"]
new_ips = payload["conditions"][0].ip_addresses
updated_ips = revised_ips + new_ips
payload["conditions"] = [core_defs.IPAddressExpression(
updated_ips)]
domain_id = '111'
group_id = '222'
name = 'name'
new_name = 'new name'
description = 'desc'
new_description = 'new desc'
ips1 = ["1.1.1.1"]
ips2 = ["2.2.2.2"]
cond1_def = core_defs.IPAddressExpression(ips1)
cond2_def = core_defs.IPAddressExpression(ips2)
updated_cond_def = core_defs.IPAddressExpression(ips1 + ips2)
original_group = {
'id': group_id,
'resource_type': 'Group',
'display_name': name,
'description': description,
'expression': [cond1_def.get_obj_dict()]}
updated_group = {
'id': group_id,
'resource_type': 'Group',
'display_name': new_name,
'description': new_description,
'expression': [updated_cond_def.get_obj_dict()]}
group_def = core_defs.GroupDef(
domain_id=domain_id,
group_id=group_id,
conditions=[cond2_def],
tenant=TEST_TENANT)
with mock.patch.object(self.policy_api, "get",
return_value=original_group),\
mock.patch.object(self.policy_api.client,
"update") as update_call:
self.resourceApi.update_with_conditions(
domain_id, group_id, name=new_name,
description=new_description,
conditions=[cond2_def], tenant=TEST_TENANT,
update_payload_cbk=update_payload_cbk)
update_call.assert_called_once_with(
group_def.get_resource_path(), updated_group)
def test_unset(self):
domain_id = '111'
group_id = '222'
description = 'new'
with self.mock_get(group_id, 'test'):
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,
headers=test_client.PARTIAL_UPDATE_HEADERS)
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, silent=False)
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, silent=False)
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, silent=False)
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, silent=False)
def test_get_path(self):
domain_id = 'd1'
group_id = 'g1'
result = self.resourceApi.get_path(domain_id, group_id,
tenant=TEST_TENANT)
expected_path = '/%s/domains/%s/groups/%s' % (
TEST_TENANT, domain_id, group_id)
self.assertEqual(expected_path, result)
def test_wait_until_realized_fail(self):
domain_id = 'd1'
group_id = 'g1'
info = {'state': constants.STATE_UNREALIZED,
'realization_specific_identifier': group_id,
'entity_type': 'RealizedGroup'}
with mock.patch.object(self.resourceApi, "_get_realization_info",
return_value=info):
self.assertRaises(nsxlib_exc.RealizationTimeoutError,
self.resourceApi.wait_until_realized,
domain_id, group_id, max_attempts=5,
sleep=0.1, tenant=TEST_TENANT)
def test_wait_until_realized_succeed(self):
domain_id = 'd1'
group_id = 'g1'
info = {'state': constants.STATE_REALIZED,
'realization_specific_identifier': group_id,
'entity_type': 'RealizedGroup'}
with mock.patch.object(self.resourceApi, "_get_realization_info",
return_value=info):
actual_info = self.resourceApi.wait_until_realized(
domain_id, group_id, max_attempts=5, sleep=0.1,
tenant=TEST_TENANT)
self.assertEqual(info, actual_info)
class TestPolicyL4Service(NsxPolicyLibTestCase):
def setUp(self, *args, **kwargs):
super(TestPolicyL4Service, self).setUp()
self.resourceApi = self.policy_lib.service
def test_create(self):
name = 's1'
description = 'desc'
protocol = constants.TCP
dest_ports = [81, 82]
source_ports = [83, 84]
tags = [{'scope': 'a', 'tag': 'b'}]
with mock.patch.object(self.policy_api,
"create_with_parent") as api_call:
result = self.resourceApi.create_or_overwrite(
name,
description=description,
protocol=protocol,
dest_ports=dest_ports,
source_ports=source_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,
source_ports=source_ports,
tenant=TEST_TENANT)
self.assert_called_with_defs(
api_call, [exp_srv_def, exp_entry_def])
self.assertIsNotNone(result)
def test_delete(self):
srv_id = '111'
with mock.patch.object(self.policy_api, "delete") as api_call:
self.resourceApi.delete(srv_id, tenant=TEST_TENANT)
expected_def = core_defs.ServiceDef(service_id=srv_id,
tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
def test_get(self):
srv_id = '111'
with mock.patch.object(self.policy_api, "get",
return_value={'id': srv_id}) as api_call:
result = self.resourceApi.get(srv_id, tenant=TEST_TENANT)
expected_def = core_defs.ServiceDef(service_id=srv_id,
tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
self.assertEqual(srv_id, result['id'])
def test_get_by_name(self):
name = 's1'
with mock.patch.object(
self.policy_api, "list",
return_value={'results': [{'display_name': name}]}) as api_call:
obj = self.resourceApi.get_by_name(name, tenant=TEST_TENANT)
self.assertIsNotNone(obj)
expected_def = core_defs.ServiceDef(tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
def test_list(self):
with mock.patch.object(self.policy_api, "list",
return_value={'results': []}) as api_call:
result = self.resourceApi.list(tenant=TEST_TENANT)
expected_def = core_defs.ServiceDef(tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
self.assertEqual([], result)
def test_update(self):
srv_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(srv_id,
name=name,
description=description,
tags=tags,
tenant=TEST_TENANT)
service_def = core_defs.ServiceDef(service_id=srv_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):
srv_id = '111'
name = 'newName'
description = 'new desc'
protocol = 'udp'
dest_ports = [555]
source_ports = [666]
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(srv_id,
name=name,
description=description,
protocol=protocol,
dest_ports=dest_ports,
source_ports=source_ports,
tenant=TEST_TENANT)
service_def = core_defs.ServiceDef(service_id=srv_id,
name=name,
description=description,
tenant=TEST_TENANT)
entry_def = core_defs.L4ServiceEntryDef(
service_id=srv_id,
entry_id=mock.ANY,
protocol=protocol,
dest_ports=dest_ports,
source_ports=source_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:
result = 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])
self.assertIsNotNone(result)
def test_delete(self):
srv_id = '111'
with mock.patch.object(self.policy_api, "delete") as api_call:
self.resourceApi.delete(srv_id, tenant=TEST_TENANT)
expected_def = core_defs.ServiceDef(service_id=srv_id,
tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
def test_get(self):
srv_id = '111'
with mock.patch.object(self.policy_api, "get",
return_value={'id': srv_id}) as api_call:
result = self.resourceApi.get(srv_id, tenant=TEST_TENANT)
expected_def = core_defs.ServiceDef(service_id=srv_id,
tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
self.assertEqual(srv_id, result['id'])
def test_get_by_name(self):
name = 's1'
with mock.patch.object(
self.policy_api, "list",
return_value={'results': [{'display_name': name}]}) as api_call:
obj = self.resourceApi.get_by_name(name, tenant=TEST_TENANT)
self.assertIsNotNone(obj)
expected_def = core_defs.ServiceDef(tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
def test_list(self):
with mock.patch.object(self.policy_api, "list",
return_value={'results': []}) as api_call:
result = self.resourceApi.list(tenant=TEST_TENANT)
expected_def = core_defs.ServiceDef(tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
self.assertEqual([], result)
def test_update(self):
srv_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(srv_id,
name=name,
description=description,
tenant=TEST_TENANT)
service_def = core_defs.ServiceDef(service_id=srv_id,
name=name,
description=description,
tenant=TEST_TENANT)
entry_def = core_defs.IcmpServiceEntryDef(
service_id=srv_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):
srv_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(srv_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=srv_id,
name=name,
description=description,
tenant=TEST_TENANT)
entry_def = core_defs.IcmpServiceEntryDef(
service_id=srv_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])
def test_icmp_type_and_code_in_obj_dict(self):
icmp_type, icmp_code = 0, 0
entry_def = core_defs.IcmpServiceEntryDef(
icmp_type=icmp_type, icmp_code=icmp_code)
body = entry_def.get_obj_dict()
self.assertEqual(icmp_type, body["icmp_type"])
self.assertEqual(icmp_code, body["icmp_code"])
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:
result = 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])
self.assertIsNotNone(result)
def test_delete(self):
srv_id = '111'
with mock.patch.object(self.policy_api, "delete") as api_call:
self.resourceApi.delete(srv_id, tenant=TEST_TENANT)
expected_def = core_defs.ServiceDef(service_id=srv_id,
tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
def test_get(self):
srv_id = '111'
with mock.patch.object(self.policy_api, "get",
return_value={'id': srv_id}) as api_call:
result = self.resourceApi.get(srv_id, tenant=TEST_TENANT)
expected_def = core_defs.ServiceDef(service_id=srv_id,
tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
self.assertEqual(srv_id, result['id'])
def test_get_by_name(self):
name = 's1'
with mock.patch.object(
self.policy_api, "list",
return_value={'results': [{'display_name': name}]}) as api_call:
obj = self.resourceApi.get_by_name(name, tenant=TEST_TENANT)
self.assertIsNotNone(obj)
expected_def = core_defs.ServiceDef(tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
def test_list(self):
with mock.patch.object(self.policy_api, "list",
return_value={'results': []}) as api_call:
result = self.resourceApi.list(tenant=TEST_TENANT)
expected_def = core_defs.ServiceDef(tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
self.assertEqual([], result)
def test_update(self):
srv_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(srv_id,
name=name,
description=description,
tenant=TEST_TENANT)
service_def = core_defs.ServiceDef(service_id=srv_id,
name=name,
description=description,
tenant=TEST_TENANT)
entry_def = core_defs.IPProtocolServiceEntryDef(
service_id=srv_id,
entry_id='entry',
tenant=TEST_TENANT)
self.assert_called_with_defs(update_call, [service_def, entry_def])
def test_update_all(self):
srv_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(srv_id,
name=name,
description=description,
protocol_number=protocol_number,
tenant=TEST_TENANT)
service_def = core_defs.ServiceDef(service_id=srv_id,
name=name,
description=description,
tenant=TEST_TENANT)
entry_def = core_defs.IPProtocolServiceEntryDef(
service_id=srv_id,
entry_id='entry',
protocol_number=protocol_number,
tenant=TEST_TENANT)
self.assert_called_with_defs(service_update_call,
[service_def, entry_def])
def test_protocol_number_in_obj_dict(self):
protocol_number = 0
entry_def = core_defs.IPProtocolServiceEntryDef(
protocol_number=protocol_number)
body = entry_def.get_obj_dict()
self.assertEqual(protocol_number, body["protocol_number"])
class TestPolicyMixedService(NsxPolicyLibTestCase):
def setUp(self, *args, **kwargs):
super(TestPolicyMixedService, self).setUp()
self.l4ServiceApi = self.policy_lib.service
self.icmpServiceApi = self.policy_lib.icmp_service
self.ipServiceApi = self.policy_lib.ip_protocol_service
self.resourceApi = self.policy_lib.mixed_service
def test_create_service_only(self):
name = 's1'
srv_id = '111'
description = 'desc'
tags = [{'scope': 'a', 'tag': 'b'}]
with mock.patch.object(self.policy_api,
"create_or_update") as api_call:
result = self.resourceApi.create_or_overwrite(
name,
srv_id,
description=description,
tags=tags,
tenant=TEST_TENANT)
exp_srv_def = core_defs.ServiceDef(
service_id=srv_id,
name=name,
description=description,
tags=tags,
tenant=TEST_TENANT)
self.assert_called_with_def(api_call, exp_srv_def)
self.assertIsNotNone(result)
def test_create_with_entries(self):
name = 's1'
srv_id = '111'
description = 'desc'
tags = [{'scope': 'a', 'tag': 'b'}]
protocol = constants.TCP
dest_ports = [81, 82]
source_ports = [83, 84]
icmp_type = 2
protocol_number = 2
l4_entry = self.l4ServiceApi.build_entry(
'l4_entry', srv_id, 'l4_entry', protocol=protocol,
dest_ports=dest_ports, source_ports=source_ports,
tenant=TEST_TENANT)
icmp_entry = self.icmpServiceApi.build_entry(
'icmp_entry', srv_id, 'icmp_entry', icmp_type=icmp_type,
tenant=TEST_TENANT)
ip_entry = self.ipServiceApi.build_entry(
'ip_entry', srv_id, 'ip_entry',
protocol_number=protocol_number, tenant=TEST_TENANT)
with mock.patch.object(self.policy_api,
"create_with_parent") as api_call:
result = self.resourceApi.create_or_overwrite(
name,
srv_id,
description=description,
entries=[l4_entry, icmp_entry, ip_entry],
tags=tags,
tenant=TEST_TENANT)
service_def = core_defs.ServiceDef(
service_id=srv_id,
name=name,
description=description,
tags=tags,
tenant=TEST_TENANT)
self.assert_called_with_defs(
api_call, [service_def, l4_entry, icmp_entry, ip_entry])
self.assertIsNotNone(result)
def test_delete(self):
srv_id = '111'
with mock.patch.object(self.policy_api, "delete") as api_call:
self.resourceApi.delete(srv_id, tenant=TEST_TENANT)
expected_def = core_defs.ServiceDef(service_id=srv_id,
tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
def test_get(self):
srv_id = '111'
with mock.patch.object(self.policy_api, "get",
return_value={'id': srv_id}) as api_call:
result = self.resourceApi.get(srv_id, tenant=TEST_TENANT)
expected_def = core_defs.ServiceDef(service_id=srv_id,
tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
self.assertEqual(srv_id, result['id'])
def test_get_by_name(self):
name = 's1'
with mock.patch.object(
self.policy_api, "list",
return_value={'results': [{'display_name': name}]}) as api_call:
obj = self.resourceApi.get_by_name(name, tenant=TEST_TENANT)
self.assertIsNotNone(obj)
expected_def = core_defs.ServiceDef(tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
def test_list(self):
with mock.patch.object(self.policy_api, "list",
return_value={'results': []}) as api_call:
result = self.resourceApi.list(tenant=TEST_TENANT)
expected_def = core_defs.ServiceDef(tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
self.assertEqual([], result)
def test_update(self):
name = 'newName'
srv_id = '111'
description = 'new desc'
tags = [{'scope': 'c', 'tag': 'd'}]
protocol = constants.UDP
dest_ports = [91, 92]
source_ports = [93, 94]
icmp_type = 3
protocol_number = 3
l4_entry = self.l4ServiceApi.build_entry(
'l4_entry', srv_id, 'l4_entry', protocol=protocol,
dest_ports=dest_ports, source_ports=source_ports,
tenant=TEST_TENANT)
icmp_entry = self.icmpServiceApi.build_entry(
'icmp_entry', srv_id, 'icmp_entry', icmp_type=icmp_type,
tenant=TEST_TENANT)
ip_entry = self.ipServiceApi.build_entry(
'ip_entry', srv_id, 'ip_entry',
protocol_number=protocol_number, tenant=TEST_TENANT)
with mock.patch.object(self.policy_api, "get",
return_value={}),\
mock.patch.object(self.policy_api,
"create_with_parent") as update_call:
self.resourceApi.update(
srv_id,
name=name,
description=description,
entries=[l4_entry, icmp_entry, ip_entry],
tags=tags,
tenant=TEST_TENANT)
service_def = core_defs.ServiceDef(
service_id=srv_id,
name=name,
description=description,
tags=tags,
tenant=TEST_TENANT)
self.assert_called_with_defs(
update_call, [service_def, l4_entry, icmp_entry, ip_entry])
class TestPolicyCommunicationMap(NsxPolicyLibTestCase):
def setUp(self, *args, **kwargs):
super(TestPolicyCommunicationMap, self).setUp()
self.resourceApi = self.policy_lib.comm_map
self.mapDef = core_defs.CommunicationMapDef
self.entryDef = core_defs.CommunicationMapEntryDef
self.resource_type = 'SecurityPolicy'
self.path_name = 'security-policies'
def test_create_another(self):
domain_id = '111'
map_id = '222'
name = 'cm1'
description = 'desc'
source_group = 'g1'
dest_group = 'g2'
seq_num = 7
map_seq_num = 10
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):
result = 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,
map_sequence_number=map_seq_num,
tenant=TEST_TENANT)
map_def = self.mapDef(
domain_id=domain_id,
map_id=map_id,
name=name,
description=description,
category=constants.CATEGORY_APPLICATION,
map_sequence_number=map_seq_num,
tenant=TEST_TENANT)
entry_def = self.entryDef(
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])
self.assertEqual(map_id, result)
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):
result = 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 = self.mapDef(
domain_id=domain_id,
map_id=map_id,
name=name,
description=description,
category=category,
tenant=TEST_TENANT)
entry_def = self.entryDef(
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])
self.assertEqual(map_id, result)
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:
result = 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 = self.mapDef(
domain_id=domain_id,
map_id=mock.ANY,
name=name,
description=description,
category=constants.CATEGORY_APPLICATION,
tenant=TEST_TENANT)
expected_entry_def = self.entryDef(
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])
self.assertIsNotNone(result)
def test_create_map_only(self):
domain_id = '111'
name = 'cm1'
description = 'desc'
map_seq_num = 10
with mock.patch.object(self.policy_api,
"create_or_update") as api_call:
result = self.resourceApi.create_or_overwrite_map_only(
name, domain_id, description=description,
map_sequence_number=map_seq_num, tenant=TEST_TENANT)
expected_map_def = self.mapDef(
domain_id=domain_id,
map_id=mock.ANY,
name=name,
description=description,
category=constants.CATEGORY_APPLICATION,
map_sequence_number=map_seq_num,
tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_map_def)
self.assertIsNotNone(result)
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'
tag = 'abc1234'
tags = ['dummy']
with mock.patch.object(self.policy_api,
"create_or_update") as api_call:
result = 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,
ip_protocol=nsx_constants.IPV4,
tag=tag, tags=tags,
tenant=TEST_TENANT)
expected_entry_def = self.entryDef(
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,
ip_protocol=nsx_constants.IPV4,
scope=None,
logged=False,
tag=tag,
tags=tags,
tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_entry_def)
self.assertIsNotNone(result)
def test_create_entry_no_service(self):
domain_id = '111'
map_id = '333'
name = 'cm1'
description = 'desc'
source_group = 'g1'
dest_group = 'g2'
tag = 'abc1234'
tags = ['abc1234']
with mock.patch.object(self.policy_api,
"create_or_update") as api_call:
result = self.resourceApi.create_entry(
name, domain_id, map_id,
description=description,
source_groups=[source_group],
dest_groups=[dest_group],
sequence_number=1,
tag=tag, tags=tags,
tenant=TEST_TENANT)
expected_entry_def = self.entryDef(
domain_id=domain_id,
map_id=map_id,
entry_id=mock.ANY,
name=name,
action=constants.ACTION_ALLOW,
direction=nsx_constants.IN_OUT,
ip_protocol=nsx_constants.IPV4_IPV6,
description=description,
sequence_number=1,
service_ids=None,
source_groups=[source_group],
dest_groups=[dest_group],
scope=None,
logged=False,
tag=tag,
tags=tags,
tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_entry_def)
self.assertIsNotNone(result)
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}]}
tag = 'abc1234'
tags = ['dummy']
with mock.patch.object(self.policy_api,
"create_or_update") as api_call,\
mock.patch.object(self.policy_api,
"get", return_value=ret_comm):
result = 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,
tag=tag, tags=tags,
tenant=TEST_TENANT)
expected_entry_def = self.entryDef(
domain_id=domain_id,
map_id=map_id,
entry_id=mock.ANY,
name=name,
action=constants.ACTION_ALLOW,
direction=nsx_constants.IN_OUT,
ip_protocol=nsx_constants.IPV4_IPV6,
description=description,
service_ids=[service1_id, service2_id],
source_groups=[source_group],
dest_groups=[dest_group],
sequence_number=seq_num + 1,
scope=None,
logged=False,
tag=tag,
tags=tags,
tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_entry_def)
self.assertIsNotNone(result)
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'
ip_protocol = nsx_constants.IPV4
map_seq_num = 10
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,
ip_protocol=ip_protocol)
self.assertEqual(rule_id, entry1.get_id())
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, ip_protocol=ip_protocol)
self.assertEqual(rule_id, entry2.get_id())
with mock.patch.object(self.policy_api,
"create_with_parent") as api_call:
result = self.resourceApi.create_with_entries(
name, domain_id,
map_id=map_id,
description=description,
entries=[entry1, entry2],
category=category,
map_sequence_number=map_seq_num,
tenant=TEST_TENANT)
expected_def = self.mapDef(
domain_id=domain_id,
map_id=map_id,
name=name,
description=description,
category=category,
map_sequence_number=map_seq_num,
tenant=TEST_TENANT)
self.assert_called_with_defs(api_call,
[expected_def, entry1, entry2])
self.assertEqual(map_id, result)
def test_create_with_entries_no_id(self):
domain_id = '111'
map_id = '222'
name = 'cm1'
description = 'desc'
source_group = 'g1'
dest_group = 'g2'
service_id = 'c1'
category = 'Emergency'
ip_protocol = nsx_constants.IPV4
map_seq_num = 10
rule_id = 1
entry1 = self.resourceApi.build_entry(
'DHCP Reply', domain_id, map_id,
sequence_number=rule_id, service_ids=[service_id],
action=constants.ACTION_DENY,
source_groups=None,
dest_groups=[dest_group],
direction=nsx_constants.IN,
ip_protocol=ip_protocol)
self.assertIsNotNone(entry1.get_id())
rule_id += 1
entry2 = self.resourceApi.build_entry(
'DHCP Request', domain_id, map_id,
sequence_number=rule_id, service_ids=None,
action=constants.ACTION_DENY,
source_groups=[source_group],
dest_groups=None,
direction=nsx_constants.OUT, ip_protocol=ip_protocol)
self.assertIsNotNone(entry2.get_id())
with mock.patch.object(self.policy_api,
"create_with_parent") as api_call:
result = self.resourceApi.create_with_entries(
name, domain_id,
map_id=map_id,
description=description,
entries=[entry1, entry2],
category=category,
map_sequence_number=map_seq_num,
tenant=TEST_TENANT)
expected_def = self.mapDef(
domain_id=domain_id,
map_id=map_id,
name=name,
description=description,
category=category,
map_sequence_number=map_seq_num,
tenant=TEST_TENANT)
self.assert_called_with_defs(api_call,
[expected_def, entry1, entry2])
self.assertEqual(map_id, result)
def test_delete(self):
domain_id = '111'
map_id = '222'
with mock.patch.object(self.policy_api, "delete") as api_call:
self.resourceApi.delete(domain_id, map_id, tenant=TEST_TENANT)
expected_def = self.mapDef(
domain_id=domain_id,
map_id=map_id,
tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
def test_delete_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 = self.entryDef(
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'
map_id = '222'
with mock.patch.object(self.policy_api, "get",
return_value={'id': map_id}) as api_call:
result = self.resourceApi.get(domain_id, map_id,
tenant=TEST_TENANT)
expected_def = self.mapDef(
domain_id=domain_id,
map_id=map_id,
tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
self.assertEqual(map_id, result['id'])
def test_get_entry(self):
domain_id = '111'
map_id = '222'
entry_id = '333'
with mock.patch.object(self.policy_api, "get",
return_value={'id': entry_id}) as api_call:
result = self.resourceApi.get_entry(domain_id, map_id,
entry_id, tenant=TEST_TENANT)
expected_def = self.entryDef(
domain_id=domain_id,
map_id=map_id,
entry_id=entry_id,
tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
self.assertEqual(entry_id, result['id'])
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 = self.mapDef(
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",
return_value={'results': []}) as api_call:
result = self.resourceApi.list(domain_id, tenant=TEST_TENANT)
expected_def = self.mapDef(
domain_id=domain_id,
tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
self.assertEqual([], result)
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'
category = constants.CATEGORY_EMERGENCY
with mock.patch.object(self.policy_api, "get",
return_value={}),\
mock.patch.object(self.resourceApi, "get",
return_value={'category': category}),\
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 = self.mapDef(
domain_id=domain_id,
map_id=map_id,
name=name,
description=description,
category=category,
tenant=TEST_TENANT)
entry_def = self.entryDef(
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_update_entry(self):
domain_id = '111'
map_id = '222'
entry_id = 'entry'
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_or_update") as update_call:
self.resourceApi.update_entry(
domain_id, map_id, entry_id,
name=name,
description=description,
service_ids=[service1_id, service2_id],
source_groups=[source_group],
dest_groups=[dest_group],
tenant=TEST_TENANT)
entry_def = self.entryDef(
domain_id=domain_id,
map_id=map_id,
entry_id=entry_id,
name=name,
description=description,
service_ids=[service1_id, service2_id],
source_groups=[source_group],
dest_groups=[dest_group],
tenant=TEST_TENANT)
self.assert_called_with_def(update_call, entry_def)
def test_update_entries(self):
domain_id = '111'
map_id = '222'
entries = "fake_entries"
with mock.patch.object(self.resourceApi,
"update_with_entries") as update_call:
self.resourceApi.update_entries(
domain_id, map_id, entries, tenant=TEST_TENANT)
update_call.assert_called_once_with(
domain_id, map_id, entries,
category=constants.CATEGORY_APPLICATION,
use_child_rules=True,
tenant=TEST_TENANT)
def test_update_with_entries(self):
domain_id = '111'
map_id = '222'
entry1_id = 'entry1'
entry2_id = 'entry2'
entry3_id = 'entry3'
entry1 = self.entryDef(
domain_id=domain_id,
map_id=map_id,
entry_id=entry1_id,
scope=['new_scope1'],
tenant=TEST_TENANT)
entry2 = self.entryDef(
domain_id=domain_id,
map_id=map_id,
entry_id=entry2_id,
scope=['scope2'],
tenant=TEST_TENANT)
original_map = {
'id': map_id,
'resource_type': self.resource_type,
'category': constants.CATEGORY_APPLICATION,
'display_name': 'map_name',
'rules': [
{'id': entry1_id, 'resource_type': 'Rule',
'display_name': 'name1', 'scope': ['scope1']},
{'id': entry2_id, 'resource_type': 'Rule',
'display_name': 'name2', 'scope': ['scope2']},
{'id': entry3_id, 'resource_type': 'Rule',
'display_name': 'name3', 'scope': ['scope3']}]}
updated_map = {
'id': map_id,
'resource_type': self.resource_type,
'category': constants.CATEGORY_APPLICATION,
'display_name': 'new_map_name',
'rules': [
{'id': entry1_id, 'resource_type': 'Rule',
'display_name': 'name1', 'scope': ['new_scope1']},
{'id': entry2_id, 'resource_type': 'Rule',
'display_name': 'name2', 'scope': ['scope2']}]}
map_def = self.mapDef(
domain_id=domain_id,
map_id=map_id,
tenant=TEST_TENANT)
with mock.patch.object(self.policy_api, "get",
return_value=original_map),\
mock.patch.object(self.policy_api.client,
"update") as update_call:
self.resourceApi.update_with_entries(
domain_id, map_id, entries=[entry1, entry2],
name='new_map_name', tenant=TEST_TENANT)
update_call.assert_called_once_with(
map_def.get_resource_path(), updated_map)
def test_update_with_entries_for_IGNORE_entries(self):
domain_id = '111'
map_id = '222'
entry1_id = 'entry1'
entry2_id = 'entry2'
entry3_id = 'entry3'
original_map = {
'id': map_id,
'resource_type': self.resource_type,
'category': constants.CATEGORY_APPLICATION,
'display_name': 'map_name',
'rules': [
{'id': entry1_id, 'resource_type': 'Rule',
'display_name': 'name1', 'scope': ['scope1'],
'_created_time': 1},
{'id': entry2_id, 'resource_type': 'Rule',
'display_name': 'name2', 'scope': ['scope2']},
{'id': entry3_id, 'resource_type': 'Rule',
'display_name': 'name3', 'scope': ['scope3']}]}
updated_map = {
'id': map_id,
'resource_type': self.resource_type,
'category': constants.CATEGORY_APPLICATION,
'display_name': 'new_map_name',
'rules': [
{'id': entry1_id, 'resource_type': 'Rule',
'display_name': 'name1', 'scope': ['scope1'],
'_created_time': 1},
{'id': entry2_id, 'resource_type': 'Rule',
'display_name': 'name2', 'scope': ['scope2']},
{'id': entry3_id, 'resource_type': 'Rule',
'display_name': 'name3', 'scope': ['scope3']}]}
map_def = self.mapDef(
domain_id=domain_id,
map_id=map_id,
tenant=TEST_TENANT)
with mock.patch.object(self.policy_api, "get",
return_value=original_map),\
mock.patch.object(self.policy_api.client,
"update") as update_call:
self.resourceApi.update_with_entries(
domain_id, map_id, name='new_map_name', tenant=TEST_TENANT)
update_call.assert_called_once_with(
map_def.get_resource_path(), updated_map)
def test_unset(self):
name = 'hello'
domain_id = 'test'
map_id = '111'
dest_groups = ['/infra/stuff']
category = constants.CATEGORY_EMERGENCY
# 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}),\
mock.patch.object(self.resourceApi, "get",
return_value={'category': category}):
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,
'category': category,
'resource_type': self.resource_type,
'rules': [{
'display_name': name,
'id': 'entry',
'resource_type': 'Rule',
'services': ["ANY"],
'source_groups': ["ANY"],
'destination_groups': dest_groups}]
}
url = '%s/domains/%s/%s/%s' % (TEST_TENANT,
domain_id,
self.path_name,
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 = self.mapDef(
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/%s/%s" % (
TEST_TENANT, domain_id, self.path_name, map_id)
api_get.assert_called_once_with(path, silent=False)
class TestPolicyGatewayPolicy(TestPolicyCommunicationMap):
def setUp(self, *args, **kwargs):
super(TestPolicyGatewayPolicy, self).setUp()
self.resourceApi = self.policy_lib.gateway_policy
self.mapDef = core_defs.GatewayPolicyDef
self.entryDef = core_defs.GatewayPolicyRuleDef
self.resource_type = 'GatewayPolicy'
self.path_name = 'gateway-policies'
def test_build_entry(self):
domain_id = '111'
map_id = '222'
name = 'rule1'
desc = 'desc'
dest_group = 'g1'
service_id = 's1'
policy_id = 'policy1'
ip_protocol = nsx_constants.IPV4
rule_id = 1
entry1 = self.resourceApi.build_entry(
name, domain_id, map_id, entry_id=rule_id, description=desc,
sequence_number=rule_id, service_ids=[service_id],
action=constants.ACTION_DENY,
scope=policy_id,
source_groups=None, dest_groups=[dest_group],
direction=nsx_constants.IN,
ip_protocol=ip_protocol)
expected_dict1 = {
'display_name': 'rule1',
'id': 1,
'description': 'desc',
'resource_type': 'Rule',
'scope': 'policy1',
'ip_protocol': 'IPV4',
'sequence_number': 1,
'action': 'DROP',
'source_groups': ['ANY'],
'destination_groups': ['/infra/domains/111/groups/g1'],
'direction': 'IN',
'logged': False,
'services': ['/infra/services/s1'],
'tag': None}
self.assertEqual(entry1.get_obj_dict(), expected_dict1)
entry2 = self.resourceApi.build_entry(
name, domain_id, map_id, entry_id=rule_id, description=desc,
sequence_number=rule_id, service_ids=[service_id],
action=constants.ACTION_DENY,
scope=policy_id,
dest_groups=[dest_group],
direction=nsx_constants.IN,
ip_protocol=ip_protocol,
plain_groups=True)
expected_dict2 = {
'display_name': 'rule1',
'id': 1,
'description': 'desc',
'resource_type': 'Rule',
'scope': 'policy1',
'ip_protocol': 'IPV4',
'sequence_number': 1,
'action': 'DROP',
'source_groups': ['ANY'],
'destination_groups': ['g1'],
'direction': 'IN',
'logged': False,
'services': ['/infra/services/s1'],
'tag': None}
self.assertEqual(entry2.get_obj_dict(), expected_dict2)
with mock.patch.object(self.resourceApi, 'version', '0.0.0'):
self.assertRaises(nsxlib_exc.NsxLibInvalidInput,
self.resourceApi.build_entry,
name, domain_id, map_id, entry_id=rule_id,
description=desc, sequence_number=rule_id,
service_ids=[service_id], scope=policy_id,
dest_groups=[dest_group], plain_groups=True)
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/%s/%s" % (
TEST_TENANT, domain_id, self.path_name, map_id)
api_get.assert_called_once_with(path, silent=False)
def test_wait_until_realized_failed(self):
domain_id = 'd1'
map_id = '111'
gw_section_id = 'realized_111'
info = {'state': constants.STATE_UNREALIZED,
'realization_specific_identifier': gw_section_id}
with mock.patch.object(self.resourceApi, "_get_realization_info",