"""WebUI API 测试：状态快照、启动/中止/配置、会话列表与回放数据源。""" # pylint: disable=redefined-outer-name # pytest 夹具名必须与测试函数形参同名，非遮蔽缺陷 import http.client import json import time import pytest from referee import Referee from referee.server import ActionInterfaceServer from referee.webui import WebUIServer 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 @pytest.fixture() def webui(tmp_path): referee = Referee(action_port=0, data_dir=tmp_path / "data") ActionInterfaceServer(referee) server = WebUIServer(referee, 0) import threading threading.Thread(target=server.serve_forever, daemon=True).start() port = server.httpd.server_address[1] yield referee, port server.shutdown() def call(port, method, path, obj=None): conn = http.client.HTTPConnection("127.0.0.1", port, timeout=5) body = json.dumps(obj).encode("utf-8") if obj is not None else None conn.request(method, path, body=body, headers={"Content-Type": "application/json"}) response = conn.getresponse() data = response.read() conn.close() return response.status, json.loads(data) if data else None def test_state_idle_snapshot(webui): _referee, port = webui status, state = call(port, "GET", "/ui/api/state") assert status == 200 assert state["phase"] == "idle" assert set(state["modules"]) == {"q3_practice", "q4_practice", "q3_formal", "q4_formal"} assert state["modules"]["q3_formal"]["attempts_left"] == 3 assert state["spec"]["omega_radius_m"] == 1800 def test_start_countdown_open_abort_cycle(webui): referee, port = webui status, body = call(port, "POST", "/ui/api/start", {"module": "q3_practice", "robot_id": "2026001", "countdown_s": 0.05, "window_s": 5}) assert status == 200 and body["started"] is True assert wait_until(lambda: referee.session.phase == "open") status, state = call(port, "GET", "/ui/api/state") assert state["phase"] == "open" and state["robot_id"] == "2026001" assert "virtual_time_s" in state and "window_remaining_s" in state status, body = call(port, "POST", "/ui/api/abort", {}) assert body["aborted"] is True status, state = call(port, "GET", "/ui/api/state") assert state["phase"] == "finished" and state["end_reason"] == "manual_abort" assert state["truth"]["revealed"] is True # 演练结束显示真值统计 def test_start_rejects_running_session_and_bad_module(webui): _referee, port = webui call(port, "POST", "/ui/api/start", {"module": "q3_practice", "robot_id": "2026001", "countdown_s": 5}) status, _body = call(port, "POST", "/ui/api/start", {"module": "q3_practice", "robot_id": "2026001"}) assert status == 409 status, _body = call(port, "POST", "/ui/api/start", {"module": "nope", "robot_id": "2026001"}) assert status == 409 call(port, "POST", "/ui/api/abort", {}) def test_config_endpoints(webui): _referee, port = webui status, body = call(port, "POST", "/ui/api/config", {"display_limit": 100}) assert body["configured"]["display_limit"] == 100 status, body = call(port, "POST", "/ui/api/config", {"display_limit": 1}) assert status == 409 status, body = call(port, "POST", "/ui/api/config", {"action_port": 8080}) assert body["configured"]["action_port"] == 8080 def test_sessions_list_and_replay_source(webui): referee, port = webui call(port, "POST", "/ui/api/start", {"module": "q3_practice", "robot_id": "2026001", "countdown_s": 0.05, "window_s": 5, "seed": 11, "source_count": 10}) assert wait_until(lambda: referee.session.phase == "open") session = referee.session from referee.interface import InterfaceResult from referee.jsonio import dumps referee.record_action(session, "POST", "/measure", {"request_id": "m1", "position": {"x": 5000, "y": 0}, "channel": 1}, InterfaceResult(200, dumps({"accepted": True, "virtual_time_s": 1005.0, "measure_result": "no_signal"} ).encode(), {"accepted": True, "virtual_time_s": 1005.0, "measure_result": "no_signal"})) call(port, "POST", "/ui/api/abort", {}) status, data = call(port, "GET", "/ui/api/sessions") assert status == 200 and len(data["sessions"]) == 1 case = data["sessions"][0]["case_code"] assert data["sessions"][0]["robot_id"] == "2026001" status, replay = call(port, "GET", f"/ui/api/replay?case={case}") assert status == 200 kinds = [event["event"] for event in replay["events"]] assert kinds[0] == "session_start" and kinds[-1] == "session_end" assert any(event["event"] == "action" for event in replay["events"]) assert replay["meta"]["module"] == "q3_practice" assert len(replay["meta"]["truth"]) == 10 # 回放含真值（评测机侧） assert call(port, "GET", "/ui/api/replay?case=NOPE-CODE-0000")[0] == 404 assert call(port, "GET", "/ui/api/replay?case=../etc")[0] == 400 def test_static_page_served(webui): _, port = webui conn = http.client.HTTPConnection("127.0.0.1", port, timeout=5) conn.request("GET", "/") response = conn.getresponse() page = response.read().decode("utf-8") conn.close() assert response.status == 200 assert "text/html" in response.getheader("Content-Type") assert "评测机" in page and "replay.js" in page # 回放器：手动选定 JSONL 的入口（文件选择按钮 + 拖拽容器）必须存在 assert 'id="rp-file"' in page and 'id="rp-open-file"' in page assert 'id="rp-stage"' in page def test_replay_script_served_with_file_import(webui): _, port = webui conn = http.client.HTTPConnection("127.0.0.1", port, timeout=5) conn.request("GET", "/replay.js") response = conn.getresponse() script = response.read().decode("utf-8") conn.close() assert response.status == 200 # 双格式归一化：评测机 action 事件 + 客户端 accepted 事件 assert "normalizeEvents" in script and "loadFromText" in script assert 'ev.event === "accepted"' in script