import importlib.util import json from pathlib import Path import subprocess import tempfile import unittest from unittest.mock import patch import uuid ROOT = Path(__file__).resolve().parents[1] def load(name, path): spec = importlib.util.spec_from_file_location(name, ROOT / path) module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) return module agent = load("reconcile", "ops/gitops/reconcile.py") with patch.dict("sys.modules", {"reconcile": agent}): host = load("host_run", "ops/ssh/host_run.py") class SSHRunTests(unittest.TestCase): def setUp(self): self.root = Path(self.enterContext(tempfile.TemporaryDirectory())) self.request = {"schema_version": 1, "request_id": str(uuid.uuid4()), "revision": "a" * 40, "image": agent.IMAGE_REPOSITORY + "@sha256:" + "b" * 64} self.command = f"run {self.request['request_id']} {self.request['revision']} {self.request['image']}" def test_only_fixed_command_is_accepted(self): self.assertEqual(host.parse_command(self.command), self.request) for command in ("", "bash", "migrate", self.command + " ; id", self.command.replace("@sha256:" + "b" * 64, ":latest"), self.command.replace("run ", "run; ", 1)): with self.subTest(command=command), self.assertRaises(ValueError): host.parse_command(command) def test_host_returns_pipeline_exit_and_keeps_output_private(self): def command(*args, **kwargs): if "inspect" in args: return json.dumps([{"Config": {"Labels": {"org.opencontainers.image.revision": "a" * 40}}}]) return "" def launch(args, **kwargs): self.assertEqual(kwargs["stderr"], subprocess.STDOUT) kwargs["stdout"].write("Private collection output\n") return subprocess.CompletedProcess(args, 3) with patch.object(agent, "command", side_effect=command), \ patch.object(agent.subprocess, "run", side_effect=launch) as child: self.assertEqual(host.run(self.root, self.request), 3) self.assertEqual(host.run(self.root, self.request), 3) self.assertEqual(child.call_count, 1) log = self.root / "state/requests" / f"{self.request['request_id']}.log" self.assertIn("Private collection output", log.read_text()) self.assertEqual(log.stat().st_mode & 0o777, 0o600) def test_unknown_prior_attempt_blocks_launch(self): state = self.root / "state/requests" state.mkdir(parents=True) agent.atomic_json(state / "old.json", {"status": "attempted"}) with patch.object(host, "attempt") as attempt, self.assertRaises(ValueError): host.run(self.root, self.request) attempt.assert_not_called() def test_busy_lock_refuses_another_job(self): import fcntl (self.root / "state").mkdir() with (self.root / "state/agent.lock").open("a") as lock: fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB) with self.assertRaises(ValueError): host.run(self.root, self.request)