import importlib.util import copy import io 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): # Diagnostic annotation tests must not emit real CI workflow commands. self.enterContext(patch("sys.stdout", new_callable=io.StringIO)) 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 "--format" in args: return json.dumps({"services": {"app": {"environment": {"COLLECT_MAX_DETAILS": "all"}}}}) if "inspect" in args: return json.dumps([{"Config": {"Labels": {"org.opencontainers.image.revision": "a" * 40}}}]) return "" def launch(args, **kwargs): self.assertIn("car-tracker-pipeline", args) self.assertNotIn("--max-details", args) self.assertEqual(kwargs["stderr"], subprocess.STDOUT) kwargs["stdout"].write("Private collection output\n") return subprocess.CompletedProcess(args, 3) with patch.object(host, "command", side_effect=command), \ 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) def test_collection_all_omits_detail_limit_but_keeps_page_limit(self): args = host.collection_command({"COLLECT_MAX_DETAILS": "all"}) self.assertNotIn("--max-details", args) self.assertEqual(args[args.index("--max-pages") + 1], "1") self.assertEqual(args[args.index("--family") + 1], "saab-900") def test_explicit_settings_and_all_families(self): args = host.collection_command({"COLLECT_FAMILY": "all", "COLLECT_MAX_PAGES": "3", "COLLECT_MAX_DETAILS": "20", "COLLECT_DELAY": "6"}) self.assertNotIn("--family", args) self.assertEqual(args[args.index("--max-details") + 1], "20") self.assertEqual(args[args.index("--max-pages") + 1], "3") def test_incremental_policy_is_host_configurable_and_validated(self): args = host.collection_command({"COLLECT_DETAIL_POLICY": "incremental", "COLLECT_DETAIL_REFRESH_DAYS": "7"}) self.assertEqual(args[args.index("--detail-policy") + 1], "incremental") self.assertEqual(args[args.index("--detail-refresh-days") + 1], "7") self.assertNotIn("--detail-policy", host.collection_command({})) for settings in ({"COLLECT_DETAIL_POLICY": "unknown"}, {"COLLECT_DETAIL_REFRESH_DAYS": "0"}): with self.assertRaises(ValueError): host.collection_command(settings) def test_fresh_unchanged_skips_are_not_source_failures(self): manifest, pipeline = self.bounded_outcome() manifest["searches"][0] = {"status": "complete"} manifest["details"]["listings"]["1"] = {"status": "not_attempted", "reason": "fresh_unchanged"} self.assertEqual(host.limit_reasons(manifest, pipeline), ["fresh_unchanged"]) self.assertIn("fresh unchanged details reused=1", " ".join(host.outcome_summary(manifest, pipeline))) def test_invalid_settings_are_refused(self): for key, value in [("COLLECT_MAX_DETAILS", "0"), ("COLLECT_MAX_DETAILS", "-1"), ("COLLECT_MAX_DETAILS", "all; id"), ("COLLECT_MAX_PAGES", "all"), ("COLLECT_MAX_PAGES", "0"), ("COLLECT_DELAY", "nan"), ("COLLECT_DELAY", "inf"), ("COLLECT_DELAY", "0.5"), ("COLLECT_FAMILY", "")]: with self.subTest(key=key, value=value), self.assertRaises(ValueError): host.collection_command({key: value}) def bounded_outcome(self): manifest = {"run_id": "a" * 36, "status": "partial", "planned_searches": [{"family": "saab-900"}], "searches": [{"status": "partial", "reason": "page_limit"}], "details": {"status": "complete", "listings": {"1": {"status": "complete"}}}} pipeline = {"collection_status": "partial", "import_status": "succeeded", "import_summary": {"run_id": "a" * 36, "rejected": 0}} return manifest, pipeline def test_only_intentional_limits_are_warnings(self): manifest, pipeline = self.bounded_outcome() self.assertEqual(host.limit_reasons(manifest, pipeline), ["page_limit"]) manifest["details"] = {"status": "partial", "listings": { "1": {"status": "not_attempted", "reason": "detail_limit"}}} self.assertEqual(host.limit_reasons(manifest, pipeline), ["detail_limit", "page_limit"]) for mutate in ( lambda m, p: m.update(error_type="SourceAccessError"), lambda m, p: m["details"]["listings"]["1"].update(reason="run_stopped"), lambda m, p: m["details"]["listings"]["1"].update(status="unavailable"), lambda m, p: m["searches"][0].update(reason="TimeoutError"), lambda m, p: m["planned_searches"].append({"family": "another"}), lambda m, p: p.update(import_status="unknown"), lambda m, p: p["import_summary"].update(rejected=1), lambda m, p: p["import_summary"].update(run_id="different"), ): m, p = copy.deepcopy(manifest), copy.deepcopy(pipeline) mutate(m, p) self.assertEqual(host.limit_reasons(m, p), []) def test_verified_limit_outcome_warns_and_returns_success(self): raw = self.root / "raw" path = raw / "2026-10-01" / ("a" * 36) path.mkdir(parents=True) manifest, pipeline = self.bounded_outcome() (path / "manifest.json").write_text(json.dumps(manifest)) (path / "pipeline.json").write_text(json.dumps(pipeline)) log = self.root / f"{self.request['request_id']}.log" log.write_text(f"Manifest: /data/2026-10-01/{'a' * 36}/manifest.json\n") with patch("sys.stdout", new_callable=io.StringIO) as output: self.assertEqual(host.actions_outcome(1, self.request, self.root, raw), 0) self.assertIn("::warning::", output.getvalue()) self.assertIn("page limit", output.getvalue()) (path / "pipeline.json").unlink() self.assertEqual(host.actions_outcome(1, self.request, self.root, raw), 1) self.assertEqual(host.actions_outcome(3, self.request, self.root, raw), 3) def test_unknown_or_unsafe_manifest_path_remains_failure(self): log = self.root / f"{self.request['request_id']}.log" for value in ("", "Manifest: /data/../../secret\n", "Manifest: /data/unknown/manifest.json\n"): log.write_text(value) self.assertEqual(host.actions_outcome(1, self.request, self.root, self.root), 1) def test_source_403_is_explained_and_remains_failure(self): raw = self.root / "raw" path = raw / "2026-10-01" / ("a" * 36) path.mkdir(parents=True) manifest, pipeline = self.bounded_outcome() manifest["error_type"] = "SourceAccessError" manifest["details"] = {"status": "partial", "listings": { "private-ad-id": {"status": "complete", "url": "https://private.example/ad"}, "other-private-id": {"status": "failed", "reason": "SourceAccessError", "http_status": 403}, "third-private-id": {"status": "not_attempted", "reason": "run_stopped"}, }} pipeline["import_summary"].update(accepted=2, field_issues=4) (path / "manifest.json").write_text(json.dumps(manifest)) (path / "pipeline.json").write_text(json.dumps(pipeline)) (self.root / f"{self.request['request_id']}.log").write_text( f"Manifest: /data/2026-10-01/{'a' * 36}/manifest.json\n") with patch("sys.stdout", new_callable=io.StringIO) as output: self.assertEqual(host.actions_outcome(1, self.request, self.root, raw), 1) message = output.getvalue() self.assertIn("::error::", message) self.assertIn("HTTP 403", message) self.assertIn("Import: succeeded", message) self.assertIn("completed=1, failed=1, not attempted=1", message) self.assertNotIn("private-ad-id", message) self.assertNotIn("private.example", message) self.assertNotIn("::warning::", message) def test_diagnostics_do_not_echo_unknown_errors_or_workflow_commands(self): manifest, pipeline = self.bounded_outcome() manifest["error_type"] = "secret\n::warning::injected" manifest["status"] = "secret\n::error::injected" pipeline["import_summary"].update(accepted="secret", field_issues="secret") lines = " ".join(host.outcome_summary(manifest, pipeline)) self.assertNotIn("secret", lines) self.assertNotIn("::", lines) def test_private_capture_is_announced_without_contents(self): manifest, pipeline = self.bounded_outcome() manifest["details"]["listings"] = {"1": {"status": "failed", "http_status": 403}} manifest["diagnostics"] = {"capture_status": "partial", "directory": "private-value"} lines = " ".join(host.outcome_summary(manifest, pipeline)) self.assertIn("diagnostics saved (partial)", lines) self.assertNotIn("private-value", lines) def test_failed_import_summary_and_missing_manifest_do_not_hide_failure(self): manifest, pipeline = self.bounded_outcome() pipeline.update(import_status="unknown", error_type="OperationalError") lines = " ".join(host.outcome_summary(manifest, pipeline)) self.assertIn("Import completion is uncertain", lines) with patch("sys.stdout", new_callable=io.StringIO) as output: self.assertEqual(host.actions_outcome(3, self.request, self.root, self.root), 3) self.assertIn("summary is unavailable", output.getvalue())