tegwick
628f984a10
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a07b5b-ea58-7ad2-bdbb-0b1c995cfc35
105 lines
3.3 KiB
Python
105 lines
3.3 KiB
Python
import json
|
|
import os
|
|
import signal
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
|
|
|
|
def cli(*args, env=None):
|
|
return subprocess.run(
|
|
[sys.executable, "-m", "coordination_engine.cli", *args],
|
|
capture_output=True,
|
|
text=True,
|
|
env=env,
|
|
timeout=10,
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def cli_env(tmp_path):
|
|
env = dict(
|
|
os.environ,
|
|
PYTHONPATH=str(ROOT / "src"),
|
|
COORDINATION_STATE_DIR=str(tmp_path / "state"),
|
|
COORDINATION_SOCKET=str(tmp_path / "coord.sock"),
|
|
COORDINATION_CONFIG=str(tmp_path / "config.toml"),
|
|
)
|
|
env.pop("STATEHUB_API_BASE", None)
|
|
return env
|
|
|
|
|
|
@pytest.mark.parametrize("flag", ["--help", "-h", "--version", "-V"])
|
|
def test_help_version(flag, cli_env):
|
|
result = cli(flag, env=cli_env)
|
|
assert result.returncode == 0
|
|
assert not Path(cli_env["COORDINATION_STATE_DIR"]).exists()
|
|
|
|
|
|
@pytest.mark.parametrize("shell", ["bash", "zsh", "fish"])
|
|
def test_completion(shell, cli_env):
|
|
result = cli("completion", shell, env=cli_env)
|
|
assert result.returncode == 0
|
|
assert "coordination-engine" in result.stdout
|
|
|
|
|
|
def test_safe_default_refuses_unconfigured_service(cli_env):
|
|
result = cli("once", env=cli_env)
|
|
assert result.returncode == 1
|
|
assert result.stderr
|
|
|
|
|
|
def test_database_commands(cli_env):
|
|
assert cli("db-version", env=cli_env).stdout.strip() == "1"
|
|
result = cli("backup", env=cli_env)
|
|
assert result.returncode == 0
|
|
assert Path(result.stdout.strip()).exists()
|
|
assert cli("history", env=cli_env).stdout.strip() == "[]"
|
|
|
|
|
|
def test_service_lifecycle_and_restart(cli_env, tmp_path):
|
|
gita = tmp_path / "gita"
|
|
gita.write_text(
|
|
'#!/bin/sh\nprintf "x,demo,/demo\\nx,coordination-engine,/coordination\\n"\n'
|
|
)
|
|
gita.chmod(0o700)
|
|
cli_env["PATH"] = str(tmp_path) + os.pathsep + cli_env["PATH"]
|
|
Path(cli_env["COORDINATION_CONFIG"]).write_text(
|
|
'[coordination]\nrepos=["demo"]\napi_base="http://127.0.0.1:1"\ntimeout=0.1\n'
|
|
)
|
|
for stop in ("stop", "signal"):
|
|
proc = subprocess.Popen(
|
|
[sys.executable, "-m", "coordination_engine.cli", "serve"],
|
|
env=cli_env,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
text=True,
|
|
)
|
|
try:
|
|
deadline = time.monotonic() + 5
|
|
while time.monotonic() < deadline:
|
|
if Path(cli_env["COORDINATION_SOCKET"]).exists():
|
|
break
|
|
if proc.poll() is not None:
|
|
pytest.fail(proc.communicate()[1])
|
|
time.sleep(0.02)
|
|
assert cli("ping", env=cli_env).returncode == 0
|
|
result = cli("status", env=cli_env)
|
|
assert json.loads(result.stdout)["schema_version"] == 1
|
|
assert cli("serve", env=cli_env).returncode == 1
|
|
if stop == "stop":
|
|
assert cli("stop", env=cli_env).returncode == 0
|
|
else:
|
|
proc.send_signal(signal.SIGTERM)
|
|
proc.communicate(timeout=5)
|
|
assert proc.returncode == 0
|
|
assert not Path(cli_env["COORDINATION_SOCKET"]).exists()
|
|
finally:
|
|
if proc.poll() is None:
|
|
proc.kill()
|
|
proc.communicate()
|