Merge "Improve unit test coverage for dcmanager's orchestrator/states/kube_rootca"
This commit is contained in:
commit
15fb726a6b
@ -1,8 +1,9 @@
|
|||||||
#
|
#
|
||||||
# Copyright (c) 2021 Wind River Systems, Inc.
|
# Copyright (c) 2021, 2024 Wind River Systems, Inc.
|
||||||
#
|
#
|
||||||
# SPDX-License-Identifier: Apache-2.0
|
# SPDX-License-Identifier: Apache-2.0
|
||||||
#
|
#
|
||||||
|
|
||||||
from dcmanager.common import consts
|
from dcmanager.common import consts
|
||||||
from dcmanager.tests.unit.common import fake_strategy
|
from dcmanager.tests.unit.common import fake_strategy
|
||||||
from dcmanager.tests.unit.orchestrator.states.kube_rootca.test_base \
|
from dcmanager.tests.unit.orchestrator.states.kube_rootca.test_base \
|
||||||
@ -12,25 +13,37 @@ from dcmanager.tests.unit.orchestrator.states.test_creating_vim_strategy \
|
|||||||
|
|
||||||
|
|
||||||
class TestCreatingVIMKubeRootCAUpgradeStrategyStage(
|
class TestCreatingVIMKubeRootCAUpgradeStrategyStage(
|
||||||
CreatingVIMStrategyStageMixin,
|
CreatingVIMStrategyStageMixin, TestKubeRootCaUpgradeState
|
||||||
TestKubeRootCaUpgradeState):
|
):
|
||||||
"""Test create vim kube rootca upgrade strategy during kube rootca orch"""
|
"""Test create vim kube rootca upgrade strategy during kube rootca orch"""
|
||||||
|
|
||||||
def setUp(self):
|
def setUp(self):
|
||||||
super(TestCreatingVIMKubeRootCAUpgradeStrategyStage, self).setUp()
|
super().setUp()
|
||||||
|
|
||||||
self.set_state(
|
self.set_state(
|
||||||
consts.STRATEGY_STATE_CREATING_VIM_KUBE_ROOTCA_UPDATE_STRATEGY,
|
consts.STRATEGY_STATE_CREATING_VIM_KUBE_ROOTCA_UPDATE_STRATEGY,
|
||||||
consts.STRATEGY_STATE_APPLYING_VIM_KUBE_ROOTCA_UPDATE_STRATEGY)
|
consts.STRATEGY_STATE_APPLYING_VIM_KUBE_ROOTCA_UPDATE_STRATEGY
|
||||||
|
)
|
||||||
|
|
||||||
|
def test_create_strategy_succeeds_with_extra_args(self):
|
||||||
|
"""Test create strategy succeeds with extra_args"""
|
||||||
|
|
||||||
def test_create_with_extra_args(self):
|
|
||||||
# Create a strategy with extra_args
|
# Create a strategy with extra_args
|
||||||
extra_args = {
|
extra_args = {
|
||||||
"expiry-date": "2020:01:31",
|
"expiry-date": "2020:01:31",
|
||||||
"subject": "C=CA ST=ON L=OTT O=WR OU=STX CN=AL_RULES"
|
"subject": "C=CA ST=ON L=OTT O=WR OU=STX CN=AL_RULES"
|
||||||
}
|
}
|
||||||
self.strategy = fake_strategy.create_fake_strategy(
|
self.strategy = fake_strategy.create_fake_strategy(
|
||||||
self.ctx,
|
self.ctx, self.DEFAULT_STRATEGY_TYPE, extra_args=extra_args
|
||||||
self.DEFAULT_STRATEGY_TYPE,
|
)
|
||||||
extra_args=extra_args)
|
|
||||||
# Call the 'success' test
|
# Call the 'success' test
|
||||||
self.test_creating_vim_strategy_success()
|
self.test_creating_vim_strategy_success()
|
||||||
|
|
||||||
|
def test_create_strategy_succeeds_without_extra_argst(self):
|
||||||
|
"""Test create strategy succeeds without extra_args"""
|
||||||
|
|
||||||
|
self.strategy = fake_strategy.create_fake_strategy(
|
||||||
|
self.ctx, self.DEFAULT_STRATEGY_TYPE, extra_args=None
|
||||||
|
)
|
||||||
|
|
||||||
|
self.test_creating_vim_strategy_success()
|
||||||
|
@ -1,21 +1,23 @@
|
|||||||
#
|
#
|
||||||
# Copyright (c) 2021-2022 Wind River Systems, Inc.
|
# Copyright (c) 2021-2022, 2024 Wind River Systems, Inc.
|
||||||
#
|
#
|
||||||
# SPDX-License-Identifier: Apache-2.0
|
# SPDX-License-Identifier: Apache-2.0
|
||||||
#
|
#
|
||||||
|
|
||||||
|
|
||||||
import mock
|
import mock
|
||||||
|
|
||||||
from dcmanager.common.consts \
|
from dcmanager.common.consts import \
|
||||||
import STRATEGY_STATE_CREATING_VIM_KUBE_ROOTCA_UPDATE_STRATEGY
|
STRATEGY_STATE_CREATING_VIM_KUBE_ROOTCA_UPDATE_STRATEGY
|
||||||
from dcmanager.common.consts import STRATEGY_STATE_FAILED
|
from dcmanager.common.consts import STRATEGY_STATE_FAILED
|
||||||
from dcmanager.common.consts \
|
from dcmanager.common.consts import \
|
||||||
import STRATEGY_STATE_KUBE_ROOTCA_UPDATE_UPLOAD_CERT
|
STRATEGY_STATE_KUBE_ROOTCA_UPDATE_UPLOAD_CERT
|
||||||
|
from dcmanager.db import api as db_api
|
||||||
from dcmanager.tests.unit.common import fake_strategy
|
from dcmanager.tests.unit.common import fake_strategy
|
||||||
from dcmanager.tests.unit.orchestrator.states.kube_rootca.test_base \
|
from dcmanager.tests.unit.orchestrator.states.kube_rootca.test_base \
|
||||||
import TestKubeRootCaUpgradeState
|
import TestKubeRootCaUpgradeState
|
||||||
|
|
||||||
# Only the 'error' field is error checked on upload_cert
|
# Only the 'error' field is checked on upload_cert
|
||||||
ERROR_UPLOADING_CERT = {"error": "File not found"}
|
ERROR_UPLOADING_CERT = {"error": "File not found"}
|
||||||
SUCCESS_UPLOADING_CERT = {"success": "Success Upload"}
|
SUCCESS_UPLOADING_CERT = {"success": "Success Upload"}
|
||||||
|
|
||||||
@ -23,16 +25,16 @@ FAKE_CERT_FILE = "some_fake_cert_file.pem"
|
|||||||
|
|
||||||
|
|
||||||
class TestUploadCertStage(TestKubeRootCaUpgradeState):
|
class TestUploadCertStage(TestKubeRootCaUpgradeState):
|
||||||
|
|
||||||
def setUp(self):
|
def setUp(self):
|
||||||
super(TestUploadCertStage, self).setUp()
|
super().setUp()
|
||||||
|
|
||||||
# Add the subcloud being processed by this unit test
|
# Add the subcloud being processed by this unit test
|
||||||
self.subcloud = self.setup_subcloud()
|
self.subcloud = self.setup_subcloud()
|
||||||
|
|
||||||
# Add the strategy_step state being processed by this unit test
|
# Add the strategy_step state being processed by this unit test
|
||||||
self.strategy_step = self.setup_strategy_step(
|
self.strategy_step = self.setup_strategy_step(
|
||||||
self.subcloud.id, STRATEGY_STATE_KUBE_ROOTCA_UPDATE_UPLOAD_CERT)
|
self.subcloud.id, STRATEGY_STATE_KUBE_ROOTCA_UPDATE_UPLOAD_CERT
|
||||||
|
)
|
||||||
|
|
||||||
self.sysinv_client.kube_rootca_update_upload_cert = mock.MagicMock()
|
self.sysinv_client.kube_rootca_update_upload_cert = mock.MagicMock()
|
||||||
|
|
||||||
@ -41,15 +43,15 @@ class TestUploadCertStage(TestKubeRootCaUpgradeState):
|
|||||||
"cert-file": FAKE_CERT_FILE
|
"cert-file": FAKE_CERT_FILE
|
||||||
}
|
}
|
||||||
self.strategy = fake_strategy.create_fake_strategy(
|
self.strategy = fake_strategy.create_fake_strategy(
|
||||||
self.ctx,
|
self.ctx, self.DEFAULT_STRATEGY_TYPE, extra_args=extra_args
|
||||||
self.DEFAULT_STRATEGY_TYPE,
|
)
|
||||||
extra_args=extra_args)
|
|
||||||
|
|
||||||
def test_upload_cert_fails(self):
|
def test_upload_cert_fails(self):
|
||||||
"""Test upload cert failed the sysinv operation
|
"""Test upload cert failed the sysinv operation
|
||||||
|
|
||||||
The state should fail
|
The state should fail
|
||||||
"""
|
"""
|
||||||
|
|
||||||
self.sysinv_client.kube_rootca_update_upload_cert.return_value = \
|
self.sysinv_client.kube_rootca_update_upload_cert.return_value = \
|
||||||
ERROR_UPLOADING_CERT
|
ERROR_UPLOADING_CERT
|
||||||
|
|
||||||
@ -63,14 +65,15 @@ class TestUploadCertStage(TestKubeRootCaUpgradeState):
|
|||||||
|
|
||||||
# Verify the strategy failed
|
# Verify the strategy failed
|
||||||
self.assert_step_updated(
|
self.assert_step_updated(
|
||||||
self.strategy_step.subcloud_id,
|
self.strategy_step.subcloud_id, STRATEGY_STATE_FAILED
|
||||||
STRATEGY_STATE_FAILED)
|
)
|
||||||
|
|
||||||
def test_upload_cert_pass(self):
|
def test_upload_cert_pass(self):
|
||||||
"""Test upload cert passes the sysinv operation
|
"""Test upload cert passes the sysinv operation
|
||||||
|
|
||||||
The state should transition to the vim creation state
|
The state should transition to the vim creation state
|
||||||
"""
|
"""
|
||||||
|
|
||||||
self.sysinv_client.kube_rootca_update_upload_cert.return_value = \
|
self.sysinv_client.kube_rootca_update_upload_cert.return_value = \
|
||||||
SUCCESS_UPLOADING_CERT
|
SUCCESS_UPLOADING_CERT
|
||||||
|
|
||||||
@ -85,4 +88,28 @@ class TestUploadCertStage(TestKubeRootCaUpgradeState):
|
|||||||
# Verify the expected next state happened
|
# Verify the expected next state happened
|
||||||
self.assert_step_updated(
|
self.assert_step_updated(
|
||||||
self.strategy_step.subcloud_id,
|
self.strategy_step.subcloud_id,
|
||||||
STRATEGY_STATE_CREATING_VIM_KUBE_ROOTCA_UPDATE_STRATEGY)
|
STRATEGY_STATE_CREATING_VIM_KUBE_ROOTCA_UPDATE_STRATEGY
|
||||||
|
)
|
||||||
|
|
||||||
|
def test_upload_cert_pass_without_extra_args(self):
|
||||||
|
"""Test upload cert passes the sysinv operation without extra args
|
||||||
|
|
||||||
|
The state should transition to the vim creation state
|
||||||
|
"""
|
||||||
|
|
||||||
|
db_api.sw_update_strategy_destroy(self.ctx)
|
||||||
|
self.strategy = fake_strategy.create_fake_strategy(
|
||||||
|
self.ctx, self.DEFAULT_STRATEGY_TYPE
|
||||||
|
)
|
||||||
|
|
||||||
|
self.sysinv_client.kube_rootca_update_upload_cert.return_value = \
|
||||||
|
SUCCESS_UPLOADING_CERT
|
||||||
|
|
||||||
|
self.worker.perform_state_action(self.strategy_step)
|
||||||
|
|
||||||
|
self.sysinv_client.kube_rootca_update_upload_cert.assert_not_called()
|
||||||
|
|
||||||
|
self.assert_step_updated(
|
||||||
|
self.strategy_step.subcloud_id,
|
||||||
|
STRATEGY_STATE_CREATING_VIM_KUBE_ROOTCA_UPDATE_STRATEGY
|
||||||
|
)
|
||||||
|
Loading…
Reference in New Issue
Block a user