undercooked/tests/integration/test_supervisor.py

"""Integration: the `RoomSupervisor` — create/join/leave, capacity, rejoin,
GC, and teardown, all driven by a `FakeClock` (no test sleeps).

`RoomSupervisor` is the process-wide registry (PLAN §6): code minting with
collision checks, `max_rooms` / `max_players` capacity, the one-token-
one-room index (a token joining a second room is evicted from the first,
and the rejoin path keeps *disconnected* seats in place), the 5 s GC
(empty rooms past `SEAT_GRACE_S`, idle `LOBBY` rooms past
`lobby_idle_timeout_s`, closed rooms dropped, stale tokens purged), and
`start`/`shutdown` as the only task teardown.
"""

from __future__ import annotations

import asyncio
from collections.abc import Awaitable, Callable
from typing import Any

import pytest

from undercooked.config import Settings
from undercooked.server.clock import FakeClock
from undercooked.server.codes import CROCKFORD
from undercooked.server.room import TEST_MODE_SEED, Room, RoomState
from undercooked.server.rooms import (
    GC_INTERVAL_S,
    SEAT_GRACE_S,
    PlayerNotInRoom,
    RoomCapacityError,
    RoomClosed,
    RoomFull,
    RoomNotFound,
    RoomSupervisor,
)
from undercooked.server.sessions import new_session
from undercooked.sim.content import ContentBundle
from undercooked.sim.recipe import RecipeBook

pytestmark = pytest.mark.integration


def empty_bundle() -> ContentBundle:
    """A legal empty campaign: the supervisor's tests never need real levels."""
    return ContentBundle(
        recipe_book=RecipeBook(recipes={}, ingredients=[]),
        levels=(),
    )


def make_supervisor(
    **settings_overrides: Any,
) -> tuple[RoomSupervisor, FakeClock, ContentBundle]:
    """A supervisor on a `FakeClock`, ready to be driven by hand."""
    settings = Settings(**settings_overrides)
    clock = FakeClock()
    bundle = empty_bundle()
    supervisor = RoomSupervisor(settings, clock, bundle)
    return supervisor, clock, bundle


def advancing_sleep(clock: FakeClock) -> Callable[[float], Awaitable[None]]:
    """A sleep that advances the `FakeClock` instead of waiting in real time."""

    async def sleep(seconds: float) -> None:
        clock.advance(seconds)
        await asyncio.sleep(0)

    return sleep


# ---------------------------------------------------------------- create / join


async def test_create_mints_a_code_and_seats_the_creator_as_host() -> None:
    supervisor, clock, _ = make_supervisor()
    session = new_session("Ada", clock)
    room = supervisor.create_room(session, "Ada")

    assert len(room.code) == 5
    assert set(room.code) <= set(CROCKFORD)
    assert room.state is RoomState.LOBBY
    assert room.level_id is None  # the empty campaign has no default level
    seat = room.seats[0]
    assert seat.token == session.token
    assert seat.host and seat.seat_index == 0 and seat.connected
    assert seat.display_name == "Ada"
    assert room.created_at == clock.now()

    assert supervisor.get(room.code) is room
    assert supervisor.get(f" {room.code.lower()}-") is room  # normalized lookup
    assert supervisor.by_token(session.token) is room


async def test_create_uses_the_session_name_when_no_name_is_typed() -> None:
    supervisor, clock, _ = make_supervisor()
    session = new_session("Ada", clock)
    room = supervisor.create_room(session, None)
    assert room.seats[0].display_name == "Ada"
    anonymous = new_session(None, clock)
    room = supervisor.create_room(anonymous, None)
    assert room.seats[0].display_name == "Cook #1"


async def test_create_evicts_a_token_from_its_previous_room() -> None:
    """One token, one room: the creator's old seat is dropped first."""
    supervisor, clock, _ = make_supervisor()
    session = new_session("Ada", clock)
    first = supervisor.create_room(session, "Ada")
    supervisor.create_room(new_session("Bob", clock), "Bob")
    second = supervisor.create_room(session, "Ada")

    assert first.seats == []
    assert first.empty_since == clock.now()
    assert supervisor.by_token(session.token) is second


async def test_join_seats_the_player_and_normalizes_the_code() -> None:
    supervisor, clock, _ = make_supervisor()
    room = supervisor.create_room(new_session("Ada", clock), "Ada")
    bob = new_session("Bob", clock)

    joined = supervisor.join(f" {room.code.lower()}-", bob.token, "Bob")

    assert joined is room
    seat = room.seat_for_token(bob.token)
    assert seat is not None
    assert seat.seat_index == 1 and not seat.host
    assert seat.display_name == "Bob"
    assert supervisor.by_token(bob.token) is room
    assert room.last_activity == clock.now()


async def test_join_unknown_and_malformed_codes_are_indistinguishable() -> None:
    """US-02 criterion 4: a bad format must fail exactly like an unknown code,
    so the supervisor (and the API above it) is not a room-existence oracle."""
    supervisor, _, _ = make_supervisor()
    for bad in ("ZZZZZ", "!!!", "", "ABC", "ABCDEF", "ABCDE?", "AB.CD"):
        with pytest.raises(RoomNotFound):
            supervisor.join(bad, "token", "x")
    assert supervisor.get("ZZZZZ") is supervisor.get("!!!") is None


async def test_join_a_full_room_raises_room_full() -> None:
    supervisor, clock, _ = make_supervisor(max_players=2)
    room = supervisor.create_room(new_session("Ada", clock), "Ada")
    supervisor.join(room.code, new_session("Bob", clock).token, "Bob")

    with pytest.raises(RoomFull, match="full"):
        supervisor.join(room.code, new_session("Cy", clock).token, "Cy")
    assert len(room.seats) == 2


async def test_create_at_max_rooms_raises_room_capacity_error() -> None:
    supervisor, clock, _ = make_supervisor(max_rooms=3)
    for name in ("Ada", "Bob", "Cy"):
        supervisor.create_room(new_session(name, clock), name)

    with pytest.raises(RoomCapacityError, match="capacity"):
        supervisor.create_room(new_session("Dee", clock), "Dee")
    assert supervisor.stats()["rooms"] == 3


# ------------------------------------------------------------------ leave / rejoin


async def test_leave_removes_the_seat_and_starts_the_grace_clock() -> None:
    supervisor, clock, _ = make_supervisor()
    ada = new_session("Ada", clock)
    room = supervisor.create_room(ada, "Ada")
    bob = new_session("Bob", clock)
    supervisor.join(room.code, bob.token, "Bob")

    supervisor.leave(room.code, bob.token)
    assert not room.is_empty and room.empty_since is None
    assert supervisor.by_token(bob.token) is None

    supervisor.leave(room.code, ada.token)
    assert room.is_empty and room.empty_since == clock.now()
    assert supervisor.stats()["tokens"] == 0


async def test_leave_by_a_stranger_raises_player_not_in_room() -> None:
    supervisor, clock, _ = make_supervisor()
    room = supervisor.create_room(new_session("Ada", clock), "Ada")
    with pytest.raises(PlayerNotInRoom):
        supervisor.leave(room.code, "stranger")


async def test_mark_disconnected_keeps_the_seat_for_the_rejoin_path() -> None:
    supervisor, clock, _ = make_supervisor()
    room = supervisor.create_room(new_session("Ada", clock), "Ada")
    supervisor.mark_disconnected(room.code, room.seats[0].token)

    seat = room.seats[0]
    assert seat.connected is False
    assert seat.last_seen == clock.now()  # the seat is kept, not removed


async def test_mark_disconnected_by_a_stranger_raises() -> None:
    supervisor, clock, _ = make_supervisor()
    room = supervisor.create_room(new_session("Ada", clock), "Ada")
    with pytest.raises(PlayerNotInRoom):
        supervisor.mark_disconnected(room.code, "stranger")


async def test_rejoin_reclaims_a_disconnected_seat() -> None:
    supervisor, clock, _ = make_supervisor()
    room = supervisor.create_room(new_session("Ada", clock), "Ada")
    bob = new_session("Bob", clock)
    supervisor.join(room.code, bob.token, "Bob")
    supervisor.mark_disconnected(room.code, bob.token)

    clock.advance(10.0)
    again = supervisor.join(room.code, bob.token, None)

    assert again is room
    seat = room.seat_for_token(bob.token)
    assert seat is not None
    assert seat.connected is True
    assert seat.seat_index == 1  # the same seat, not a second one
    assert seat.last_seen == 10.0
    assert supervisor.stats()["seats"] == 2


async def test_join_moves_a_token_between_rooms() -> None:
    """One token, one room: joining room B evicts the seat in room A."""
    supervisor, clock, _ = make_supervisor()
    room_a = supervisor.create_room(new_session("Ada", clock), "Ada")
    room_b = supervisor.create_room(new_session("Bob", clock), "Bob")
    ada = room_a.seats[0].token

    supervisor.join(room_b.code, ada, None)

    assert room_a.seats == []
    assert room_a.empty_since == clock.now()
    assert supervisor.by_token(ada) is room_b
    assert room_b.seat_for_token(ada) is not None


# ------------------------------------------------------------- stats / queries


async def test_stats_and_all_rooms_are_a_consistent_census() -> None:
    supervisor, clock, _ = make_supervisor(max_rooms=7)
    room_a = supervisor.create_room(new_session("Ada", clock), "Ada")
    room_b = supervisor.create_room(new_session("Bob", clock), "Bob")
    supervisor.join(room_a.code, new_session("Cy", clock).token, "Cy")
    supervisor.mark_disconnected(room_a.code, room_a.seats[0].token)

    assert supervisor.stats() == {
        "rooms": 2,
        "max_rooms": 7,
        "seats": 3,
        "states": {"lobby": 2},
        "tokens": 3,
        "spectators": 0,
    }
    assert [room.code for room in supervisor.all_rooms()] == sorted(
        [room_a.code, room_b.code]
    )


async def test_close_room_keeps_the_room_visible_until_the_next_gc_pass() -> None:
    supervisor, clock, _ = make_supervisor()
    room = supervisor.create_room(new_session("Ada", clock), "Ada")

    await supervisor.close_room(room.code)

    assert room.state is RoomState.CLOSED
    assert room.closed_at == clock.now()
    with pytest.raises(RoomClosed):
        supervisor.join(room.code, new_session("Bob", clock).token, "Bob")

    await supervisor.close_room(room.code)  # idempotent
    assert await supervisor.gc_pass() == 0  # the pass drops it, closes nothing
    assert supervisor.get(room.code) is None
    with pytest.raises(RoomNotFound):
        supervisor.join(room.code, new_session("Bob", clock).token, "Bob")


# ------------------------------------------------------------------ lifecycle


async def test_room_close_cancels_its_tick_task_and_is_idempotent() -> None:
    clock = FakeClock()
    room = Room(
        code="TEST1", settings=Settings(), clock=clock, sleep=advancing_sleep(clock)
    )
    room.start_ticking()
    first_task = room.task
    assert first_task is not None
    room.start_ticking()  # idempotent while the task runs
    assert room.task is first_task

    await room.close()
    assert room.state is RoomState.CLOSED
    assert room.task is not None
    assert room.task.cancelled()
    assert room.seats == []
    await room.close()  # idempotent
    assert room.closed_at is not None


async def test_tick_once_in_a_fresh_occupied_lobby_moves_nothing() -> None:
    """Task 02: `tick_once` is a real tick (the FSM body, crash-isolated),
    but a fresh room with a seat and no idle past a threshold stays
    `LOBBY` — the idle rules are the only things a `LOBBY` tick does."""
    clock = FakeClock()
    room = Room(code="TEST1", settings=Settings(), clock=clock)
    room.add_seat("tok-a", "Ada", True)
    room.tick_once()
    room.tick_once(now=1.0)  # 1 s < the 900 s lobby idle timeout
    assert room.state is RoomState.LOBBY  # nothing moved


async def test_shutdown_cancels_the_gc_loop_and_every_room_task() -> None:
    clock = FakeClock()
    supervisor = RoomSupervisor(
        Settings(), clock, empty_bundle(), sleep=advancing_sleep(clock)
    )
    await supervisor.start()
    assert supervisor.gc_task is not None and not supervisor.gc_task.done()

    room = supervisor.create_room(new_session("Ada", clock), "Ada")
    room.start_ticking()
    assert room.task is not None and not room.task.done()

    await supervisor.shutdown()

    assert room.state is RoomState.CLOSED
    assert room.task is not None
    assert room.task.cancelled()
    assert supervisor.gc_task is None
    assert supervisor.stats()["rooms"] == 0
    assert all(room_.state is RoomState.CLOSED for room_ in supervisor.all_rooms())


async def test_start_is_idempotent_and_a_second_start_restarts_the_gc_loop() -> None:
    clock = FakeClock()
    supervisor = RoomSupervisor(
        Settings(), clock, empty_bundle(), sleep=advancing_sleep(clock)
    )
    await supervisor.start()
    first = supervisor.gc_task
    assert first is not None
    await supervisor.start()
    assert supervisor.gc_task is first  # not restarted while it runs
    await supervisor.shutdown()
    await supervisor.start()
    assert supervisor.gc_task is not first
    assert supervisor.gc_task is not None


# --------------------------------------------------------------------------- GC


async def test_gc_closes_exactly_what_it_should() -> None:
    """The orphan rule, at the default 900 s idle timeout: the 30 s grace is
    the only clock that can fire, and it must fire on the empty room alone."""
    supervisor, clock, _ = make_supervisor()
    # A: empties at t=0 (grace clock starts then)
    ada = new_session("Ada", clock)
    room_a = supervisor.create_room(ada, "Ada")
    bob = new_session("Bob", clock)
    supervisor.join(room_a.code, bob.token, "Bob")
    supervisor.leave(room_a.code, bob.token)
    supervisor.leave(room_a.code, ada.token)
    assert room_a.empty_since == 0.0
    # C: a lobby with a seat, activity at t=0 (idle 900 s — survives)
    supervisor.create_room(new_session("Cy", clock), "Cy")
    # E: a PLAYING room — the idle-lobby rule must never touch it
    room_e = supervisor.create_room(new_session("Eve", clock), "Eve")
    room_e.state = RoomState.PLAYING

    assert await supervisor.gc_pass() == 0  # t=0: nothing is past a threshold
    clock.advance(5.0)
    assert await supervisor.gc_pass() == 0  # t=5: 5 < 30 grace
    clock.advance(20.0)  # t=25
    assert await supervisor.gc_pass() == 0  # 25 < 30 grace: still no

    clock.advance(5.5)  # t=30.5
    assert await supervisor.gc_pass() == 1  # A only: grace 30.5 > 30
    assert supervisor.get(room_a.code) is None
    assert supervisor.get(room_e.code) is not None
    assert supervisor.stats() == {
        "rooms": 2,
        "max_rooms": 200,
        "seats": 2,
        "states": {"lobby": 1, "playing": 1},
        "tokens": 2,
        "spectators": 0,
    }
    assert SEAT_GRACE_S == 30.0  # the constant the test measures against


async def test_gc_closes_idle_lobbies_past_the_timeout() -> None:
    """The idle-lobby rule: a `LOBBY` room with no activity for longer than
    `lobby_idle_timeout_s` is closed — an empty room stays `LOBBY`, so the
    idle rule reaps it too (the grace rule is the *other* way an empty
    room dies, covered above)."""
    supervisor, clock, _ = make_supervisor(lobby_idle_timeout_s=10.0)
    # A: empties at t=0 (stays LOBBY, last activity t=0)
    ada = new_session("Ada", clock)
    room_a = supervisor.create_room(ada, "Ada")
    supervisor.leave(room_a.code, ada.token)
    # C: an idle lobby with a seat (last activity t=0)
    room_c = supervisor.create_room(new_session("Cy", clock), "Cy")
    # D: a lobby with recent activity (t=1)
    room_d = supervisor.create_room(new_session("Dee", clock), "Dee")
    room_d.note_activity(1.0)
    # E: a PLAYING room — the idle-lobby rule must never touch it
    room_e = supervisor.create_room(new_session("Eve", clock), "Eve")
    room_e.state = RoomState.PLAYING

    assert await supervisor.gc_pass() == 0  # t=0: nothing is past 10 s
    clock.advance(10.5)
    assert await supervisor.gc_pass() == 2  # A and C: idle 10.5 > 10
    assert supervisor.get(room_a.code) is None
    assert supervisor.get(room_c.code) is None
    assert supervisor.get(room_d.code) is not None  # idle 9.5 < 10
    assert supervisor.get(room_e.code) is not None  # not a LOBBY

    clock.advance(20.0)  # t=30.5
    assert await supervisor.gc_pass() == 1  # D: idle 29.5 > 10
    assert supervisor.get(room_d.code) is None
    assert supervisor.get(room_e.code) is not None
    assert supervisor.stats()["rooms"] == 1


async def test_gc_purges_stale_token_entries_but_keeps_disconnected_ones() -> None:
    supervisor, clock, _ = make_supervisor()
    ada = new_session("Ada", clock)
    room = supervisor.create_room(ada, "Ada")
    bob = new_session("Bob", clock)
    supervisor.join(room.code, bob.token, "Bob")
    supervisor.mark_disconnected(room.code, ada.token)  # Ada's seat is kept

    await supervisor.close_room(room.code)
    assert supervisor.by_token(bob.token) is room  # registry still holds it
    assert supervisor.by_token(ada.token) is room

    clock.advance(1.0)
    await supervisor.gc_pass()  # drops the closed room, purges both tokens
    assert supervisor.get(room.code) is None
    assert supervisor.by_token(bob.token) is None
    assert supervisor.by_token(ada.token) is None
    assert supervisor.stats()["tokens"] == 0

    # a merely disconnected seat in a *live* room keeps its index entry
    live = supervisor.create_room(new_session("Cy", clock), "Cy")
    supervisor.mark_disconnected(live.code, live.seats[0].token)
    clock.advance(5.0)
    await supervisor.gc_pass()
    assert supervisor.by_token(live.seats[0].token) is live


async def test_gc_loop_runs_on_the_injected_cadence_and_is_cancellable() -> None:
    clock = FakeClock()
    loop_task: dict[asyncio.Task[None], None] = {}

    async def stopper_sleep(seconds: float) -> None:
        assert seconds == GC_INTERVAL_S
        clock.advance(seconds)
        if clock.now() >= 3 * GC_INTERVAL_S:
            next(iter(loop_task)).cancel()
        await asyncio.sleep(0)

    supervisor = RoomSupervisor(
        Settings(), clock, empty_bundle(), sleep=stopper_sleep
    )
    loop_task[asyncio.create_task(supervisor.gc_loop())] = None

    with pytest.raises(asyncio.CancelledError):
        await next(iter(loop_task))
    assert clock.now() == 3 * GC_INTERVAL_S  # exactly three 5-second cycles


# ---------------------------------------------------------------- the test-mode seed


async def test_create_room_in_test_mode_uses_the_documented_fixed_seed() -> None:
    """The test-only seam: under `UNDERCOOKED_TEST_MODE=1` the room is born
    with the documented fixed seed, so a `?seed=20240701` page plays the
    seed the room's sim actually draws (the e2e flagship's determinism)."""
    supervisor, clock, _ = make_supervisor(test_mode=True)
    room = supervisor.create_room(new_session("Ada", clock), "Ada")
    assert room._seed_cursor == TEST_MODE_SEED


async def test_create_room_outside_test_mode_draws_a_fresh_seed() -> None:
    """A production room never pins the seam's value: the base seed is a
    fresh draw from `secrets.randbelow(_SEED_MODULUS)` (the range check —
    a value equal to the constant is possible in principle, so the
    assertion is the range, not an inequality)."""
    supervisor, clock, _ = make_supervisor()  # test_mode defaults to False
    room = supervisor.create_room(new_session("Ada", clock), "Ada")
    assert isinstance(room._seed_cursor, int)
    assert 0 <= room._seed_cursor < 2**63