image-push/tests/test_action.py

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()