Refuse prune apply when requested image inventories are unavailable
All checks were successful
CI Smoke / host-smoke (push) Successful in 0s
CI Smoke / container-smoke (push) Successful in 1s

Assistant: codex
Assistant-Model: gpt-6-astra
Assistant-Session: 01a06ecb-456a-71c2-b41e-0755d336e883
This commit is contained in:
codex 2026-09-05 02:00:50 +02:00
parent 0349a08e1b
commit f637989a69
4 changed files with 58 additions and 4 deletions

View file

@ -121,8 +121,8 @@ def collect_live_images_from_files(
output from another cluster). This closes the multi-cluster gap
(ACTIVITY-WP-0020-T07): the prune host's kubectl only sees its own
cluster, so every other production cluster exports its live images to a
file that is merged here. A missing file is a WARN, not a failure
but it means reduced protection coverage, so the note must surface.
file that is merged here. Unavailable or empty exports produce notes;
main refuses apply when any requested export cannot provide coverage.
"""
protected: set[tuple[str, str, str]] = set()
notes: list[str] = []
@ -131,13 +131,22 @@ def collect_live_images_from_files(
if not path.is_file():
notes.append(f"live-images file missing: {path}")
continue
for line in path.read_text(encoding="utf-8").splitlines():
try:
lines = path.read_text(encoding="utf-8").splitlines()
except (OSError, UnicodeError):
notes.append(f"live-images file unreadable: {path}")
continue
has_images = False
for line in lines:
image = line.strip()
if not image or image.startswith("#"):
continue
has_images = True
match = FORGEJO_IMAGE_RE.match(image)
if match and match.group("tag"):
protected.add(("container", match.group("name"), match.group("tag")))
if not has_images:
notes.append(f"live-images file empty: {path}")
return protected, notes
@ -509,6 +518,12 @@ def main(argv: list[str] | None = None) -> int:
apply = bool(args.apply)
dry_run = not apply
package_types = [part.strip() for part in args.types.split(",") if part.strip()]
file_live, file_notes = collect_live_images_from_files(args.live_images_files)
if apply and file_notes:
for note in file_notes:
print(f" ERROR: {note}", file=sys.stderr)
print("Refusing apply: requested live-image inventory is unavailable or empty", file=sys.stderr)
return 2
token = load_token()
protected = collect_protected_versions(args.apps_root.expanduser())
protect_notes: list[str] = []
@ -519,7 +534,6 @@ def main(argv: list[str] | None = None) -> int:
for note in protect_notes:
print(f" WARN: {note}", file=sys.stderr)
if args.live_images_files:
file_live, file_notes = collect_live_images_from_files(args.live_images_files)
protected |= file_live
protect_notes.extend(file_notes)
print(f"Protected exported live tags: {len(file_live)}", file=sys.stderr)