aboutsummaryrefslogtreecommitdiffstats
path: root/aria/multivim-plugin/neutron_plugin/tests/test_port.py
diff options
context:
space:
mode:
authorSeshu Kumar M <seshu.kumar.m@huawei.com>2017-08-10 06:48:03 +0000
committerGerrit Code Review <gerrit@onap.org>2017-08-10 06:48:03 +0000
commitb2356570d9aa82f9e8ea259f9e282499c98efcbc (patch)
treeebccf8677dcc645ef1c27a09a646acd013fb033e /aria/multivim-plugin/neutron_plugin/tests/test_port.py
parentd48b750fc08b11a758b4c71ac71fe6cd2a2437d3 (diff)
parent9981f55920a6f1c1f20396d42e35b075b22f6a8f (diff)
Merge "ARIA multivim plugin initial checkin"
Diffstat (limited to 'aria/multivim-plugin/neutron_plugin/tests/test_port.py')
-rw-r--r--aria/multivim-plugin/neutron_plugin/tests/test_port.py156
1 files changed, 156 insertions, 0 deletions
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)