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 <Raphael.Lima@windriver.com>
This commit is contained in:
rlima
Raphael Lima
parent
b754865fad
commit
42ef0762bc
|
|
@ -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"""
|
||||
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
|
|||
Loading…
Reference in New Issue