diff options
Diffstat (limited to 'aria/multivim-plugin/neutron_plugin/tests/test.py')
| -rw-r--r-- | aria/multivim-plugin/neutron_plugin/tests/test.py | 220 |
1 files changed, 0 insertions, 220 deletions
diff --git a/aria/multivim-plugin/neutron_plugin/tests/test.py b/aria/multivim-plugin/neutron_plugin/tests/test.py deleted file mode 100644 index 459c23a6cd..0000000000 --- a/aria/multivim-plugin/neutron_plugin/tests/test.py +++ /dev/null @@ -1,220 +0,0 @@ -######### -# 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() |