"""会话状态机测试：官方流程（倒计时/窗口/结束原因/真值策略/次数/日志）。""" import json import time import pytest from referee import (FORMAL_ATTEMPTS, FieldSpec, Referee, RefereeError, Spec) from referee.interface import InterfaceResult from referee.session import MODULES def wait_until(condition, timeout=5.0, interval=0.02): deadline = time.monotonic() + timeout while time.monotonic() < deadline: if condition(): return True time.sleep(interval) return False def fake_result(response, status=200): from referee.jsonio import dumps return InterfaceResult(status, dumps(response).encode("utf-8"), response) def test_countdown_then_open_then_window_timeout(tmp_path): referee = Referee(action_port=0, data_dir=tmp_path) opened = [] referee.set_interface_callbacks(lambda s: opened.append(1), lambda s: None) session = referee.start("q3_practice", "2026001", countdown_s=0.05, window_s=0.3) assert session.phase == "countdown" assert wait_until(lambda: session.phase == "open") assert opened == [1] assert session.simulator is not None assert wait_until(lambda: session.phase == "finished", timeout=3) assert session.end_reason == "window_timeout" def test_abort_during_countdown_and_running(tmp_path): referee = Referee(action_port=0, data_dir=tmp_path) session = referee.start("q3_practice", "2026001", countdown_s=5) assert referee.abort() is True assert session.phase == "finished" and session.end_reason == "manual_abort" assert referee.abort() is False # 已结束，不能重复中止 session = referee.start("q3_practice", "2026001", countdown_s=0.05, window_s=30) assert wait_until(lambda: session.phase == "open") assert referee.abort() is True assert session.end_reason == "manual_abort" def test_cannot_start_while_running_and_restart_after_finish(tmp_path): referee = Referee(action_port=0, data_dir=tmp_path) first = referee.start("q3_practice", "2026001", countdown_s=5) with pytest.raises(RefereeError): referee.start("q3_practice", "2026001", countdown_s=0.05) assert referee.abort() second = referee.start("q3_practice", "2026001", countdown_s=0.05, window_s=1) assert second.case_code != first.case_code # 新的一局、新的案例编码 referee.abort() # 结束后不能复位继续原测试，只能开始新的一局 [题面 4.5] assert referee.session.phase == "finished" def test_formal_attempts_accounting(tmp_path): referee = Referee(action_port=0, data_dir=tmp_path) for _ in range(FORMAL_ATTEMPTS): session = referee.start("q3_formal", "2026001", countdown_s=0.05) referee.abort() # 正式中止仍占用机会 [题面 4.5] with pytest.raises(RefereeError): referee.start("q3_formal", "2026001", countdown_s=0.05) # 演练不限次数 for _ in range(FORMAL_ATTEMPTS + 2): referee.start("q3_practice", "2026001", countdown_s=0.05) referee.abort() def test_truth_policy_practice_vs_formal(tmp_path): referee = Referee(action_port=0, data_dir=tmp_path) practice = referee.start("q3_practice", "2026001", countdown_s=0.05, window_s=5, seed=42, source_count=12) assert wait_until(lambda: practice.phase == "open") referee.abort() state = referee.snapshot() assert state["truth"]["revealed"] is True # 演练结束显示真值统计 assert state["truth"]["source_count"] == 12 assert {"omni_count", "directional_count"} <= set(state["truth"]) formal = referee.start("q4_formal", "2026001", countdown_s=0.05, seed=42) assert wait_until(lambda: formal.phase == "open") # 运行中考官视图也不显示 [题面 4.6：正式测试任何界面不显示真值] assert referee.snapshot(examiner=True)["truth"]["revealed"] is False referee.abort() assert referee.snapshot(examiner=True)["truth"]["revealed"] is False assert "sources" not in referee.snapshot()["truth"] def test_exit_finishes_session_with_user_exit(tmp_path): referee = Referee(action_port=0, data_dir=tmp_path) session = referee.start("q3_practice", "2026001", countdown_s=0.05, window_s=20) assert wait_until(lambda: session.phase == "open") payload = dict(arena_id="default", robot_id="2026001", request_id="enter") session.simulator.handle("/enter", payload) session.simulator.handle("/exit", {**payload, "request_id": "exit"}) assert wait_until(lambda: session.phase == "finished") assert session.end_reason == "user_exit" def test_program_timeout_after_enter(tmp_path): spec = Spec(field=FieldSpec(max_real_duration_s=0.3)) referee = Referee(action_port=0, data_dir=tmp_path, spec=spec) session = referee.start("q3_practice", "2026001", countdown_s=0.05, window_s=10) assert wait_until(lambda: session.phase == "open") session.simulator.handle("/enter", dict(arena_id="default", robot_id="2026001", request_id="enter")) assert wait_until(lambda: session.phase == "finished", timeout=5) assert session.end_reason == "program_timeout" def test_record_action_updates_feed_trail_and_counts(tmp_path): referee = Referee(action_port=0, data_dir=tmp_path) session = referee.start("q3_practice", "2026001", countdown_s=0.05, window_s=5) assert wait_until(lambda: session.phase == "open") sim = session.simulator referee.record_action(session, "POST", "/measure", {"request_id": "m1", "position": {"x": 5000, "y": 0}, "channel": 1}, fake_result({"accepted": True, "virtual_time_s": 105.0, "measure_result": "no_signal"})) referee.record_action(session, "POST", "/clear", {"request_id": "c1", "position": {"x": 5000, "y": 0}, "channel": 2}, fake_result({"accepted": True, "virtual_time_s": 113.0, "clear_result": "no_target_in_range"})) assert session.request_count == 2 assert session.measure_count == 1 and session.clear_count == 1 assert len(session.trail) == 2 state = referee.snapshot(feed_limit=10) assert len(state["feed"]) == 2 assert state["feed"][-1]["response"]["clear_result"] == "no_target_in_range" def test_behavior_log_and_meta_written(tmp_path): referee = Referee(action_port=0, data_dir=tmp_path) session = referee.start("q3_practice", "2026001", countdown_s=0.05, window_s=1, seed=7, source_count=10) assert wait_until(lambda: session.phase == "open") referee.record_action(session, "POST", "/measure", {"request_id": "m1", "position": {"x": 0, "y": 0}, "channel": 1}, fake_result({"accepted": True, "virtual_time_s": 5.0, "measure_result": "no_signal"})) referee.abort() log_lines = [json.loads(line) for line in session.log_path.read_text(encoding="utf-8").splitlines()] events = [entry["event"] for entry in log_lines] assert events[0] == "session_start" and events[-1] == "session_end" assert "action" in events assert log_lines[0]["truth"][0]["channel"] >= 1 # 评测机侧真值留档 meta = json.loads(session.log_path.with_name("meta.json").read_text(encoding="utf-8")) assert meta["module"] == "q3_practice" and len(meta["truth"]) == 10 def test_display_limit_and_port_settings(tmp_path): referee = Referee(action_port=2026, data_dir=tmp_path) referee.set_display_limit(100) with pytest.raises(RefereeError): referee.set_display_limit(50) with pytest.raises(RefereeError): referee.set_display_limit(5001) session = referee.start("q3_practice", "2026001", countdown_s=5) with pytest.raises(RefereeError): referee.set_action_port(8080) # 测试进行中不能改端口 referee.abort() referee.set_action_port(8080) assert referee.action_port == 8080 def test_robot_id_validation(tmp_path): referee = Referee(action_port=0, data_dir=tmp_path) with pytest.raises(RefereeError): referee.start("q3_practice", "", countdown_s=0.05) with pytest.raises(RefereeError): referee.start("q3_practice", "队号\n", countdown_s=0.05) with pytest.raises(RefereeError): referee.start("q3_practice", "x" * 65, countdown_s=0.05) def test_modules_registry(): assert set(MODULES) == {"q3_practice", "q4_practice", "q3_formal", "q4_formal"} assert MODULES["q3_formal"].formal and not MODULES["q3_practice"].formal