import json import os from pathlib import Path import subprocess import sys import tempfile import unittest import yaml class PruningContract(unittest.TestCase): @classmethod def setUpClass(cls): cls.action = yaml.safe_load((Path(__file__).resolve().parents[1] / "action.yaml").read_text()) cls.steps = cls.action["runs"]["steps"] def test_default_preserves_existing_callers_and_only_prune_is_conditional(self): self.assertEqual(self.action["inputs"]["prune"]["default"], "true") gated = [step for step in self.steps if "if" in step] self.assertEqual(len(gated), 1) self.assertEqual(gated[0]["name"], "Prune old tags") self.assertEqual(gated[0]["if"], "${{ inputs.prune == 'true' }}") remote = [step for step in self.steps if "/packages/" in step.get("run", "")] self.assertEqual(remote, gated) self.assertIn("docker push", next(step for step in self.steps if step["name"] == "Push")["run"]) self.assertNotIn("if", next(step for step in self.steps if step["name"] == "Untag local image")) def test_invalid_policy_stops_before_credential_step(self): validation = self.steps[0] self.assertEqual(validation["name"], "Validate pruning policy") self.assertEqual(validation["env"], {"PRUNE": "${{ inputs.prune }}"}) self.assertNotIn("uses", validation) for value, success in (("true", True), ("false", True), ("False", False), ("", False), ("yes", False), ("false\ntrue", False)): with self.subTest(value=value): result = subprocess.run(["bash", "-c", validation["run"]], env={"PATH": os.defpath, "PRUNE": value}, capture_output=True) self.assertEqual(result.returncode == 0, success) class PruneLoginContract(unittest.TestCase): @classmethod def setUpClass(cls): cls.action = yaml.safe_load((Path(__file__).resolve().parents[1] / "action.yaml").read_text()) cls.prune = next(step for step in cls.action["runs"]["steps"] if step["name"] == "Prune old tags") def run_prune(self, login_exit=0, keep="3", existing_login=False): # Only this fake executable can receive the synthetic token. Do not # inherit the caller's environment or use its real Tea configuration. with tempfile.TemporaryDirectory(prefix="image-push-fake-tea-") as directory: root = Path(directory) binary = root / "bin" binary.mkdir() config = root / "config" config.mkdir() marker = config / "fake-login" if existing_login: marker.write_text("previous fake login") calls = root / "calls.jsonl" fake = binary / "tea" fake.write_text(f"#!{sys.executable}\n" + r''' import json import os from pathlib import Path import sys args = sys.argv[1:] with Path(os.environ["FAKE_CALLS"]).open("a") as calls: calls.write(json.dumps(args) + "\n") marker = Path(os.environ["FAKE_CONFIG_ROOT"]) / "fake-login" if args[:2] == ["login", "add"]: failure = int(os.environ["FAKE_LOGIN_EXIT"]) if failure: # Deliberately sensitive-looking synthetic output must stay suppressed. print("SYNTHETIC_TOKEN_VALUE stdout config=/fake/private") print("SYNTHETIC_TOKEN_VALUE stderr config=/fake/private", file=sys.stderr) sys.exit(failure) marker.write_text("new fake login") sys.exit(0) if args[:1] == ["api"] and marker.exists(): if len(args) == 2 and args[1].startswith("/packages/example?type=container&limit=50&page="): page = args[1].rsplit("=", 1)[1] pages = { "1": [ {"name": "widget", "version": "9"}, {"name": "widget", "version": "latest"}, {"name": "other", "version": "0"}, {"name": "widget", "version": "2"}, {"name": "widget", "version": "001"}, {"name": "widget", "version": "12"}, ], "2": [ {"name": "widget", "version": "4"}, {"name": "other", "version": "100"}, {"name": "widget", "version": "release-1"}, ], "3": [], } if page not in pages: sys.exit(92) print(json.dumps(pages[page])) sys.exit(0) if len(args) == 4 and args[:3] == ["api", "-X", "DELETE"]: # Log a fake request only: this executable never sends a network request. sys.exit(0) sys.exit(93) ''') fake.chmod(0o700) result = subprocess.run( ["bash", "-c", self.prune["run"]], env={ "PATH": str(binary) + os.pathsep + os.defpath, "FAKE_CONFIG_ROOT": str(config), "TOKEN": "SYNTHETIC_TOKEN_VALUE", "ORG": "example", "NAME": "widget", "KEEP": keep, "FAKE_CALLS": str(calls), "FAKE_LOGIN_EXIT": str(login_exit), }, capture_output=True, text=True, timeout=10, ) recorded = [json.loads(line) for line in calls.read_text().splitlines()] return result, recorded def test_failed_setup_stops_all_dependent_calls_without_leaking_output(self): for failure in (1, 7): for previous in (False, True): with self.subTest(failure=failure, previous=previous): result, calls = self.run_prune(login_exit=failure, existing_login=previous) self.assertEqual(result.returncode, 1) self.assertEqual(len(calls), 1) self.assertEqual(calls[0], [ "login", "add", "--name", "ci", "--url", "https://code.fritzlab.net", "--token", "SYNTHETIC_TOKEN_VALUE", "--no-version-check", ]) self.assertEqual(result.stdout, "") self.assertEqual(result.stderr, "::warning::image-push prune failed at login_setup; package cleanup skipped\n") self.assertNotIn("SYNTHETIC_TOKEN_VALUE", result.stdout + result.stderr) self.assertNotIn("/fake/private", result.stdout + result.stderr) def test_success_keeps_pagination_selection_numeric_order_and_retention(self): for keep, removed in (("3", ["001", "2"]), ("1", ["001", "2", "4", "9"]), ("5", [])): with self.subTest(keep=keep): result, calls = self.run_prune(keep=keep) self.assertEqual(result.returncode, 0, result.stderr) lists = [args[1] for args in calls if args[:1] == ["api"] and len(args) == 2] self.assertEqual(lists, [ f"/packages/example?type=container&limit=50&page={page}" for page in (1, 2, 3) ]) deletes = [args for args in calls if args[:3] == ["api", "-X", "DELETE"]] self.assertEqual(deletes, [ ["api", "-X", "DELETE", f"/packages/example/container/widget/{tag}"] for tag in removed ]) self.assertEqual(result.stdout, "".join(f"deleting widget:{tag}\n" for tag in removed)) self.assertEqual(result.stderr, "") self.assertNotIn("SYNTHETIC_TOKEN_VALUE", result.stdout + result.stderr) def test_cleanup_failure_remains_optional_and_login_contract_unchanged(self): self.assertIs(self.prune["continue-on-error"], True) self.assertEqual(self.prune["if"], "${{ inputs.prune == 'true' }}") self.assertEqual(self.prune["env"], { "TOKEN": "${{ inputs.token }}", "ORG": "${{ inputs.org }}", "NAME": "${{ inputs.name }}", "KEEP": "${{ inputs.keep }}", }) if __name__ == "__main__": unittest.main()