Add failure-safe cluster image capture hook

Assistant: codex
Assistant-Model: gpt-6-astra
Assistant-Session: 01a06ecb-456a-71c2-b41e-0755d336e883

tests/test_live_images_inventory.py

@@ -1,5 +1,7 @@
 import importlib.util
+import os
 from pathlib import Path
+import subprocess
 import tempfile
 import unittest
 
@@ -11,6 +13,31 @@ SPEC.loader.exec_module(inventory)
 
 
 class InventoryTests(unittest.TestCase):
+    def test_bare_image_names_are_valid(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            source = root / "source"
+            source.write_text("nginx:stable\n")
+            self.assertEqual(inventory.refresh(root / "all", [source])["images"], 1)
+
+    def test_capture_failure_keeps_prior_inventory(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            output = root / "all"
+            output.write_text("forgejo.example/org/app:old\n")
+            kubectl = root / "kubectl"
+            kubectl.write_text('#!/bin/sh\necho forgejo.example/org/app:new\nexit 1\n')
+            kubectl.chmod(0o755)
+            env = {**os.environ, "PATH": str(root) + ":" + os.environ["PATH"], "OUT": str(output)}
+            command = Path(__file__).resolve().parents[1] / "tools/cmd/refresh-live-images"
+            failed = subprocess.run(["bash", str(command)], env=env, capture_output=True)
+            self.assertNotEqual(failed.returncode, 0)
+            self.assertEqual(output.read_text(), "forgejo.example/org/app:old\n")
+            kubectl.write_text('#!/bin/sh\necho forgejo.example/org/app:new\n')
+            succeeded = subprocess.run(["bash", str(command)], env=env, capture_output=True)
+            self.assertEqual(succeeded.returncode, 0, succeeded.stderr)
+            self.assertEqual(len(output.read_text().splitlines()), 2)
+
     def test_refresh_preserves_other_clusters_and_previous_tags(self):
         with tempfile.TemporaryDirectory() as tmp:
             root = Path(tmp)