laily
adaa8e4edd
agent-runtime executes fixed tester node operations over JSON stdin. deployer prepare brings up the local test Compose service the same way. test-preflight now passes the tester binding; tester select skips unauthorized cases before creating a runner.
89 lines
3.4 KiB
Python
89 lines
3.4 KiB
Python
#!/usr/bin/env python3
|
|
|
|
from __future__ import annotations
|
|
|
|
import importlib.util
|
|
import json
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
|
|
SCRIPT = Path(__file__).resolve().parents[1] / "scripts" / "deploy" / "prepare.py"
|
|
SPEC = importlib.util.spec_from_file_location("deployer_prepare", SCRIPT)
|
|
assert SPEC and SPEC.loader
|
|
prepare = importlib.util.module_from_spec(SPEC)
|
|
SPEC.loader.exec_module(prepare)
|
|
|
|
|
|
class PrepareTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.temp = tempfile.TemporaryDirectory()
|
|
self.root = Path(self.temp.name)
|
|
deployer = self.root / ".pouch" / "deployer" / "test"
|
|
deployer.mkdir(parents=True)
|
|
(deployer / "compose.yaml").write_text("services:\n app:\n image: example:test\n", encoding="utf-8")
|
|
(deployer / "_config.yaml").write_text("node: devbox\nname: writes-test\n", encoding="utf-8")
|
|
self.request = {"operation": "prepare", "environment": "test", "project_root": str(self.root), "host": "http://test.local"}
|
|
|
|
def tearDown(self) -> None:
|
|
self.temp.cleanup()
|
|
|
|
def test_prod_is_rejected_without_inspecting_or_executing_compose(self):
|
|
calls = []
|
|
result = prepare.prepare_test({**self.request, "environment": "prod"}, executor=lambda command, cwd: calls.append(command) or 0)
|
|
self.assertEqual(result["status"], "blocked")
|
|
self.assertEqual(calls, [])
|
|
|
|
def test_missing_config_is_blocked(self):
|
|
(self.root / ".pouch" / "deployer" / "test" / "_config.yaml").unlink()
|
|
result = prepare.prepare_test(self.request, local_node=lambda node: True)
|
|
self.assertEqual(result["status"], "blocked")
|
|
self.assertIn("discoverable", result["observed"])
|
|
|
|
def test_mock_compose_prepare_is_ready(self):
|
|
calls = []
|
|
|
|
def executor(command, cwd):
|
|
calls.append((command, cwd))
|
|
return 0
|
|
|
|
result = prepare.prepare_test(self.request, executor=executor, local_node=lambda node: node == "devbox")
|
|
self.assertEqual(result["status"], "ready")
|
|
self.assertEqual(calls[0][0][-2:], ["config", "--quiet"])
|
|
self.assertEqual(calls[1][0][-3:], ["up", "-d", "--force-recreate"])
|
|
self.assertEqual(calls[2][0][-1], "ps")
|
|
self.assertEqual(calls[0][1], self.root / ".pouch" / "deployer" / "test")
|
|
|
|
def test_compose_failures_are_environment_fail(self):
|
|
for failed_step in range(3):
|
|
calls = []
|
|
|
|
def executor(command, cwd):
|
|
del cwd
|
|
calls.append(command)
|
|
return 1 if len(calls) == failed_step + 1 else 0
|
|
|
|
result = prepare.prepare_test(self.request, executor=executor, local_node=lambda node: True)
|
|
self.assertEqual(result["status"], "environment-fail")
|
|
self.assertNotEqual(result["status"], "ready")
|
|
|
|
def test_json_stdout_has_no_diagnostics(self):
|
|
process = subprocess.run(
|
|
[sys.executable, str(SCRIPT)],
|
|
input=json.dumps({**self.request, "environment": "prod"}),
|
|
text=True,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
check=False,
|
|
)
|
|
self.assertEqual(process.returncode, 0)
|
|
self.assertEqual(json.loads(process.stdout)["status"], "blocked")
|
|
self.assertEqual(process.stderr, "")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|