undercooked/tests/sim/test_perf_full_pipeline.py

"""The headless full-pipeline perf budget (phase 04, task 06, marked `slow`).

Four scripted cooks on the shipped level 1, 3 000 ticks (150 s) through the
*whole* pipeline — the movement, the action gate, the station state
machines, the freshness/burn and the dirty loop, the order rail, the
scoring and the round state, every substep (`test_perf_smoke.py`'s random
walk only exercises the movement half of it). The cooks run the "serve
everything possible" policy of `helpers.py` with a deep shelf stock, so
the stations are working, the rail is spawning and the stock pair is
washing the whole run — the worst case the phase's pipeline can put on a
tick.

The loop times `step()` directly (the controller frames are built outside
the timed section, so the number is the kernel's tick cost, exactly what
the §10 budget (mean ≤ 2 ms/room) is about — the full-loop wall time is
printed alongside for reference). The budget is the task's: mean step
**< 0.5 ms** and total **< 1 500 ms** — 2 000+ ticks/s, four times the
20 Hz the rooms run at — and the measured numbers are printed in the
report so a slow machine shows up in the gate, not as a silent failure.
"""

from __future__ import annotations

import time

import pytest
from helpers import CookController, load_book, make_level_1_state, shelf_id_for

from undercooked.sim.model import EventKind, ItemKind
from undercooked.sim.step import STEP_DT, step

pytestmark = [pytest.mark.slow]

TICKS = 3000  # level 1's full 150 s budget
COOKS = 4

#: The task's budget: mean step < 0.5 ms, total < 1 500 ms on the dev machine.
BUDGET_MEAN_S = 0.0005
BUDGET_TOTAL_S = 1.5


def test_four_cook_full_pipeline_stays_within_the_perf_budget() -> None:
    state = make_level_1_state(seats=COOKS, shelf_stock=40)
    book = load_book()
    tomato = shelf_id_for(state, ItemKind.TOMATO)
    onion = shelf_id_for(state, ItemKind.ONION)
    shelves = [tomato, onion, tomato, onion]
    cooks = [CookController(index, shelves[index]) for index in range(COOKS)]
    current = state
    kinds: set[str] = set()
    step_total = 0.0
    loop_start = time.perf_counter()
    for tick in range(TICKS):
        frames = tuple(cook.next_frame(current, tick) for cook in cooks)
        tick_start = time.perf_counter()
        current, events = step(current, STEP_DT, frames, book)
        step_total += time.perf_counter() - tick_start
        kinds.update(e.kind for e in events)
    loop_total = time.perf_counter() - loop_start
    mean = step_total / TICKS
    print(
        f"perf full pipeline (level 1, {COOKS} cooks, {TICKS} ticks): "
        f"step total {step_total * 1000:.1f} ms, mean {mean * 1000:.4f} ms/tick, "
        f"{1 / mean:,.0f} ticks/s (full loop incl. the scripted controllers: "
        f"{loop_total * 1000:.1f} ms)"
    )
    assert mean < BUDGET_MEAN_S, (
        f"mean step {mean * 1000:.4f} ms is over the {BUDGET_MEAN_S * 1000:.1f} ms budget"
    )
    assert step_total < BUDGET_TOTAL_S, (
        f"{TICKS} steps took {step_total * 1000:.1f} ms, over the "
        f"{BUDGET_TOTAL_S * 1000:.0f} ms budget"
    )
    # The run exercised the whole pipeline: chopping, cooking and plating
    # all ran (the station state machines), a dish was served (the scoring
    # and the dirty loop's stock moved) — the budget is measured on a run
    # that used the mechanism, not an idle kitchen. (The four-cook policy
    # stalls before it needs the in-place wash on this single-pipeline
    # kitchen — the wash cycle's own test drives it, `test_wash_cycle_required`.)
    for kind in (EventKind.CHOPPED, EventKind.COOKED, EventKind.READY, EventKind.SERVED):
        assert kind.value in kinds, f"no {kind.value} in the run"
    # The kernel ran the whole window (no early stop: the round's verdict,
    # wherever it landed, does not shorten the loop).
    assert current.tick == TICKS