openstack
/
murano
Code Issues Proposed changes

Merge "Increase unit test coverage for db services environments and engine system"

This commit is contained in:
Jenkins 2016-10-14 12:39:11 +00:00 committed by Gerrit Code Review
parent 0846f5cee6 8c999e7444
commit 84058d79b3
4 changed files with 347 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

86
murano/tests/unit/db/services/test_environments.py
View File

@ -12,6 +12,8 @@
# License for the specific language governing permissions and limitations
# under the License.
import datetime as dt
import mock
import uuid
from oslo_utils import timeutils
@ -20,9 +22,22 @@ from murano.db.services import environments
from murano.db import session as db_session
from murano.services import states
from murano.tests.unit import base
from murano.tests.unit import utils
OLD_VERSION = 0
LATEST_VERSION = 1
class TestEnvironmentServices(base.MuranoWithDBTestCase):
def setUp(self):
super(TestEnvironmentServices, self).setUp()
self.environment = models.Environment(
name='test_environment', tenant_id='test_tenant_id',
version=LATEST_VERSION
)
self.env_services = environments.EnvironmentServices()
def test_environment_ready_if_last_session_deployed_after_failed(self):
"""Test environment ready status
@ -31,27 +46,21 @@ class TestEnvironmentServices(base.MuranoWithDBTestCase):
Bug: #1413260
"""
OLD_VERSION = 0
LATEST_VERSION = 1
session = db_session.get_session()
environment = models.Environment(
name='test_environment', tenant_id='test_tenant_id',
version=LATEST_VERSION
)
session.add(environment)
session.add(self.environment)
now = timeutils.utcnow()
session_1 = models.Session(
environment=environment, user_id='test_user_id_1',
environment=self.environment, user_id='test_user_id_1',
version=OLD_VERSION,
state=states.SessionState.DEPLOY_FAILURE,
updated=now, description={}
)
session_2 = models.Session(
environment=environment, user_id='test_user_id_2',
environment=self.environment, user_id='test_user_id_2',
version=LATEST_VERSION,
state=states.SessionState.DEPLOYED,
updated=now + dt.timedelta(minutes=1), description={}
@ -61,7 +70,64 @@ class TestEnvironmentServices(base.MuranoWithDBTestCase):
expected_status = states.EnvironmentStatus.READY
actual_status = environments.EnvironmentServices.get_status(
environment.id
self.environment.id
)
self.assertEqual(expected_status, actual_status)
@mock.patch("murano.db.services.environments.auth_utils")
def test_get_network_driver(self, mock_authentication):
self.tenant_id = str(uuid.uuid4())
self.context = utils.dummy_context(tenant_id=self.tenant_id)
driver_context = self.env_services.get_network_driver(self.context)
self.assertEqual(driver_context, "neutron")
def test_get_status(self):
session = db_session.get_session()
session.add(self.environment)
now = timeutils.utcnow()
session_1 = models.Session(
environment=self.environment, user_id='test_user_id_1',
version=OLD_VERSION,
state=states.SessionState.DEPLOY_FAILURE,
updated=now, description={}
)
session.add(session_1)
session.flush()
expected_status = states.EnvironmentStatus.DEPLOY_FAILURE
self.assertEqual(expected_status, self.env_services.get_status(
self.environment.id))
def test_delete_failure_get_description(self):
session = db_session.get_session()
session.add(self.environment)
now = timeutils.utcnow()
session_1 = models.Session(
environment=self.environment, user_id='test_user_id_1',
version=OLD_VERSION,
state=states.SessionState.DELETE_FAILURE,
updated=now, description={}
)
session.add(session_1)
session.flush()
expected_status = states.EnvironmentStatus.DELETE_FAILURE
self.assertEqual(expected_status, self.env_services.get_status(
self.environment.id))
env_id = self.environment.id
description = (self.env_services.
get_environment_description(env_id,
session_id=None,
inner=False))
self.assertEqual({}, description)

14
murano/tests/unit/engine/system/test_agent.py
View File

@ -378,3 +378,17 @@ class TestExecutionPlan(base.MuranoTestCase):
rtn_template = self.agent._build_v1_execution_plan(template,
self.resources)
self.assertEqual(template, rtn_template)
def test_get_array_item(self):
array = [1, 2, 3]
index = 2
self.assertEqual(array[2], self.agent._get_array_item(array, index))
index = 3
self.assertIsNone(self.agent._get_array_item(array, index))
def test_execution_plan_error(self):
template = None
self.assertRaises(ValueError,
self.agent.build_execution_plan,
template, self.resources)

98
murano/tests/unit/engine/system/test_net_explorer.py Normal file
View File

@ -0,0 +1,98 @@
# Copyright (c) 2016 AT&T
#
# 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
from oslo_config import cfg
from murano.dsl import murano_method
from murano.dsl import murano_type
from murano.engine.system import net_explorer
from murano.tests.unit import base
CONF = cfg.CONF
class TestNetExplorer(base.MuranoTestCase):
def setUp(self):
super(TestNetExplorer, self).setUp()
self.mock_class = mock.MagicMock(spec=murano_type.MuranoClass)
self.mock_method = mock.MagicMock(spec=murano_method.MuranoMethod)
self._this = mock.MagicMock()
self.region_name = "test-region"
self.addCleanup(mock.patch.stopall)
@mock.patch("murano.engine.system.net_explorer.nclient")
@mock.patch("murano.engine.system.net_explorer.auth_utils")
@mock.patch("murano.dsl.helpers.get_execution_session")
def test_get_available_cidr(self, execution_session,
mock_authentication, mock_nclient):
ne = net_explorer.NetworkExplorer(self._this, self.region_name)
router_id = 12
net_id = 144
self.assertIsNotNone(ne.get_available_cidr(router_id, net_id))
self.assertTrue(execution_session.called)
@mock.patch("murano.engine.system.net_explorer.nclient")
@mock.patch("murano.engine.system.net_explorer.auth_utils")
@mock.patch("murano.dsl.helpers.get_execution_session")
def test_list(self, execution_session, mock_authentication, mock_nclient):
ne = net_explorer.NetworkExplorer(self._this, self.region_name)
self.assertEqual(ne.list_networks(),
ne._client.list_networks()['networks'])
self.assertEqual(ne.list_subnetworks(),
ne._client.list_subnets()['subnets'])
self.assertEqual(ne.list_ports(), ne._client.list_ports()['ports'])
self.assertEqual(ne.list_neutron_extensions(),
ne._client.list_extensions()['extensions'])
self.assertEqual(ne.get_default_dns(), ne._settings.default_dns)
@mock.patch("murano.engine.system.net_explorer.nclient")
@mock.patch("murano.engine.system.net_explorer.auth_utils")
@mock.patch("murano.dsl.helpers.get_execution_session")
def test_get_router_error(self, execution_session,
mock_authentication, mock_nclient):
ne = net_explorer.NetworkExplorer(self._this, self.region_name)
self.assertRaises(KeyError, ne.get_default_router)
@mock.patch("murano.engine.system.net_explorer.nclient")
@mock.patch("murano.engine.system.net_explorer.auth_utils")
@mock.patch("murano.dsl.helpers.get_execution_session")
def test_get_ext_network_id_router(self, execution_session,
mock_authentication, mock_nclient):
ne = net_explorer.NetworkExplorer(self._this, self.region_name)
router_id = 12
self.assertIsNone(ne.get_external_network_id_for_router(router_id))
@mock.patch("murano.engine.system.net_explorer.nclient")
@mock.patch("murano.engine.system.net_explorer.auth_utils")
@mock.patch("murano.dsl.helpers.get_execution_session")
def test_get_ext_network_id_network(self, execution_session,
mock_authentication, mock_nclient):
ne = net_explorer.NetworkExplorer(self._this, self.region_name)
net_id = 144
self.assertEqual(net_id,
ne.get_external_network_id_for_network(net_id))
@mock.patch("murano.engine.system.net_explorer.nclient")
@mock.patch("murano.engine.system.net_explorer.auth_utils")
@mock.patch("murano.dsl.helpers.get_execution_session")
def test_get_cidr_none_router(self, execution_session,
mock_authentication, mock_nclient):
ne = net_explorer.NetworkExplorer(self._this, self.region_name)
router_id = None
self.assertEqual([], ne._get_cidrs_taken_by_router(router_id))

159
murano/tests/unit/engine/system/test_workflowclient.py Normal file
View File

@ -0,0 +1,159 @@
# Copyright (c) 2016 AT&T
# 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.
try:
from mistralclient.api import client as mistralcli
except ImportError as mistral_import_error:
mistralcli = None
import mock
from oslo_config import cfg
from murano.dsl import murano_method
from murano.dsl import murano_type
from murano.engine.system import workflowclient
from murano.tests.functional.common import utils as test_utils
from murano.tests.unit import base
CONF = cfg.CONF
rand_name = test_utils.DeployTestMixin.rand_name
class TestMistralClient(base.MuranoTestCase):
def setUp(self):
super(TestMistralClient, self).setUp()
self.mistral_client_mock = mock.Mock()
self.mistral_client_mock.client = mock.MagicMock(
spec=mistralcli.client)
self._patch_client()
self.mock_class = mock.MagicMock(spec=murano_type.MuranoClass)
self.mock_method = mock.MagicMock(spec=murano_method.MuranoMethod)
self._this = mock.MagicMock()
self._this.owner = None
self.addCleanup(mock.patch.stopall)
def _patch_client(self):
self.mock_client = mock.Mock(return_value=self.mistral_client_mock)
self.client_patcher = mock.patch.object(workflowclient.MistralClient,
'_client', self.mock_client)
self.client_patcher.start()
self.mock_create_client = mock.Mock(
return_value=self.mistral_client_mock)
self.create_client_patcher = mock.patch.object(
workflowclient.MistralClient, '_create_client',
self.mock_create_client)
self.create_client_patcher.start()
def _unpatch_client(self):
self.client_patcher.stop()
self.create_client_patcher.stop()
def test_run_with_execution_success_state(self):
test_output = '{"openstack": "foo", "__execution": "bar", "task":'\
' "baz"}'
mock_execution = mock.MagicMock(
id='123', state='SUCCESS', output=test_output)
self.mock_client.executions.create.return_value = mock_execution
self.mock_client.executions.get.return_value = mock_execution
run_name = rand_name('test')
timeout = 1
mc = workflowclient.MistralClient(self._this, 'regionOne')
output = mc.run(run_name, timeout)
for prop in ['openstack', '__execution', 'task']:
self.assertFalse(hasattr(output, prop))
self.assertEqual({}, output)
def test_run_with_execution_error_state(self):
mock_execution = mock.MagicMock(
id='123', state='ERROR', output="{'test_attr': 'test_val'}")
self.mock_client.executions.create.return_value = mock_execution
self.mock_client.executions.get.return_value = mock_execution
run_name = rand_name('test')
timeout = 1
mc = workflowclient.MistralClient(self._this, 'regionOne')
expected_error_msg = 'Mistral execution completed with ERROR.'\
' Execution id: {0}. Output: {1}'\
.format(mock_execution.id, mock_execution.output)
with self.assertRaisesRegexp(workflowclient.MistralError,
expected_error_msg):
mc.run(run_name, timeout)
def test_run_except_timeout_error(self):
mock_execution = mock.MagicMock(
id='123', state='TEST_STATE', output="{'test_attr': 'test_val'}")
self.mock_client.executions.create.return_value = mock_execution
self.mock_client.executions.get.return_value = mock_execution
run_name = rand_name('test')
timeout = 1
mc = workflowclient.MistralClient(self._this, 'regionOne')
expected_error_msg = 'Mistral run timed out. Execution id: {0}.'\
.format(mock_execution.id)
with self.assertRaisesRegexp(workflowclient.MistralError,
expected_error_msg):
mc.run(run_name, timeout)
def test_run_with_immediate_timeout(self):
mock_execution = mock.MagicMock(
id='123', state='ERROR', output="{'test_attr': 'test_val'}")
self.mock_client.executions.create.return_value = mock_execution
run_name = rand_name('test')
timeout = 0
mc = workflowclient.MistralClient(self._this, 'regionOne')
self.assertEqual(mock_execution.id, mc.run(run_name, timeout))
def test_upload(self):
mc = workflowclient.MistralClient(self._this, 'regionOne')
definition = rand_name('test')
self.assertIsNone(mc.upload(definition))
self.assertTrue(workflowclient.MistralClient.
_client.workflows.create.called)
@mock.patch('murano.engine.system.workflowclient.auth_utils')
def test_client_property(self, _):
self._unpatch_client()
test_mistral_settings = {
'url': rand_name('test_mistral_url'),
'project_id': rand_name('test_project_id'),
'endpoint_type': rand_name('test_endpoint_type'),
'auth_token': rand_name('test_auth_token'),
'user_id': rand_name('test_user_id'),
'insecure': rand_name('test_insecure'),
'cacert': rand_name('test_ca_cert')
}
with mock.patch('murano.engine.system.workflowclient.CONF')\
as mock_conf:
mock_conf.mistral = mock.MagicMock(**test_mistral_settings)
region_name = rand_name('test_region_name')
mc = workflowclient.MistralClient(self._this, region_name)
mistral_client = mc._client
self.assertIsNotNone(mistral_client)