ariadne/tests/test_game_mode.py

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