test(ariadne): cover Jenkins workspace cleanup edges

ariadne/services/jenkins_workspace_cleanup.py

@@ -105,11 +105,7 @@ def _validate_cleanup_settings() -> tuple[str, str, bool, int]:
     return namespace, prefix, dry_run, max_deletions
 
 
-def _planned_removed_pv_names_dry_run(
-    stale_pvcs: list[_CleanupCandidate],
-    stale_pvs: list[_CleanupCandidate],
-    max_deletions: int,
-) -> set[str]:
+def _planned_removed_pv_names_dry_run(stale_pvcs: list[_CleanupCandidate], stale_pvs: list[_CleanupCandidate], max_deletions: int) -> set[str]:
     remaining = max(max_deletions - len(stale_pvcs), 0)
     if remaining == 0:
         return set()
@@ -117,14 +113,7 @@ def _planned_removed_pv_names_dry_run(
     return set(names[:remaining])
 
 
-def _delete_candidates(
-    candidates: list[_CleanupCandidate],
-    *,
-    deletion_budget: int | None,
-    failure_log: str,
-    failure_field: str,
-    removed_pv_names: set[str] | None = None,
-) -> tuple[int, int, int, int | None]:
+def _delete_candidates(candidates: list[_CleanupCandidate], *, deletion_budget: int | None, failure_log: str, failure_field: str, removed_pv_names: set[str] | None = None) -> tuple[int, int, int, int | None]:
     deleted = 0
     skipped = 0
     failures = 0
@@ -152,14 +141,7 @@ def _delete_candidates(
     return deleted, skipped, failures, budget
 
 
-def _record_guard_cap(
-    *,
-    max_deletions: int,
-    stale_pvcs: list[_CleanupCandidate],
-    stale_pvs: list[_CleanupCandidate],
-    stale_volumes: list[_CleanupCandidate],
-    dry_run: bool,
-) -> None:
+def _record_guard_cap(*, max_deletions: int, stale_pvcs: list[_CleanupCandidate], stale_pvs: list[_CleanupCandidate], stale_volumes: list[_CleanupCandidate], dry_run: bool) -> None:
     planned_total = len(stale_pvcs) + len(stale_pvs) + len(stale_volumes)
     if planned_total <= max_deletions:
         return
@@ -179,14 +161,7 @@ def _record_guard_cap(
     )
 
 
-def _dry_run_summary(
-    *,
-    namespace: str,
-    max_deletions: int,
-    stale_pvcs: list[_CleanupCandidate],
-    stale_pvs: list[_CleanupCandidate],
-    all_pv_names: set[str],
-) -> JenkinsWorkspaceCleanupSummary:
+def _dry_run_summary(*, namespace: str, max_deletions: int, stale_pvcs: list[_CleanupCandidate], stale_pvs: list[_CleanupCandidate], all_pv_names: set[str]) -> JenkinsWorkspaceCleanupSummary:
     simulated_removed = _planned_removed_pv_names_dry_run(stale_pvcs, stale_pvs, max_deletions)
     stale_volumes = _workspace_longhorn_candidates(settings, get_json, all_pv_names, simulated_removed)
     _record_guard_cap(
@@ -222,14 +197,7 @@ def _dry_run_summary(
     )
 
 
-def _delete_run_summary(
-    *,
-    namespace: str,
-    max_deletions: int,
-    stale_pvcs: list[_CleanupCandidate],
-    stale_pvs: list[_CleanupCandidate],
-    all_pv_names: set[str],
-) -> JenkinsWorkspaceCleanupSummary:
+def _delete_run_summary(*, namespace: str, max_deletions: int, stale_pvcs: list[_CleanupCandidate], stale_pvs: list[_CleanupCandidate], all_pv_names: set[str]) -> JenkinsWorkspaceCleanupSummary:
     removed_pv_names: set[str] = set()
     deletion_budget: int | None = max_deletions
     pvcs_deleted, pvc_skipped, pvc_failures, deletion_budget = _delete_candidates(

tests/unit/services/test_jenkins_workspace_cleanup_edges.py

@@ -0,0 +1,117 @@
+from __future__ import annotations
+
+import types
+
+import pytest
+
+from ariadne.services import jenkins_workspace_cleanup as cleanup_module
+
+
+def _settings(**overrides):
+    values = {
+        "jenkins_workspace_namespace": "jenkins",
+        "jenkins_workspace_pvc_prefix": "pvc-workspace-",
+        "jenkins_workspace_cleanup_min_age_hours": 1.0,
+        "jenkins_workspace_cleanup_dry_run": False,
+        "jenkins_workspace_cleanup_max_deletions_per_run": 10,
+    }
+    values.update(overrides)
+    return types.SimpleNamespace(**values)
+
+
+def _candidate(name: str, kind: str = "pv"):
+    return cleanup_module._CleanupCandidate(
+        name=name,
+        kind=kind,
+        path=f"/delete/{name or 'empty'}",
+        created_at=None,
+    )
+
+
+def test_cleanup_summary_properties() -> None:
+    summary = cleanup_module.JenkinsWorkspaceCleanupSummary(
+        pvs_planned=1,
+        pvcs_planned=2,
+        volumes_planned=3,
+        pvs_deleted=4,
+        pvcs_deleted=5,
+        volumes_deleted=6,
+        skipped=0,
+        failures=0,
+        dry_run=False,
+    )
+
+    assert summary.planned == 6
+    assert summary.deleted == 15
+
+
+@pytest.mark.parametrize(
+    ("overrides", "message"),
+    [
+        ({"jenkins_workspace_namespace": "  "}, "namespace is empty"),
+        ({"jenkins_workspace_pvc_prefix": "  "}, "pvc prefix is empty"),
+        ({"jenkins_workspace_cleanup_min_age_hours": 0.5}, "min age must be"),
+        ({"jenkins_workspace_cleanup_max_deletions_per_run": 0}, "max deletions must be"),
+    ],
+)
+def test_validate_cleanup_settings_rejects_bad_config(monkeypatch, overrides, message) -> None:
+    monkeypatch.setattr(cleanup_module, "settings", _settings(**overrides))
+
+    with pytest.raises(ValueError, match=message):
+        cleanup_module._validate_cleanup_settings()
+
+
+def test_planned_removed_pv_names_respects_remaining_budget() -> None:
+    pvcs = [_candidate("pvc-a", "pvc")]
+    pvs = [_candidate("pv-a"), _candidate("pv-b")]
+
+    assert cleanup_module._planned_removed_pv_names_dry_run(pvcs, pvs, max_deletions=1) == set()
+    assert cleanup_module._planned_removed_pv_names_dry_run(pvcs, pvs, max_deletions=2) == {"pv-a"}
+
+
+def test_delete_candidates_skips_empty_names_and_budget_exhaustion(monkeypatch) -> None:
+    deleted_paths: list[str] = []
+    monkeypatch.setattr(cleanup_module, "delete_json", lambda path: deleted_paths.append(path))
+    removed: set[str] = set()
+
+    deleted, skipped, failures, budget = cleanup_module._delete_candidates(
+        [_candidate(""), _candidate("pv-a"), _candidate("pv-b")],
+        deletion_budget=1,
+        failure_log="delete failed",
+        failure_field="pv",
+        removed_pv_names=removed,
+    )
+
+    assert (deleted, skipped, failures, budget) == (1, 2, 0, 0)
+    assert deleted_paths == ["/delete/pv-a"]
+    assert removed == {"pv-a"}
+
+
+def test_record_metrics_captures_skipped_and_failure_paths() -> None:
+    summary = cleanup_module.JenkinsWorkspaceCleanupSummary(
+        pvs_planned=0,
+        pvcs_planned=0,
+        volumes_planned=0,
+        pvs_deleted=0,
+        pvcs_deleted=0,
+        volumes_deleted=0,
+        skipped=2,
+        failures=1,
+        dry_run=False,
+    )
+
+    cleanup_module._record_metrics(summary)
+
+    assert cleanup_module.JENKINS_WORKSPACE_CLEANUP_LAST_SKIPPED._value.get() == 2
+    assert cleanup_module.JENKINS_WORKSPACE_CLEANUP_LAST_FAILURES._value.get() == 1
+
+
+def test_cleanup_records_metrics_before_reraising(monkeypatch) -> None:
+    monkeypatch.setattr(cleanup_module, "settings", _settings(jenkins_workspace_cleanup_dry_run=True))
+    monkeypatch.setattr(cleanup_module, "_validate_cleanup_settings", lambda: ("jenkins", "pvc-workspace-", True, 10))
+    monkeypatch.setattr(cleanup_module, "_active_workspace_claims", lambda *_args: (_ for _ in ()).throw(RuntimeError("api down")))
+
+    with pytest.raises(RuntimeError, match="api down"):
+        cleanup_module.cleanup_jenkins_workspace_storage()
+
+    assert cleanup_module.JENKINS_WORKSPACE_CLEANUP_LAST_FAILURES._value.get() >= 1