from __future__ import annotations import importlib.util import sys import tempfile import unittest from pathlib import Path from unittest import mock REPO_DIR = Path(__file__).resolve().parents[1] SPEC = importlib.util.spec_from_file_location( "forgejo_package_prune", REPO_DIR / "scripts/forgejo_package_prune.py" ) prune = importlib.util.module_from_spec(SPEC) assert SPEC.loader is not None sys.modules[SPEC.name] = prune SPEC.loader.exec_module(prune) class ForgejoPackagePruneTests(unittest.TestCase): def test_collect_protected_versions_from_helm_values(self) -> None: import tempfile with tempfile.TemporaryDirectory() as tmpdir: root = Path(tmpdir) helm = root / "helm" helm.mkdir() (helm / "vergabe-teilnahme-values.yaml").write_text( """ image: repository: forgejo.coulomb.social/coulomb/vergabe-teilnahme tag: "064d295" """.lstrip(), encoding="utf-8", ) protected = prune.collect_protected_versions(root) self.assertIn(("container", "vergabe-teilnahme", "064d295"), protected) def test_build_delete_plans_respects_depth_and_protection(self) -> None: protected = {("container", "vergabe-teilnahme", "old-prod")} # Forgejo's package list endpoint returns one entry per (name, version); # build_delete_plans groups these by name (no separate /versions call). packages = [ {"name": "vergabe-teilnahme", "version": "v4", "created_at": "2026-07-12T10:00:00Z"}, {"name": "vergabe-teilnahme", "version": "v3", "created_at": "2026-07-11T10:00:00Z"}, {"name": "vergabe-teilnahme", "version": "v2", "created_at": "2026-07-10T10:00:00Z"}, {"name": "vergabe-teilnahme", "version": "v1", "created_at": "2026-07-09T10:00:00Z"}, {"name": "vergabe-teilnahme", "version": "old-prod", "created_at": "2026-07-08T10:00:00Z"}, ] with mock.patch.object(prune, "list_packages", return_value=packages): plans, errors = prune.build_delete_plans( base_url="https://forgejo.example", token="token", owner="coulomb", package_types=["container"], max_versions=3, protected=protected, ) self.assertEqual(errors, []) deletable = {(p.name, p.version) for p in plans if not p.protected} protected_plans = {(p.name, p.version) for p in plans if p.protected} self.assertEqual(deletable, {("vergabe-teilnahme", "v1")}) self.assertEqual(protected_plans, {("vergabe-teilnahme", "old-prod")}) def test_collect_live_cluster_versions_protects_forgejo_tags(self) -> None: images = "\n".join([ "forgejo.coulomb.social/coulomb/state-hub:f2e042a", "forgejo.coulomb.social/coulomb/vergabe-teilnahme:064d295", "registry.k8s.io/pause:3.9", # unrelated registry → ignored "gitea.coulomb.social/coulomb/old-app:x", # non-forgejo → ignored "", ]) with mock.patch.object(prune.shutil, "which", return_value="/usr/bin/kubectl"), \ mock.patch.object( prune.subprocess, "run", return_value=mock.Mock(stdout=images, returncode=0), ): protected, notes = prune.collect_live_cluster_versions() self.assertEqual(notes, []) self.assertEqual( protected, { ("container", "state-hub", "f2e042a"), ("container", "vergabe-teilnahme", "064d295"), }, ) def test_collect_live_cluster_versions_skips_without_kubectl(self) -> None: with mock.patch.object(prune.shutil, "which", return_value=None): protected, notes = prune.collect_live_cluster_versions() self.assertEqual(protected, set()) self.assertTrue(notes and "kubectl not found" in notes[0]) def test_load_token_prefers_explicit_env(self) -> None: with mock.patch.dict(prune.os.environ, {"FORGEJO_TOKEN": " env-token "}, clear=True): with mock.patch.object(prune, "_load_token_from_openbao") as load_openbao: self.assertEqual(prune.load_token(), "env-token") load_openbao.assert_not_called() def test_load_token_reads_explicit_token_file_env(self) -> None: with tempfile.TemporaryDirectory() as tmpdir: token_file = Path(tmpdir) / "forgejo-admin.token" token_file.write_text(" file-token \n", encoding="utf-8") with mock.patch.dict( prune.os.environ, {"FORGEJO_TOKEN_FILE": str(token_file)}, clear=True, ): with mock.patch.object(prune, "_load_token_from_openbao") as load_openbao: self.assertEqual(prune.load_token(), "file-token") load_openbao.assert_not_called() def test_load_token_uses_openbao_lane_by_default(self) -> None: with mock.patch.dict(prune.os.environ, {}, clear=True): with mock.patch.object(prune.shutil, "which", return_value="/usr/bin/bao"), \ mock.patch.object( prune.subprocess, "run", return_value=mock.Mock(stdout=" bao-token \n"), ) as run_mock: self.assertEqual(prune.load_token(), "bao-token") run_mock.assert_called_once_with( [ "bao", "kv", "get", f"-field={prune.DEFAULT_FORGEJO_ADMIN_BAO_FIELD}", prune.DEFAULT_FORGEJO_ADMIN_BAO_PATH, ], capture_output=True, text=True, check=True, ) def test_load_token_uses_legacy_file_only_when_opted_in(self) -> None: with tempfile.TemporaryDirectory() as tmpdir: legacy_file = Path(tmpdir) / "forgejo-tegwick-api-token" legacy_file.write_text("legacy-token\n", encoding="utf-8") with mock.patch.dict( prune.os.environ, {"FORGEJO_ALLOW_LEGACY_FILE_FALLBACK": "1"}, clear=True, ): with mock.patch.object(prune.shutil, "which", return_value=None), \ mock.patch.object(prune, "LEGACY_FORGEJO_TOKEN_FILE", legacy_file): self.assertEqual(prune.load_token(), "legacy-token") def test_load_token_error_mentions_openbao_and_break_glass(self) -> None: with mock.patch.dict(prune.os.environ, {}, clear=True): with mock.patch.object(prune.shutil, "which", return_value=None): with self.assertRaises(SystemExit) as exc_info: prune.load_token() message = str(exc_info.exception) self.assertIn("forgejo-admin-workload-kv-read", message) self.assertIn("FORGEJO_ALLOW_LEGACY_FILE_FALLBACK=1", message) self.assertIn(prune.DEFAULT_FORGEJO_ADMIN_BAO_PATH, message) if __name__ == "__main__": unittest.main() def test_collect_live_images_from_files(tmp_path): from scripts.forgejo_package_prune import collect_live_images_from_files export = tmp_path / "live-coulombcore.txt" export.write_text( "# exported 2026-07-18\n" "forgejo.coulomb.social/coulomb/issue-core:0.2.1\n" "forgejo.coulomb.social/coulomb/state-hub:f2e042a\n" "gitea.coulomb.social/coulomb/core-hub:3ed8531\n" "nats:2.10-alpine\n" "\n" ) protected, notes = collect_live_images_from_files( [export, tmp_path / "missing.txt"] ) assert ("container", "issue-core", "0.2.1") in protected assert ("container", "state-hub", "f2e042a") in protected # non-forgejo registries and bare images are ignored assert all(name != "core-hub" for (_, name, _) in protected) assert len(protected) == 2 assert any("missing.txt" in n for n in notes)