add base structure for test from yaml

setup.cfg

@@ -52,6 +52,7 @@ unittests =
     pytest-cov
     pytest-mock
     pytest-asyncio
+    pytest-subtests
     aiounittest
     gevent
 

tests/unit/apps_planning/test_apps_plan.py

@@ -0,0 +1,33 @@
+#  Copyright 2023 Canonical Limited
+#
+#  Licensed under the Apache License, Version 2.0 (the "License");
+#  you may not use this file except in compliance with the License.
+#  You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+#  Unless required by applicable law or agreed to in writing, software
+#  distributed under the License is distributed on an "AS IS" BASIS,
+#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#  See the License for the specific language governing permissions and
+#  limitations under the License.
+"""Test all sample plans."""
+from unittest.mock import patch
+
+import pytest
+
+from cou.commands import CLIargs
+from cou.steps.plan import generate_plan
+
+
+@pytest.mark.asyncio
+@patch("cou.steps.plan.filter_hypervisors_machines")
+async def test_sample_plans_no_inputs(_, subtests, sample_plans):
+    """Testing all sample plans."""
+    args = CLIargs("plan", auto_approve=True)
+
+    for analysis_create_coro, exp_plan, file in sample_plans:
+        with subtests.test(msg=file.name):
+            analysis_results = await analysis_create_coro
+            plan = await generate_plan(analysis_results, args)
+            assert str(plan) == exp_plan

tests/unit/conftest.py

@@ -13,26 +13,14 @@
 # limitations under the License.
 
 from pathlib import Path
+from typing import Generator
 from unittest.mock import AsyncMock, MagicMock, PropertyMock, patch
 
 import pytest
-from juju.client.client import FullStatus
 
 from cou.commands import CLIargs
-
-
-def get_status():
-    """Help function to load Juju status from json file."""
-    current_path = Path(__file__).parent.resolve()
-    with open(current_path / "jujustatus.json", "r") as file:
-        status = file.read().rstrip()
-
-    return FullStatus.from_json(status)
-
-
-async def get_charm_name(value: str):
-    """Help function to get charm name."""
-    return value
+from cou.steps.analyze import Analysis
+from tests.unit.utils import get_charm_name, get_sample_plan, get_status
 
 
 @pytest.fixture
@@ -71,3 +59,14 @@ def cli_args() -> MagicMock:
     """
     # spec_set needs an instantiated class to be strict with the fields.
     return MagicMock(spec_set=CLIargs(command="plan"))()
+
+
+@pytest.fixture
+def sample_plans(model) -> Generator[tuple[Analysis, str], None, None]:
+    """Fixture that returns all sample plans in a directory."""
+    directory = Path(__file__).parent / "sample_plans"
+    sample_plans = [
+        get_sample_plan(model, sample_file) for sample_file in directory.glob("*.yaml")
+    ]
+    model.get_applications = AsyncMock(side_effect=[apps for apps, _, _ in sample_plans])
+    return [(Analysis.create(model), exp_plan, file) for _, exp_plan, file in sample_plans]

tests/unit/sample_plans/base.yaml

@@ -0,0 +1,147 @@
+plan: |
+    Upgrade cloud from 'ussuri' to 'victoria'
+        Verify that all OpenStack applications are in idle state
+        Back up MySQL databases
+        Control Plane principal(s) upgrade plan
+            Upgrade plan for 'keystone' to 'victoria'
+                Upgrade software packages of 'keystone' from the current APT repositories
+                    Upgrade software packages on unit 'keystone/0'
+                Refresh 'keystone' to the latest revision of 'ussuri/stable'
+                Change charm config of 'keystone' 'action-managed-upgrade' to 'False'
+                Upgrade 'keystone' to the new channel: 'victoria/stable'
+                Change charm config of 'keystone' 'openstack-origin' to 'cloud:focal-victoria'
+                Wait for up to 1800s for model 'test_model' to reach the idle state
+                Verify that the workload of 'keystone' has been upgraded on units: keystone/0
+        Control Plane subordinate(s) upgrade plan
+            Upgrade plan for 'keystone-ldap' to 'victoria'
+                Refresh 'keystone-ldap' to the latest revision of 'ussuri/stable'
+                Upgrade 'keystone-ldap' to the new channel: 'victoria/stable'
+        Upgrading all applications deployed on machines with hypervisor.
+            Upgrade plan for 'az-1' to 'victoria'
+                Upgrade software packages of 'nova-compute' from the current APT repositories
+                    Upgrade software packages on unit 'nova-compute/0'
+                Refresh 'nova-compute' to the latest revision of 'ussuri/stable'
+                Change charm config of 'nova-compute' 'action-managed-upgrade' to 'True'
+                Upgrade 'nova-compute' to the new channel: 'victoria/stable'
+                Change charm config of 'nova-compute' 'source' to 'cloud:focal-victoria'
+                Upgrade plan for units: nova-compute/0
+                    Upgrade plan for unit 'nova-compute/0'
+                        Disable nova-compute scheduler from unit: 'nova-compute/0'
+                        Verify that unit 'nova-compute/0' has no VMs running
+                        ├── Pause the unit: 'nova-compute/0'
+                        ├── Upgrade the unit: 'nova-compute/0'
+                        ├── Resume the unit: 'nova-compute/0'
+                        Enable nova-compute scheduler from unit: 'nova-compute/0'
+                Wait for up to 1800s for model 'test_model' to reach the idle state
+                Verify that the workload of 'nova-compute' has been upgraded on units: nova-compute/0
+        Remaining Data Plane principal(s) upgrade plan
+            Upgrade plan for 'ceph-osd' to 'victoria'
+                Verify that all 'nova-compute' units had been upgraded
+                Upgrade software packages of 'ceph-osd' from the current APT repositories
+                    Upgrade software packages on unit 'ceph-osd/0'
+                Change charm config of 'ceph-osd' 'source' to 'cloud:focal-victoria'
+                Wait for up to 300s for app 'ceph-osd' to reach the idle state
+                Verify that the workload of 'ceph-osd' has been upgraded on units: ceph-osd/0
+        Data Plane subordinate(s) upgrade plan
+            Upgrade plan for 'ovn-chassis' to 'victoria'
+                Refresh 'ovn-chassis' to the latest revision of '22.03/stable'
+
+applications:
+  keystone:
+    can_upgrade_to: ussuri/stable
+    charm: keystone
+    channel: ussuri/stable
+    config:
+      openstack-origin:
+        value: distro
+      action-managed-upgrade:
+        value: true
+    origin: ch
+    series: focal
+    subordinate_to: []
+    workload_version: 17.0.1
+    units:
+      keystone/0:
+        name: keystone/0
+        machine: '0'
+        workload_version: 17.0.1
+        os_version: ussuri
+    machines:
+      '0':
+        id: '0'
+        apps: !!python/tuple ['keystone-ldap']
+        az: az-0
+
+  keystone-ldap:
+    can_upgrade_to: ussuri/stable
+    charm: keystone-ldap
+    channel: ussuri/stable
+    config: {}
+    origin: ch
+    series: focal
+    subordinate_to:
+    - keystone
+    workload_version: 17.0.1
+    units: {}
+    machines: {}
+
+  ceph-osd:
+    can_upgrade_to: octopus/stable
+    charm: ceph-osd
+    channel: octopus/stable
+    config:
+      source:
+        value: distro
+    origin: ch
+    series: focal
+    subordinate_to: []
+    workload_version: 17.0.1
+    units:
+      ceph-osd/0:
+        name: ceph-osd/0
+        machine: '2'
+        workload_version: 17.0.1
+        os_version: xena
+    machines:
+      '2':
+        id: '2'
+        apps: !!python/tuple []
+        az: az-0
+
+  nova-compute:
+    can_upgrade_to: ussuri/stable
+    charm: nova-compute
+    channel: ussuri/stable
+    config:
+      source:
+        value: distro
+      action-managed-upgrade:
+        value: false
+    origin: ch
+    series: focal
+    subordinate_to: []
+    workload_version: 21.0.0
+    units:
+      nova-compute/0:
+        name: nova-compute/0
+        machine: '1'
+        workload_version: 21.0.0
+        os_version: ussuri
+    machines:
+      '1':
+        id: '1'
+        apps: !!python/tuple ['ovn-chassis']
+        az: az-0
+
+  ovn-chassis:
+    can_upgrade_to: 22.03/stable
+    charm: ovn-chassis
+    channel: 22.03/stable
+    config: {}
+    origin: ch
+    series: focal
+    subordinate_to:
+    - nova-compute
+    workload_version: '22.3'
+    units: {}
+    machines: {}

tests/unit/utils.py

@@ -12,11 +12,16 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 """Module to provide helper for writing unit tests."""
+from pathlib import Path
 from textwrap import dedent
 from unittest.mock import MagicMock
 
+import yaml
+from juju.client.client import FullStatus
+
+from cou.apps.base import OpenStackApplication
 from cou.steps import BaseStep
-from cou.utils.juju_utils import Machine
+from cou.utils.juju_utils import Application, Machine, Model, Unit
 
 
 def assert_steps(step_1: BaseStep, step_2: BaseStep) -> None:
@@ -38,3 +43,73 @@ def dedent_plan(plan: str) -> str:
     result = dedent(plan)
     result = result.replace("    ", "\t")  # replace 4 spaces with tap
     return result
+
+
+def get_status():
+    """Help function to load Juju status from json file."""
+    current_path = Path(__file__).parent.resolve()
+    with open(current_path / "jujustatus.json", "r") as file:
+        status = file.read().rstrip()
+
+    return FullStatus.from_json(status)
+
+
+async def get_charm_name(value: str):
+    """Help function to get charm name."""
+    return value
+
+
+def get_sample_plan(
+    model: Model, source: Path
+) -> tuple[dict[str, OpenStackApplication], str, Path]:
+    """Help function to get dict of Applications and expected upgrade plan from file.
+
+    This function can load applications from yaml format, where each app is string representation
+    of OpenStackApplication (str(OpenStackApplication), see OpenStackApplication.__str__).
+
+    applications:
+      <app_1_name>:
+        model_name: ...
+        can_upgrade_to: ...
+        ...
+      <app_1_name>:
+        model_name: ...
+        can_upgrade_to: ...
+        ...
+    plan: |
+      ...
+    """
+    with open(source, "r") as file:
+        data = yaml.load(file, Loader=yaml.Loader)
+
+    # Note(rgildein): We need to get machines first, since they are used in Unit object.
+    machines = {
+        machine_id: Machine(machine["id"], machine["apps"], machine["az"])
+        for app_data in data["applications"].values()
+        for machine_id, machine in app_data["machines"].items()
+    }
+
+    return (
+        {
+            name: Application(
+                name=name,
+                can_upgrade_to=app_data["can_upgrade_to"],
+                charm=app_data["charm"],
+                channel=app_data["channel"],
+                config=app_data["config"],
+                machines={machine_id: machines[machine_id] for machine_id in app_data["machines"]},
+                model=model,
+                origin=app_data["origin"],
+                series=app_data["series"],
+                subordinate_to=app_data["subordinate_to"],
+                units={
+                    name: Unit(name, machines[unit["machine"]], unit["workload_version"])
+                    for name, unit in app_data["units"].items()
+                },
+                workload_version=app_data["workload_version"],
+            )
+            for name, app_data in data["applications"].items()
+        },
+        dedent_plan(data["plan"]),
+        source,
+    )