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() -> SimpleNamespace: return SimpleNamespace( game_mode_node_name="titan-24", game_mode_displace_workloads=[ { "kind": "Deployment", "namespace": "openclaw", "name": "openclaw-ollama", "restoreReplicas": 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() started = svc.start("Arc Raiders") assert started["active"] is True assert calls[-1][1] == {"spec": {"replicas": 0}} stopped = svc.stop() assert stopped["active"] is False assert calls[-1][1] == {"spec": {"replicas": 1}} def test_game_mode_status_reports_workload(monkeypatch) -> None: monkeypatch.setattr(game_mode_module, "settings", _settings()) monkeypatch.setattr( game_mode_module, "get_json", lambda _path: {"spec": {"replicas": 0}, "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["status"] == "active" assert status["workloads"][0]["name"] == "openclaw-ollama" def test_game_mode_supports_statefulset_workload(monkeypatch) -> None: monkeypatch.setattr( game_mode_module, "settings", SimpleNamespace( game_mode_node_name="titan-24", game_mode_displace_workloads=[ { "kind": "StatefulSet", "namespace": "hermes", "name": "hermes-llm", "restoreReplicas": "2", } ], ), ) calls: list[str] = [] monkeypatch.setattr( game_mode_module, "get_json", lambda _path: {"spec": {"replicas": 2}, "status": {"replicas": 2}}, ) def fake_patch_json(path, _payload): calls.append(path) return {"ok": True} 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) GameModeService().start("wolf") assert calls == ["/apis/apps/v1/namespaces/hermes/statefulsets/hermes-llm/scale"]