- added unit tests

cou/steps/analyze.py

@@ -162,6 +162,9 @@ def __str__(self) -> str:
     def min_os_release_apps(apps: list[OpenStackApplication]) -> Optional[OpenStackRelease]:
         """Get the minimal OpenStack release from a list of applications.
 
+        - subordinates or channel based apps are not considered if not using release channels
+        - other apps are considered even if not using release channels
+
         :param apps: Applications.
         :type apps: list[OpenStackApplication]
         :return: OpenStack release.

cou/utils/app_utils.py

@@ -15,7 +15,7 @@
 """Application utilities."""
 import json
 import logging
-from typing import Iterable, Optional, Protocol, TypeVar
+from typing import Any, Iterable, Optional
 
 from cou.exceptions import RunUpgradeError
 from cou.utils.juju_utils import Model
@@ -135,22 +135,13 @@ async def _get_current_osd_release(unit: str, model: Model) -> str:
     return current_osd_release
 
 
-class NamedClass(Protocol):  # pylint: disable=too-few-public-methods
-    """Class to help lint allowing stringify any object containing name."""
-
-    name: str
-
-
-NM = TypeVar("NM", bound=NamedClass)
-
-
-def stringify_class(instances: Iterable[NM]) -> str:
+def stringify_class(instances: Iterable[Any]) -> str:
     """Convert any class with name attribute into a comma-separated string of names, sorted.
 
-    :param instances: An iterable of NamedClass objects to be converted.
+    :param instances: An iterable of objects to be converted that has name attribute.
     :type instances: Iterable[NamedClass]
     :return: A comma-separated string of sorted unit names.
     :rtype: str
     """
-    sorted_names = sorted([instance.name for instance in instances])
+    sorted_names = sorted([instance.name for instance in instances if hasattr(instance, "name")])
     return ", ".join(sorted_names)

tests/unit/apps/test_base.py

@@ -541,6 +541,7 @@ def test_get_charmhub_migration_step(current_os_release, model):
 def test_get_change_to_openstack_channels_step(current_os_release, model):
     """Applications using latest/stable should be switched to a release-specific channel."""
     current_os_release.return_value = OpenStackRelease("ussuri")
+    target = OpenStackRelease("victoria")
 
     app = OpenStackApplication(
         name="app",
@@ -556,7 +557,7 @@ def test_get_change_to_openstack_channels_step(current_os_release, model):
         units={},
         workload_version="1",
     )
-    assert app._get_change_to_openstack_channels_step() == PreUpgradeStep(
+    assert app._get_change_to_openstack_channels_step(target) == PreUpgradeStep(
         f"WARNING: Changing '{app.name}' channel from {app.channel} to "
         f"{app.expected_current_channel}. This may be a charm downgrade, "
         "which is generally not supported.",
@@ -696,7 +697,7 @@ def test_get_refresh_charm_step_change_to_openstack_channels(
     assert app._get_refresh_charm_step(target) == expected_result
 
     mock_ch_migration.assert_not_called()
-    mock_change_os_channels.assert_called_once()
+    mock_change_os_channels.assert_called_once_with(target)
     mock_refresh_current_channel.assert_not_called()
 
 
@@ -772,3 +773,27 @@ def test_need_current_channel_refresh(model, can_upgrade_to, channel, exp_result
         workload_version="1",
     )
     assert app._need_current_channel_refresh(target) is exp_result
+
+
+@pytest.mark.parametrize(
+    "channel, origin, exp_result",
+    [("latest/stable", "ch", False), ("ussuri/stable", "ch", True), ("stable", "cs", False)],
+)
+def test_using_release_channel(model, channel, origin, exp_result):
+    """Test if application is using release channel."""
+    app = OpenStackApplication(
+        name="app",
+        can_upgrade_to="",
+        charm="app",
+        channel=channel,
+        config={},
+        machines={},
+        model=model,
+        origin=origin,
+        series="focal",
+        subordinate_to=[],
+        units={},
+        workload_version="1",
+    )
+
+    assert app.using_release_channel is exp_result

tests/unit/apps/test_channel_based.py

@@ -23,7 +23,7 @@
 from cou.utils import app_utils
 from cou.utils.juju_utils import Machine, Unit
 from cou.utils.openstack import OpenStackRelease
-from tests.unit.utils import assert_steps
+from tests.unit.utils import assert_steps, dedent_plan
 
 
 def test_application_versionless(model):
@@ -61,7 +61,20 @@ def test_application_versionless(model):
 
 def test_channel_based_application_latest_stable(model):
     """Test application without version."""
-    target = OpenStackRelease("victoria")
+    target = OpenStackRelease("wallaby")
+
+    exp_plan = dedent_plan(
+        """\
+        Upgrade plan for 'glance-simplestreams-sync' to 'wallaby'
+            Upgrade software packages of 'glance-simplestreams-sync' from the current APT repositories
+                Upgrade software packages on unit 'glance-simplestreams-sync/0'
+            WARNING: Changing 'glance-simplestreams-sync' channel from latest/stable to victoria/stable. \
+This may be a charm downgrade, which is generally not supported.
+            Upgrade 'glance-simplestreams-sync' to the new channel: 'wallaby/stable'
+            Change charm config of 'glance-simplestreams-sync' 'openstack-origin' to 'cloud:focal-wallaby'
+    """  # noqa: E501 line too long
+    )
+
     machines = {"0": MagicMock(spec_set=Machine)}
     units = {
         "glance-simplestreams-sync/0": Unit(
@@ -86,9 +99,12 @@ def test_channel_based_application_latest_stable(model):
         units=units,
         workload_version="",
     )
+    # app is considered ussuri because is using latest/stable, but it won't be considered to
+    # calculate the cloud minimum OpenStack release. It will refresh the charm channel to whatever
+    # the minimum version of other components.
+    assert app.current_os_release == "ussuri"
     plan = app.generate_upgrade_plan(target, False)
-    print(plan)
-    assert 1 == 0
+    assert str(plan) == exp_plan
 
 
 def test_application_gnocchi_ussuri(model):

tests/unit/steps/test_analyze.py

@@ -18,6 +18,7 @@
 
 from cou.apps.auxiliary import RabbitMQServer
 from cou.apps.base import OpenStackApplication
+from cou.apps.channel_based import ChannelBasedApplication
 from cou.apps.core import Keystone
 from cou.apps.subordinate import SubordinateApplication
 from cou.steps import analyze
@@ -536,3 +537,153 @@ def test_split_apps(exp_control_plane, exp_data_plane):
     control_plane, data_plane = Analysis._split_apps(all_apps)
     assert exp_control_plane == control_plane
     assert exp_data_plane == data_plane
+
+
+def test_min_os_release_apps_release_channel(model):
+    """Test to evaluate the Openstack release using release channels on apps."""
+    machines = {f"{i}": generate_cou_machine(f"{i}") for i in range(3)}
+    keystone = Keystone(
+        name="keystone",
+        can_upgrade_to="",
+        charm="keystone",
+        channel="wallaby/stable",
+        config={"source": {"value": "cloud:focal-wallaby"}},
+        machines=machines,
+        model=model,
+        origin="ch",
+        series="focal",
+        subordinate_to=[],
+        units={
+            "keystone/0": Unit(
+                name="keystone/0",
+                workload_version="19.1.0",
+                machine=machines["0"],
+            )
+        },
+        workload_version="19.1.0",
+    )
+
+    ch_subordinate_using_release = SubordinateApplication(
+        name="keystone-ldap",
+        can_upgrade_to="",
+        charm="keystone-ldap",
+        channel="victoria/stable",
+        config={},
+        machines=machines,
+        model=model,
+        origin="ch",
+        series="focal",
+        subordinate_to=["keystone"],
+        units={},
+        workload_version="1",
+    )
+
+    channel_app_using_release = ChannelBasedApplication(
+        name="channel_app_using_release",
+        can_upgrade_to="",
+        charm="channel_app_using_release",
+        channel="ussuri/stable",
+        config={},
+        machines=machines,
+        model=model,
+        origin="ch",
+        series="focal",
+        subordinate_to=[],
+        units={
+            "channel_app_using_release": Unit(
+                name="channel_app_using_release",
+                workload_version="",
+                machine=machines["1"],
+            )
+        },
+        workload_version="",
+    )
+
+    assert keystone.current_os_release == "wallaby"
+    assert ch_subordinate_using_release.current_os_release == "victoria"
+    assert channel_app_using_release.current_os_release == "ussuri"
+
+    # channel_app_using_release and ch_subordinate_using_release are considered because are
+    # using release channel
+    assert (
+        Analysis.min_os_release_apps(
+            [keystone, channel_app_using_release, ch_subordinate_using_release]
+        )
+        == "ussuri"
+    )
+
+    assert Analysis.min_os_release_apps([keystone, ch_subordinate_using_release]) == "victoria"
+
+
+def test_min_os_release_apps_not_release_channel(model):
+    """Test to evaluate the Openstack release not using release channels on apps."""
+    machines = {f"{i}": generate_cou_machine(f"{i}") for i in range(3)}
+    keystone_latest_stable = Keystone(
+        name="keystone",
+        can_upgrade_to="",
+        charm="keystone",
+        channel="latest/stable",
+        config={"source": {"value": "cloud:focal-wallaby"}},
+        machines=machines,
+        model=model,
+        origin="ch",
+        series="focal",
+        subordinate_to=[],
+        units={
+            "keystone/0": Unit(
+                name="keystone/0",
+                workload_version="19.1.0",
+                machine=machines["0"],
+            )
+        },
+        workload_version="19.1.0",
+    )
+
+    cs_subordinate = SubordinateApplication(
+        name="keystone-ldap",
+        can_upgrade_to="",
+        charm="keystone-ldap",
+        channel="stable",
+        config={},
+        machines=machines,
+        model=model,
+        origin="cs",
+        series="focal",
+        subordinate_to=["keystone"],
+        units={},
+        workload_version="1",
+    )
+
+    channel_app_latest_stable = ChannelBasedApplication(
+        name="channel_app_latest_stable",
+        can_upgrade_to="",
+        charm="channel_app_latest_stable",
+        channel="latest/stable",
+        config={},
+        machines=machines,
+        model=model,
+        origin="ch",
+        series="focal",
+        subordinate_to=[],
+        units={
+            "channel_app_latest_stable/0": Unit(
+                name="channel_app_latest_stable/0",
+                workload_version="",
+                machine=machines["1"],
+            )
+        },
+        workload_version="",
+    )
+
+    assert keystone_latest_stable.current_os_release == "wallaby"
+    assert cs_subordinate.current_os_release == "ussuri"
+    assert channel_app_latest_stable.current_os_release == "ussuri"
+
+    # channel_app_latest_stable is skipped because is using latest/stable
+    # cs_subordinate is disconsidered because is from charmstore
+    assert (
+        Analysis.min_os_release_apps(
+            [keystone_latest_stable, channel_app_latest_stable, cs_subordinate]
+        )
+        == "wallaby"
+    )