m4
98 lines
3.2 KiB
Python
98 lines
3.2 KiB
Python
"""Tests for the standalone updater runner (run_plan / wait_for_pid_exit)."""
|
|
|
|
import json
|
|
import os
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
from EvoScientist.updater import run_plan, wait_for_pid_exit
|
|
|
|
|
|
def test_wait_for_pid_exit_true_when_gone():
|
|
alive = {"n": 3}
|
|
|
|
def pid_alive(pid):
|
|
alive["n"] -= 1
|
|
return alive["n"] > 0
|
|
|
|
assert wait_for_pid_exit(1234, timeout=10, poll=0, sleep=lambda s: None, pid_alive=pid_alive)
|
|
|
|
|
|
def test_wait_for_pid_exit_timeout():
|
|
assert not wait_for_pid_exit(1234, timeout=0.01, poll=0, sleep=lambda s: None, pid_alive=lambda p: True)
|
|
|
|
|
|
def _write_plan(tmp_path, **over):
|
|
target = tmp_path / "v1"
|
|
target.mkdir()
|
|
plan = {
|
|
"version": "1.0.0",
|
|
"artifact": str(target / "pkg.whl"),
|
|
"install_command": ["true"],
|
|
"restart": {"strategy": "cli", "unit": None},
|
|
"respawn_command": ["echo", "restarted"],
|
|
"cwd": str(tmp_path),
|
|
"deploy_mode": "full",
|
|
"previous_version": "0.9.9",
|
|
"result_path": str(target / "update-result.json"),
|
|
"log_path": str(target / "updater.log"),
|
|
}
|
|
plan.update(over)
|
|
plan_path = target / "plan.json"
|
|
plan_path.write_text(json.dumps(plan), encoding="utf-8")
|
|
return plan_path
|
|
|
|
|
|
def test_run_plan_success(tmp_path):
|
|
plan_path = _write_plan(tmp_path)
|
|
rc = run_plan(plan_path, parent_pid=os.getpid(), pid_alive=lambda p: False)
|
|
assert rc == 0
|
|
result = json.loads(Path(json.loads(plan_path.read_text())["result_path"]).read_text())
|
|
assert result["status"] == "success"
|
|
assert result["version"] == "1.0.0"
|
|
last = json.loads((tmp_path / "last-result.json").read_text())
|
|
assert last["status"] == "success"
|
|
|
|
|
|
def test_run_plan_install_failure(tmp_path):
|
|
plan_path = _write_plan(
|
|
tmp_path,
|
|
install_command=[sys.executable, "-c", "import sys; sys.exit(3)"],
|
|
)
|
|
rc = run_plan(plan_path, parent_pid=os.getpid(), pid_alive=lambda p: False)
|
|
assert rc == 1
|
|
result = json.loads(Path(json.loads(plan_path.read_text())["result_path"]).read_text())
|
|
assert result["status"] == "failed"
|
|
assert result["stage"] == "install"
|
|
|
|
|
|
def test_run_plan_restart_failure(tmp_path):
|
|
plan_path = _write_plan(tmp_path, respawn_command=["/nonexistent/evosci-binary"])
|
|
rc = run_plan(plan_path, parent_pid=os.getpid(), pid_alive=lambda p: False)
|
|
assert rc == 1
|
|
result = json.loads(Path(json.loads(plan_path.read_text())["result_path"]).read_text())
|
|
assert result["status"] == "installed_restart_failed"
|
|
|
|
|
|
def test_run_plan_releases_plan_lock(tmp_path):
|
|
plan_path = _write_plan(tmp_path)
|
|
lock = plan_path.with_suffix(".lock")
|
|
lock.write_text("")
|
|
run_plan(plan_path, parent_pid=os.getpid(), pid_alive=lambda p: False)
|
|
assert not lock.exists()
|
|
|
|
|
|
def test_run_plan_parent_never_exits(tmp_path, monkeypatch):
|
|
monkeypatch.setattr("EvoScientist.updater.PID_WAIT_TIMEOUT", 0.01)
|
|
plan_path = _write_plan(tmp_path)
|
|
rc = run_plan(
|
|
plan_path,
|
|
parent_pid=os.getpid(),
|
|
sleep=lambda s: None,
|
|
pid_alive=lambda p: True,
|
|
)
|
|
assert rc == 1
|
|
result = json.loads(Path(json.loads(plan_path.read_text())["result_path"]).read_text())
|
|
assert result["status"] == "failed"
|
|
assert result["stage"] == "wait"
|