Add data-plane pre upgrade sanity checks

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 Namespace, 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: Namespace) -> 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: Namespace
     :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: Namespace) -> 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: Namespace
     """
-    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,24 +146,14 @@ 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: Namespace) -> 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: Namespace
     """
-    analysis_result, upgrade_plan = await analyze_and_plan(model_name, backup_database)
+    prompt = not args.auto_approve
+    analysis_result, upgrade_plan = await analyze_and_plan(args)
     print_and_debug(upgrade_plan)
 
     if prompt and not await continue_upgrade():
@@ -183,26 +165,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)
     manually_upgrade_data_plane(analysis_result)
     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":
-            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,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,15 +197,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 +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":

cou/exceptions.py

@@ -122,6 +122,10 @@ class WaitForApplicationsTimeout(COUException):
     """Waiting for applications hit timeout error."""
 
 
+class DataPlaneCannotUpgrade(COUException):
+    """COU exception when the cloud is inconsistent to generate a plan."""
+
+
 class InterruptError(KeyboardInterrupt):
     """COU exception when upgrade was interrupted by signal."""