Stop package cleanup after failed Tea login setup

Emit a fixed credential-free login_setup warning and fail the optional prune
step before any dependent package API calls if its existing login setup fails.
Preserve successful pruning, retention, registry upload and local cleanup.

Exercise the extracted prune shell with fake Tea and disposable configuration,
including failed setup with a previous login and unchanged successful targets.
This identifies the failed stage, not the suppressed authentication cause.

Ticket: https://agenthub.fritzlab.net/t-c1070dtgpqqy

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 \
+        if ! tea login add --name ci --url https://code.fritzlab.net \
-          --token "$TOKEN" --no-version-check >/dev/null 2>&1 || true
+          --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()