184 lines
8.1 KiB
Python
184 lines
8.1 KiB
Python
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])
|