From 42ef0762bcb7f43bde852b48defc636a2f6e599b Mon Sep 17 00:00:00 2001 From: rlima Date: Thu, 7 Dec 2023 16:32:20 -0300 Subject: [PATCH] Improve unit test coverage for dcmanager's orchestrator/states/prestage Improves unit test coverage for dcmanager's orchestrator/states/prestage functionality from 78% to 99%. Test plan: All of the tests were created taking into account the output of 'tox -c tox.ini -e cover' command Story: 2007082 Task: 49310 Change-Id: If041d9aede8f93b99b8cc1001f99088c1a1c77be Signed-off-by: rlima --- distributedcloud/dcmanager/tests/base.py | 10 +- .../states/prestage/test_states.py | 294 +++++++++--------- .../tests/unit/orchestrator/test_base.py | 5 +- 3 files changed, 162 insertions(+), 147 deletions(-) diff --git a/distributedcloud/dcmanager/tests/base.py b/distributedcloud/dcmanager/tests/base.py index d32d69006..ecf81548d 100644 --- a/distributedcloud/dcmanager/tests/base.py +++ b/distributedcloud/dcmanager/tests/base.py @@ -18,9 +18,10 @@ import base64 import builtins import json -import mock +import os.path as os_path import pecan +import mock from oslo_config import cfg from oslo_db import options from oslotest import base @@ -241,6 +242,13 @@ class DCManagerTestCase(base.BaseTestCase): self.mock_get_vault_load_files = mock_patch_object.start() self.addCleanup(mock_patch_object.stop) + def _mock_os_path_isdir(self): + """Mock os' path.isdir""" + + mock_patch_object = mock.patch.object(os_path, 'isdir') + self.mock_os_path_isdir = mock_patch_object.start() + self.addCleanup(mock_patch_object.stop) + def _mock_builtins_open(self): """Mock builtins' open""" diff --git a/distributedcloud/dcmanager/tests/unit/orchestrator/states/prestage/test_states.py b/distributedcloud/dcmanager/tests/unit/orchestrator/states/prestage/test_states.py index 1d75d04c2..d070158c4 100644 --- a/distributedcloud/dcmanager/tests/unit/orchestrator/states/prestage/test_states.py +++ b/distributedcloud/dcmanager/tests/unit/orchestrator/states/prestage/test_states.py @@ -5,12 +5,13 @@ # import base64 +import copy import threading import mock +from dccommon.utils import AnsiblePlaybook from dcmanager.common import consts -from dcmanager.common.consts import DEPLOY_STATE_DONE from dcmanager.common.consts import STRATEGY_STATE_COMPLETE from dcmanager.common.consts import STRATEGY_STATE_FAILED from dcmanager.common.consts import STRATEGY_STATE_PRESTAGE_IMAGES @@ -21,199 +22,206 @@ from dcmanager.db.sqlalchemy import api as db_api from dcmanager.tests.unit.common import fake_strategy from dcmanager.tests.unit.orchestrator.test_base import TestSwUpdate -OAM_FLOATING_IP = "10.10.10.12" -FAKE_PASSWORD = (base64.b64encode("testpass".encode("utf-8"))).decode("ascii") +FAKE_PASSWORD = (base64.b64encode('testpass'.encode('utf-8'))).decode('ascii') +OAM_FLOATING_IP = '10.10.10.12' +REQUIRED_EXTRA_ARGS = { + 'sysadmin_password': FAKE_PASSWORD, + 'force': False +} class TestPrestage(TestSwUpdate): # Setting DEFAULT_STRATEGY_TYPE to prestage will setup the prestage upgrade - # orchestration worker, and will mock away the other orch threads + # orchestration worker and will mock away the other orch threads DEFAULT_STRATEGY_TYPE = consts.SW_UPDATE_TYPE_PRESTAGE + strategy_step = None def setUp(self): - super(TestPrestage, self).setUp() + super().setUp() + + self.strategy_step = None + # Add the subcloud being processed by this unit test + # The subcloud is online, managed with deploy_state 'installed' + self.subcloud = self.setup_subcloud(deploy_status=consts.DEPLOY_STATE_DONE) + + self.required_extra_args_with_oam = copy.copy(REQUIRED_EXTRA_ARGS) + self.required_extra_args_with_oam["oam_floating_ip_dict"] = { + self.subcloud.name: OAM_FLOATING_IP + } + + def _setup_strategy_step(self, strategy_step): + self.strategy_step = self.setup_strategy_step( + self.subcloud.id, strategy_step + ) + + def _setup_and_assert(self, next_state, extra_args=None): + self.strategy = fake_strategy.create_fake_strategy( + self.ctx, self.DEFAULT_STRATEGY_TYPE, extra_args=extra_args + ) + + # invoke the strategy state operation on the orch thread + self.worker.perform_state_action(self.strategy_step) + + # Verify the transition to the expected next state + self.assert_step_updated(self.strategy_step.subcloud_id, next_state) class TestPrestagePreCheckState(TestPrestage): def setUp(self): - super(TestPrestagePreCheckState, self).setUp() + super().setUp() - # Add the subcloud being processed by this unit test - # The subcloud is online, managed with deploy_state 'installed' - self.subcloud = self.setup_subcloud() + self._setup_strategy_step(STRATEGY_STATE_PRESTAGE_PRE_CHECK) - p = mock.patch("dcmanager.common.prestage.validate_prestage") - self.mock_prestage_subcloud = p.start() + self._mock_validate_prestage() + self._mock_threading_start() + + def _mock_validate_prestage(self): + """Mock dcmanager's common validate_prestage method + + The validate_prestage method is mocked because the focus is on testing + the orchestrator logic only. Any specifc prestage functionality is covered on + individual tests. + """ + + mock_class = mock.patch('dcmanager.common.prestage.validate_prestage') + self.mock_prestage_subcloud = mock_class.start() self.mock_prestage_subcloud.return_value = OAM_FLOATING_IP - self.addCleanup(p.stop) + self.addCleanup(mock_class.stop) - t = mock.patch.object(threading.Thread, "start") - self.mock_thread_start = t.start() - self.addCleanup(t.stop) + def _mock_threading_start(self): + """Mock threading's Thread.start""" - # Add the strategy_step state being processed by this unit test - self.strategy_step = self.setup_strategy_step( - self.subcloud.id, STRATEGY_STATE_PRESTAGE_PRE_CHECK - ) + mock_thread = mock.patch.object(threading.Thread, 'start') + self.mock_thread_start = mock_thread.start() + self.addCleanup(mock_thread.stop) - def test_prestage_prepare_no_extra_args(self): - next_state = STRATEGY_STATE_FAILED - # Update the subcloud to have deploy state as "complete" - db_api.subcloud_update( - self.ctx, self.subcloud.id, deploy_status=DEPLOY_STATE_DONE - ) + def test_prestage_pre_check_without_extra_args(self): + """Test prestage pre check without extra args""" - self.strategy = fake_strategy.create_fake_strategy( - self.ctx, self.DEFAULT_STRATEGY_TYPE - ) + self._setup_and_assert(STRATEGY_STATE_FAILED) - # invoke the strategy state operation on the orch thread - self.worker.perform_state_action(self.strategy_step) + def test_prestage_pre_check_validate_failed_with_orch_skip_false(self): + """Test prestage pre check validate failed with orch skip as false""" - # Verify the transition to the expected next state - self.assert_step_updated(self.strategy_step.subcloud_id, next_state) - - def test_prestage_prepare_validate_failed(self): - next_state = STRATEGY_STATE_FAILED - # Update the subcloud to have deploy state as "complete" - db_api.subcloud_update( - self.ctx, self.subcloud.id, deploy_status=DEPLOY_STATE_DONE - ) - - self.mock_prestage_subcloud.side_effect = ( + self.mock_prestage_subcloud.side_effect = \ exceptions.PrestagePreCheckFailedException( - subcloud=None, orch_skip=False, details="test" + subcloud=None, orch_skip=False, details='test' ) + + self._setup_and_assert(STRATEGY_STATE_FAILED, extra_args=REQUIRED_EXTRA_ARGS) + + new_strategy_step = db_api.strategy_step_get( + self.ctx, self.subcloud.id ) - extra_args = { - "sysadmin_password": FAKE_PASSWORD, - "force": False, - "oam_floating_ip": OAM_FLOATING_IP, - } - self.strategy = fake_strategy.create_fake_strategy( - self.ctx, self.DEFAULT_STRATEGY_TYPE, extra_args=extra_args - ) - # invoke the strategy state operation on the orch thread - self.worker.perform_state_action(self.strategy_step) + # The strategy step details field should be updated with the Exception string + self.assertTrue('test' in str(new_strategy_step.details)) - new_strategy_step = db_api.strategy_step_get(self.ctx, self.subcloud.id) - # Verify the transition to the expected next state - self.assert_step_updated(self.strategy_step.subcloud_id, next_state) + def test_prestage_pre_check_validate_failed_with_orch_skip_true(self): + """Test prestage pre check validate failed with orch skip as true""" + + self.mock_prestage_subcloud.side_effect = \ + exceptions.PrestagePreCheckFailedException( + subcloud=None, orch_skip=True, details='test' + ) + + self._setup_and_assert( + STRATEGY_STATE_COMPLETE, extra_args=REQUIRED_EXTRA_ARGS + ) + + new_strategy_step = db_api.strategy_step_get( + self.ctx, self.subcloud.id + ) # The strategy step details field should be updated with the Exception string self.assertTrue("test" in str(new_strategy_step.details)) - def test_prestage_prepare_validate_failed_skipped(self): - next_state = STRATEGY_STATE_COMPLETE - # Update the subcloud to have deploy state as "complete" - db_api.subcloud_update( - self.ctx, self.subcloud.id, deploy_status=DEPLOY_STATE_DONE + def test_prestage_pre_check_fails_with_generic_exception(self): + """Test prestage pre check fails with generic exception""" + + self.mock_prestage_subcloud.side_effect = Exception() + + self._setup_and_assert(STRATEGY_STATE_FAILED, extra_args=REQUIRED_EXTRA_ARGS) + + def test_prestage_pre_check_succeeds(self): + """Test prestage pre check succeeds""" + + self._setup_and_assert( + STRATEGY_STATE_PRESTAGE_PACKAGES, extra_args=REQUIRED_EXTRA_ARGS ) - self.mock_prestage_subcloud.side_effect = ( - exceptions.PrestagePreCheckFailedException( - subcloud=None, orch_skip=True, details="test" - ) + def test_prestage_pre_check_succeeds_with_oam_floating_ip_dict(self): + """Test prestage pre check succeeds with oam floating ip dict""" + + self._setup_and_assert( + STRATEGY_STATE_PRESTAGE_PACKAGES, + extra_args=self.required_extra_args_with_oam ) - extra_args = { - "sysadmin_password": FAKE_PASSWORD, - "force": False, - "oam_floating_ip": OAM_FLOATING_IP, - } - self.strategy = fake_strategy.create_fake_strategy( - self.ctx, self.DEFAULT_STRATEGY_TYPE, extra_args=extra_args + def test_prestage_pre_check_succeds_with_prestage_software_version(self): + """Test prestage pre check succeeds with prestage software version""" + + extra_args = copy.copy(REQUIRED_EXTRA_ARGS) + extra_args['prestage-software-version'] = '22.3' + + self._setup_and_assert( + STRATEGY_STATE_PRESTAGE_PACKAGES, extra_args=extra_args ) - # invoke the strategy state operation on the orch thread - self.worker.perform_state_action(self.strategy_step) - - new_strategy_step = db_api.strategy_step_get(self.ctx, self.subcloud.id) - - # Verify the transition to the expected next state - self.assert_step_updated(self.strategy_step.subcloud_id, next_state) - - # The strategy step details field should be updated with the Exception string - self.assertTrue("test" in str(new_strategy_step.details)) -class TestPrestagePackageState(TestPrestage): +class TestPrestagePackagesState(TestPrestage): def setUp(self): - super(TestPrestagePackageState, self).setUp() + super().setUp() - # Add the subcloud being processed by this unit test - # The subcloud is online, managed with deploy_state 'installed' - self.subcloud = self.setup_subcloud() + self._setup_strategy_step(STRATEGY_STATE_PRESTAGE_PACKAGES) - p = mock.patch("dcmanager.common.prestage.prestage_packages") - self.mock_prestage_packages = p.start() - self.addCleanup(p.stop) + self._mock_builtins_open() + self._mock_ansible_playbook() - # Add the strategy_step state being processed by this unit test - self.strategy_step = self.setup_strategy_step( - self.subcloud.id, STRATEGY_STATE_PRESTAGE_PACKAGES + def _mock_ansible_playbook(self): + mock_patch_object = mock.patch.object(AnsiblePlaybook, 'run_playbook') + self.mock_ansible_playbook = mock_patch_object.start() + self.addCleanup(mock_patch_object.stop) + + def test_prestage_package_succeeds(self): + """Test prestage package succeeds""" + + self._setup_and_assert( + STRATEGY_STATE_PRESTAGE_IMAGES, + extra_args=self.required_extra_args_with_oam ) - def test_prestage_prestage_package(self): - next_state = STRATEGY_STATE_PRESTAGE_IMAGES - # Update the subcloud to have deploy state as "complete" - db_api.subcloud_update( - self.ctx, self.subcloud.id, deploy_status=DEPLOY_STATE_DONE + def test_prestage_package_succeeds_with_prestage_software_version(self): + """Test prestage package succeeds with prestage software version""" + + extra_args = copy.copy(self.required_extra_args_with_oam) + extra_args['prestage-software-version'] = '22.3' + + self._setup_and_assert( + STRATEGY_STATE_PRESTAGE_IMAGES, extra_args=extra_args ) - oam_floating_ip_dict = {self.subcloud.name: OAM_FLOATING_IP} - extra_args = { - "sysadmin_password": FAKE_PASSWORD, - "force": False, - "oam_floating_ip_dict": oam_floating_ip_dict, - } - self.strategy = fake_strategy.create_fake_strategy( - self.ctx, self.DEFAULT_STRATEGY_TYPE, extra_args=extra_args - ) - - # invoke the strategy state operation on the orch thread - self.worker.perform_state_action(self.strategy_step) - - # Verify the transition to the expected next state - self.assert_step_updated(self.strategy_step.subcloud_id, next_state) - class TestPrestageImagesState(TestPrestage): def setUp(self): - super(TestPrestageImagesState, self).setUp() + super().setUp() - # Add the subcloud being processed by this unit test - # The subcloud is online, managed with deploy_state 'installed' - self.subcloud = self.setup_subcloud() + self._setup_strategy_step(STRATEGY_STATE_PRESTAGE_IMAGES) - p = mock.patch("dcmanager.common.prestage.prestage_images") - self.mock_prestage_packages = p.start() - self.addCleanup(p.stop) + self._mock_os_path_isdir() + self.mock_os_path_isdir.return_value = False - # Add the strategy_step state being processed by this unit test - self.strategy_step = self.setup_strategy_step( - self.subcloud.id, STRATEGY_STATE_PRESTAGE_IMAGES + def test_prestage_images_succeeds(self): + """Test prestage images succeeds""" + + self._setup_and_assert( + STRATEGY_STATE_COMPLETE, extra_args=self.required_extra_args_with_oam ) - def test_prestage_prestage_images(self): - next_state = STRATEGY_STATE_COMPLETE - # Update the subcloud to have deploy state as "complete" - db_api.subcloud_update( - self.ctx, self.subcloud.id, deploy_status=DEPLOY_STATE_DONE - ) + def test_prestage_images_succeeds_with_prestage_software_version(self): + """Test prestage images succeeds with prestage software version""" - oam_floating_ip_dict = {self.subcloud.name: OAM_FLOATING_IP} - extra_args = { - "sysadmin_password": FAKE_PASSWORD, - "force": False, - "oam_floating_ip_dict": oam_floating_ip_dict, - } - self.strategy = fake_strategy.create_fake_strategy( - self.ctx, self.DEFAULT_STRATEGY_TYPE, extra_args=extra_args - ) + extra_args = copy.copy(self.required_extra_args_with_oam) + extra_args['prestage-software-version'] = '22.3' - # invoke the strategy state operation on the orch thread - self.worker.perform_state_action(self.strategy_step) - - # Verify the transition to the expected next state - self.assert_step_updated(self.strategy_step.subcloud_id, next_state) + self._setup_and_assert(STRATEGY_STATE_COMPLETE, extra_args=extra_args) diff --git a/distributedcloud/dcmanager/tests/unit/orchestrator/test_base.py b/distributedcloud/dcmanager/tests/unit/orchestrator/test_base.py index 6f8000e14..67595efa6 100644 --- a/distributedcloud/dcmanager/tests/unit/orchestrator/test_base.py +++ b/distributedcloud/dcmanager/tests/unit/orchestrator/test_base.py @@ -202,10 +202,9 @@ class TestSwUpdate(base.DCManagerTestCase): return worker - def setup_subcloud(self): + def setup_subcloud(self, deploy_status=consts.DEPLOY_STATE_INSTALLED): subcloud_id = fake_subcloud.create_fake_subcloud( - self.ctx, - deploy_status=consts.DEPLOY_STATE_INSTALLED, + self.ctx, deploy_status=deploy_status, ).id return db_api.subcloud_update( self.ctx,