122 lines
6.4 KiB
Python
122 lines
6.4 KiB
Python
from tests.unit.app.app_route_helpers import *
|
|
|
|
|
|
def test_game_stream_profile_me(monkeypatch) -> None:
|
|
ctx = AuthContext(username="brad", email="", groups=["game-stream-users"], claims={})
|
|
client = _client(monkeypatch, ctx)
|
|
|
|
resp = client.get("/api/game-stream/me", headers={"Authorization": "Bearer token"})
|
|
assert resp.status_code == 200
|
|
assert resp.json()["allowed"] is True
|
|
|
|
|
|
def test_game_mode_admin_start_and_stop(monkeypatch) -> None:
|
|
ctx = AuthContext(username="bstein", email="", groups=["admin"], claims={})
|
|
client = _client(monkeypatch, ctx)
|
|
calls = []
|
|
|
|
monkeypatch.setattr(app_module.game_mode, "start", lambda game, note=None: calls.append(("start", game, note)) or {"status": "active"})
|
|
monkeypatch.setattr(app_module.game_mode, "stop", lambda game, note=None: calls.append(("stop", game, note)) or {"status": "idle"})
|
|
monkeypatch.setattr(app_module.storage, "record_task_run", lambda *args, **kwargs: None)
|
|
monkeypatch.setattr(app_module, "_record_event", lambda *args, **kwargs: None)
|
|
|
|
start = client.post("/api/admin/game-mode/start", headers={"Authorization": "Bearer token"}, json={"game": "arc", "note": "now"})
|
|
stop = client.post("/api/admin/game-mode/stop", headers={"Authorization": "Bearer token"}, json={"game": "arc"})
|
|
assert start.status_code == 200
|
|
assert stop.status_code == 200
|
|
assert calls[0] == ("start", "arc", "now")
|
|
|
|
|
|
def test_game_mode_hook_requires_token(monkeypatch) -> None:
|
|
ctx = AuthContext(username="", email="", groups=[], claims={})
|
|
client = _client(monkeypatch, ctx)
|
|
monkeypatch.setattr(app_module, "settings", dataclasses.replace(app_module.settings, game_mode_hook_token="secret"))
|
|
|
|
resp = client.post("/api/game-mode/start", json={"game": "arc"})
|
|
assert resp.status_code == 401
|
|
|
|
|
|
def test_game_mode_hook_requires_configured_token(monkeypatch) -> None:
|
|
ctx = AuthContext(username="", email="", groups=[], claims={})
|
|
client = _client(monkeypatch, ctx)
|
|
monkeypatch.setattr(app_module, "settings", dataclasses.replace(app_module.settings, game_mode_hook_token=""))
|
|
|
|
resp = client.post("/api/game-mode/start", headers={"Authorization": "Bearer secret"}, json={"game": "arc"})
|
|
assert resp.status_code == 503
|
|
|
|
|
|
def test_game_mode_hook_start_and_stop(monkeypatch) -> None:
|
|
ctx = AuthContext(username="", email="", groups=[], claims={})
|
|
client = _client(monkeypatch, ctx)
|
|
monkeypatch.setattr(app_module, "settings", dataclasses.replace(app_module.settings, game_mode_hook_token="secret"))
|
|
monkeypatch.setattr(app_module.game_mode, "start", lambda game, note=None: {"status": "active", "game": game})
|
|
monkeypatch.setattr(app_module.game_mode, "stop", lambda game, note=None: {"status": "idle", "game": game})
|
|
monkeypatch.setattr(app_module.storage, "record_task_run", lambda *args, **kwargs: None)
|
|
monkeypatch.setattr(app_module, "_record_event", lambda *args, **kwargs: None)
|
|
|
|
start = client.post("/api/game-mode/start", headers={"Authorization": "Bearer secret"}, json={"game": "arc"})
|
|
stop = client.post("/api/game-mode/stop", headers={"x-ariadne-game-mode-token": "secret"}, json={"game": "arc"})
|
|
assert start.status_code == 200
|
|
assert stop.status_code == 200
|
|
|
|
|
|
def test_game_mode_status_error(monkeypatch) -> None:
|
|
ctx = AuthContext(username="bstein", email="", groups=["admin"], claims={})
|
|
client = _client(monkeypatch, ctx)
|
|
monkeypatch.setattr(app_module.game_mode, "status", lambda: (_ for _ in ()).throw(RuntimeError("boom")))
|
|
|
|
resp = client.get("/api/admin/game-mode/status", headers={"Authorization": "Bearer token"})
|
|
assert resp.status_code == 502
|
|
|
|
|
|
def test_game_mode_action_error_records(monkeypatch) -> None:
|
|
ctx = AuthContext(username="bstein", email="", groups=["admin"], claims={})
|
|
client = _client(monkeypatch, ctx)
|
|
recorded = []
|
|
monkeypatch.setattr(app_module.game_mode, "start", lambda *_args, **_kwargs: (_ for _ in ()).throw(RuntimeError("boom")))
|
|
monkeypatch.setattr(app_module.storage, "record_task_run", lambda *args, **kwargs: recorded.append(args))
|
|
monkeypatch.setattr(app_module, "_record_event", lambda *args, **kwargs: None)
|
|
|
|
resp = client.post("/api/admin/game-mode/start", headers={"Authorization": "Bearer token"}, json={"game": "arc"})
|
|
assert resp.status_code == 502
|
|
assert recorded
|
|
|
|
|
|
def test_wolf_oauth2_ensure(monkeypatch) -> None:
|
|
ctx = AuthContext(username="bstein", email="", groups=["admin"], claims={})
|
|
client = _client(monkeypatch, ctx)
|
|
|
|
monkeypatch.setattr(app_module.oauth2_proxy, "ensure_wolf", lambda: {"status": "ok", "client_id": "wolf"})
|
|
monkeypatch.setattr(app_module.storage, "record_task_run", lambda *args, **kwargs: None)
|
|
monkeypatch.setattr(app_module, "_record_event", lambda *args, **kwargs: None)
|
|
|
|
resp = client.post("/api/admin/game-stream/wolf/oauth2/ensure", headers={"Authorization": "Bearer token"})
|
|
assert resp.status_code == 200
|
|
assert resp.json()["client_id"] == "wolf"
|
|
|
|
|
|
def test_wolf_oauth2_ensure_error_paths(monkeypatch) -> None:
|
|
ctx = AuthContext(username="bstein", email="", groups=["admin"], claims={})
|
|
client = _client(monkeypatch, ctx)
|
|
monkeypatch.setattr(app_module.storage, "record_task_run", lambda *args, **kwargs: None)
|
|
monkeypatch.setattr(app_module, "_record_event", lambda *args, **kwargs: None)
|
|
|
|
monkeypatch.setattr(app_module.oauth2_proxy, "ensure_wolf", lambda: {"status": "error", "detail": "missing"})
|
|
resp = client.post("/api/admin/game-stream/wolf/oauth2/ensure", headers={"Authorization": "Bearer token"})
|
|
assert resp.status_code == 502
|
|
|
|
monkeypatch.setattr(app_module.oauth2_proxy, "ensure_wolf", lambda: (_ for _ in ()).throw(RuntimeError("boom")))
|
|
alias = client.post("/api/admin/game-stream/sunshine/oauth2/ensure", headers={"Authorization": "Bearer token"})
|
|
assert alias.status_code == 502
|
|
|
|
|
|
def test_record_simple_task_swallows_storage_errors(monkeypatch) -> None:
|
|
ctx = AuthContext(username="bstein", email="", groups=["admin"], claims={})
|
|
client = _client(monkeypatch, ctx)
|
|
monkeypatch.setattr(app_module.game_mode, "start", lambda game, note=None: {"status": "active", "game": game})
|
|
monkeypatch.setattr(app_module.storage, "record_task_run", lambda *args, **kwargs: (_ for _ in ()).throw(RuntimeError("db")))
|
|
monkeypatch.setattr(app_module, "_record_event", lambda *args, **kwargs: None)
|
|
|
|
resp = client.post("/api/admin/game-mode/start", headers={"Authorization": "Bearer token"}, json={"game": "arc"})
|
|
assert resp.status_code == 200
|