156 lines
6.4 KiB
Python
156 lines
6.4 KiB
Python
"""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": "TEAM-0001",
|
|
"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"] == "TEAM-0001"
|
|
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": "TEAM-0001", "countdown_s": 5})
|
|
status, _body = call(port, "POST", "/ui/api/start",
|
|
{"module": "q3_practice", "robot_id": "TEAM-0001"})
|
|
assert status == 409
|
|
status, _body = call(port, "POST", "/ui/api/start",
|
|
{"module": "nope", "robot_id": "TEAM-0001"})
|
|
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": "TEAM-0001",
|
|
"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"] == "TEAM-0001"
|
|
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
|