undercooked/tests/sim/test_movement_kernel.py

"""The movement stage through the real tick, on the levels the game actually ships with.

`test_movement.py` proves the geometry and `test_movement_stress.py` proves the model under
a random walk on a synthetic maze. This is the seam between them and the content pipeline:
`load_content` reads `levels/*.yaml`, `build_state` turns that into a kitchen, and `step()`
runs the whole tick — timers, orders, hazards, movement, in that fixed order — with four
cooks walking around inside the authored geometry.

It earns its place by being the only test that would notice any of these:

* a level whose tiles make a cook contain-able (a spawn the body footprint does not fit
  in, a station placed where the body cannot clear it),
* the movement stage drifting out of the tick's subsystem order, or a whole substep of
  movement going missing because a hook signature changed under the caller,
* authored content making the kernel non-deterministic — the replay contract is per-level,
  not per-fixture,
* the round timer freezing movement: a walk that runs past `time_limit` stops, and the
  progress assertions below would then be measuring a paused kitchen instead of a stuck
  cook. TICKS therefore stays well inside the shorter level's clock.

The kitchen here comes from YAML, so nothing in this file names a tile, a station or a
wall: it says what must be true of any level that passes the validator.
"""

from __future__ import annotations

import math
from pathlib import Path

import pytest

from undercooked.sim import movement
from undercooked.sim.collision import body_blocker, body_cells
from undercooked.sim.content import load_content
from undercooked.sim.level import Level
from undercooked.sim.model import KitchenState, TileKind, Vec2
from undercooked.sim.rng import Rng
from undercooked.sim.state_build import SeatSpec, build_state
from undercooked.sim.step import STEP_DT, PlayerInput, step

pytestmark = pytest.mark.unit

REPO_ROOT = Path(__file__).resolve().parents[2]
SHIPPED_LEVELS = REPO_ROOT / "levels"
RECIPES = "recipes.yaml"

TICKS = 1200
SEED = 0x0D1C
COOK_COUNT = 4

#: How much of the level each cook must have stood on by the end of the walk. Measured on
#: the shipped campaign at a quarter of TICKS: 16-41 cells per cook. Well clear of the
#: 2x2 the pair starts stacked in, and well short of anything that would flap on a redesign.
MIN_CELLS_VISITED = 10
DIRECTIONS: tuple[Vec2, ...] = (
    Vec2(1, 0),
    Vec2(1, 1),
    Vec2(0, 1),
    Vec2(-1, 1),
    Vec2(-1, 0),
    Vec2(-1, -1),
    Vec2(0, -1),
    Vec2(1, -1),
    Vec2(0, 0),
)


#: The shipped campaign, read once at import. Content that will not load should stop
#: collection rather than fail one test with a confusing AttributeError.
CAMPAIGN = load_content(SHIPPED_LEVELS, SHIPPED_LEVELS / RECIPES)


def shipped_levels() -> list[tuple[str, Level]]:
    """Every level in `levels/`, as (id, level) pairs for parametrization."""
    return [(level.id, level) for level in CAMPAIGN.levels]


def seats(count: int) -> tuple[SeatSpec, ...]:
    """Four anonymous seats, in seat order, for a full lobby on any level."""
    return tuple(
        SeatSpec(player_id=f"p{index}", seat=index, display_name=f"Cook {index}")
        for index in range(count)
    )


def tick_budget(cooks: int) -> float:
    """The most one cook may move in a substep: one tick of walking, plus every push."""
    body = 2 * movement.RADIUS
    return movement.MAX_SPEED * STEP_DT + movement.PUSH_ITERATIONS * (cooks - 1) * body * 2


def walk_level(
    level: Level,
    ticks: int = TICKS,
    visited: dict[str, set[tuple[int, int]]] | None = None,
) -> KitchenState:
    """Run `ticks` real substeps of a wandering four-cook kitchen on `level`."""
    state = build_state(level, seats(COOK_COUNT), SEED, CAMPAIGN.recipe_book)
    blocked = body_blocker(state.grid, state.stations)
    assert state.players, "a level with no cooks cannot test movement"
    budget = tick_budget(len(state.players))
    rng = Rng(SEED)
    heading = {player.id: Vec2(0, 0) for player in state.players}

    for tick in range(1, ticks + 1):
        frames: list[PlayerInput] = []
        for player in state.players:
            if rng.next_float() < 0.2:
                heading[player.id] = rng.choice(DIRECTIONS)
            frames.append(PlayerInput(
                player_id=player.id,
                seq=tick,
                move=heading[player.id],
                received_at_tick=tick,
            ))

        before = {player.id: player.pos for player in state.players}
        state, _events = step(state, STEP_DT, frames)

        assert state.tick == tick, "the movement stage must not own the tick counter"
        assert state.result is None, "the walk outran the round clock: raise time_limit"
        for player in state.players:
            assert math.isfinite(player.pos.x) and math.isfinite(player.pos.y)
            assert math.isfinite(player.vel.x) and math.isfinite(player.vel.y)
            assert not any(
                blocked(x, y) for x, y in body_cells(player.pos, movement.RADIUS)
            ), f"{player.id} is inside {level.id} at tick {tick}: {player.pos}"
            assert player.vel.length <= movement.MAX_SPEED + 1e-9
            assert (player.pos - before[player.id]).length <= budget
            if visited is not None:
                visited.setdefault(player.id, set()).update(
                    body_cells(player.pos, movement.RADIUS)
                )

    return state


@pytest.mark.parametrize(("level_id", "level"), shipped_levels())
def test_a_walk_on_a_shipped_level_stays_inside_it(level_id: str, level: Level) -> None:
    """Containment, bounded velocity and tick accounting, after every single substep."""
    del level_id
    final = walk_level(level)
    assert final.tick == TICKS
    assert all(player.vel.length <= movement.MAX_SPEED + 1e-9 for player in final.players)


@pytest.mark.parametrize(("level_id", "level"), shipped_levels())
def test_a_walk_on_a_shipped_level_goes_somewhere(level_id: str, level: Level) -> None:
    """Each cook explores authored geometry instead of wearing out its spawn tile."""
    del level_id
    visited: dict[str, set[tuple[int, int]]] = {}
    walk_level(level, visited=visited)
    assert len(visited) == 4
    for player_id, cells in sorted(visited.items()):
        assert len(cells) >= MIN_CELLS_VISITED, f"{player_id} only ever saw {sorted(cells)}"


@pytest.mark.parametrize(("level_id", "level"), shipped_levels())
def test_a_replay_of_a_shipped_level_reproduces_itself(level_id: str, level: Level) -> None:
    """Determinism is per level: same seed and script, identical hash, on every map."""
    del level_id
    assert walk_level(level).state_hash() == walk_level(level).state_hash()


def test_the_shipped_kitchen_has_something_to_walk_on() -> None:
    """Guard for the two tests above: the floor is floor, and it is not one cell."""
    assert len(CAMPAIGN.levels) >= 2
    for level in CAMPAIGN.levels:
        floor = sum(1 for tile in level.grid.tiles if tile is TileKind.FLOOR)
        assert floor > 4 * MIN_CELLS_VISITED, f"{level.id} is too small to walk"