m4
197 lines
7.1 KiB
Python
197 lines
7.1 KiB
Python
"""HTTP-layer tests for the system update routes."""
|
|
|
|
import json
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from starlette.applications import Starlette
|
|
from starlette.routing import Route
|
|
from starlette.testclient import TestClient
|
|
|
|
from EvoScientist import update_check
|
|
from EvoScientist.langgraph_dev import http as http_mod
|
|
|
|
|
|
async def _allow(request):
|
|
return None
|
|
|
|
|
|
@pytest.fixture
|
|
def client(monkeypatch, tmp_path):
|
|
monkeypatch.setattr(http_mod, "_authenticate_system", _allow, raising=False)
|
|
monkeypatch.setattr(http_mod, "_authenticate_system_write", _allow, raising=False)
|
|
monkeypatch.setattr(http_mod, "_schedule_self_exit", lambda delay=0.5: None)
|
|
monkeypatch.setenv("EVOSCIENTIST_UPDATE_STAGING_DIR", str(tmp_path))
|
|
app = Starlette(
|
|
routes=[
|
|
Route("/internal/system/update", http_mod.post_system_update, methods=["POST"]),
|
|
Route(
|
|
"/internal/system/update/status",
|
|
http_mod.get_system_update_status,
|
|
methods=["GET"],
|
|
),
|
|
Route(
|
|
"/internal/system/rollback-versions",
|
|
http_mod.get_system_rollback_versions,
|
|
methods=["GET"],
|
|
),
|
|
Route(
|
|
"/internal/system/update/rollback",
|
|
http_mod.post_system_update_rollback,
|
|
methods=["POST"],
|
|
),
|
|
]
|
|
)
|
|
return TestClient(app)
|
|
|
|
|
|
def _info(has_update, breaking_db=False):
|
|
return {
|
|
"current_version": "0.2.8",
|
|
"latest_version": "0.3.0",
|
|
"has_update": has_update,
|
|
"release_url": None,
|
|
"release_notes": None,
|
|
"published_at": None,
|
|
"cached": False,
|
|
"warning": None,
|
|
"breaking_db": breaking_db,
|
|
}
|
|
|
|
|
|
def _probe_docker():
|
|
from EvoScientist.updater import SystemProbe
|
|
|
|
return SystemProbe(in_container=True, invocation_id=None, uv_tool_pkg=False, pipx_pkg=False)
|
|
|
|
|
|
def _probe_uvtool():
|
|
from EvoScientist.updater import SystemProbe
|
|
|
|
return SystemProbe(in_container=False, invocation_id=None, uv_tool_pkg=True, pipx_pkg=False)
|
|
|
|
|
|
def test_update_409_when_no_update(client, monkeypatch):
|
|
monkeypatch.setattr(update_check, "get_update_info", lambda force=False: _info(has_update=False))
|
|
res = client.post("/internal/system/update")
|
|
assert res.status_code == 409
|
|
assert res.json()["code"] == "ALREADY_UP_TO_DATE"
|
|
|
|
|
|
def test_update_docker_returns_guidance(client, monkeypatch):
|
|
monkeypatch.setattr(update_check, "get_update_info", lambda force=False: _info(has_update=True))
|
|
monkeypatch.setattr(http_mod, "probe_system", lambda environ=None: _probe_docker())
|
|
res = client.post("/internal/system/update")
|
|
assert res.status_code == 200
|
|
assert res.json()["status"] == "manual"
|
|
assert "docker compose" in res.json()["guidance"]
|
|
|
|
|
|
def test_update_202_spawns_updater(client, monkeypatch, tmp_path):
|
|
spawned = {}
|
|
monkeypatch.setattr(update_check, "get_update_info", lambda force=False: _info(has_update=True))
|
|
monkeypatch.setattr(http_mod, "probe_system", lambda environ=None: _probe_uvtool())
|
|
monkeypatch.setattr(
|
|
update_check,
|
|
"download_update",
|
|
lambda version=None: {
|
|
"version": "0.3.0",
|
|
"file": "EvoScientist-0.3.0-py3-none-any.whl",
|
|
"path": str(tmp_path / "v0.3.0" / "EvoScientist-0.3.0-py3-none-any.whl"),
|
|
"suggested_command": "...",
|
|
},
|
|
)
|
|
(tmp_path / "v0.3.0").mkdir(parents=True)
|
|
(tmp_path / "v0.3.0" / "EvoScientist-0.3.0-py3-none-any.whl").write_bytes(b"x")
|
|
monkeypatch.setattr(
|
|
http_mod,
|
|
"spawn_updater",
|
|
lambda u, p, pid: spawned.setdefault("ok", (str(u), str(p), pid)),
|
|
)
|
|
res = client.post("/internal/system/update")
|
|
assert res.status_code == 202
|
|
body = res.json()
|
|
assert body["status"] == "applying"
|
|
assert body["need_restart"] is True
|
|
assert spawned["ok"][2] > 0 # parent pid
|
|
plan = json.loads((tmp_path / "v0.3.0" / "plan.json").read_text())
|
|
assert plan["version"] == "0.3.0"
|
|
assert plan["respawn_command"]
|
|
|
|
|
|
def test_update_409_when_update_in_progress(client, monkeypatch, tmp_path):
|
|
monkeypatch.setattr(update_check, "get_update_info", lambda force=False: _info(has_update=True))
|
|
monkeypatch.setattr(http_mod, "probe_system", lambda environ=None: _probe_uvtool())
|
|
monkeypatch.setattr(
|
|
update_check,
|
|
"download_update",
|
|
lambda version=None: {
|
|
"version": "0.3.0",
|
|
"file": "f.whl",
|
|
"path": str(tmp_path / "v0.3.0" / "f.whl"),
|
|
"suggested_command": "...",
|
|
},
|
|
)
|
|
(tmp_path / "v0.3.0").mkdir(parents=True)
|
|
(tmp_path / "v0.3.0" / "f.whl").write_bytes(b"x")
|
|
(tmp_path / "v0.3.0" / "plan.lock").write_text("") # lock held
|
|
monkeypatch.setattr(http_mod, "spawn_updater", lambda u, p, pid: None)
|
|
res = client.post("/internal/system/update")
|
|
assert res.status_code == 409
|
|
assert res.json()["code"] == "UPDATE_IN_PROGRESS"
|
|
|
|
|
|
def test_update_breaking_requires_confirm(client, monkeypatch):
|
|
monkeypatch.setattr(
|
|
update_check,
|
|
"get_update_info",
|
|
lambda force=False: _info(has_update=True, breaking_db=True),
|
|
)
|
|
res = client.post("/internal/system/update")
|
|
assert res.status_code == 409
|
|
assert res.json()["code"] == "BREAKING_DB_CONFIRM_REQUIRED"
|
|
|
|
|
|
def test_status_none_then_result(client, tmp_path):
|
|
assert client.get("/internal/system/update/status").json()["status"] == "none"
|
|
(tmp_path / "last-result.json").write_text(json.dumps({"status": "success", "version": "0.3.0"}))
|
|
assert client.get("/internal/system/update/status").json()["status"] == "success"
|
|
|
|
|
|
def test_rollback_versions_passthrough(client, monkeypatch):
|
|
monkeypatch.setattr(
|
|
update_check,
|
|
"list_rollback_versions",
|
|
lambda limit=3: [{"version": "0.2.9", "published_at": "p", "release_url": "u"}],
|
|
)
|
|
res = client.get("/internal/system/rollback-versions")
|
|
assert res.json()["versions"][0]["version"] == "0.2.9"
|
|
|
|
|
|
def test_rollback_rejects_disallowed_version(client, monkeypatch):
|
|
monkeypatch.setattr(update_check, "is_allowed_rollback", lambda v: False)
|
|
res = client.post("/internal/system/update/rollback", json={"version": "0.0.1"})
|
|
assert res.status_code == 400
|
|
|
|
|
|
def test_rollback_202_uses_same_pipeline(client, monkeypatch, tmp_path):
|
|
monkeypatch.setattr(update_check, "is_allowed_rollback", lambda v: True)
|
|
monkeypatch.setattr(http_mod, "probe_system", lambda environ=None: _probe_uvtool())
|
|
monkeypatch.setattr(
|
|
update_check,
|
|
"download_update",
|
|
lambda version=None: {
|
|
"version": version,
|
|
"file": "f.whl",
|
|
"path": str(tmp_path / f"v{version}" / "f.whl"),
|
|
"suggested_command": "...",
|
|
},
|
|
)
|
|
(tmp_path / "v0.2.9").mkdir(parents=True)
|
|
(tmp_path / "v0.2.9" / "f.whl").write_bytes(b"x")
|
|
monkeypatch.setattr(http_mod, "spawn_updater", lambda u, p, pid: None)
|
|
res = client.post("/internal/system/update/rollback", json={"version": "0.2.9"})
|
|
assert res.status_code == 202
|
|
plan = json.loads((tmp_path / "v0.2.9" / "plan.json").read_text())
|
|
assert plan["version"] == "0.2.9"
|