Stop optional image pruning after failed Tea login setup

Guard failed cleanup login setup before dependent package operations; preserve successful and optional cleanup semantics. Required contract CI 20382 passed. Exact task-specific Human review-count exception is recorded on t-c1070dtgpqqy; official architect review 8504 approved this head. No authentication repair, tag movement or consumer activation.

Authored-By: Codex (GPT-5) <noreply@openai.com>

.gitea/workflows/pr.yaml

@@ -0,0 +1,17 @@
+name: pr
+on:
+  pull_request:
+jobs:
+  contract:
+    runs-on: fritzlab
+    timeout-minutes: 5
+    steps:
+      - uses: actions/checkout@v4
+        with:
+          persist-credentials: false
+      - name: validate composite contract without registry operations
+        run: |
+          python3 -m venv /tmp/image-push-tests
+          /tmp/image-push-tests/bin/pip install --quiet -r tests/requirements.txt
+          /tmp/image-push-tests/bin/python -m unittest discover -s tests
+          git diff --check

.gitignore

@@ -0,0 +1 @@
+__pycache__/

README.md

@@ -31,6 +31,7 @@ local Docker daemon, image-push uploads it.
 | `org` | yes | — | Gitea org for package API (`fritzlab`, `dns`). |
 | `name` | yes | — | Package name as registered in the registry. |
 | `latest` | no | `true` | Also push a `:latest` tag. |
+| `prune` | no | `true` | Prune old numeric registry tags. Set `false` for retained or independently pinned artifacts. |
 | `keep` | no | `3` | Numeric tags to retain. Older are deleted. |
 
 ## Behavior
@@ -38,6 +39,25 @@ local Docker daemon, image-push uploads it.
 1. `docker login code.fritzlab.net` as `ci-bot`.
 2. `docker push <image>:<tag>`.
 3. If `latest=true`, also `docker push <image>:latest`.
-4. Prune: list numeric tags from Gitea package API, keep the newest `keep`,
+4. If `prune=true`, list numeric tags from Gitea package API, keep the newest `keep`,
    delete the rest. Failures here do not fail the workflow
    (`continue-on-error: true`).
+
+Cleanup first sets up its existing Tea login. If that setup fails, the optional
+prune step exits with a fixed `login_setup` warning and does not list or delete
+packages, even if a previous login is available. Login output remains suppressed
+to avoid exposing credentials or configuration. This identifies the failing
+stage, not the underlying authentication or configuration cause; it does not
+repair login setup or change the successful cleanup's targets or retention.
+
+`prune` accepts only the strings `true` and `false`; invalid input fails before
+registry login or push. With `false`, remote package listing and deletion are
+skipped. Image upload and local Docker cleanup remain unchanged. The action does
+not discover deployment pins; owners choosing pruning must account for that
+retention contract.
+
+Run `python3 -m unittest discover -s tests` with the pinned test dependencies in
+`tests/requirements.txt`. Tests inspect the composite action, execute its
+credential-free validation step, and exercise the prune shell with a fake Tea
+client and disposable configuration. No real login, registry operation, or
+package deletion runs.

action.yaml

@@ -21,6 +21,10 @@ inputs:
     description: Also push a :latest tag
     required: false
     default: 'true'
+  prune:
+    description: Prune old numeric registry tags; false retains every remote tag
+    required: false
+    default: 'true'
   keep:
     description: Numeric tags to retain after prune; older ones are deleted
     required: false
@@ -28,6 +32,16 @@ inputs:
 runs:
   using: composite
   steps:
+    - name: Validate pruning policy
+      shell: bash
+      env:
+        PRUNE: ${{ inputs.prune }}
+      run: |
+        case "$PRUNE" in
+          true|false) ;;
+          *) echo "prune must be true or false" >&2; exit 1 ;;
+        esac
+
     - name: Log in to registry
       uses: docker/login-action@v3
       with:
@@ -54,6 +68,7 @@ runs:
         fi
 
     - name: Prune old tags
+      if: ${{ inputs.prune == 'true' }}
       continue-on-error: true
       shell: bash
       env:
@@ -63,8 +78,11 @@ runs:
         KEEP: ${{ inputs.keep }}
       run: |
         set -euo pipefail
-        tea login add --name ci --url https://code.fritzlab.net \
-          --token "$TOKEN" --no-version-check >/dev/null 2>&1 || true
+        if ! tea login add --name ci --url https://code.fritzlab.net \
+          --token "$TOKEN" --no-version-check >/dev/null 2>&1; then
+          echo "::warning::image-push prune failed at login_setup; package cleanup skipped" >&2
+          exit 1
+        fi
         # Paginate the package list. The org has more container versions than a
         # single API page returns (Gitea caps page size at MAX_RESPONSE_ITEMS),
         # so an unpaginated call silently drops a package's older tags from the

tests/requirements.txt

@@ -0,0 +1 @@
+PyYAML==6.0.3

tests/test_action.py

@@ -0,0 +1,171 @@
+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()