da42f764c0
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a06ecb-456a-71c2-b41e-0755d336e883
69 lines
3.3 KiB
Python
69 lines
3.3 KiB
Python
import importlib.util
|
|
import os
|
|
from pathlib import Path
|
|
import subprocess
|
|
import tempfile
|
|
import unittest
|
|
|
|
SPEC = importlib.util.spec_from_file_location(
|
|
"refresh_live_images", Path(__file__).resolve().parents[1] / "scripts/refresh_live_images.py"
|
|
)
|
|
inventory = importlib.util.module_from_spec(SPEC)
|
|
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)
|
|
output = root / "durable" / "images.txt"
|
|
first, second = root / "first", root / "second"
|
|
first.write_text("forgejo.example/org/app:old\n")
|
|
second.write_text("forgejo.example/org/other:live\n")
|
|
inventory.refresh(output, [first, second])
|
|
first.write_text("forgejo.example/org/app:new\n")
|
|
receipt = inventory.refresh(output, [first])
|
|
self.assertEqual(receipt["images"], 3)
|
|
self.assertIn("forgejo.example/org/other:live", output.read_text())
|
|
self.assertIn("forgejo.example/org/app:old", output.read_text())
|
|
self.assertEqual(inventory.refresh(output, [first]), receipt)
|
|
|
|
def test_failed_export_preserves_inventory(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
output, source = root / "images", root / "export"
|
|
output.write_text("forgejo.example/org/app:live\n")
|
|
before = output.read_bytes()
|
|
for content in ["", "# no images\n", "error fetching pods\n"]:
|
|
source.write_text(content)
|
|
with self.assertRaises(ValueError):
|
|
inventory.refresh(output, [source])
|
|
self.assertEqual(output.read_bytes(), before)
|
|
with self.assertRaises(OSError):
|
|
inventory.refresh(output, [root / "missing"])
|
|
self.assertEqual(output.read_bytes(), before)
|