109 lines
5.0 KiB
Python
109 lines
5.0 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 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")
|