192 lines
8.7 KiB
Python
192 lines
8.7 KiB
Python
"""会话状态机测试：官方流程（倒计时/窗口/结束原因/真值策略/次数/日志）。"""
|
|
|
|
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", "TEAM-0001", 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", "TEAM-0001", 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", "TEAM-0001", 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", "TEAM-0001", countdown_s=5)
|
|
with pytest.raises(RefereeError):
|
|
referee.start("q3_practice", "TEAM-0001", countdown_s=0.05)
|
|
assert referee.abort()
|
|
second = referee.start("q3_practice", "TEAM-0001", 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", "TEAM-0001", countdown_s=0.05)
|
|
referee.abort() # 正式中止仍占用机会 [题面 4.5]
|
|
with pytest.raises(RefereeError):
|
|
referee.start("q3_formal", "TEAM-0001", countdown_s=0.05)
|
|
# 演练不限次数
|
|
for _ in range(FORMAL_ATTEMPTS + 2):
|
|
referee.start("q3_practice", "TEAM-0001", 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", "TEAM-0001", 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", "TEAM-0001", 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", "TEAM-0001", countdown_s=0.05, window_s=20)
|
|
assert wait_until(lambda: session.phase == "open")
|
|
payload = dict(arena_id="default", robot_id="TEAM-0001", 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", "TEAM-0001", countdown_s=0.05, window_s=10)
|
|
assert wait_until(lambda: session.phase == "open")
|
|
session.simulator.handle("/enter", dict(arena_id="default", robot_id="TEAM-0001",
|
|
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", "TEAM-0001", 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", "TEAM-0001", 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", "TEAM-0001", 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
|