252 lines
11 KiB
Python
252 lines
11 KiB
Python
from __future__ import annotations
|
|
|
|
from types import SimpleNamespace
|
|
|
|
from ariadne.services import game_mode as game_mode_module
|
|
from ariadne.services.game_mode import GameModeService
|
|
|
|
|
|
def _settings(workloads=None) -> SimpleNamespace:
|
|
return SimpleNamespace(
|
|
game_mode_node_name="titan-24",
|
|
game_mode_displace_workloads=workloads
|
|
or [{"kind": "Deployment", "namespace": "openclaw", "name": "openclaw-ollama", "restoreReplicas": 1}],
|
|
)
|
|
|
|
|
|
def _handoff_settings() -> SimpleNamespace:
|
|
return SimpleNamespace(
|
|
game_mode_node_name="titan-24",
|
|
game_mode_displace_workloads=[],
|
|
game_mode_lease_namespace="hermes",
|
|
game_mode_lease_name="titan-24-gpu-owner",
|
|
game_mode_ollama_url="http://ollama.test:11434",
|
|
game_mode_ollama_model="gpt-oss:20b",
|
|
game_mode_ollama_request_timeout_sec=30.0,
|
|
game_mode_transition_timeout_sec=2.0,
|
|
game_mode_poll_interval_sec=0.1,
|
|
)
|
|
|
|
|
|
def test_game_mode_start_and_stop_patch_scale(monkeypatch) -> None:
|
|
monkeypatch.setattr(game_mode_module, "settings", _settings())
|
|
calls: list[tuple[str, dict]] = []
|
|
replicas = {"desired": 1, "current": 1}
|
|
|
|
def fake_get_json(_path):
|
|
return {"spec": {"replicas": replicas["desired"]}, "status": {"replicas": replicas["current"]}}
|
|
|
|
def fake_patch_json(path, payload):
|
|
calls.append((path, payload))
|
|
replicas["desired"] = payload["spec"]["replicas"]
|
|
replicas["current"] = payload["spec"]["replicas"]
|
|
return {"ok": True}
|
|
|
|
monkeypatch.setattr(game_mode_module, "get_json", fake_get_json)
|
|
monkeypatch.setattr(game_mode_module, "patch_json", fake_patch_json)
|
|
monkeypatch.setattr(game_mode_module, "set_game_mode_state", lambda *args, **kwargs: None)
|
|
monkeypatch.setattr(game_mode_module, "set_game_mode_managed_replicas", lambda *args, **kwargs: None)
|
|
monkeypatch.setattr(game_mode_module, "record_game_mode_transition", lambda *args, **kwargs: None)
|
|
|
|
svc = GameModeService()
|
|
assert svc.start("Arc Raiders")["active"] is True
|
|
assert calls[-1][1] == {"spec": {"replicas": 0}}
|
|
assert svc.stop()["active"] is False
|
|
assert calls[-1][1] == {"spec": {"replicas": 1}}
|
|
|
|
|
|
def test_game_mode_supports_statefulset_workload(monkeypatch) -> None:
|
|
workload = [{"kind": "StatefulSet", "namespace": "hermes", "name": "hermes-llm", "restoreReplicas": "2"}]
|
|
monkeypatch.setattr(game_mode_module, "settings", _settings(workload))
|
|
calls: list[str] = []
|
|
monkeypatch.setattr(game_mode_module, "get_json", lambda _path: {"spec": {"replicas": 2}, "status": {"replicas": 2}})
|
|
monkeypatch.setattr(game_mode_module, "patch_json", lambda path, _payload: calls.append(path) or {"ok": True})
|
|
monkeypatch.setattr(game_mode_module, "set_game_mode_state", lambda *args, **kwargs: None)
|
|
monkeypatch.setattr(game_mode_module, "set_game_mode_managed_replicas", lambda *args, **kwargs: None)
|
|
monkeypatch.setattr(game_mode_module, "record_game_mode_transition", lambda *args, **kwargs: None)
|
|
|
|
GameModeService().start("wolf")
|
|
assert calls == ["/apis/apps/v1/namespaces/hermes/statefulsets/hermes-llm/scale"]
|
|
|
|
|
|
def test_game_mode_ignores_invalid_workloads_and_fallback_replicas(monkeypatch) -> None:
|
|
workloads = [
|
|
{"kind": "Deployment", "namespace": "", "name": "missing"},
|
|
{"kind": "Deployment", "namespace": "openclaw", "name": "ollama", "restoreReplicas": "bad"},
|
|
]
|
|
monkeypatch.setattr(game_mode_module, "settings", _settings(workloads))
|
|
monkeypatch.setattr(game_mode_module, "get_json", lambda _path: {"spec": {"replicas": None}, "status": {}})
|
|
monkeypatch.setattr(game_mode_module, "set_game_mode_state", lambda *args, **kwargs: None)
|
|
monkeypatch.setattr(game_mode_module, "set_game_mode_managed_replicas", lambda *args, **kwargs: None)
|
|
|
|
status = GameModeService().status()
|
|
assert status["workloads"][0]["restore_replicas"] == 1
|
|
assert status["workloads"][0]["desired_replicas"] is None
|
|
|
|
|
|
def test_game_mode_status_uses_current_when_desired_missing(monkeypatch) -> None:
|
|
monkeypatch.setattr(game_mode_module, "settings", _settings())
|
|
monkeypatch.setattr(game_mode_module, "get_json", lambda _path: {"spec": {}, "status": {"replicas": 0}})
|
|
monkeypatch.setattr(game_mode_module, "set_game_mode_state", lambda *args, **kwargs: None)
|
|
monkeypatch.setattr(game_mode_module, "set_game_mode_managed_replicas", lambda *args, **kwargs: None)
|
|
|
|
status = GameModeService().status()
|
|
|
|
assert status["active"] is True
|
|
assert status["workloads"][0]["effective_replicas"] == 0
|
|
|
|
|
|
def test_game_mode_rejects_unsupported_kind(monkeypatch) -> None:
|
|
monkeypatch.setattr(game_mode_module, "settings", _settings([{"kind": "Job", "namespace": "x", "name": "y"}]))
|
|
monkeypatch.setattr(game_mode_module, "record_game_mode_transition", lambda *args, **kwargs: None)
|
|
|
|
try:
|
|
GameModeService().start("arc")
|
|
except ValueError as exc:
|
|
assert "unsupported" in str(exc)
|
|
else:
|
|
raise AssertionError("unsupported kind should fail")
|
|
|
|
|
|
def test_game_mode_records_stop_errors(monkeypatch) -> None:
|
|
monkeypatch.setattr(game_mode_module, "settings", _settings())
|
|
transitions = []
|
|
monkeypatch.setattr(game_mode_module, "record_game_mode_transition", lambda *args: transitions.append(args))
|
|
monkeypatch.setattr(game_mode_module, "patch_json", lambda *_args, **_kwargs: (_ for _ in ()).throw(RuntimeError("boom")))
|
|
|
|
try:
|
|
GameModeService().stop("arc")
|
|
except RuntimeError:
|
|
pass
|
|
|
|
assert transitions[-1] == ("stop", "error", "arc")
|
|
|
|
|
|
def test_game_mode_handoff_gates_before_unload_and_warms_before_release(monkeypatch) -> None:
|
|
monkeypatch.setattr(game_mode_module, "settings", _handoff_settings())
|
|
events: list[tuple[str, str]] = []
|
|
state = {"owner": "hermes", "loaded": True, "game": "unknown"}
|
|
|
|
def fake_get_json(_path):
|
|
return {
|
|
"metadata": {"annotations": {"ai.bstein.dev/game": state["game"]}},
|
|
"spec": {"holderIdentity": state["owner"]},
|
|
}
|
|
|
|
def fake_patch_json(_path, payload):
|
|
state["owner"] = payload["spec"]["holderIdentity"]
|
|
state["game"] = payload["metadata"]["annotations"]["ai.bstein.dev/game"]
|
|
events.append(("owner", state["owner"]))
|
|
return {"ok": True}
|
|
|
|
def fake_post(path, payload):
|
|
assert path == "/api/generate"
|
|
if payload.get("prompt"):
|
|
events.append(("canary", state["owner"]))
|
|
return {"response": "READY"}
|
|
state["loaded"] = payload["keep_alive"] != 0
|
|
events.append(("model", "loaded" if state["loaded"] else "unloaded"))
|
|
return {"response": ""}
|
|
|
|
monkeypatch.setattr(game_mode_module, "get_json", fake_get_json)
|
|
monkeypatch.setattr(game_mode_module, "patch_json", fake_patch_json)
|
|
monkeypatch.setattr(GameModeService, "_ollama_post", lambda self, path, payload: fake_post(path, payload))
|
|
monkeypatch.setattr(GameModeService, "_model_loaded", lambda self: state["loaded"])
|
|
monkeypatch.setattr(game_mode_module, "set_game_mode_state", lambda *args, **kwargs: None)
|
|
monkeypatch.setattr(game_mode_module, "record_game_mode_transition", lambda *args, **kwargs: None)
|
|
|
|
svc = GameModeService()
|
|
started = svc.start("Arc Raiders", "priority test")
|
|
assert started["gpu_owner"] == "wolf"
|
|
assert started["inference_path"] == "fallback"
|
|
assert events[:3] == [("owner", "wolf-draining"), ("model", "unloaded"), ("owner", "wolf")]
|
|
|
|
events.clear()
|
|
stopped = svc.stop("Arc Raiders")
|
|
assert stopped["gpu_owner"] == "hermes"
|
|
assert stopped["local_inference_ready"] is True
|
|
assert events == [
|
|
("owner", "hermes-warming"),
|
|
("model", "loaded"),
|
|
("canary", "hermes-warming"),
|
|
("owner", "hermes"),
|
|
]
|
|
|
|
|
|
def test_game_mode_handoff_failure_stays_on_fallback(monkeypatch) -> None:
|
|
monkeypatch.setattr(game_mode_module, "settings", _handoff_settings())
|
|
owners: list[str] = []
|
|
monkeypatch.setattr(
|
|
game_mode_module,
|
|
"patch_json",
|
|
lambda _path, payload: owners.append(payload["spec"]["holderIdentity"]) or {"ok": True},
|
|
)
|
|
monkeypatch.setattr(GameModeService, "_unload_model", lambda self: (_ for _ in ()).throw(RuntimeError("busy")))
|
|
monkeypatch.setattr(game_mode_module, "record_game_mode_transition", lambda *args, **kwargs: None)
|
|
|
|
try:
|
|
GameModeService().start("wolf")
|
|
except RuntimeError as exc:
|
|
assert str(exc) == "busy"
|
|
else:
|
|
raise AssertionError("failed unload should stop the Wolf transition")
|
|
|
|
assert owners == ["wolf-draining", "fallback-error"]
|
|
|
|
|
|
def test_game_mode_status_degrades_to_fallback_when_dependencies_fail(monkeypatch) -> None:
|
|
monkeypatch.setattr(game_mode_module, "settings", _handoff_settings())
|
|
monkeypatch.setattr(game_mode_module, "get_json", lambda _path: (_ for _ in ()).throw(RuntimeError("api down")))
|
|
monkeypatch.setattr(GameModeService, "_model_loaded", lambda self: (_ for _ in ()).throw(RuntimeError("ollama down")))
|
|
monkeypatch.setattr(game_mode_module, "set_game_mode_state", lambda *args, **kwargs: None)
|
|
|
|
status = GameModeService().status()
|
|
|
|
assert status["gpu_owner"] == "unavailable"
|
|
assert status["local_inference_ready"] is False
|
|
assert status["inference_path"] == "fallback"
|
|
assert "api down" in status["error"]
|
|
assert "ollama down" in status["error"]
|
|
|
|
|
|
def test_game_mode_ollama_client_and_loaded_model_detection(monkeypatch) -> None:
|
|
monkeypatch.setattr(game_mode_module, "settings", _handoff_settings())
|
|
calls: list[tuple[str, str, object]] = []
|
|
|
|
class FakeResponse:
|
|
def __init__(self, payload):
|
|
self.payload = payload
|
|
|
|
def raise_for_status(self):
|
|
return None
|
|
|
|
def json(self):
|
|
return self.payload
|
|
|
|
class FakeClient:
|
|
def __init__(self, timeout):
|
|
calls.append(("timeout", "", timeout))
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *_args):
|
|
return False
|
|
|
|
def get(self, url):
|
|
calls.append(("get", url, None))
|
|
return FakeResponse({"models": [{"model": "gpt-oss:20b"}]})
|
|
|
|
def post(self, url, json):
|
|
calls.append(("post", url, json))
|
|
return FakeResponse({"response": "READY"})
|
|
|
|
monkeypatch.setattr(game_mode_module.httpx, "Client", FakeClient)
|
|
svc = GameModeService()
|
|
|
|
assert svc._ollama_get("/api/ps")["models"]
|
|
assert svc._ollama_post("/api/generate", {"prompt": "test"}) == {"response": "READY"}
|
|
assert svc._model_loaded() is True
|
|
assert ("get", "http://ollama.test:11434/api/ps", None) in calls
|
|
assert ("post", "http://ollama.test:11434/api/generate", {"prompt": "test"}) in calls
|