Custom COU Namespace

cou/cli.py

@@ -13,18 +13,16 @@
 # limitations under the License.
 
 """Entrypoint for 'charmed-openstack-upgrader'."""
-import argparse
 import asyncio
 import logging
 import logging.handlers
 import sys
 from enum import Enum
 from signal import SIGINT, SIGTERM
-from typing import Optional
 
 from juju.errors import JujuError
 
-from cou.commands import parse_args
+from cou.commands import CLIargs, parse_args
 from cou.exceptions import COUException, HighestReleaseAchieved, TimeoutException
 from cou.logging import setup_logging
 from cou.steps import UpgradePlan
@@ -107,19 +105,15 @@ async def continue_upgrade() -> bool:
     return False
 
 
-async def analyze_and_plan(
-    model_name: Optional[str], backup_database: bool
-) -> tuple[Analysis, UpgradePlan]:
+async def analyze_and_plan(args: CLIargs) -> tuple[Analysis, UpgradePlan]:
     """Analyze cloud and generate the upgrade plan with steps.
 
-    :param model_name: Model name inputted by user.
-    :type model_name: Optional[str]
-    :param backup_database: Whether to create database backup before upgrade.
-    :type backup_database: bool
+    :param args: CLI arguments
+    :type args: CLIargs
     :return: Generated analysis and upgrade plan.
     :rtype: tuple[Analysis, UpgradePlan]
     """
-    model = COUModel(model_name)
+    model = COUModel(args.model_name)
     progress_indicator.start(f"Connecting to '{model.name}' model...")
     await model.connect()
     logger.info("Using model: %s", model.name)
@@ -131,21 +125,19 @@ async def analyze_and_plan(
     progress_indicator.succeed()
 
     progress_indicator.start("Generating upgrade plan...")
-    upgrade_plan = await generate_plan(analysis_result, backup_database)
+    upgrade_plan = await generate_plan(analysis_result, args)
     progress_indicator.succeed()
 
     return analysis_result, upgrade_plan
 
 
-async def get_upgrade_plan(model_name: Optional[str], backup_database: bool) -> None:
+async def get_upgrade_plan(args: CLIargs) -> None:
     """Get upgrade plan and print to console.
 
-    :param model_name: Model name inputted by user.
-    :type model_name: Optional[str]
-    :param backup_database: Whether to create database backup before upgrade.
-    :type backup_database: bool
+    :param args: CLI arguments
+    :type args: CLIargs
     """
-    analysis_result, upgrade_plan = await analyze_and_plan(model_name, backup_database)
+    analysis_result, upgrade_plan = await analyze_and_plan(args)
     print_and_debug(upgrade_plan)
     manually_upgrade_data_plane(analysis_result)
     print(
@@ -154,27 +146,16 @@ async def get_upgrade_plan(model_name: Optional[str], backup_database: bool) ->
     )
 
 
-async def run_upgrade(
-    model_name: Optional[str],
-    backup_database: bool,
-    prompt: bool,
-    quiet: bool,
-) -> None:
+async def run_upgrade(args: CLIargs) -> None:
     """Run cloud upgrade.
 
-    :param model_name: Model name inputted by user.
-    :type model_name: Optional[str]
-    :param backup_database: Whether to create database backup before upgrade.
-    :type backup_database: bool
-    :param prompt: Whether to prompt to run upgrade interactively.
-    :type prompt: bool
-    :param quiet: Whether to run upgrade in quiet mode.
-    :type quiet: bool
+    :param args: CLI arguments
+    :type args: CLIargs
     """
-    analysis_result, upgrade_plan = await analyze_and_plan(model_name, backup_database)
+    analysis_result, upgrade_plan = await analyze_and_plan(args)
     print_and_debug(upgrade_plan)
 
-    if prompt and not await continue_upgrade():
+    if args.prompt and not await continue_upgrade():
         return
 
     # NOTE(rgildein): add handling upgrade plan canceling for SIGINT (ctrl+c) and SIGTERM
@@ -183,26 +164,25 @@ async def run_upgrade(
     loop.add_signal_handler(SIGTERM, interrupt_handler, upgrade_plan, loop, 143)
 
     # don't print plan if in quiet mode
-    if not quiet:
+    if not args.quiet:
         print("Running cloud upgrade...")
 
-    await apply_step(upgrade_plan, prompt)
+    await apply_step(upgrade_plan, args.prompt)
     manually_upgrade_data_plane(analysis_result)
     print("Upgrade completed.")
 
 
-async def _run_command(args: argparse.Namespace) -> None:
+async def _run_command(args: CLIargs) -> None:
     """Run 'charmed-openstack-upgrade' command.
 
     :param args: CLI arguments
-    :type args: argparse.Namespace
+    :type args: CLIargs
     """
     match args.command:
         case "plan":
-            await get_upgrade_plan(args.model_name, args.backup)
+            await get_upgrade_plan(args)
         case "upgrade":
-            prompt = not args.auto_approve
-            await run_upgrade(args.model_name, args.backup, prompt, args.quiet)
+            await run_upgrade(args)
 
 
 def entrypoint() -> None:

cou/commands.py

@@ -14,10 +14,14 @@
 
 """Command line arguments parsing for 'charmed-openstack-upgrader'."""
 import argparse
+from dataclasses import dataclass
 from typing import Any, Iterable, Optional
 
 import pkg_resources
 
+CONTROL_PLANE = "control-plane"
+DATA_PLANE = "data-plane"
+
 
 class CapitalizeHelpFormatter(argparse.RawTextHelpFormatter):
     """Capitalize message prefix."""
@@ -194,15 +198,15 @@ def create_plan_subparser(
         # help="For more information about a upgrade group, run 'cou plan <upgrade-group>' -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]",
         parents=[subcommand_common_opts_parser],
         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 +320,44 @@ def create_subparsers(parser: argparse.ArgumentParser) -> argparse._SubParsersAc
     return subparsers
 
 
-def parse_args(args: Any) -> argparse.Namespace:  # pylint: disable=inconsistent-return-statements
+@dataclass(frozen=True)
+class CLIargs:
+    """Wrap CLI arguments instead of using argparse.Namespace.
+
+    Keep in sync with the argument parser defined in parse_args and check types.
+    """
+
+    # pylint: disable=too-many-instance-attributes
+
+    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
+
+    @property
+    def prompt(self) -> bool:
+        """Whether if COU should prompt to the user.
+
+        :return: Prompt if auto_approve is True, otherwise don't prompt.
+        :rtype: bool
+        """
+        return not self.auto_approve
+
+
+def parse_args(args: Any) -> CLIargs:  # pylint: disable=inconsistent-return-statements
     """Parse cli arguments.
 
     :param args: Arguments parser.
     :type args: Any
-    :return: argparse.Namespace
-    :rtype: argparse.Namespace
+    :return: CLIargs custom object.
+    :rtype: CLIargs
     :raises argparse.ArgumentError: Unexpected arguments input.
     """
     # Configure top level argparser and its options
@@ -354,7 +389,7 @@ def parse_args(args: Any) -> argparse.Namespace:  # pylint: disable=inconsistent
         parser.exit()
 
     try:
-        parsed_args = parser.parse_args(args)
+        parsed_args = CLIargs(**vars(parser.parse_args(args)))
 
         # print help messages for an available sub-command
         if parsed_args.command == "help":

cou/steps/plan.py

@@ -37,6 +37,7 @@
     OpenStackSubordinateApplication,
     SubordinateBaseClass,
 )
+from cou.commands import CLIargs
 from cou.exceptions import (
     HaltUpgradePlanGeneration,
     HighestReleaseAchieved,
@@ -52,18 +53,13 @@
 logger = logging.getLogger(__name__)
 
 
-async def generate_plan(analysis_result: Analysis, backup_database: bool) -> UpgradePlan:
+async def generate_plan(analysis_result: Analysis, args: CLIargs) -> 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
-    :raises NoTargetError: When cannot find target to upgrade.
-    :raises HighestReleaseAchieved: When the highest possible OpenStack release is
-    already achieved.
-    :raises OutOfSupportRange: When the OpenStack release or Ubuntu series is out of the current
-    supporting range.
+    :param args: CLI arguments
+    :type args: CLIargs
     :return: Plan with all upgrade steps necessary based on the Analysis.
     :rtype: UpgradePlan
     """
@@ -89,7 +85,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",

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 CLIargs
 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 CLIargs.
+
+    :return: MagicMock of the COU CLIargs got from the cli.
+    :rtype: MagicMock
+    """
+    # spec_set needs an instantiated class to be strict with the fields.
+    return MagicMock(spec_set=CLIargs(command="plan"))()

tests/unit/steps/test_steps_plan.py

@@ -131,7 +131,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 +142,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(