272 lines
12 KiB
Python
272 lines
12 KiB
Python
"""评测引擎一致性：时限、状态机、观测判定、幂等与量化（移植自参赛队仓库并扩展）。"""
|
|
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
from dataclasses import replace
|
|
from threading import Event
|
|
|
|
import pytest
|
|
|
|
from referee import FieldSpec, LocalSimulator, SimulatorClosed, Source, Spec
|
|
|
|
|
|
def payload(id="e", **extra):
|
|
return dict(arena_id="default", robot_id="2026001", request_id=id, **extra)
|
|
|
|
|
|
def measure(id="m", x=0, y=0, channel=1):
|
|
return payload(id, position=dict(x=x, y=y), channel=channel)
|
|
|
|
|
|
# ---------------------------------------------------------------- 时限 [题面 附件1 2.5]
|
|
def test_late_enter_and_deadline():
|
|
now = [1400.25]
|
|
sim = LocalSimulator([], clock=lambda: now[0], wall_clock=lambda: 1234.5,
|
|
window_started_at=0)
|
|
response = sim.handle("/enter", payload())[1]
|
|
assert response["remaining_real_duration_s"] == 99 # 窗口剩余取早
|
|
assert response["real_timestamp_ms"] == 1234500
|
|
assert response["max_virtual_duration_s"] == 360000
|
|
assert response["max_real_duration_s"] == 1200
|
|
now[0] = 1500
|
|
for path, body in [("/measure", measure()), ("/exit", payload("x"))]:
|
|
with pytest.raises(SimulatorClosed):
|
|
sim.handle(path, body)
|
|
assert sim.virtual_time_s == 0
|
|
|
|
|
|
def test_program_deadline_and_subsecond_enter():
|
|
now = [100]
|
|
sim = LocalSimulator([], clock=lambda: now[0], window_started_at=0)
|
|
assert sim.handle("/enter", payload())[1]["remaining_real_duration_s"] == 1200
|
|
now[0] = 1300
|
|
with pytest.raises(SimulatorClosed):
|
|
sim.handle("/measure", measure())
|
|
late = LocalSimulator([], clock=lambda: 1499.5, window_started_at=0)
|
|
assert late.handle("/enter", payload())[1]["remaining_real_duration_s"] == 0
|
|
expired = LocalSimulator([], clock=lambda: 1500, window_started_at=0)
|
|
with pytest.raises(SimulatorClosed):
|
|
expired.handle("/enter", payload())
|
|
|
|
|
|
def test_stress_mode_is_explicit():
|
|
spec = Spec(field=FieldSpec(max_real_duration_s=1800))
|
|
official = LocalSimulator([], spec=spec, clock=lambda: 0)
|
|
stress = LocalSimulator([], spec=spec, clock=lambda: 0, stress_mode=True)
|
|
assert official.handle("/enter", payload())[1]["remaining_real_duration_s"] == 1200
|
|
assert stress.handle("/enter", payload())[1]["remaining_real_duration_s"] == 1800
|
|
|
|
|
|
def test_virtual_crossing_action_finishes_then_connection_closes():
|
|
sim = LocalSimulator([])
|
|
sim.handle("/enter", payload())
|
|
assert sim.handle("/measure", measure(x=1_800_000))[1]["virtual_time_s"] == 360005
|
|
for path, body in [("/measure", measure("m2")), ("/exit", payload("x"))]:
|
|
with pytest.raises(SimulatorClosed):
|
|
sim.handle(path, body)
|
|
assert not sim.exited
|
|
|
|
|
|
def test_explicit_exit_closes_new_requests():
|
|
sim = LocalSimulator([])
|
|
sim.handle("/enter", payload())
|
|
assert sim.handle("/exit", payload("exit"))[1]["exit_reason"] == "user_exit"
|
|
assert sim.closed_reason == "user_exit"
|
|
with pytest.raises(SimulatorClosed):
|
|
sim.handle("/measure", measure())
|
|
|
|
|
|
# ---------------------------------------------------------------- 量化与幂等
|
|
def test_microseconds_accumulate_and_signed_zero_is_same_location():
|
|
sim = LocalSimulator([Source(1, 1000, 0, 1500)])
|
|
sim.handle("/enter", payload())
|
|
a = sim.handle("/measure", measure("a", 0.0))[1]
|
|
b = sim.handle("/measure", measure("b", -0.0))[1]
|
|
assert a["svd_deg"] == b["svd_deg"]
|
|
for i in range(100):
|
|
point = 1 if i % 2 == 0 else 0
|
|
sim.handle("/measure", measure(str(i), point, point))
|
|
assert sim.virtual_time_s == 10 + 100 * 5282843 / 1000000
|
|
|
|
|
|
def test_error_rejection_does_not_consume_id():
|
|
sim = LocalSimulator([])
|
|
for bad in ["e\n", "e\u200b", "e\u0085", "e\ud800"]:
|
|
assert sim.handle("/enter", payload(bad))[0] == 400
|
|
assert sim._idempotency == {}
|
|
assert sim.handle("/enter", payload())[1]["accepted"]
|
|
|
|
|
|
def test_idempotent_replay_and_conflict():
|
|
sim = LocalSimulator([])
|
|
sim.handle("/enter", payload("enter"))
|
|
first = sim.handle("/measure", measure("m1", 100, 0))[1]
|
|
replay = sim.handle("/measure", measure("m1", 100, 0))[1]
|
|
assert replay == first # 同 ID 同内容：回放首次响应
|
|
assert sim.virtual_time_s == 25 # 100m/5=20s + 检测5s；不重复推进
|
|
conflict = measure("m1", 200, 0)
|
|
assert sim.handle("/measure", conflict)[0] == 409 # 同 ID 不同内容
|
|
assert sim.handle("/measure", measure("m2", 200, 0))[1]["accepted"]
|
|
|
|
|
|
def test_idempotency_only_accepted_actions_occupy_id():
|
|
sim = LocalSimulator([])
|
|
sim.handle("/enter", payload("enter"))
|
|
# 未知字段：200 accepted=false，不占用 ID [题面 附件2 5.3]
|
|
rejected = payload("m1", extra=1, position=dict(x=0, y=0), channel=1)
|
|
assert sim.handle("/measure", rejected)[1]["accepted"] is False
|
|
assert sim.handle("/measure", measure("m1"))[1]["accepted"]
|
|
|
|
|
|
def test_concurrent_registration_and_action_finishing_after_deadline():
|
|
now = [0]
|
|
sim = LocalSimulator([], clock=lambda: now[0])
|
|
sim.handle("/enter", payload())
|
|
started, release = Event(), Event()
|
|
observe = sim._observe
|
|
calls = []
|
|
|
|
def blocking_observe(pos, channel):
|
|
calls.append(1)
|
|
started.set()
|
|
assert release.wait(5)
|
|
return observe(pos, channel)
|
|
|
|
sim._observe = blocking_observe
|
|
with ThreadPoolExecutor(max_workers=2) as pool:
|
|
first = pool.submit(sim.handle, "/measure", measure())
|
|
try:
|
|
assert started.wait(5)
|
|
assert sim.handle("/measure", measure("other"))[0] == 409
|
|
assert sim.handle("/measure", measure(x=1))[0] == 409
|
|
waiting = Event()
|
|
completed = sim._pending["done"]
|
|
|
|
class ObservedEvent:
|
|
def wait(self):
|
|
waiting.set()
|
|
return completed.wait(5)
|
|
|
|
def set(self):
|
|
completed.set()
|
|
|
|
sim._pending["done"] = ObservedEvent()
|
|
duplicate = pool.submit(sim.handle, "/measure", measure())
|
|
assert waiting.wait(5)
|
|
finally:
|
|
release.set()
|
|
assert first.result(timeout=5) == duplicate.result(timeout=5)
|
|
assert calls == [1]
|
|
assert sim.virtual_time_s == 5
|
|
# 执行中跨过现实截止仍返回结果，后续新请求断连。
|
|
sim._observe = lambda pos, channel: (now.__setitem__(0, 1200) or observe(pos, channel))
|
|
assert sim.handle("/measure", measure("last"))[1]["accepted"]
|
|
with pytest.raises(SimulatorClosed):
|
|
sim.handle("/exit", payload("exit"))
|
|
|
|
|
|
# ---------------------------------------------------------------- 校验与业务拒绝
|
|
def test_huge_numeric_fields_return_bad_request():
|
|
sim = LocalSimulator([])
|
|
sim.handle("/enter", payload())
|
|
huge = 10 ** 400
|
|
assert sim.handle("/measure", measure(x=huge))[0] == 400
|
|
body = measure()
|
|
body["channel"] = huge
|
|
assert sim.handle("/measure", body)[0] == 400
|
|
assert sim.handle("/measure", measure())[1]["accepted"]
|
|
|
|
|
|
@pytest.mark.parametrize("body,expect", [
|
|
(dict(arena_id="default", robot_id="2026001", request_id="e", extra=1), False),
|
|
(dict(arena_id="other", robot_id="2026001", request_id="e"), False),
|
|
(dict(arena_id="default", robot_id="someone-else", request_id="e"), False),
|
|
])
|
|
def test_unknown_fields_and_mismatch_are_business_rejected(body, expect):
|
|
sim = LocalSimulator([])
|
|
status, response = sim.handle("/enter", body)
|
|
assert status == 200 and response["accepted"] is False
|
|
assert response["virtual_time_s"] == 0
|
|
|
|
|
|
def test_channel_and_position_validation():
|
|
sim = LocalSimulator([])
|
|
sim.handle("/enter", payload())
|
|
cases = [
|
|
dict(position=dict(x=0, y=0), channel=0),
|
|
dict(position=dict(x=0, y=0), channel=21),
|
|
dict(position=dict(x=0, y=0), channel=1.5),
|
|
dict(position=dict(x=0, y=0), channel=True),
|
|
dict(position=dict(x=float("nan"), y=0), channel=1),
|
|
dict(position=dict(x=0, y=float("inf")), channel=1),
|
|
dict(position=dict(x=2_000_001, y=0), channel=1),
|
|
dict(position=dict(x=0), channel=1),
|
|
dict(position="origin", channel=1),
|
|
]
|
|
for extra in cases:
|
|
assert sim.handle("/measure", payload("m", **extra))[0] == 400
|
|
assert sim.handle("/measure", payload("ok", position=dict(x=0, y=0), channel=1.0))[1]["accepted"]
|
|
|
|
|
|
def test_actions_before_enter_are_business_rejected():
|
|
sim = LocalSimulator([])
|
|
for path, body in [("/measure", measure()), ("/clear", payload("c", position=dict(x=0, y=0), channel=1)),
|
|
("/exit", payload("x"))]:
|
|
status, response = sim.handle(path, body)
|
|
assert status == 200 and response["accepted"] is False
|
|
|
|
|
|
def test_clear_rules():
|
|
sim = LocalSimulator([Source(5, 10, 0, 1500), Source(6, 15, 0, 1500, directional=True,
|
|
heading_deg=180)])
|
|
sim.handle("/enter", payload())
|
|
# 20m 内清除成功，且与定向朝向无关（源6 朝向 180°，机器人从西侧接近仍可清除）
|
|
ok = sim.handle("/clear", payload("c5", position=dict(x=25, y=0), channel=5))[1]
|
|
assert ok["clear_result"] == "success"
|
|
again = sim.handle("/clear", payload("c5b", position=dict(x=10, y=0), channel=5))[1]
|
|
assert again["clear_result"] == "no_target_in_range" # 同一源只能清除一次
|
|
assert sim.handle("/clear", payload("c6", position=dict(x=15, y=0), channel=6))[1]["clear_result"] == "success"
|
|
miss = sim.handle("/clear", payload("c9", position=dict(x=0, y=0), channel=9))[1]
|
|
assert miss["clear_result"] == "no_target_in_range"
|
|
assert sim.channel == 1 # /clear 不切换测向机频道 [题面 附件1 2.3]
|
|
assert sim.position == (0.0, 0.0)
|
|
|
|
|
|
def test_measure_observation_matrix():
|
|
sources = [Source(1, 1000, 0, 1500), # 全向，r_eff=1500
|
|
Source(2, 100, 0, 1500, directional=True, heading_deg=0), # 定向朝东
|
|
Source(3, 5000, 0, 1500)]
|
|
sim = LocalSimulator(sources, error_half_deg=0.0)
|
|
sim.handle("/enter", payload("enter"))
|
|
# 超出有效接收半径
|
|
assert sim.handle("/measure", measure("a", -600, 0, channel=1))[1]["measure_result"] == "no_signal"
|
|
# 接收半径边界：distance == r_eff 仍有信号 [题面 附件2 2.1 "超过"]
|
|
assert sim.handle("/measure", measure("b", -500, 0, channel=1))[1]["measure_result"] == "direction"
|
|
# 覆盖角边界：源2 朝东，正东方向（bearing=0°）恰在 ±90° 覆盖内
|
|
assert sim.handle("/measure", measure("c", 200, 0, channel=2))[1]["measure_result"] == "direction"
|
|
# 覆盖角边界：bearing=90°（正北）仍在覆盖内（含边界）[题面 附件2 2.2]
|
|
assert sim.handle("/measure", measure("c2", 100, 100, channel=2))[1]["measure_result"] == "direction"
|
|
# 覆盖角外（bearing≈135° > 90°）：无信号
|
|
assert sim.handle("/measure", measure("d", 90, 10, channel=2))[1]["measure_result"] == "no_signal"
|
|
# 距离过近：不返回 svd_deg [题面 附件2 2.4]
|
|
near = sim.handle("/measure", measure("near", 998, 0, channel=1))[1]
|
|
assert near["measure_result"] == "near" and "svd_deg" not in near
|
|
# 有源但超有效接收半径（源3 在 5000m 外，测点在原点）
|
|
assert sim.handle("/measure", measure("f", 0, 0, channel=3))[1]["measure_result"] == "no_signal"
|
|
|
|
|
|
def test_svd_decimals_and_normalization():
|
|
sim = LocalSimulator([Source(1, -10, 0, 1500)], error_half_deg=0.0)
|
|
sim.handle("/enter", payload())
|
|
response = sim.handle("/measure", measure("m", 10, 0))[1]
|
|
assert response["measure_result"] == "direction"
|
|
assert response["svd_deg"] == 180.0 # [0,360) 归一化，西为 180°
|
|
|
|
|
|
def test_error_field_is_fixed_per_location():
|
|
sim = LocalSimulator([Source(1, 1000, 0, 1500)], error_seed=7)
|
|
sim.handle("/enter", payload())
|
|
first = sim.handle("/measure", measure("m1", 500, 0))[1]["svd_deg"]
|
|
sim.handle("/measure", measure("away", 300, 300))
|
|
again = sim.handle("/measure", measure("m1", 500, 0))[1]["svd_deg"]
|
|
assert first == again # 同一地点重复测量读数不变 [题面]
|