diff options
Diffstat (limited to 'aria/multivim-plugin/neutron_plugin/tests')
4 files changed, 492 insertions, 0 deletions
diff --git a/aria/multivim-plugin/neutron_plugin/tests/__init__.py b/aria/multivim-plugin/neutron_plugin/tests/__init__.py new file mode 100644 index 0000000000..04cb21f745 --- /dev/null +++ b/aria/multivim-plugin/neutron_plugin/tests/__init__.py @@ -0,0 +1 @@ +__author__ = 'idanmo' diff --git a/aria/multivim-plugin/neutron_plugin/tests/test.py b/aria/multivim-plugin/neutron_plugin/tests/test.py new file mode 100644 index 0000000000..459c23a6cd --- /dev/null +++ b/aria/multivim-plugin/neutron_plugin/tests/test.py @@ -0,0 +1,220 @@ +######### +# Copyright (c) 2014 GigaSpaces Technologies Ltd. All rights reserved +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# * See the License for the specific language governing permissions and +# * limitations under the License. + +import mock +import random +import string +import unittest + +from cloudify.exceptions import NonRecoverableError +from cloudify.context import BootstrapContext + +from cloudify.mocks import MockCloudifyContext + +import openstack_plugin_common as common +import openstack_plugin_common.tests.test as common_test + +import neutron_plugin +import neutron_plugin.network +import neutron_plugin.port +import neutron_plugin.router +import neutron_plugin.security_group + + +class ResourcesRenamingTest(unittest.TestCase): + def setUp(self): + neutron_plugin.port._find_network_in_related_nodes = mock.Mock() + # *** Configs from files ******************** + common.Config.get = mock.Mock() + common.Config.get.return_value = {} + # *** Neutron ******************** + self.neutron_mock = mock.Mock() + + def neutron_mock_connect(unused_self, unused_cfg): + return self.neutron_mock + common.NeutronClient.connect = neutron_mock_connect + + self.neutron_mock.cosmo_list = mock.Mock() + self.neutron_mock.cosmo_list.return_value = [] + + def _setup_ctx(self, obj_type): + ctx = common_test.create_mock_ctx_with_provider_info( + node_id='__cloudify_id_something_001', + properties={ + obj_type: { + 'name': obj_type + '_name', + }, + 'rules': [] # For security_group + } + ) + return ctx + + def _test(self, obj_type): + ctx = self._setup_ctx(obj_type) + attr = getattr(self.neutron_mock, 'create_' + obj_type) + attr.return_value = { + obj_type: { + 'id': obj_type + '_id', + } + } + getattr(neutron_plugin, obj_type).create(ctx) + calls = attr.mock_calls + self.assertEquals(len(calls), 1) # Exactly one object created + # Indexes into call[]: + # 0 - the only call + # 1 - regular arguments + # 0 - first argument + arg = calls[0][1][0] + self.assertEquals(arg[obj_type]['name'], 'p2_' + obj_type + '_name') + + def test_network(self): + self._test('network') + + def test_port(self): + self._test('port') + + def test_router(self): + self._test('router') + + def test_security_group(self): + self._test('security_group') + + # Network chosen arbitrary for this test. + # Just testing something without prefix. + def test_network_no_prefix(self): + ctx = self._setup_ctx('network') + for pctx in common_test.BOOTSTRAP_CONTEXTS_WITHOUT_PREFIX: + ctx._bootstrap_context = BootstrapContext(pctx) + self.neutron_mock.create_network.reset_mock() + self.neutron_mock.create_network.return_value = { + 'network': { + 'id': 'network_id', + } + } + neutron_plugin.network.create(ctx) + calls = self.neutron_mock.create_network.mock_calls + self.assertEquals(len(calls), 1) # Exactly one network created + # Indexes into call[]: + # 0 - the only call + # 1 - regular arguments + # 0 - first argument + arg = calls[0][1][0] + self.assertEquals(arg['network']['name'], 'network_name', + "Failed with context: " + str(pctx)) + + +def _rand_str(n): + chars = string.ascii_uppercase + string.digits + return ''.join(random.choice(chars) for _ in range(n)) + + +class SecurityGroupTest(unittest.TestCase): + def setUp(self): + # *** Configs from files ******************** + common.Config.get = mock.Mock() + common.Config.get.return_value = {} + # *** Neutron ******************** + self.neutron_mock = mock.Mock() + + def neutron_mock_connect(unused_self, unused_cfg): + return self.neutron_mock + common.NeutronClient.connect = neutron_mock_connect + neutron_plugin.security_group._rules_for_sg_id = mock.Mock() + neutron_plugin.security_group._rules_for_sg_id.return_value = [] + + def _setup_ctx(self): + sg_name = _rand_str(6) + '_new' + ctx = MockCloudifyContext(properties={ + 'security_group': { + 'name': sg_name, + 'description': 'blah' + }, + 'rules': [{'port': 80}], + 'disable_default_egress_rules': True, + }) + return ctx + + def test_sg_new(self): + ctx = self._setup_ctx() + self.neutron_mock.cosmo_list = mock.Mock() + self.neutron_mock.cosmo_list.return_value = [] + self.neutron_mock.create_security_group = mock.Mock() + self.neutron_mock.create_security_group.return_value = { + 'security_group': { + 'description': 'blah', + 'id': ctx['security_group']['name'] + '_id', + } + } + neutron_plugin.security_group.create(ctx) + self.assertTrue(self.neutron_mock.create_security_group.mock_calls) + + def test_sg_use_existing(self): + ctx = self._setup_ctx() + self.neutron_mock.cosmo_list = mock.Mock() + self.neutron_mock.cosmo_list.return_value = [{ + 'id': ctx['security_group']['name'] + '_existing_id', + 'description': 'blah', + 'security_group_rules': [{ + 'remote_group_id': None, + 'direction': 'ingress', + 'protocol': 'tcp', + 'ethertype': 'IPv4', + 'port_range_max': 80, + 'port_range_min': 80, + 'remote_ip_prefix': '0.0.0.0/0', + }] + }] + self.neutron_mock.create_security_group = mock.Mock() + self.neutron_mock.create_security_group.return_value = { + 'security_group': { + 'description': 'blah', + 'id': ctx['security_group']['name'] + '_id', + } + } + neutron_plugin.security_group.create(ctx) + self.assertFalse(self.neutron_mock.create_security_group.mock_calls) + + def test_sg_use_existing_with_other_rules(self): + ctx = self._setup_ctx() + self.neutron_mock.cosmo_list = mock.Mock() + self.neutron_mock.cosmo_list.return_value = [{ + 'id': ctx['security_group']['name'] + '_existing_id', + 'description': 'blah', + 'security_group_rules': [{ + 'remote_group_id': None, + 'direction': 'ingress', + 'protocol': 'tcp', + 'ethertype': 'IPv4', + 'port_range_max': 81, # Note the different port! + 'port_range_min': 81, # Note the different port! + 'remote_ip_prefix': '0.0.0.0/0', + }] + }] + self.neutron_mock.create_security_group = mock.Mock() + self.neutron_mock.create_security_group.return_value = { + 'security_group': { + 'description': 'blah', + 'id': ctx['security_group']['name'] + '_id', + } + } + self.assertRaises( + NonRecoverableError, + neutron_plugin.security_group.create, + ctx + ) + + +if __name__ == '__main__': + unittest.main() diff --git a/aria/multivim-plugin/neutron_plugin/tests/test_port.py b/aria/multivim-plugin/neutron_plugin/tests/test_port.py new file mode 100644 index 0000000000..1acee3d05d --- /dev/null +++ b/aria/multivim-plugin/neutron_plugin/tests/test_port.py @@ -0,0 +1,156 @@ +######## +# Copyright (c) 2014 GigaSpaces Technologies Ltd. All rights reserved +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# * See the License for the specific language governing permissions and +# * limitations under the License. + +import unittest + +import mock + +import neutron_plugin.port +from cloudify.mocks import (MockCloudifyContext, + MockNodeInstanceContext, + MockRelationshipSubjectContext) +from openstack_plugin_common import (NeutronClientWithSugar, + OPENSTACK_ID_PROPERTY) +from cloudify.exceptions import OperationRetry + + +class TestPort(unittest.TestCase): + + def test_fixed_ips_no_fixed_ips(self): + node_props = {'fixed_ip': ''} + + with mock.patch( + 'neutron_plugin.port.' + 'get_openstack_id_of_single_connected_node_by_openstack_type', + self._get_connected_subnet_mock(return_empty=True)): + with mock.patch( + 'neutron_plugin.port.ctx', + self._get_mock_ctx_with_node_properties(node_props)): + + port = {} + neutron_plugin.port._handle_fixed_ips(port) + + self.assertNotIn('fixed_ips', port) + + def test_fixed_ips_subnet_only(self): + node_props = {'fixed_ip': ''} + + with mock.patch( + 'neutron_plugin.port.' + 'get_openstack_id_of_single_connected_node_by_openstack_type', + self._get_connected_subnet_mock(return_empty=False)): + with mock.patch( + 'neutron_plugin.port.ctx', + self._get_mock_ctx_with_node_properties(node_props)): + + port = {} + neutron_plugin.port._handle_fixed_ips(port) + + self.assertEquals([{'subnet_id': 'some-subnet-id'}], + port.get('fixed_ips')) + + def test_fixed_ips_ip_address_only(self): + node_props = {'fixed_ip': '1.2.3.4'} + + with mock.patch( + 'neutron_plugin.port.' + 'get_openstack_id_of_single_connected_node_by_openstack_type', + self._get_connected_subnet_mock(return_empty=True)): + with mock.patch( + 'neutron_plugin.port.ctx', + self._get_mock_ctx_with_node_properties(node_props)): + + port = {} + neutron_plugin.port._handle_fixed_ips(port) + + self.assertEquals([{'ip_address': '1.2.3.4'}], + port.get('fixed_ips')) + + def test_fixed_ips_subnet_and_ip_address(self): + node_props = {'fixed_ip': '1.2.3.4'} + + with mock.patch( + 'neutron_plugin.port.' + 'get_openstack_id_of_single_connected_node_by_openstack_type', + self._get_connected_subnet_mock(return_empty=False)): + with mock.patch( + 'neutron_plugin.port.ctx', + self._get_mock_ctx_with_node_properties(node_props)): + + port = {} + neutron_plugin.port._handle_fixed_ips(port) + + self.assertEquals([{'ip_address': '1.2.3.4', + 'subnet_id': 'some-subnet-id'}], + port.get('fixed_ips')) + + @staticmethod + def _get_connected_subnet_mock(return_empty=True): + return lambda *args, **kw: None if return_empty else 'some-subnet-id' + + @staticmethod + def _get_mock_ctx_with_node_properties(properties): + return MockCloudifyContext(node_id='test_node_id', + properties=properties) + + +class MockNeutronClient(NeutronClientWithSugar): + """A fake neutron client with hard-coded test data.""" + def __init__(self, update): + self.update = update + self.body = {'port': {'id': 'test-id', 'security_groups': []}} + + def show_port(self, *_): + return self.body + + def update_port(self, _, b, **__): + if self.update: + self.body.update(b) + return + + def cosmo_get(self, *_, **__): + return self.body['port'] + + +class TestPortSG(unittest.TestCase): + @mock.patch('openstack_plugin_common._put_client_in_kw') + def test_connect_sg_to_port(self, *_): + mock_neutron = MockNeutronClient(update=True) + ctx = MockCloudifyContext( + source=MockRelationshipSubjectContext(node=mock.MagicMock(), + instance=mock.MagicMock()), + target=MockRelationshipSubjectContext(node=mock.MagicMock(), + instance=mock.MagicMock())) + + with mock.patch('neutron_plugin.port.ctx', ctx): + neutron_plugin.port.connect_security_group(mock_neutron) + self.assertIsNone(ctx.operation._operation_retry) + + @mock.patch('openstack_plugin_common._put_client_in_kw') + def test_connect_sg_to_port_race_condition(self, *_): + mock_neutron = MockNeutronClient(update=False) + + ctx = MockCloudifyContext( + source=MockRelationshipSubjectContext(node=mock.MagicMock(), + instance=mock.MagicMock()), + target=MockRelationshipSubjectContext( + node=mock.MagicMock(), + instance=MockNodeInstanceContext( + runtime_properties={ + OPENSTACK_ID_PROPERTY: 'test-sg-id'}))) + with mock.patch('neutron_plugin.port.ctx', ctx): + neutron_plugin.port.connect_security_group(mock_neutron, ctx=ctx) + self.assertIsInstance(ctx.operation._operation_retry, + OperationRetry) diff --git a/aria/multivim-plugin/neutron_plugin/tests/test_security_group.py b/aria/multivim-plugin/neutron_plugin/tests/test_security_group.py new file mode 100644 index 0000000000..e958cddb33 --- /dev/null +++ b/aria/multivim-plugin/neutron_plugin/tests/test_security_group.py @@ -0,0 +1,115 @@ +# -*- coding: utf-8 -*- +######### +# Copyright (c) 2016 GigaSpaces Technologies Ltd. All rights reserved +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# * See the License for the specific language governing permissions and +# * limitations under the License. + +import unittest + +from mock import Mock, patch +from requests.exceptions import RequestException + +from neutron_plugin import security_group + +from cloudify.exceptions import NonRecoverableError +from cloudify.state import current_ctx + +from cloudify.mocks import MockCloudifyContext + + +class FakeException(Exception): + pass + + +@patch('openstack_plugin_common.OpenStackClient._validate_auth_params') +@patch('openstack_plugin_common.NeutronClientWithSugar') +class TestSecurityGroup(unittest.TestCase): + + def setUp(self): + super(TestSecurityGroup, self).setUp() + self.nova_client = Mock() + + self.ctx = MockCloudifyContext( + node_id='test', + deployment_id='test', + properties={ + 'description': 'The best Security Group. Great', + 'rules': [], + 'resource_id': 'mock_sg', + 'security_group': { + }, + 'server': {}, + 'openstack_config': { + 'auth_url': 'things/v3', + }, + }, + operation={'retry_number': 0}, + provider_context={'resources': {}} + ) + current_ctx.set(self.ctx) + self.addCleanup(current_ctx.clear) + + findctx = patch( + 'openstack_plugin_common._find_context_in_kw', + return_value=self.ctx, + ) + findctx.start() + self.addCleanup(findctx.stop) + + def test_set_sg_runtime_properties(self, mock_nc, *_): + security_group.create( + nova_client=self.nova_client, + ctx=self.ctx, + args={}, + ) + + self.assertEqual( + { + 'external_type': 'security_group', + 'external_id': mock_nc().get_id_from_resource(), + 'external_name': mock_nc().get_name_from_resource(), + }, + self.ctx.instance.runtime_properties + ) + + def test_create_sg_wait_timeout(self, mock_nc, *_): + mock_nc().show_security_group.side_effect = RequestException + + with self.assertRaises(NonRecoverableError): + security_group.create( + nova_client=self.nova_client, + ctx=self.ctx, + args={}, + status_attempts=3, + status_timeout=0.001, + ) + + @patch( + 'neutron_plugin.security_group.delete_resource_and_runtime_properties') + def test_dont_duplicate_if_failed_rule(self, mock_del_res, mock_nc, *_): + self.ctx.node.properties['rules'] = [ + { + 'port': '🍷', + }, + ] + mock_nc().create_security_group_rule.side_effect = FakeException + mock_del_res.side_effect = FakeException('the 2nd') + + with self.assertRaises(NonRecoverableError) as e: + security_group.create( + nova_client=self.nova_client, + ctx=self.ctx, + args={}, + ) + + self.assertIn('the 2nd', str(e.exception)) |