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"]