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.
6820 lines
276 KiB
6820 lines
276 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 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' |
|
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, |
|
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, |
|
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' |
|
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, |
|
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, |
|
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' |
|
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, |
|
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, |
|
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< |