diff --git a/cou/cli.py b/cou/cli.py index ff6b4f5e..a0f42d80 100644 --- a/cou/cli.py +++ b/cou/cli.py @@ -13,7 +13,6 @@ # limitations under the License. """Entrypoint for 'charmed-openstack-upgrader'.""" -import argparse import asyncio import logging import logging.handlers @@ -23,17 +22,13 @@ from juju.errors import JujuError -from cou.commands import parse_args +from cou.commands import Namespace, parse_args from cou.exceptions import COUException, HighestReleaseAchieved, TimeoutException from cou.logging import setup_logging from cou.steps import UpgradePlan from cou.steps.analyze import Analysis from cou.steps.execute import apply_step -from cou.steps.plan import ( - generate_plan, - manually_upgrade_data_plane, - pre_plan_sane_checks, -) +from cou.steps.plan import generate_plan, manually_upgrade_data_plane from cou.utils import print_and_debug, progress_indicator, prompt_input from cou.utils.cli import interrupt_handler from cou.utils.juju_utils import COUModel @@ -110,11 +105,11 @@ async def continue_upgrade() -> bool: return False -async def analyze_and_plan(args: argparse.Namespace) -> tuple[Analysis, UpgradePlan]: +async def analyze_and_plan(args: Namespace) -> tuple[Analysis, UpgradePlan]: """Analyze cloud and generate the upgrade plan with steps. :param args: CLI arguments - :type args: argparse.Namespace + :type args: Namespace :return: Generated analysis and upgrade plan. :rtype: tuple[Analysis, UpgradePlan] """ @@ -129,22 +124,18 @@ async def analyze_and_plan(args: argparse.Namespace) -> tuple[Analysis, UpgradeP logger.info(analysis_result) progress_indicator.succeed() - progress_indicator.start("Making sane checks to upgrade...") - pre_plan_sane_checks(args.upgrade_group, analysis_result) - progress_indicator.succeed() - progress_indicator.start("Generating upgrade plan...") - upgrade_plan = await generate_plan(analysis_result, args.backup) + upgrade_plan = await generate_plan(analysis_result, args) progress_indicator.succeed() return analysis_result, upgrade_plan -async def get_upgrade_plan(args: argparse.Namespace) -> None: +async def get_upgrade_plan(args: Namespace) -> None: """Get upgrade plan and print to console. :param args: CLI arguments - :type args: argparse.Namespace + :type args: Namespace """ analysis_result, upgrade_plan = await analyze_and_plan(args) print_and_debug(upgrade_plan) @@ -155,11 +146,11 @@ async def get_upgrade_plan(args: argparse.Namespace) -> None: ) -async def run_upgrade(args: argparse.Namespace) -> None: +async def run_upgrade(args: Namespace) -> None: """Run cloud upgrade. :param args: CLI arguments - :type args: argparse.Namespace + :type args: Namespace """ prompt = not args.auto_approve analysis_result, upgrade_plan = await analyze_and_plan(args) @@ -182,11 +173,11 @@ async def run_upgrade(args: argparse.Namespace) -> None: print("Upgrade completed.") -async def _run_command(args: argparse.Namespace) -> None: +async def _run_command(args: Namespace) -> None: """Run 'charmed-openstack-upgrade' command. :param args: CLI arguments - :type args: argparse.Namespace + :type args: Namespace """ match args.command: case "plan": diff --git a/cou/commands.py b/cou/commands.py index 1d36de2e..ac820f9e 100644 --- a/cou/commands.py +++ b/cou/commands.py @@ -14,10 +14,13 @@ """Command line arguments parsing for 'charmed-openstack-upgrader'.""" import argparse -from typing import Any, Iterable, Optional +from typing import Any, Iterable, NamedTuple, Optional import pkg_resources +CONTROL_PLANE = "control-plane" +DATA_PLANE = "data-plane" + class CapitalizeHelpFormatter(argparse.RawTextHelpFormatter): """Capitalize message prefix.""" @@ -194,7 +197,7 @@ def create_plan_subparser( # help="For more information about a upgrade group, run 'cou plan ' -h.", ) plan_subparser.add_parser( - "control-plane", + CONTROL_PLANE, description="Show the steps for upgrading the control-plane components.", help="Show the steps for upgrading the control-plane components.", usage="cou plan control-plane [options]", @@ -202,7 +205,7 @@ def create_plan_subparser( formatter_class=CapitalizeHelpFormatter, ) plan_subparser.add_parser( - "data-plane", + DATA_PLANE, description="Show the steps for upgrading the data-plane components.\nThis is possible " "only if control-plane has been fully upgraded,\notherwise an error will be thrown.", help="Show the steps for upgrading the data-plane components.\nThis is possible " @@ -316,13 +319,32 @@ def create_subparsers(parser: argparse.ArgumentParser) -> argparse._SubParsersAc return subparsers -def parse_args(args: Any) -> argparse.Namespace: # pylint: disable=inconsistent-return-statements +class Namespace(NamedTuple): + """Mock Namespace used for type hinting purposes. + + Keep in sync with the argument parser defined in parse_args + """ + + command: str + verbosity: int = 0 + backup: bool = True + quiet: bool = False + auto_approve: bool = False + model_name: Optional[str] = None + upgrade_group: Optional[str] = None + subcommand: Optional[str] = None # for help option + machines: Optional[list[str]] = None + hostnames: Optional[list[str]] = None + availability_zones: Optional[list[str]] = None + + +def parse_args(args: Any) -> Namespace: # pylint: disable=inconsistent-return-statements """Parse cli arguments. :param args: Arguments parser. :type args: Any - :return: argparse.Namespace - :rtype: argparse.Namespace + :return: Namespace custom object. + :rtype: Namespace :raises argparse.ArgumentError: Unexpected arguments input. """ # Configure top level argparser and its options @@ -354,7 +376,7 @@ def parse_args(args: Any) -> argparse.Namespace: # pylint: disable=inconsistent parser.exit() try: - parsed_args = parser.parse_args(args) + parsed_args = Namespace(**vars(parser.parse_args(args))) # print help messages for an available sub-command if parsed_args.command == "help": diff --git a/cou/steps/plan.py b/cou/steps/plan.py index 9332615f..caf61b0d 100644 --- a/cou/steps/plan.py +++ b/cou/steps/plan.py @@ -15,7 +15,7 @@ """Upgrade planning utilities.""" import logging -from typing import Callable, Optional +from typing import Callable # NOTE we need to import the modules to register the charms with the register_application # decorator @@ -37,6 +37,7 @@ OpenStackSubordinateApplication, SubordinateBaseClass, ) +from cou.commands import DATA_PLANE, Namespace from cou.exceptions import ( DataPlaneCannotUpgrade, HaltUpgradePlanGeneration, @@ -53,24 +54,20 @@ logger = logging.getLogger(__name__) -def pre_plan_sane_checks(upgrade_group: Optional[str], analysis_result: Analysis) -> None: +def pre_plan_sane_checks(args: Namespace, analysis_result: Analysis) -> None: """Pre checks to generate the upgrade plan. - :param upgrade_group: Upgrade group passed by the user using the cli. - :type upgrade_group: Optional[str] + :param args: CLI arguments + :type args: Namespace :param analysis_result: Analysis result. :type analysis_result: Analysis """ - checks = [ - is_valid_openstack_cloud, - is_supported_series, - is_highest_release_achieved, - is_target_supported, - ] - if upgrade_group == "data-plane": - checks.append(is_data_plane_ready_to_upgrade) - for check in checks: - check(analysis_result) + is_valid_openstack_cloud(analysis_result) + is_supported_series(analysis_result) + is_highest_release_achieved(analysis_result) + + if args.upgrade_group == DATA_PLANE: + is_data_plane_ready_to_upgrade(analysis_result) def is_valid_openstack_cloud(analysis_result: Analysis) -> None: @@ -81,15 +78,13 @@ def is_valid_openstack_cloud(analysis_result: Analysis) -> None: :raises NoTargetError: When cannot determine the current OS release or Ubuntu series. """ - current_os_release = analysis_result.current_cloud_os_release - current_series = analysis_result.current_cloud_series - if not current_os_release: + if not analysis_result.current_cloud_os_release: raise NoTargetError( "Cannot determine the current OS release in the cloud. " "Is this a valid OpenStack cloud?" ) - if not current_series: + if not analysis_result.current_cloud_series: raise NoTargetError( "Cannot determine the current Ubuntu series in the cloud. " "Is this a valid OpenStack cloud?" @@ -103,10 +98,11 @@ def is_supported_series(analysis_result: Analysis) -> None: :type analysis_result: Analysis :raises OutOfSupportRange: When series is not supported. """ - current_series = analysis_result.current_cloud_series supporting_lts_series = ", ".join(LTS_TO_OS_RELEASE) # series already checked at is_valid_openstack_cloud - if current_series not in supporting_lts_series: # type: ignore + if ( + current_series := analysis_result.current_cloud_series + ) and current_series not in LTS_TO_OS_RELEASE: raise OutOfSupportRange( f"Cloud series '{current_series}' is not a Ubuntu LTS series supported by COU. " f"The supporting series are: {supporting_lts_series}" @@ -120,48 +116,18 @@ def is_highest_release_achieved(analysis_result: Analysis) -> None: :type analysis_result: Analysis :raises HighestReleaseAchieved: When the OpenStack release is the last supported by the series. """ - # series and current OS release already checked at is_valid_openstack_cloud - current_os_release = analysis_result.current_cloud_os_release - current_series = analysis_result.current_cloud_series - if str(current_os_release) == LTS_TO_OS_RELEASE[current_series][-1]: # type: ignore + if ( + (current_os_release := analysis_result.current_cloud_os_release) + and (current_series := analysis_result.current_cloud_series) + and str(current_os_release) == LTS_TO_OS_RELEASE[current_series][-1] + ): raise HighestReleaseAchieved( f"No upgrades available for OpenStack {str(current_os_release).capitalize()} on " - f"Ubuntu {current_series.capitalize()}.\nNewer OpenStack releases " # type: ignore + f"Ubuntu {current_series.capitalize()}.\nNewer OpenStack releases " "may be available after upgrading to a later Ubuntu series." ) -def is_target_supported(analysis_result: Analysis) -> None: - """Check if the target to upgrade is supported. - - :param analysis_result: Analysis result. - :type analysis_result: Analysis - :raises NoTargetError: When cannot find target to upgrade - :raises OutOfSupportRange: When the upgrade scope is not supported - by the current series. - """ - current_os_release = analysis_result.current_cloud_os_release - current_series = analysis_result.current_cloud_series - if current_os_release and current_series: - target = current_os_release.next_release - if not target: - raise NoTargetError( - "Cannot find target to upgrade. Current minimum OS release is " - f"'{str(current_os_release)}'. Current Ubuntu series is '{current_series}'." - ) - - supporting_os_release = ", ".join(LTS_TO_OS_RELEASE[current_series]) - if ( - str(current_os_release) not in supporting_os_release - or str(target) not in supporting_os_release - ): - raise OutOfSupportRange( - f"Unable to upgrade cloud from `{current_series}` to '{target}'. " - "Both the from and to releases need to be supported by the current " - "Ubuntu series '{current_series}': {supporting_os_release}." - ) - - def is_data_plane_ready_to_upgrade(analysis_result: Analysis) -> None: """Check if data plane is ready to upgrade. @@ -173,7 +139,7 @@ def is_data_plane_ready_to_upgrade(analysis_result: Analysis) -> None: """ if not analysis_result.min_os_version_data_plane: raise DataPlaneCannotUpgrade( - "Cannot find data-plane charms. Is this a valid OpenStack cloud?" + "Cannot find data-plane apps. Is this a valid OpenStack cloud?" ) if not is_control_plane_upgraded(analysis_result): raise DataPlaneCannotUpgrade("Please, upgrade control-plane before data-plane") @@ -193,23 +159,57 @@ def is_control_plane_upgraded(analysis_result: Analysis) -> bool: control_plane = analysis_result.min_os_version_control_plane data_plane = analysis_result.min_os_version_data_plane - if control_plane and data_plane: - return control_plane > data_plane - return False + return bool(control_plane and data_plane and control_plane > data_plane) + + +def determine_upgrade_target(analysis_result: Analysis) -> OpenStackRelease: + """Determine the target release to upgrade to. + + :param analysis_result: Analysis result. + :type analysis_result: Analysis + :raises NoTargetError: When cannot find target to upgrade + :raises OutOfSupportRange: When the upgrade scope is not supported + by the current series. + :return: The target OS release to upgrade the cloud to. + :rtype: OpenStackRelease + """ + if ( + (current_os_release := analysis_result.current_cloud_os_release) + and (current_series := analysis_result.current_cloud_series) + and not (target := current_os_release.next_release) + ): + raise NoTargetError( + "Cannot find target to upgrade. Current minimum OS release is " + f"'{str(current_os_release)}'. Current Ubuntu series is '{current_series}'." + ) + + if ( + current_series + and (supporting_os_release := ", ".join(LTS_TO_OS_RELEASE[current_series])) + and str(current_os_release) not in supporting_os_release + or str(target) not in supporting_os_release + ): + raise OutOfSupportRange( + f"Unable to upgrade cloud from `{current_series}` to '{target}'. " + "Both the from and to releases need to be supported by the current " + "Ubuntu series '{current_series}': {supporting_os_release}." + ) + + return target # type: ignore -async def generate_plan(analysis_result: Analysis, backup_database: bool) -> UpgradePlan: +async def generate_plan(analysis_result: Analysis, args: Namespace) -> UpgradePlan: """Generate plan for upgrade. :param analysis_result: Analysis result. :type analysis_result: Analysis - :param backup_database: Whether to create database backup before upgrade. - :type backup_database: bool + :param args: CLI arguments + :type args: Namespace :return: Plan with all upgrade steps necessary based on the Analysis. :rtype: UpgradePlan """ - # target already checked on pre_plan_sane_checks - target = analysis_result.current_cloud_os_release.next_release # type: ignore + pre_plan_sane_checks(args, analysis_result) + target = determine_upgrade_target(analysis_result) plan = UpgradePlan( f"Upgrade cloud from '{analysis_result.current_cloud_os_release}' to '{target}'" @@ -229,7 +229,7 @@ async def generate_plan(analysis_result: Analysis, backup_database: bool) -> Upg ), ) ) - if backup_database: + if args.backup: plan.add_step( PreUpgradeStep( description="Backup mysql databases", @@ -241,7 +241,7 @@ async def generate_plan(analysis_result: Analysis, backup_database: bool) -> Upg control_plane_principal_upgrade_plan = await create_upgrade_group( apps=analysis_result.apps_control_plane, description="Control Plane principal(s) upgrade plan", - target=target, # type: ignore + target=target, filter_function=lambda app: not isinstance(app, SubordinateBaseClass), ) plan.add_step(control_plane_principal_upgrade_plan) @@ -249,7 +249,7 @@ async def generate_plan(analysis_result: Analysis, backup_database: bool) -> Upg control_plane_subordinate_upgrade_plan = await create_upgrade_group( apps=analysis_result.apps_control_plane, description="Control Plane subordinate(s) upgrade plan", - target=target, # type: ignore + target=target, filter_function=lambda app: isinstance(app, SubordinateBaseClass), ) plan.add_step(control_plane_subordinate_upgrade_plan) diff --git a/tests/unit/conftest.py b/tests/unit/conftest.py index cf9b71dc..f1dd7f07 100644 --- a/tests/unit/conftest.py +++ b/tests/unit/conftest.py @@ -25,6 +25,7 @@ from cou.apps.base import ApplicationUnit, OpenStackApplication from cou.apps.core import Keystone from cou.apps.subordinate import OpenStackSubordinateApplication +from cou.commands import Namespace from cou.utils.openstack import OpenStackRelease @@ -575,3 +576,14 @@ def cou_data(tmp_path_factory): cou_test = tmp_path_factory.mktemp("cou_test") with patch("cou.utils.COU_DATA", cou_test): yield + + +@pytest.fixture +def cli_args() -> MagicMock: + """Magic Mock of the COU Namespace. + + :return: MagicMock of the COU Namespace got from the cli. + :rtype: MagicMock + """ + # spec_set needs an instantiated class to be strict with the fields. + return MagicMock(spec_set=Namespace(command="plan"))() diff --git a/tests/unit/steps/test_steps_plan.py b/tests/unit/steps/test_steps_plan.py index fdfc6d89..552bdfd5 100644 --- a/tests/unit/steps/test_steps_plan.py +++ b/tests/unit/steps/test_steps_plan.py @@ -17,6 +17,7 @@ from cou.apps.base import OpenStackApplication from cou.exceptions import ( + DataPlaneCannotUpgrade, HaltUpgradePlanGeneration, HighestReleaseAchieved, NoTargetError, @@ -35,7 +36,9 @@ create_upgrade_group, determine_upgrade_target, generate_plan, + is_control_plane_upgraded, manually_upgrade_data_plane, + pre_plan_sane_checks, ) from cou.utils import app_utils from cou.utils.openstack import OpenStackRelease @@ -131,7 +134,7 @@ def generate_expected_upgrade_plan_subordinate(app, target, model): @pytest.mark.asyncio -async def test_generate_plan(apps, model): +async def test_generate_plan(apps, model, cli_args): target = OpenStackRelease("victoria") app_keystone = apps["keystone_ussuri"] app_cinder = apps["cinder_ussuri"] @@ -142,7 +145,7 @@ async def test_generate_plan(apps, model): apps_data_plane=[], ) - upgrade_plan = await generate_plan(analysis_result, backup_database=True) + upgrade_plan = await generate_plan(analysis_result, cli_args) expected_plan = UpgradePlan("Upgrade cloud from 'ussuri' to 'victoria'") expected_plan.add_step( @@ -180,23 +183,26 @@ async def test_generate_plan(apps, model): @pytest.mark.parametrize( - "current_os_release, current_series, next_release", + "upgrade_group, expected_call", [ - (OpenStackRelease("victoria"), "focal", "wallaby"), - (OpenStackRelease("xena"), "focal", "yoga"), + (None, False), + ("control-plane", False), + ("data-plane", True), ], ) -def test_determine_upgrade_target(current_os_release, current_series, next_release): - target = determine_upgrade_target(current_os_release, current_series) - - assert target == next_release - - -def test_determine_upgrade_target_no_upgrade_available(): - current_os_release = OpenStackRelease("yoga") - current_series = "focal" - with pytest.raises(HighestReleaseAchieved): - determine_upgrade_target(current_os_release, current_series) +@patch("cou.steps.plan.is_data_plane_ready_to_upgrade") +def test_pre_plan_sane_checks( + mock_is_data_plane_ready_to_upgrade, upgrade_group, expected_call, cli_args +): + mock_analysis_result = MagicMock(spec=Analysis)() + mock_analysis_result.current_cloud_os_release = OpenStackRelease("ussuri") + mock_analysis_result.current_cloud_series = "focal" + cli_args.upgrade_group = upgrade_group + pre_plan_sane_checks(cli_args, mock_analysis_result) + if expected_call: + mock_is_data_plane_ready_to_upgrade.assert_called_once_with(mock_analysis_result) + else: + mock_is_data_plane_ready_to_upgrade.assert_not_called() @pytest.mark.parametrize( @@ -216,15 +222,107 @@ def test_determine_upgrade_target_no_upgrade_available(): ), # current_series is None ], ) -def test_determine_upgrade_target_invalid_input(current_os_release, current_series, exp_error_msg): +def test_pre_plan_sane_checks_is_valid_openstack_cloud( + current_os_release, current_series, exp_error_msg, cli_args +): with pytest.raises(NoTargetError, match=exp_error_msg): - determine_upgrade_target(current_os_release, current_series) + mock_analysis_result = MagicMock(spec=Analysis)() + mock_analysis_result.current_cloud_series = current_series + mock_analysis_result.current_cloud_os_release = current_os_release + pre_plan_sane_checks(cli_args, mock_analysis_result) + + +@pytest.mark.parametrize( + "current_os_release, current_series", + [ + (OpenStackRelease("yoga"), "jammy"), + (OpenStackRelease("train"), "bionic"), + ], +) +def test_pre_plan_sane_checks_is_supported_series(current_os_release, current_series, cli_args): + mock_analysis_result = MagicMock(spec=Analysis)() + with pytest.raises(OutOfSupportRange): + mock_analysis_result.current_cloud_os_release = current_os_release + mock_analysis_result.current_cloud_series = current_series + pre_plan_sane_checks(cli_args, mock_analysis_result) + + +def test_pre_plan_sane_checks_is_highest_release_achieved(cli_args): + mock_analysis_result = MagicMock(spec=Analysis)() + mock_analysis_result.current_cloud_os_release = OpenStackRelease("yoga") + mock_analysis_result.current_cloud_series = "focal" + with pytest.raises(HighestReleaseAchieved): + pre_plan_sane_checks(cli_args, mock_analysis_result) + + +@pytest.mark.parametrize( + "min_os_version_control_plane, min_os_version_data_plane, exp_error_msg", + [ + ( + OpenStackRelease("ussuri"), + OpenStackRelease("ussuri"), + "Please, upgrade control-plane before data-plane", + ), + ( + OpenStackRelease("ussuri"), + None, + "Cannot find data-plane apps. Is this a valid OpenStack cloud?", + ), + ], +) +def test_pre_plan_sane_checks_is_data_plane_ready_to_upgrade_error( + min_os_version_control_plane, min_os_version_data_plane, exp_error_msg, cli_args +): + cli_args.upgrade_group = "data-plane" + mock_analysis_result = MagicMock(spec=Analysis)() + mock_analysis_result.current_cloud_series = "focal" + mock_analysis_result.min_os_version_control_plane = min_os_version_control_plane + mock_analysis_result.min_os_version_data_plane = min_os_version_data_plane + with pytest.raises(DataPlaneCannotUpgrade, match=exp_error_msg): + pre_plan_sane_checks(cli_args, mock_analysis_result) + + +@pytest.mark.parametrize( + "min_os_version_control_plane, min_os_version_data_plane, expected_result", + [ + (OpenStackRelease("ussuri"), OpenStackRelease("ussuri"), False), + (OpenStackRelease("ussuri"), OpenStackRelease("victoria"), False), + (OpenStackRelease("ussuri"), None, False), + (OpenStackRelease("victoria"), OpenStackRelease("ussuri"), True), + ], +) +def test_is_control_plane_upgraded( + min_os_version_control_plane, min_os_version_data_plane, expected_result +): + mock_analysis_result = MagicMock(spec=Analysis)() + mock_analysis_result.min_os_version_control_plane = min_os_version_control_plane + mock_analysis_result.min_os_version_data_plane = min_os_version_data_plane + assert is_control_plane_upgraded(mock_analysis_result) is expected_result + + +@pytest.mark.parametrize( + "current_os_release, current_series, next_release", + [ + (OpenStackRelease("victoria"), "focal", "wallaby"), + (OpenStackRelease("xena"), "focal", "yoga"), + ], +) +def test_determine_upgrade_target(current_os_release, current_series, next_release): + mock_analysis_result = MagicMock(spec=Analysis)() + mock_analysis_result.current_cloud_os_release = current_os_release + mock_analysis_result.current_cloud_series = current_series + + target = determine_upgrade_target(mock_analysis_result) + + assert target == next_release def test_determine_upgrade_target_no_next_release(): + mock_analysis_result = MagicMock(spec=Analysis)() + mock_analysis_result.current_cloud_series = "focal" + exp_error_msg = "Cannot find target to upgrade. Current minimum OS release is " "'ussuri'. Current Ubuntu series is 'focal'." - current_series = "focal" with pytest.raises(NoTargetError, match=exp_error_msg), patch( "cou.utils.openstack.OpenStackRelease.next_release", new_callable=PropertyMock @@ -233,20 +331,17 @@ def test_determine_upgrade_target_no_next_release(): current_os_release = OpenStackRelease( "ussuri" ) # instantiate OpenStackRelease with any valid codename - determine_upgrade_target(current_os_release, current_series) + mock_analysis_result.current_cloud_os_release = current_os_release + determine_upgrade_target(mock_analysis_result) -@pytest.mark.parametrize( - "current_os_release, current_series", - [ - (OpenStackRelease("yoga"), "jammy"), - (OpenStackRelease("train"), "bionic"), - (OpenStackRelease("zed"), "focal"), - ], -) -def test_determine_upgrade_target_release_out_of_range(current_os_release, current_series): +def test_determine_upgrade_target_out_support_range(): + mock_analysis_result = MagicMock(spec=Analysis)() + mock_analysis_result.current_cloud_series = "focal" + mock_analysis_result.current_cloud_os_release = OpenStackRelease("zed") + with pytest.raises(OutOfSupportRange): - determine_upgrade_target(current_os_release, current_series) + determine_upgrade_target(mock_analysis_result) @pytest.mark.asyncio diff --git a/tests/unit/test_cli.py b/tests/unit/test_cli.py index e79b2875..47cfbd9d 100644 --- a/tests/unit/test_cli.py +++ b/tests/unit/test_cli.py @@ -63,17 +63,20 @@ def test_get_log_level(quiet, verbosity, level): @patch("cou.cli.COUModel") @patch("cou.cli.generate_plan", new_callable=AsyncMock) @patch("cou.cli.Analysis.create", new_callable=AsyncMock) -async def test_analyze_and_plan(mock_analyze, mock_generate_plan, cou_model): +async def test_analyze_and_plan(mock_analyze, mock_generate_plan, cou_model, cli_args): """Test analyze_and_plan function with different model_name arguments.""" + cli_args.model_name = None + cli_args.backup = False + cou_model.return_value.connect.side_effect = AsyncMock() analysis_result = Analysis(model=cou_model, apps_control_plane=[], apps_data_plane=[]) mock_analyze.return_value = analysis_result - await cli.analyze_and_plan(None, False) + await cli.analyze_and_plan(cli_args) cou_model.assert_called_once_with(None) mock_analyze.assert_awaited_once_with(cou_model.return_value) - mock_generate_plan.assert_awaited_once_with(analysis_result, False) + mock_generate_plan.assert_awaited_once_with(analysis_result, cli_args) @pytest.mark.asyncio @@ -81,7 +84,7 @@ async def test_analyze_and_plan(mock_analyze, mock_generate_plan, cou_model): @patch("cou.cli.analyze_and_plan", new_callable=AsyncMock) @patch("cou.cli.print_and_debug") async def test_get_upgrade_plan( - mock_print_and_debug, mock_analyze_and_plan, mock_manually_upgrade + mock_print_and_debug, mock_analyze_and_plan, mock_manually_upgrade, cli_args ): """Test get_upgrade_plan function.""" plan = UpgradePlan(description="Upgrade cloud from 'ussuri' to 'victoria'") @@ -89,9 +92,9 @@ async def test_get_upgrade_plan( mock_analysis_result = MagicMock() mock_analyze_and_plan.return_value = (mock_analysis_result, plan) - await cli.get_upgrade_plan(None, True) + await cli.get_upgrade_plan(cli_args) - mock_analyze_and_plan.assert_awaited_once_with(None, True) + mock_analyze_and_plan.assert_awaited_once_with(cli_args) mock_print_and_debug.assert_called_once_with(plan) mock_manually_upgrade.assert_called_once() @@ -117,16 +120,19 @@ async def test_run_upgrade_quiet( mock_manually_upgrade, quiet, expected_print_count, + cli_args, ): """Test get_upgrade_plan function in either quiet or non-quiet mode.""" + cli_args.quiet = quiet + plan = UpgradePlan(description="Upgrade cloud from 'ussuri' to 'victoria'") plan.add_step(PreUpgradeStep(description="backup mysql databases", parallel=False)) mock_analysis_result = MagicMock() mock_analyze_and_plan.return_value = (mock_analysis_result, plan) - await cli.run_upgrade(model_name=None, backup_database=True, prompt=False, quiet=quiet) + await cli.run_upgrade(cli_args) - mock_analyze_and_plan.assert_awaited_once_with(None, True) + mock_analyze_and_plan.assert_awaited_once_with(cli_args) mock_print_and_debug.assert_called_once_with(plan) mock_apply_step.assert_called_once_with(plan, False) mock_print.call_count == expected_print_count @@ -143,16 +149,20 @@ async def test_run_upgrade_with_prompt_continue( mock_apply_step, mock_analyze_and_plan, mock_manually_upgrade, + cli_args, ): + cli_args.auto_approve = False + cli_args.quiet = True + plan = UpgradePlan(description="Upgrade cloud from 'ussuri' to 'victoria'") plan.add_step(PreUpgradeStep(description="backup mysql databases", parallel=False)) mock_analysis_result = MagicMock() mock_analyze_and_plan.return_value = (mock_analysis_result, plan) mock_continue_upgrade.return_value = True - await cli.run_upgrade(model_name=None, backup_database=True, prompt=True, quiet=False) + await cli.run_upgrade(cli_args) - mock_analyze_and_plan.assert_awaited_once_with(None, True) + mock_analyze_and_plan.assert_awaited_once_with(cli_args) mock_continue_upgrade.assert_awaited_once_with() mock_apply_step.assert_called_once_with(plan, True) mock_manually_upgrade.assert_called_once() @@ -168,16 +178,20 @@ async def test_run_upgrade_with_prompt_abort( mock_apply_step, mock_analyze_and_plan, mock_manually_upgrade, + cli_args, ): + cli_args.auto_approve = False + cli_args.quiet = True + plan = UpgradePlan(description="Upgrade cloud from 'ussuri' to 'victoria'") plan.add_step(PreUpgradeStep(description="backup mysql databases", parallel=False)) mock_analysis_result = MagicMock() mock_analyze_and_plan.return_value = (mock_analysis_result, plan) mock_continue_upgrade.return_value = False - await cli.run_upgrade(model_name=None, backup_database=True, prompt=True, quiet=False) + await cli.run_upgrade(cli_args) - mock_analyze_and_plan.assert_awaited_once_with(None, True) + mock_analyze_and_plan.assert_awaited_once_with(cli_args) mock_continue_upgrade.assert_awaited_once_with() mock_apply_step.assert_not_awaited() mock_manually_upgrade.assert_not_called() @@ -193,15 +207,19 @@ async def test_run_upgrade_with_no_prompt( mock_apply_step, mock_analyze_and_plan, mock_manually_upgrade, + cli_args, ): + cli_args.auto_approve = True + cli_args.quiet = True + plan = UpgradePlan(description="Upgrade cloud from 'ussuri' to 'victoria'") plan.add_step(PreUpgradeStep(description="backup mysql databases", parallel=False)) mock_analysis_result = MagicMock() mock_analyze_and_plan.return_value = (mock_analysis_result, plan) - await cli.run_upgrade(model_name=None, backup_database=True, prompt=False, quiet=False) + await cli.run_upgrade(cli_args) - mock_analyze_and_plan.assert_awaited_once_with(None, True) + mock_analyze_and_plan.assert_awaited_once_with(cli_args) mock_continue_upgrade.assert_not_awaited() mock_apply_step.assert_called_once_with(plan, False) mock_manually_upgrade.assert_called_once() @@ -234,22 +252,17 @@ async def test_continue_upgrade( @pytest.mark.parametrize("command", ["plan", "upgrade", "other1", "other2"]) @patch("cou.cli.get_upgrade_plan") @patch("cou.cli.run_upgrade") -async def test_run_command(mock_run_upgrade, mock_get_upgrade_plan, command): +async def test_run_command(mock_run_upgrade, mock_get_upgrade_plan, command, cli_args): """Test run command function.""" - args = MagicMock(spec="argparse.Namespace")() - args.command = command - prompt = not args.auto_approve + cli_args.command = command - await cli._run_command(args) + await cli._run_command(cli_args) if command == "plan": - mock_get_upgrade_plan.assert_awaited_once_with( - args.model_name, - args.backup, - ) + mock_get_upgrade_plan.assert_awaited_once_with(cli_args) mock_run_upgrade.assert_not_called() elif command == "upgrade": - mock_run_upgrade.assert_awaited_once_with(args.model_name, args.backup, prompt, args.quiet) + mock_run_upgrade.assert_awaited_once_with(cli_args) mock_get_upgrade_plan.assert_not_called() else: mock_run_upgrade.assert_not_called() diff --git a/tests/unit/test_commands.py b/tests/unit/test_commands.py index 5f4d7287..1410a84c 100644 --- a/tests/unit/test_commands.py +++ b/tests/unit/test_commands.py @@ -12,12 +12,13 @@ # See the License for the specific language governing permissions and # limitations under the License. -from argparse import ArgumentParser, Namespace +from argparse import ArgumentParser from unittest.mock import patch import pytest from cou import commands +from cou.commands import Namespace @pytest.mark.parametrize( @@ -349,7 +350,6 @@ def test_parse_args_plan(args, expected_namespace): def test_parse_args_upgrade(args, expected_namespace): """Test parsing 'run' subcommand and its arguments/options.""" parsed_args = commands.parse_args(args) - assert parsed_args == expected_namespace