Custom COU Namespace (canonical#221)

- create a new dataclass to be the COU Namespace. With that, type
is still checked and few arguments are necessary to pass to the
functions that use the arguments passed by the user. Moreover,
docstrings are shorter and easier to maintain and frozen don't let
accidentally change the object. 

- Global variables for "data-plane" and "control-plane".

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(