# Copyright 2013: Mirantis Inc. # 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. """Test for deploy engines.""" import mock from rally import consts from rally import deploy from rally import exceptions from tests.unit import test def make_fake_deployment(**kwargs): values = dict({ "uuid": "1359befb-8737-4f4e-bea9-492416106977", "config": { "name": "fake", }, "status": consts.DeployStatus.DEPLOY_INIT, }, **kwargs) return FakeDeployment(values=values) class FakeDeployment(object): def __init__(self, values={}): self._values = values def __getitem__(self, name): return self._values[name] def update_status(self, status): self._values["status"] = status def set_started(self): pass def set_completed(self): pass def delete(self): pass class FakeEngine(deploy.EngineFactory): """Fake deployment engine. Used for tests. """ deployed = False cleanuped = False def __init__(self, deployment): super(FakeEngine, self).__init__(deployment) self.deployment = deployment def deploy(self): self.deployed = True return self def cleanup(self): self.cleanuped = True class EngineMixIn(object): def deploy(self): pass def cleanup(self): pass class EngineFake1(EngineMixIn, deploy.EngineFactory): """Fake deployment engine. Used for tests. """ pass class EngineFake2(EngineMixIn, deploy.EngineFactory): """Fake deployment engine. Used for tests. """ pass class EngineFake3(EngineFake2): """Fake deployment engine. Used for tests. """ pass class EngineFactoryTestCase(test.TestCase): FAKE_ENGINES = [EngineFake1, EngineFake2, EngineFake3] def test_get_engine_not_found(self): deployment = make_fake_deployment() self.assertRaises(exceptions.NoSuchEngine, deploy.EngineFactory.get_engine, "non_existing_engine", deployment) self.assertEqual(consts.DeployStatus.DEPLOY_FAILED, deployment["status"]) @mock.patch.object(FakeDeployment, "set_completed") @mock.patch.object(FakeDeployment, "set_started") def test_make_deploy(self, mock_set_started, mock_set_completed): deployment = make_fake_deployment() engine = FakeEngine(deployment) endpoint = engine.make_deploy() self.assertEqual(engine, endpoint) self.assertTrue(endpoint.deployed) self.assertFalse(endpoint.cleanuped) mock_set_started.assert_called_once_with() mock_set_completed.assert_called_once_with() @mock.patch.object(FakeDeployment, "set_started") @mock.patch.object(FakeEngine, "deploy") def test_make_deploy_failed(self, mock_deploy, mock_set_started): class DeployFailed(Exception): pass deployment = make_fake_deployment() engine = FakeEngine(deployment) mock_deploy.side_effect = DeployFailed() self.assertRaises(DeployFailed, engine.make_deploy) mock_set_started.assert_called_once_with() @mock.patch.object(FakeDeployment, "update_status") def test_make_cleanup(self, mock_update_status): deployment = make_fake_deployment() engine = FakeEngine(deployment) engine.make_cleanup() self.assertTrue(engine.cleanuped) self.assertFalse(engine.deployed) mock_update_status.assert_has_calls([ mock.call(consts.DeployStatus.CLEANUP_STARTED), mock.call(consts.DeployStatus.CLEANUP_FINISHED), ]) self.assertTrue(engine.cleanuped) @mock.patch.object(FakeDeployment, "update_status") @mock.patch.object(FakeEngine, "cleanup") def test_make_cleanup_failed(self, mock_cleanup, mock_update_status): class CleanUpFailed(Exception): pass deployment = make_fake_deployment() engine = FakeEngine(deployment) mock_cleanup.side_effect = CleanUpFailed() self.assertRaises(CleanUpFailed, engine.make_cleanup) mock_update_status.assert_has_calls([ mock.call(consts.DeployStatus.CLEANUP_STARTED), ]) self.assertFalse(engine.cleanuped) @mock.patch.object(FakeDeployment, "update_status") def test_with_statement(self, mock_update_status): deployment = make_fake_deployment() engine = FakeEngine(deployment) with engine as deployer: self.assertEqual(engine, deployer) self.assertFalse(mock_update_status.called) self.assertFalse(engine.cleanuped) self.assertFalse(engine.deployed) def test_with_statement_failed_on_init(self): self._assert_changed_status_on_error( consts.DeployStatus.DEPLOY_INIT, consts.DeployStatus.DEPLOY_FAILED) def test_with_statement_failed_on_started(self): self._assert_changed_status_on_error( consts.DeployStatus.DEPLOY_STARTED, consts.DeployStatus.DEPLOY_FAILED) def test_with_statement_failed_on_finished(self): self._assert_changed_status_on_error( consts.DeployStatus.DEPLOY_FINISHED, consts.DeployStatus.DEPLOY_INCONSISTENT) def test_with_statement_failed_on_cleanup(self): self._assert_changed_status_on_error( consts.DeployStatus.CLEANUP_STARTED, consts.DeployStatus.CLEANUP_FAILED) @mock.patch.object(FakeDeployment, "update_status") def _assert_changed_status_on_error(self, initial, final, mock_update_status): # NOTE(akscram): The assertRaises of testtools can't be used as # a context manager in python26: # with self.assertRaises(SomeError): # with engine as deployer: # raise SomeError() # instead of: # self.assertRaises(SomeError, # context_with_error, # SomeError(), engine) def context_with_error(error, manager): with manager: raise error class SomeError(Exception): pass deployment = make_fake_deployment(status=initial) engine = FakeEngine(deployment) self.assertRaises(SomeError, context_with_error, SomeError(), engine) mock_update_status.assert_called_once_with(final) self.assertFalse(engine.cleanuped) self.assertFalse(engine.deployed) def test_get_engine(self): deployment = make_fake_deployment() engines = EngineFactoryTestCase.FAKE_ENGINES for e in engines: engine_inst = deploy.EngineFactory.get_engine(e.__name__, deployment) self.assertIsInstance(engine_inst, e) def test_get_by_name(self): engines = EngineFactoryTestCase.FAKE_ENGINES for e in engines: self.assertEqual(e, deploy.EngineFactory.get_by_name(e.__name__)) def test_get_by_name_not_found(self): self.assertRaises(exceptions.NoSuchEngine, deploy.EngineFactory.get_by_name, "NonExistingEngine") def test_get_available_engines(self): engines = set([e.__name__ for e in EngineFactoryTestCase.FAKE_ENGINES]) real_engines = set(deploy.EngineFactory.get_available_engines()) self.assertEqual(engines & real_engines, engines) def test_engine_factory_is_abstract(self): self.assertRaises(TypeError, deploy.EngineFactory)