Sane checks for cli inputs

- introduce machine class
- sane checks for machine, hostname and azs filtration

cou/apps/base.py

@@ -54,8 +54,36 @@ class ApplicationUnit:
 
     name: str
     os_version: OpenStackRelease
+    machine: Machine
     workload_version: str = ""
-    machine: str = ""
+
+
+@dataclass
+class Machine:
+    """Representation of a juju machine."""
+
+    machine_id: str
+    hostname: str
+    az: Optional[str]
+    is_data_plane: bool = False
+
+    def __hash__(self) -> int:
+        """Hash magic method for Machine.
+
+        :return: Unique hash identifier for Machine object.
+        :rtype: int
+        """
+        return hash(self.machine_id)
+
+    def __eq__(self, other: Any) -> bool:
+        """Equal magic method for Application.
+
+        :param other: Application object to compare.
+        :type other: Any
+        :return: True if equal False if different.
+        :rtype: bool
+        """
+        return other.machine_id == self.machine_id
 
 
 @dataclass
@@ -99,6 +127,7 @@ class OpenStackApplication:
     config: dict
     model: COUModel
     charm: str
+    machines: dict[str, Machine]
     charm_origin: str = ""
     os_origin: str = ""
     origin_setting: Optional[str] = None
@@ -169,7 +198,7 @@ def _populate_units(self) -> None:
                         name=name,
                         workload_version=unit.workload_version,
                         os_version=compatible_os_version,
-                        machine=unit.machine,
+                        machine=self.machines[unit.machine],
                     )
                 )
 

cou/apps/factory.py

@@ -21,7 +21,7 @@
 
 from juju.client._definitions import ApplicationStatus
 
-from cou.apps.base import OpenStackApplication
+from cou.apps.base import Machine, OpenStackApplication
 from cou.utils.juju_utils import COUModel
 from cou.utils.openstack import is_charm_supported
 
@@ -41,6 +41,7 @@ def create(
         config: dict,
         model: COUModel,
         charm: str,
+        machines: dict[str, Machine],
     ) -> Optional[OpenStackApplication]:
         """Create the OpenStackApplication or registered subclasses.
 
@@ -56,13 +57,22 @@ def create(
         :type model: COUModel
         :param charm: Name of the charm
         :type charm: str
+        :param machines: Machines in the model
+        :type machines: dict[str, Machine]
         :return: The OpenStackApplication class or None if not supported.
         :rtype: Optional[OpenStackApplication]
         """
         # pylint: disable=too-many-arguments
         if is_charm_supported(charm):
             app_class = cls.charms.get(charm, OpenStackApplication)
-            return app_class(name=name, status=status, config=config, model=model, charm=charm)
+            return app_class(
+                name=name,
+                status=status,
+                config=config,
+                model=model,
+                charm=charm,
+                machines=machines,
+            )
         logger.debug(
             "'%s' is not a supported OpenStack related application and will be ignored.",
             name,

cou/cli.py

@@ -130,7 +130,7 @@ async def analyze_and_plan(args: argparse.Namespace) -> tuple[Analysis, UpgradeP
     progress_indicator.succeed()
 
     progress_indicator.start("Making sane checks to upgrade...")
-    pre_plan_sane_checks(args.upgrade_group, analysis_result)
+    pre_plan_sane_checks(args, analysis_result)
     progress_indicator.succeed()
 
     progress_indicator.start("Generating upgrade plan...")

cou/exceptions.py

@@ -67,6 +67,10 @@ class RunUpgradeError(COUException):
     """Exception raised when an upgrade fails."""
 
 
+class DataPlaneMachineFilterError(COUException):
+    """Exception raised when filtering data-plane machines fails."""
+
+
 class ActionFailed(COUException):
     """Exception raised when action fails."""
 

cou/steps/analyze.py

@@ -17,11 +17,14 @@
 
 import logging
 from dataclasses import dataclass, field
-from typing import Optional
+from typing import Any, Optional
 
-from cou.apps.base import OpenStackApplication
+from juju.client._definitions import ApplicationStatus, MachineStatus
+
+from cou.apps.base import Machine, OpenStackApplication
 from cou.apps.factory import AppFactory
 from cou.utils import juju_utils
+from cou.utils.juju_utils import COUModel
 from cou.utils.openstack import DATA_PLANE_CHARMS, UPGRADE_ORDER, OpenStackRelease
 
 logger = logging.getLogger(__name__)
@@ -37,11 +40,16 @@ class Analysis:
     :type apps_control_plane:  list[OpenStackApplication]
     :param apps_data_plane: Data plane applications in the model
     :type apps_data_plane:  list[OpenStackApplication]
+    :param machines: Machines in the model
+    :type machines: dict[str, Machine]
     """
 
+    # pylint: disable=too-many-instance-attributes
+
     model: juju_utils.COUModel
     apps_control_plane: list[OpenStackApplication]
     apps_data_plane: list[OpenStackApplication]
+    machines: dict[str, Machine]
     min_os_version_control_plane: Optional[OpenStackRelease] = None
     min_os_version_data_plane: Optional[OpenStackRelease] = None
 
@@ -56,7 +64,19 @@ def __post_init__(self) -> None:
         self.current_cloud_series = self._get_minimum_cloud_series()
 
     @staticmethod
+    def is_data_plane(charm: str) -> bool:
+        """Check if app belong to data plane.
+
+        :param charm: charm name
+        :type charm: str
+        :return: boolean
+        :rtype: bool
+        """
+        return charm in DATA_PLANE_CHARMS
+
+    @classmethod
     def _split_apps(
+        cls,
         apps: list[OpenStackApplication],
     ) -> tuple[list[OpenStackApplication], list[OpenStackApplication]]:
         """Split applications to control plane and data plane apps.
@@ -66,25 +86,9 @@ def _split_apps(
         :return: Control plane and data plane application lists.
         :rtype: tuple[list[OpenStackApplication], list[OpenStackApplication]]
         """
-
-        def is_data_plane(app: OpenStackApplication) -> bool:
-            """Check if app belong to data plane.
-
-            :param app: application
-            :type app: OpenStackApplication
-            :return: boolean
-            :rtype: bool
-            """
-            return app.charm in DATA_PLANE_CHARMS
-
         control_plane, data_plane = [], []
-        data_plane_machines = {
-            unit.machine for app in apps if is_data_plane(app) for unit in app.units
-        }
         for app in apps:
-            if is_data_plane(app):
-                data_plane.append(app)
-            elif any(unit.machine in data_plane_machines for unit in app.units):
+            if any(unit.machine.is_data_plane for unit in app.units):
                 data_plane.append(app)
             else:
                 control_plane.append(app)
@@ -101,14 +105,22 @@ async def create(cls, model: juju_utils.COUModel) -> Analysis:
         :rtype: Analysis
         """
         logger.info("Analyzing the OpenStack deployment...")
-        apps = await Analysis._populate(model)
+        machines = await cls.get_machines(model)
+        apps = await Analysis._populate(model, machines)
 
         control_plane, data_plane = cls._split_apps(apps)
 
-        return Analysis(model=model, apps_data_plane=data_plane, apps_control_plane=control_plane)
+        return Analysis(
+            model=model,
+            apps_data_plane=data_plane,
+            apps_control_plane=control_plane,
+            machines=machines,
+        )
 
     @classmethod
-    async def _populate(cls, model: juju_utils.COUModel) -> list[OpenStackApplication]:
+    async def _populate(
+        cls, model: juju_utils.COUModel, machines: dict[str, Machine]
+    ) -> list[OpenStackApplication]:
         """Analyze the applications in the model.
 
         Applications that must be upgraded in a specific order will be returned first, followed
@@ -117,6 +129,8 @@ async def _populate(cls, model: juju_utils.COUModel) -> list[OpenStackApplicatio
 
         :param model: COUModel object
         :type model: COUModel
+        :param machines: Machines in the model
+        :type machines: dict[str, Machine]
         :return: Application objects with their respective information.
         :rtype: List[OpenStackApplication]
         """
@@ -128,6 +142,7 @@ async def _populate(cls, model: juju_utils.COUModel) -> list[OpenStackApplicatio
                 config=await model.get_application_config(app),
                 model=model,
                 charm=await model.get_charm_name(app),
+                machines=cls.get_app_machines(app_status, machines),
             )
             for app, app_status in juju_status.applications.items()
             if app_status
@@ -150,6 +165,63 @@ async def _populate(cls, model: juju_utils.COUModel) -> list[OpenStackApplicatio
         )
         return sorted_apps_to_upgrade_in_order + other_o7k_apps_sorted_by_name  # type: ignore
 
+    @classmethod
+    async def get_machines(cls, model: COUModel) -> dict[str, Machine]:
+        """Get all the machines in the model.
+
+        :param model: COUModel object
+        :type model: _type_
+        :return: _description_
+        :rtype: dict[str, Machine]
+        """
+        juju_status = await model.get_status()
+        data_plane_machines = {
+            unit.machine
+            for app in juju_status.applications
+            if cls.is_data_plane(await model.get_charm_name(app))
+            for unit in app.units
+        }
+        machines = {}
+        for machine_id, raw_machine_data in juju_status.machines.items():
+            machine_data = cls.get_machine_data(raw_machine_data)
+            machines[machine_id] = Machine(
+                machine_id=machine_id,
+                hostname=machine_data["hostname"],
+                az=machine_data["az"],
+                is_data_plane=id in data_plane_machines,
+            )
+        return machines
+
+    @classmethod
+    def get_app_machines(
+        cls, app_status: ApplicationStatus, machines: dict[str, Machine]
+    ) -> dict[str, Machine]:
+        """Get the machines of an app.
+
+        :param app_status: Status of the application.
+        :type app_status: ApplicationStatus
+        :param machines: Machines in the model
+        :type machines: dict[str, Machine]
+        :return: Machines in the application
+        :rtype: dict[str, Machine]
+        """
+        return {
+            unit_status.machine: machines[unit_status.machine]
+            for unit_status in app_status.units.values()
+        }
+
+    @staticmethod
+    def get_machine_data(machine: MachineStatus) -> dict[str, Any]:
+        """Get the data of a machine.
+
+        :param machine: Machine status from juju
+        :type machine: MachineStatus
+        :return: Machine data formatted
+        :rtype: dict[str, Any]
+        """
+        hardware = dict(entry.split("=") for entry in machine["hardware"].split())
+        return {"az": hardware.get("availability-zone"), "hostname": machine["hostname"]}
+
     def __str__(self) -> str:
         """Dump as string.