Spaces:
Running
Running
File size: 2,847 Bytes
16ff49b | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 | """Never-repeat pool logic: excluded case ids are never dealt, the shared rotation is
untouched by per-player filtering, and the fallback ladder degrades gracefully."""
from __future__ import annotations
from test_case_adapter import _casefile
from case_zero.api.runtime import GameRuntime
def _pool(ids: list[str]):
base = _casefile()
return [base.model_copy(update={"case_id": cid}) for cid in ids]
def _runtime_with(ids: list[str]) -> GameRuntime:
rt = GameRuntime()
rt._prebaked_loaded = True
rt._prebaked = _pool(ids)
return rt
def test_rotation_unchanged_without_exclusions():
rt = _runtime_with(["A", "B", "C"])
dealt = [rt._take_prebaked().case_id for _ in range(3)]
assert dealt == ["A", "B", "C"]
assert rt._prebaked_idx == 3
def test_excluded_ids_never_dealt_and_rotation_untouched():
rt = _runtime_with(["A", "B", "C"])
for _ in range(20):
case = rt._take_prebaked({"A", "C"})
assert case.case_id == "B"
assert rt._prebaked_idx == 0 # per-player picks never advance the shared rotation
def test_all_excluded_returns_none():
rt = _runtime_with(["A", "B"])
assert rt._take_prebaked({"A", "B"}) is None
def test_excluded_buffered_case_stays_buffered():
rt = _runtime_with(["A"])
buffered = _pool(["FRESH"])[0]
rt._buffer = buffered
assert rt._take_buffered({"FRESH"}) is None
assert rt._buffer is buffered # still there for another player
assert rt._take_buffered(set()).case_id == "FRESH"
assert rt._buffer is None
def test_pressure_heuristic_spawns_generation(monkeypatch):
rt = _runtime_with(["A", "B", "C"])
spawned = []
monkeypatch.setattr(rt, "available", lambda: True)
monkeypatch.setattr(rt, "_spawn_gen", lambda: spawned.append(1))
monkeypatch.setattr(rt, "_register", lambda case: (case.case_id, "run"))
monkeypatch.setattr(rt, "_maybe_refill", lambda: None)
# two of three already played -> fewer than 3 unplayed -> pressure spawn
out = rt.new_generated_run(exclude=["A", "B"])
assert out == ("C", "run")
assert spawned
def test_lru_fallback_when_generation_fails(monkeypatch):
rt = _runtime_with(["A", "B"])
monkeypatch.setattr(rt, "available", lambda: True)
monkeypatch.setattr(rt, "_spawn_gen", lambda: None)
monkeypatch.setattr(rt, "_maybe_refill", lambda: None)
def boom(seed, **kw):
raise RuntimeError("no model")
monkeypatch.setattr(rt, "_generate", boom)
registered = []
def fake_load(case_id):
registered.append(case_id)
return (case_id, "run")
monkeypatch.setattr(rt, "load_generated_run", fake_load)
# everything played: generation fails -> oldest played id is re-dealt
out = rt.new_generated_run(exclude=["A", "B"])
assert out == ("A", "run")
assert registered == ["A"]
|