test(refresh): cover the manager gate and the status reconciliation (3/6)

tests/unit/test_refresh.py

@@ -1,40 +1,47 @@
 # Copyright 2026 Canonical Ltd.
 # See LICENSE file for licensing details.
 
-"""Unit tests for the refresh module's charm-specific classes."""
+"""Unit tests for the refresh module."""
 
-from pathlib import Path
+import json
+import pathlib
 from unittest.mock import MagicMock, patch
 
+import charm_refresh
 import pytest
 from charm_refresh import CharmVersion, PrecheckFailed
+from ops import ActiveStatus, BlockedStatus, MaintenanceStatus, WaitingStatus
+from single_kernel_postgresql.config.enums import Substrates
 from single_kernel_postgresql.config.exceptions import SwitchoverFailedError
-from single_kernel_postgresql.managers.refresh import PostgreSQLRefreshK8s
+from single_kernel_postgresql.managers.refresh import (
+    PostgreSQLRefreshK8s,
+    RefreshManager,
+)
 
 CHARM_VERSION = "16/1.0.0"
 
 
 @pytest.fixture
-def refresh_versions_toml(tmp_path, monkeypatch):
-    """Provide a minimal refresh_versions.toml in the working directory.
+def charm():
+    """A mock charm with the surfaces the K8s pre-refresh checks touch."""
+    return MagicMock(name="charm")
 
-    charm_refresh reads the pinned charm and workload versions from the working
-    directory when a charm-specific class is instantiated; the repository keeps the
-    file only inside the test charms.
-    """
-    monkeypatch.chdir(tmp_path)
-    Path("refresh_versions.toml").write_text(f'charm = "{CHARM_VERSION}"\nworkload = "16.14"\n')
-    return tmp_path
+
+@pytest.fixture
+def state():
+    """A mock charm state on the K8s substrate."""
+    state = MagicMock(name="state")
+    state.substrate = Substrates.K8S
+    return state
 
 
 @pytest.fixture
-def charm():
-    """A mock charm with the surfaces the K8s pre-refresh checks touch."""
-    return MagicMock(name="charm")
+def set_default_status():
+    return MagicMock(name="set_default_status")
 
 
 @pytest.fixture
-def refresh_k8s(charm, refresh_versions_toml) -> PostgreSQLRefreshK8s:
+def refresh_k8s(charm) -> PostgreSQLRefreshK8s:
     """The K8s charm-specific refresh class wired to the mock charm."""
     return PostgreSQLRefreshK8s(
         workload_name="PostgreSQL",
@@ -44,6 +51,16 @@ def refresh_k8s(charm, refresh_versions_toml) -> PostgreSQLRefreshK8s:
     )
 
 
+@pytest.fixture
+def refresh_manager(charm, state, set_default_status) -> RefreshManager:
+    return RefreshManager(
+        state=state,
+        workload=MagicMock(name="workload"),
+        charm=charm,
+        set_default_status=set_default_status,
+    )
+
+
 @pytest.mark.parametrize(
     "old_charm,new_charm,expected",
     [
@@ -109,7 +126,7 @@ def test_pre_refresh_check_after_1_unit_refreshed_member_not_running(refresh_k8s
         "postgresql-k8s-0",
         "postgresql-k8s-2",
     ]
-    with pytest.raises(PrecheckFailed, match="PostgreSQL is not running on unit 1"):
+    with pytest.raises(PrecheckFailed, match=r"PostgreSQL is not running on unit 1"):
         refresh_k8s.run_pre_refresh_checks_after_1_unit_refreshed()
 
 
@@ -160,7 +177,7 @@ def test_pre_refresh_check_after_1_unit_refreshed_switchover_failed(refresh_k8s,
     charm.patroni_manager.get_primary.return_value = "postgresql-k8s/2"
     charm.patroni_manager.switchover.side_effect = SwitchoverFailedError
 
-    with pytest.raises(PrecheckFailed, match="Unable to switch primary"):
+    with pytest.raises(PrecheckFailed, match=r"Unable to switch primary"):
         refresh_k8s.run_pre_refresh_checks_after_1_unit_refreshed()
 
 
@@ -184,3 +201,138 @@ def test_pre_refresh_check_before_any_units_refreshed_delegates(refresh_k8s, cha
         refresh_k8s.run_pre_refresh_checks_before_any_units_refreshed()
 
     delegate.assert_called_once()
+
+
+def test_refresh_manager_constructs_the_refresh_object(refresh_manager):
+    assert refresh_manager.refresh is not None
+    assert refresh_manager.can_set_app_status is True
+
+
+def test_refresh_manager_k8s_untrusted_disables_app_status(state, charm, set_default_status):
+    state.substrate = Substrates.K8S
+    with patch("charm_refresh.Kubernetes", side_effect=charm_refresh.KubernetesJujuAppNotTrusted):
+        manager = RefreshManager(
+            state=state,
+            workload=MagicMock(),
+            charm=charm,
+            set_default_status=set_default_status,
+        )
+    assert manager.refresh is None
+    assert manager.can_set_app_status is False
+
+
+def test_refresh_manager_peer_relation_not_ready(state, charm, set_default_status):
+    state.substrate = Substrates.K8S
+    with patch("charm_refresh.Kubernetes", side_effect=charm_refresh.PeerRelationNotReady):
+        manager = RefreshManager(
+            state=state,
+            workload=MagicMock(),
+            charm=charm,
+            set_default_status=set_default_status,
+        )
+    assert manager.refresh is None
+    assert manager.can_set_app_status is True
+
+
+def test_set_unit_status_suppressed_by_higher_priority(refresh_manager, charm):
+    refresh_manager.refresh.unit_status_higher_priority = MaintenanceStatus("refreshing")
+    charm.unit.status = ActiveStatus("prior status")
+
+    refresh_manager.set_unit_status(ActiveStatus("would override"))
+
+    assert charm.unit.status == ActiveStatus("prior status")
+
+
+def test_set_unit_status_writes_lower_priority_for_active_status(refresh_manager, charm):
+    lower = ActiveStatus("PostgreSQL 16.14 running")
+    refresh_manager.refresh.unit_status_lower_priority = MagicMock(return_value=lower)
+
+    refresh_manager.set_unit_status(ActiveStatus())
+
+    assert charm.unit.status == lower
+    assert pathlib.Path(".last_refresh_unit_status.json").read_text() == json.dumps(lower.message)
+
+
+def test_set_unit_status_writes_non_active_status_directly(refresh_manager, charm):
+    refresh_manager.refresh.unit_status_lower_priority = MagicMock(
+        return_value=ActiveStatus("should not be used")
+    )
+    blocked = BlockedStatus("blocked for a reason")
+
+    refresh_manager.set_unit_status(blocked)
+
+    assert charm.unit.status == blocked
+    refresh_manager.refresh.unit_status_lower_priority.assert_not_called()
+
+
+def test_set_unit_status_without_refresh_object_sets_directly(refresh_manager, charm):
+    refresh_manager.refresh = None
+    waiting = WaitingStatus("waiting")
+
+    refresh_manager.set_unit_status(waiting)
+
+    assert charm.unit.status == waiting
+
+
+def test_set_unit_status_explicit_refresh_argument_wins(refresh_manager, charm):
+    lower = ActiveStatus("explicit refresh lower priority")
+    explicit = MagicMock(name="explicit_refresh")
+    explicit.unit_status_higher_priority = None
+    explicit.unit_status_lower_priority = MagicMock(return_value=lower)
+    refresh_manager.refresh.unit_status_lower_priority = MagicMock(
+        return_value=ActiveStatus("default refresh lower priority")
+    )
+
+    refresh_manager.set_unit_status(ActiveStatus(), refresh=explicit)
+
+    assert charm.unit.status == lower
+    refresh_manager.refresh.unit_status_lower_priority.assert_not_called()
+
+
+def test_reconcile_refresh_status_sets_higher_priority_status(refresh_manager, charm):
+    higher = MaintenanceStatus("refresh in progress")
+    refresh_manager.refresh.unit_status_higher_priority = higher
+
+    refresh_manager.reconcile_refresh_status()
+
+    assert charm.unit.status == higher
+    assert pathlib.Path(".last_refresh_unit_status.json").read_text() == json.dumps(higher.message)
+
+
+def test_reconcile_refresh_status_clears_stale_cached_status(
+    refresh_manager, charm, set_default_status
+):
+    pathlib.Path(".last_refresh_unit_status.json").write_text(json.dumps("PostgreSQL 16.14"))
+    charm.unit.status = ActiveStatus("PostgreSQL 16.14")
+    refresh_manager.refresh.unit_status_lower_priority = MagicMock(return_value=None)
+
+    refresh_manager.reconcile_refresh_status()
+
+    set_default_status.assert_called_once()
+    assert pathlib.Path(".last_refresh_unit_status.json").read_text() == json.dumps(None)
+
+
+def test_reconcile_refresh_status_restores_lower_priority_from_cached_status(
+    refresh_manager, charm
+):
+    lower = ActiveStatus("PostgreSQL 16.14 running")
+    pathlib.Path(".last_refresh_unit_status.json").write_text(json.dumps("PostgreSQL 16.14"))
+    charm.unit.status = ActiveStatus("PostgreSQL 16.14")
+    refresh_manager.refresh.unit_status_lower_priority = MagicMock(return_value=lower)
+
+    refresh_manager.reconcile_refresh_status()
+
+    assert charm.unit.status == lower
+
+
+def test_reconcile_refresh_status_ignores_unrelated_status(refresh_manager, charm):
+    pathlib.Path(".last_refresh_unit_status.json").write_text(json.dumps(None))
+    charm.unit.status = BlockedStatus("unrelated")
+    refresh_manager.refresh.unit_status_lower_priority = MagicMock(
+        return_value=ActiveStatus("nope")
+    )
+
+    refresh_manager.reconcile_refresh_status()
+
+    assert charm.unit.status == BlockedStatus("unrelated")
+    refresh_manager.refresh.unit_status_lower_priority.assert_not_called()