"""Game lobby route tests via kaya's ASGI transport, on the DummyEngine.""" from __future__ import annotations import unittest from httpx import ASGITransport, AsyncClient from helpers import async_test, make_platform, make_principal, oidc_user class GamesRouteTest(unittest.TestCase): @async_test async def test_create_requires_auth(self) -> None: app, _, _ = make_platform() transport = ASGITransport(app=app) async with AsyncClient(transport=transport, base_url="http://127.0.0.1") as client: response = await client.post("/api/games", json={}) self.assertEqual(401, response.status_code) self.assertEqual({"error": "unauthenticated"}, response.json()) @async_test async def test_create_and_read_lobby(self) -> None: app, platform, _ = make_platform() transport = ASGITransport(app=app) async with AsyncClient(transport=transport, base_url="http://127.0.0.1") as client: with oidc_user(platform.oidc, "alice"): created = await client.post( "/api/games", json={"options": {"target": 5}} ) self.assertEqual(201, created.status_code) body = created.json() self.assertEqual("lobby", body["phase"]) self.assertEqual(1, body["seats_open"]) self.assertEqual(5, body["target"]) self.assertEqual("dummy", body["game_type"]) self.assertEqual("A", body["players"][0]["team"]) self.assertEqual(6, len(body["join_code"])) game_id = body["id"] with oidc_user(platform.oidc, "alice"): snapshot = await client.get(f"/api/games/{game_id}") self.assertEqual(200, snapshot.status_code) snap = snapshot.json() self.assertEqual(game_id, snap["id"]) self.assertEqual("dummy", snap["game_type"]) self.assertTrue(snap["viewer_seated"]) self.assertFalse(snap["spectator"]) self.assertIn("join_code", snap) # A logged-in non-player is a read-only spectator: the view # marks it and the join code is withheld. with oidc_user(platform.oidc, "mallory"): watched = await client.get(f"/api/games/{game_id}") self.assertEqual(200, watched.status_code) spect = watched.json() self.assertFalse(spect["viewer_seated"]) self.assertTrue(spect["spectator"]) self.assertNotIn("join_code", spect) # Anonymous users may watch too. anonymous = await client.get(f"/api/games/{game_id}") self.assertEqual(200, anonymous.status_code) self.assertTrue(anonymous.json()["spectator"]) self.assertNotIn("join_code", anonymous.json()) @async_test async def test_join_starts_game(self) -> None: app, platform, _ = make_platform() transport = ASGITransport(app=app) async with AsyncClient(transport=transport, base_url="http://127.0.0.1") as client: with oidc_user(platform.oidc, "alice"): created = await client.post("/api/games", json={}) code = created.json()["join_code"] with oidc_user(platform.oidc, "bob"): started = await client.post("/api/games/join", json={"code": code}) self.assertEqual(200, started.status_code) state = started.json() # The second join started the match, so the response is the # personalized view rather than the lobby payload. self.assertTrue(state["started"]) self.assertFalse(state["finished"]) self.assertEqual(0, state["plays"]) self.assertTrue(state["viewer_seated"]) @async_test async def test_preview_requires_auth(self) -> None: app, _, _ = make_platform() transport = ASGITransport(app=app) async with AsyncClient(transport=transport, base_url="http://127.0.0.1") as client: response = await client.get("/api/games/preview?code=ABC123") self.assertEqual(401, response.status_code) @async_test async def test_preview_lobby_and_started_game(self) -> None: app, platform, _ = make_platform() transport = ASGITransport(app=app) async with AsyncClient(transport=transport, base_url="http://127.0.0.1") as client: with oidc_user(platform.oidc, "alice"): created = await client.post( "/api/games", json={"options": {"target": 5}} ) code = created.json()["join_code"] with oidc_user(platform.oidc, "mallory"): missing = await client.get("/api/games/preview") unknown = await client.get("/api/games/preview?code=ZZZZZZ") self.assertEqual(400, missing.status_code) self.assertEqual(404, unknown.status_code) with oidc_user(platform.oidc, "mallory"): preview = await client.get(f"/api/games/preview?code={code}") self.assertEqual(200, preview.status_code) body = preview.json() self.assertEqual("dummy", body["game_type"]) self.assertEqual("Dummy game", body["game_name"]) self.assertTrue(body["game_description"]) self.assertEqual("lobby", body["phase"]) self.assertEqual(5, body["target"]) self.assertEqual(1, body["seats_open"]) self.assertEqual(make_principal("alice"), body["players"][0]["sub"]) # Join codes are case-insensitive. with oidc_user(platform.oidc, "mallory"): lower = await client.get(f"/api/games/preview?code={code.lower()}") self.assertEqual(200, lower.status_code) with oidc_user(platform.oidc, "bob"): await client.post("/api/games/join", json={"code": code}) with oidc_user(platform.oidc, "mallory"): started = await client.get(f"/api/games/preview?code={code}") self.assertEqual(200, started.status_code) self.assertEqual("playing", started.json()["phase"]) self.assertEqual(0, started.json()["seats_open"]) @async_test async def test_create_rejects_bad_options(self) -> None: app, platform, _ = make_platform() transport = ASGITransport(app=app) async with AsyncClient(transport=transport, base_url="http://127.0.0.1") as client: with oidc_user(platform.oidc, "alice"): zero = await client.post( "/api/games", json={"options": {"target": 0}} ) text = await client.post( "/api/games", json={"options": {"target": "three"}} ) non_object = await client.post( "/api/games", json={"options": [1, 2]} ) self.assertEqual(400, zero.status_code) self.assertEqual(400, text.status_code) self.assertEqual(400, non_object.status_code) @async_test async def test_join_errors(self) -> None: app, platform, _ = make_platform() transport = ASGITransport(app=app) async with AsyncClient(transport=transport, base_url="http://127.0.0.1") as client: with oidc_user(platform.oidc, "alice"): created = await client.post("/api/games", json={}) code = created.json()["join_code"] with oidc_user(platform.oidc, "bob"): unknown = await client.post("/api/games/join", json={"code": "ZZZZZZ"}) self.assertEqual(404, unknown.status_code) with oidc_user(platform.oidc, "alice"): duplicate = await client.post("/api/games/join", json={"code": code}) self.assertEqual(409, duplicate.status_code) with oidc_user(platform.oidc, "bob"): missing = await client.post("/api/games/join", json={}) self.assertEqual(400, missing.status_code) with oidc_user(platform.oidc, "bob"): await client.post("/api/games/join", json={"code": code}) with oidc_user(platform.oidc, "erin"): late = await client.post("/api/games/join", json={"code": code}) self.assertEqual(409, late.status_code) @async_test async def test_get_unknown_game(self) -> None: app, platform, _ = make_platform() transport = ASGITransport(app=app) async with AsyncClient(transport=transport, base_url="http://127.0.0.1") as client: with oidc_user(platform.oidc, "alice"): response = await client.get("/api/games/does-not-exist") self.assertEqual(404, response.status_code) @async_test async def test_my_games_lists_only_my_ongoing_games(self) -> None: app, platform, _ = make_platform() transport = ASGITransport(app=app) async with AsyncClient(transport=transport, base_url="http://127.0.0.1") as client: anonymous = await client.get("/api/me/games") self.assertEqual(401, anonymous.status_code) with oidc_user(platform.oidc, "alice"): mine = await client.post( "/api/games", json={"options": {"target": 5}} ) with oidc_user(platform.oidc, "bob"): theirs = await client.post("/api/games", json={}) self.assertEqual(201, mine.status_code) self.assertEqual(201, theirs.status_code) with oidc_user(platform.oidc, "alice"): listed = await client.get("/api/me/games") self.assertEqual(200, listed.status_code) results = listed.json()["results"] self.assertEqual([mine.json()["id"]], [g["id"] for g in results]) entry = results[0] self.assertEqual("dummy", entry["game_type"]) self.assertEqual("Dummy game", entry["game_name"]) self.assertTrue(entry["game_description"]) self.assertEqual("lobby", entry["phase"]) self.assertEqual(5, entry["target"]) self.assertEqual(1, entry["seats_open"]) self.assertEqual(0, entry["your_seat"]) self.assertNotIn("join_code", entry) self.assertEqual(make_principal("alice"), entry["players"][0]["sub"]) # Bob's own game is the only one he sees; Alice's is not his. with oidc_user(platform.oidc, "bob"): bobs = await client.get("/api/me/games") self.assertEqual( [theirs.json()["id"]], [g["id"] for g in bobs.json()["results"]], ) # Once Bob joins Alice's game it appears for both, now in the # playing phase. with oidc_user(platform.oidc, "bob"): await client.post( "/api/games/join", json={"code": mine.json()["join_code"]} ) with oidc_user(platform.oidc, "bob"): bobs = await client.get("/api/me/games") by_id = {g["id"]: g for g in bobs.json()["results"]} self.assertEqual( {mine.json()["id"], theirs.json()["id"]}, set(by_id) ) self.assertEqual("playing", by_id[mine.json()["id"]]["phase"]) self.assertEqual(1, by_id[mine.json()["id"]]["your_seat"]) self.assertEqual(0, by_id[mine.json()["id"]]["seats_open"]) class CrossProviderGamesTest(unittest.TestCase): @async_test async def test_same_sub_other_provider_does_not_inherit_player_access(self) -> None: app, platform, _ = make_platform(oidc_providers=("default", "google")) transport = ASGITransport(app=app) async with AsyncClient(transport=transport, base_url="http://127.0.0.1") as client: with oidc_user(platform.oidc, "alice"): created = await client.post("/api/games", json={}) code = created.json()["join_code"] game_id = created.json()["id"] with oidc_user(platform.oidc, "bob"): await client.post("/api/games/join", json={"code": code}) with oidc_user(platform.oidc, "alice", provider="google"): listed = await client.get("/api/me/games") watched = await client.get(f"/api/games/{game_id}") self.assertEqual([], listed.json()["results"]) self.assertTrue(watched.json()["spectator"]) self.assertFalse(watched.json()["viewer_seated"]) self.assertNotIn("join_code", watched.json()) # The original provider's identical subject remains seated. with oidc_user(platform.oidc, "alice"): owned = await client.get(f"/api/games/{game_id}") self.assertFalse(owned.json()["spectator"]) self.assertTrue(owned.json()["viewer_seated"]) self.assertIn("join_code", owned.json()) @async_test async def test_same_sub_at_two_providers_occupies_distinct_seats(self) -> None: app, platform, _ = make_platform(oidc_providers=("default", "google")) transport = ASGITransport(app=app) async with AsyncClient(transport=transport, base_url="http://127.0.0.1") as client: with oidc_user(platform.oidc, "alice"): created = await client.post("/api/games", json={}) code = created.json()["join_code"] with oidc_user(platform.oidc, "alice", provider="google"): joined = await client.post("/api/games/join", json={"code": code}) self.assertEqual(200, joined.status_code) self.assertTrue(joined.json()["viewer_seated"]) session = await platform.game_store.load(created.json()["id"]) assert session is not None self.assertEqual( { make_principal("alice"), make_principal("alice", "google"), }, {seat.user_sub for seat in session.players}, ) class GameTypesRouteTest(unittest.TestCase): @async_test async def test_lists_available_game_types(self) -> None: app, _, _ = make_platform() transport = ASGITransport(app=app) async with AsyncClient(transport=transport, base_url="http://127.0.0.1") as client: response = await client.get("/api/game-types") self.assertEqual(200, response.status_code) results = response.json()["results"] self.assertEqual(["dummy"], [g["id"] for g in results]) self.assertEqual("Dummy game", results[0]["name"]) self.assertTrue(results[0]["description"]) self.assertEqual(2, results[0]["min_players"]) self.assertEqual(2, results[0]["max_players"]) self.assertIn("target", results[0]["options_schema"]["properties"]) @async_test async def test_create_defaults_game_type(self) -> None: app, platform, _ = make_platform() transport = ASGITransport(app=app) async with AsyncClient(transport=transport, base_url="http://127.0.0.1") as client: with oidc_user(platform.oidc, "alice"): created = await client.post("/api/games", json={}) self.assertEqual(201, created.status_code) self.assertEqual("dummy", created.json()["game_type"]) @async_test async def test_create_with_explicit_game_type(self) -> None: app, platform, _ = make_platform() transport = ASGITransport(app=app) async with AsyncClient(transport=transport, base_url="http://127.0.0.1") as client: with oidc_user(platform.oidc, "alice"): created = await client.post( "/api/games", json={"game_type": "dummy"} ) self.assertEqual(201, created.status_code) body = created.json() self.assertEqual("dummy", body["game_type"]) with oidc_user(platform.oidc, "alice"): snapshot = await client.get(f"/api/games/{body['id']}") self.assertEqual("dummy", snapshot.json()["game_type"]) @async_test async def test_create_rejects_unknown_game_type(self) -> None: app, platform, _ = make_platform() transport = ASGITransport(app=app) async with AsyncClient(transport=transport, base_url="http://127.0.0.1") as client: with oidc_user(platform.oidc, "alice"): unknown = await client.post("/api/games", json={"game_type": "briscola"}) non_string = await client.post("/api/games", json={"game_type": 42}) self.assertEqual(400, unknown.status_code) self.assertEqual(400, non_string.status_code) class MeRouteTest(unittest.TestCase): @async_test async def test_me_authenticated(self) -> None: app, platform, _ = make_platform() transport = ASGITransport(app=app) async with AsyncClient(transport=transport, base_url="http://127.0.0.1") as client: with oidc_user(platform.oidc, "alice"): response = await client.get("/api/me") self.assertEqual(200, response.status_code) self.assertEqual( {"sub": make_principal("alice"), "name": "alice", "picture": None}, response.json(), ) @async_test async def test_me_unauthenticated(self) -> None: app, _, _ = make_platform() transport = ASGITransport(app=app) async with AsyncClient(transport=transport, base_url="http://127.0.0.1") as client: response = await client.get("/api/me") self.assertEqual(401, response.status_code) if __name__ == "__main__": unittest.main()