185 lines
7.9 KiB
Python
185 lines
7.9 KiB
Python
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)
|