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