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): spec = importlib.util.spec_from_file_location(name, ROOT / "ops/gitops" / f"{name}.py") module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) return module agent = load("reconcile") requester = load("request") class GitOpsTests(unittest.TestCase): def setUp(self): self.request = {"schema_version": 1, "request_id": str(uuid.uuid4()), "revision": "a" * 40, "image": agent.IMAGE_REPOSITORY + "@sha256:" + "b" * 64} self.temporary = self.enterContext(tempfile.TemporaryDirectory()) self.state = Path(self.temporary) def attempt(self): agent.attempt(self.request, self.state, self.state / "compose.yaml", self.state / "app.env") def commands(self, *args, **kwargs): if "inspect" in args: return json.dumps([{"Config": {"Labels": {"org.opencontainers.image.revision": "a" * 40}}}]) return "" def test_validate_request_rejects_mutable_tag_extra_keys_and_invalid_id(self): agent.validate_request(self.request) for field, value in [("image", agent.IMAGE_REPOSITORY + ":latest"), ("request_id", "../../secret"), ("revision", "master"), ("command", "migrate")]: with self.subTest(field=field), self.assertRaises(ValueError): agent.validate_request({**self.request, field: value}) def test_atomic_state_is_private(self): path = self.state / "outcome.json" agent.atomic_json(path, self.request) self.assertEqual(json.loads(path.read_text()), self.request) self.assertEqual(path.stat().st_mode & 0o777, 0o600) def test_attempt_is_written_before_launch_and_failure_is_not_retried(self): path = self.state / f"{self.request['request_id']}.json" def launch(*args, **kwargs): self.assertEqual(json.loads(path.read_text())["status"], "attempted") return subprocess.CompletedProcess(args, 3) with patch.object(agent, "command", side_effect=self.commands), \ patch.object(agent.subprocess, "run", side_effect=launch) as run: self.attempt() self.attempt() self.assertEqual(run.call_count, 1) self.assertEqual(json.loads(path.read_text())["exit_code"], 3) def test_crash_preserves_unknown_attempt(self): with patch.object(agent, "command", side_effect=self.commands), \ patch.object(agent.subprocess, "run", side_effect=OSError): with self.assertRaises(OSError): self.attempt() outcome = json.loads((self.state / f"{self.request['request_id']}.json").read_text()) self.assertEqual(outcome["status"], "attempted") self.assertIsNone(outcome["exit_code"]) def test_pull_failure_does_not_consume_request(self): def commands(*args, **kwargs): if "pull" in args: raise subprocess.CalledProcessError(1, args) return "" with patch.object(agent, "command", side_effect=commands), \ patch.object(agent.subprocess, "run") as run: with self.assertRaises(subprocess.CalledProcessError): self.attempt() run.assert_not_called() self.assertFalse(list(self.state.glob("*.json"))) def test_revision_mismatch_does_not_launch(self): def commands(*args, **kwargs): if "inspect" in args: return '[{"Config": {"Labels": {}}}]' return "" with patch.object(agent, "command", side_effect=commands), self.assertRaises(ValueError): self.attempt() self.assertFalse(list(self.state.glob("*.json"))) def test_request_id_cannot_be_reused_for_different_image(self): path = self.state / f"{self.request['request_id']}.json" agent.atomic_json(path, {"request": {**self.request, "revision": "c" * 40}, "status": "finished"}) with self.assertRaises(ValueError): self.attempt() def test_environment_does_not_override_external_app_env(self): with patch.dict("os.environ", {"PGHOST": "wrong", "RAW_DATA_DIR": "wrong", "APP_UID": "0"}): env = agent.compose_environment(self.request["image"]) self.assertNotIn("PGHOST", env) self.assertNotIn("RAW_DATA_DIR", env) self.assertNotIn("APP_UID", env) self.assertEqual(env["CAR_TRACKER_IMAGE"], self.request["image"]) def test_branch_names_are_not_release_inputs(self): calls = [] def command(*args): calls.append(args) if "rev-parse" in args: raise subprocess.CalledProcessError(1, args) return "" with patch.object(requester, "command", side_effect=command), \ self.assertRaises(subprocess.CalledProcessError): requester.resolve_release("master") self.assertIn(("git", "rev-parse", "--verify", "refs/tags/master^{commit}"), calls) def test_multiple_requests_between_polls_are_consumed_in_order(self): root = self.state (root / "desired.git").mkdir() second = {**self.request, "request_id": str(uuid.uuid4())} def commands(*args, **kwargs): if "log" in args: return "commit1\ncommit2\n" if "show" in args: return json.dumps(second if args[-1].startswith("commit2:") else self.request) return "" with patch.object(agent, "command", side_effect=commands), patch.object(agent, "attempt") as attempt: agent.reconcile(root) self.assertEqual([call.args[0] for call in attempt.call_args_list], [self.request, second]) def test_unknown_attempt_blocks_new_requests(self): state = self.state / "state/requests" state.mkdir(parents=True) agent.atomic_json(state / "old.json", {"status": "attempted"}) with patch.object(agent, "command") as command, self.assertRaises(ValueError): agent.reconcile(self.state) command.assert_not_called() def test_queue_creates_only_state_branch_and_retains_request_history(self): seed = self.state / "seed" remote = self.state / "remote.git" seed.mkdir() requester.command("git", "init", "--initial-branch=master", cwd=seed) (seed / "README.md").write_text("Trusted application source\n") requester.command("git", "add", "README.md", cwd=seed) requester.command("git", "-c", "user.name=Test", "-c", "user.email=test@localhost", "commit", "-m", "initial", cwd=seed) requester.command("git", "clone", "--bare", str(seed), str(remote)) before = requester.command("git", "-C", str(remote), "rev-parse", "master") real_command = requester.command def commands(*args, **kwargs): if args[:4] == ("git", "remote", "add", "origin"): args = (*args[:4], str(remote)) return real_command(*args, **kwargs) second = {**self.request, "request_id": str(uuid.uuid4())} with patch.dict("os.environ", { "GITOPS_USER": "test", "GITOPS_TOKEN": "test-only", "GITOPS_BRANCH": "gitops", "GITOPS_URL": "https://gitea.vandeplanque.com/tonio/car-tracker.git", }), patch.object(requester, "command", side_effect=commands): requester.queue(self.request) requester.queue(second) self.assertEqual(real_command("git", "-C", str(remote), "rev-parse", "master"), before) commits = real_command("git", "-C", str(remote), "log", "--reverse", "--format=%H", "gitops", "--", "deploy/request.json").splitlines() self.assertEqual(len(commits), 2) requests = [json.loads(real_command("git", "-C", str(remote), "show", f"{commit}:deploy/request.json")) for commit in commits] self.assertEqual(requests, [self.request, second])