ARIA multivim plugin initial checkin

Change-Id: I3a24ab6fc5ba54466bfecaf596a13b8907248ae8
Issue-id: SO-77
Signed-off-by: DeWayne Filppi <dewayne@gigaspaces.com>

1 files changed, 220 insertions, 0 deletions

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()