test(ariadne): cover Jenkins workspace candidate filters

ariadne/services/jenkins_workspace_candidates.py

@@ -88,11 +88,7 @@ def _active_workspace_claims(settings_obj: Any, get_json_func: Callable[[str], d
     return active
 
 
-def _workspace_pv_candidates(
+def _workspace_pv_candidates(settings_obj: Any, get_json_func: Callable[[str], dict[str, Any]], active_claims: set[str]) -> tuple[list[_CleanupCandidate], set[str]]:
-    settings_obj: Any,
-    get_json_func: Callable[[str], dict[str, Any]],
-    active_claims: set[str],
-) -> tuple[list[_CleanupCandidate], set[str]]:
     """Find releasable Jenkins workspace PVs and keep a set of all PV names."""
 
     namespace = settings_obj.jenkins_workspace_namespace
@@ -141,11 +137,7 @@ def _workspace_pv_candidates(
     return candidates, all_pv_names
 
 
-def _workspace_pvc_candidates(
+def _workspace_pvc_candidates(settings_obj: Any, get_json_func: Callable[[str], dict[str, Any]], active_claims: set[str]) -> list[_CleanupCandidate]:
-    settings_obj: Any,
-    get_json_func: Callable[[str], dict[str, Any]],
-    active_claims: set[str],
-) -> list[_CleanupCandidate]:
     """Find stale Jenkins workspace PVCs that are not actively referenced."""
 
     namespace = settings_obj.jenkins_workspace_namespace
@@ -183,10 +175,7 @@ def _workspace_pvc_candidates(
     return candidates
 
 
-def _workspace_binding_from_longhorn(
+def _workspace_binding_from_longhorn(metadata: dict[str, Any], status: dict[str, Any]) -> _LonghornBinding:
-    metadata: dict[str, Any],
-    status: dict[str, Any],
-) -> _LonghornBinding:
     labels = metadata.get("labels") if isinstance(metadata.get("labels"), dict) else {}
     kubernetes_status = status.get("kubernetesStatus") if isinstance(status.get("kubernetesStatus"), dict) else {}
     pvc_name = labels.get("kubernetes.io/created-for/pvc/name")
@@ -203,13 +192,7 @@ def _workspace_binding_from_longhorn(
     )
 
 
-def _should_delete_longhorn_volume(
+def _should_delete_longhorn_volume(settings_obj: Any, name: str, binding: _LonghornBinding, all_pv_names: set[str], removed_pv_names: set[str]) -> bool:
-    settings_obj: Any,
-    name: str,
-    binding: _LonghornBinding,
-    all_pv_names: set[str],
-    removed_pv_names: set[str],
-) -> bool:
     if name in removed_pv_names or binding.referenced_pv_name in removed_pv_names:
         return True
     if not _is_workspace_name(settings_obj, binding.pvc_name):
@@ -225,12 +208,7 @@ def _should_delete_longhorn_volume(
     )
 
 
-def _workspace_longhorn_candidates(
+def _workspace_longhorn_candidates(settings_obj: Any, get_json_func: Callable[[str], dict[str, Any]], all_pv_names: set[str], removed_pv_names: set[str]) -> list[_CleanupCandidate]:
-    settings_obj: Any,
-    get_json_func: Callable[[str], dict[str, Any]],
-    all_pv_names: set[str],
-    removed_pv_names: set[str],
-) -> list[_CleanupCandidate]:
     namespace = "longhorn-system"
     payload = get_json_func("/apis/longhorn.io/v1beta2/namespaces/longhorn-system/volumes")
     items = payload.get("items") if isinstance(payload.get("items"), list) else []

tests/unit/services/test_jenkins_workspace_candidates_edges.py

@@ -0,0 +1,170 @@
+from __future__ import annotations
+
+from datetime import datetime, timezone
+import types
+
+from ariadne.services import jenkins_workspace_candidates as candidates
+
+
+OLD_TS = "2020-01-01T00:00:00Z"
+
+
+def _settings(**overrides):
+    values = {
+        "jenkins_workspace_namespace": "jenkins",
+        "jenkins_workspace_pvc_prefix": "pvc-workspace-",
+        "jenkins_workspace_cleanup_min_age_hours": 1.0,
+    }
+    values.update(overrides)
+    return types.SimpleNamespace(**values)
+
+
+def _payload(items):
+    return {"items": items}
+
+
+def test_timestamp_and_age_helpers_handle_invalid_metadata() -> None:
+    settings = _settings()
+
+    assert candidates._parse_timestamp("not-a-date") is None
+    assert candidates._created_at({"creationTimestamp": ""}) is None
+    assert candidates._created_at({"creationTimestamp": 123}) is None
+    assert candidates._is_old_enough(settings, {}) is False
+    assert candidates._is_deleting({"deletionTimestamp": "  "}) is False
+    assert candidates._is_deleting({"deletionTimestamp": OLD_TS}) is True
+
+
+def test_active_workspace_claims_ignores_malformed_pods() -> None:
+    settings = _settings()
+
+    def get_json(_path: str):
+        return _payload(
+            [
+                None,
+                {
+                    "metadata": {"annotations": {"jenkins.io/workspace-pvc": "pvc-workspace-annotated"}},
+                    "spec": {
+                        "volumes": [
+                            None,
+                            {"emptyDir": {}},
+                            {"persistentVolumeClaim": None},
+                            {"persistentVolumeClaim": {"claimName": "not-workspace"}},
+                            {"persistentVolumeClaim": {"claimName": "pvc-workspace-mounted"}},
+                        ]
+                    },
+                },
+            ]
+        )
+
+    assert candidates._active_workspace_claims(settings, get_json) == {
+        "pvc-workspace-annotated",
+        "pvc-workspace-mounted",
+    }
+
+
+def test_workspace_pv_candidates_filter_every_skip_reason() -> None:
+    settings = _settings()
+    fresh_ts = datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")
+
+    def get_json(_path: str):
+        return _payload(
+            [
+                None,
+                {"metadata": {"name": "pv-wrong-ns", "creationTimestamp": OLD_TS}, "status": {"phase": "Released"}, "spec": {"claimRef": {"namespace": "other", "name": "pvc-workspace-wrong"}}},
+                {"metadata": {"name": "pv-not-workspace", "creationTimestamp": OLD_TS}, "status": {"phase": "Released"}, "spec": {"claimRef": {"namespace": "jenkins", "name": "data"}}},
+                {"metadata": {"name": "pv-active", "creationTimestamp": OLD_TS}, "status": {"phase": "Released"}, "spec": {"claimRef": {"namespace": "jenkins", "name": "pvc-workspace-active"}}},
+                {"metadata": {"name": "pv-deleting", "creationTimestamp": OLD_TS, "deletionTimestamp": OLD_TS}, "status": {"phase": "Released"}, "spec": {"claimRef": {"namespace": "jenkins", "name": "pvc-workspace-deleting"}}},
+                {"metadata": {"name": "pv-bound", "creationTimestamp": OLD_TS}, "status": {"phase": "Bound"}, "spec": {"claimRef": {"namespace": "jenkins", "name": "pvc-workspace-bound"}}},
+                {"metadata": {"name": "pv-fresh", "creationTimestamp": fresh_ts}, "status": {"phase": "Released"}, "spec": {"claimRef": {"namespace": "jenkins", "name": "pvc-workspace-fresh"}}},
+                {"metadata": {"creationTimestamp": OLD_TS}, "status": {"phase": "Released"}, "spec": {"claimRef": {"namespace": "jenkins", "name": "pvc-workspace-no-name"}}},
+                {"metadata": {"name": "pv-stale", "creationTimestamp": OLD_TS}, "status": {"phase": "Failed"}, "spec": {"claimRef": {"namespace": "jenkins", "name": "pvc-workspace-stale"}}},
+            ]
+        )
+
+    found, all_names = candidates._workspace_pv_candidates(settings, get_json, {"pvc-workspace-active"})
+
+    assert {item.name for item in found} == {"pv-stale"}
+    assert {"pv-wrong-ns", "pv-not-workspace", "pv-active", "pv-stale"} <= all_names
+
+
+def test_workspace_pvc_candidates_filter_edges() -> None:
+    fresh_ts = datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")
+
+    def normal_payload(_path: str):
+        return _payload(
+            [
+                None,
+                {"metadata": {"name": "data", "creationTimestamp": OLD_TS}, "status": {"phase": "Lost"}},
+                {"metadata": {"name": "pvc-workspace-deleting", "creationTimestamp": OLD_TS, "deletionTimestamp": OLD_TS}, "status": {"phase": "Lost"}},
+                {"metadata": {"name": "pvc-workspace-active", "creationTimestamp": OLD_TS}, "status": {"phase": "Lost"}},
+                {"metadata": {"name": "pvc-workspace-bound", "creationTimestamp": OLD_TS}, "status": {"phase": "Bound"}},
+                {"metadata": {"name": "pvc-workspace-fresh", "creationTimestamp": fresh_ts}, "status": {"phase": "Lost"}},
+                {"metadata": {"name": "pvc-workspace-stale", "creationTimestamp": OLD_TS}, "status": {"phase": "Lost"}},
+            ]
+        )
+
+    found = candidates._workspace_pvc_candidates(_settings(), normal_payload, {"pvc-workspace-active"})
+    assert [item.name for item in found] == ["pvc-workspace-stale"]
+
+    def empty_name_payload(_path: str):
+        return _payload([{"metadata": {"name": "", "creationTimestamp": OLD_TS}, "status": {"phase": "Lost"}}])
+
+    assert candidates._workspace_pvc_candidates(_settings(jenkins_workspace_pvc_prefix=""), empty_name_payload, []) == []
+
+
+def test_longhorn_binding_and_deletion_policy_edges() -> None:
+    settings = _settings()
+    binding = candidates._workspace_binding_from_longhorn(
+        {"labels": {"kubernetes.io/created-for/pvc/name": "pvc-workspace-label"}},
+        {"kubernetesStatus": {"namespace": "jenkins", "pvName": "pv-label"}},
+    )
+    assert binding.pvc_name == "pvc-workspace-label"
+    assert binding.pvc_namespace == "jenkins"
+
+    assert candidates._should_delete_longhorn_volume(settings, "vol", binding, set(), {"vol"}) is True
+    assert candidates._should_delete_longhorn_volume(settings, "vol", candidates._LonghornBinding("data", "jenkins", ""), set(), set()) is False
+    assert candidates._should_delete_longhorn_volume(settings, "vol", candidates._LonghornBinding("pvc-workspace-a", "jenkins", "pv-a"), {"pv-a"}, set()) is False
+    assert candidates._should_delete_longhorn_volume(settings, "vol", candidates._LonghornBinding("pvc-workspace-a", "", ""), set(), set()) is True
+
+
+def test_workspace_longhorn_candidates_filter_edges() -> None:
+    settings = _settings()
+    fresh_ts = datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")
+
+    def volume(name: str, **overrides):
+        item = {
+            "metadata": {
+                "name": name,
+                "creationTimestamp": OLD_TS,
+                "labels": {
+                    "kubernetes.io/created-for/pvc/name": "pvc-workspace-orphan",
+                    "kubernetes.io/created-for/pvc/namespace": "jenkins",
+                },
+            },
+            "status": {"state": "detached", "isAttached": False, "robustness": "unknown"},
+            "spec": {"frontend": "blockdev"},
+        }
+        for key, value in overrides.items():
+            item[key].update(value)
+        return item
+
+    def get_json(_path: str):
+        return _payload(
+            [
+                None,
+                {"metadata": {"creationTimestamp": OLD_TS}},
+                volume("vol-active-pv", status={"kubernetesStatus": {"pvName": "pv-present"}}),
+                volume("vol-deleting", metadata={"deletionTimestamp": OLD_TS}),
+                volume("vol-fresh", metadata={"creationTimestamp": fresh_ts}),
+                volume("vol-attached-state", status={"state": "attached"}),
+                volume("vol-attached-flag", status={"isAttached": True}),
+                volume("vol-healthy", status={"robustness": "healthy"}),
+                volume("vol-frontend", spec={"frontend": "iscsi"}),
+                volume("vol-stale"),
+            ]
+        )
+
+    found = candidates._workspace_longhorn_candidates(settings, get_json, {"pv-present"}, set())
+
+    assert [item.name for item in found] == ["vol-stale"]
+    assert found[0].kind == "longhorn_volume"