208 lines
9.6 KiB
Python
208 lines
9.6 KiB
Python
"""P0 验收测试：计时示例、状态机规则与字段表语义。
|
|
|
|
对应验收标准（docs/B题建模分析与实施计划.md P0 行）：
|
|
- 附件计时示例能逐步得到 105、111、194、199 秒；
|
|
- /clear 不切换测向频道；
|
|
- accepted=false 不更新状态，其 virtual_time_s=0 不当作当前虚拟时刻。
|
|
"""
|
|
|
|
import unittest
|
|
|
|
from mcm26b.config import CostConfig
|
|
from mcm26b.protocol.rules import CLEAR_NO_TARGET
|
|
from mcm26b.protocol.state import RobotState, StateError
|
|
from mcm26b.protocol.timing import clear_cost, measure_cost
|
|
from mcm26b.harness import LocalSimTransport, LocalSimulator
|
|
from mcm26b.protocol.client import ActionRejected, SimulatorClient
|
|
|
|
|
|
class TimingModelTests(unittest.TestCase):
|
|
"""附件1表2 / 附件2第10节示例逐步分解。"""
|
|
|
|
def setUp(self):
|
|
self.cost = CostConfig()
|
|
|
|
def test_measure_step2_move_500_no_switch(self):
|
|
breakdown = measure_cost((0, 0), (300, 400), 1, 1, self.cost)
|
|
self.assertAlmostEqual(breakdown.move_s, 100) # 500m / 5(m/s)
|
|
self.assertAlmostEqual(breakdown.switch_s, 0) # 初始频道即1
|
|
self.assertAlmostEqual(breakdown.action_s, 5)
|
|
self.assertAlmostEqual(breakdown.total_s, 105)
|
|
|
|
def test_measure_step3_in_place_switch_channel(self):
|
|
breakdown = measure_cost((300, 400), (300, 400), 1, 2, self.cost)
|
|
self.assertAlmostEqual(breakdown.move_s, 0)
|
|
self.assertAlmostEqual(breakdown.switch_s, 1)
|
|
self.assertAlmostEqual(breakdown.total_s, 6)
|
|
|
|
def test_clear_step4_move_400_no_target(self):
|
|
breakdown = clear_cost((300, 400), (300, 0), success=False, cost=self.cost)
|
|
self.assertAlmostEqual(breakdown.move_s, 80) # 400m / 5
|
|
self.assertAlmostEqual(breakdown.switch_s, 0) # /clear 不产生换频道耗时
|
|
self.assertAlmostEqual(breakdown.action_s, 3) # 未发现=光学定位3s
|
|
self.assertAlmostEqual(breakdown.total_s, 83)
|
|
|
|
def test_measure_step5_same_channel_after_clear(self):
|
|
# 第4步 /clear 后测向机仍是频道2，原地再测频道2无切换耗时。
|
|
breakdown = measure_cost((300, 0), (300, 0), 2, 2, self.cost)
|
|
self.assertAlmostEqual(breakdown.total_s, 5)
|
|
|
|
def test_virtual_clock_walk(self):
|
|
"""累积虚拟时刻逐步得到附件给定值 0/105/111/194/199/199。"""
|
|
virtual = 0.0
|
|
expected = [0, 105, 111, 194, 199, 199]
|
|
sequence = [
|
|
measure_cost((0, 0), (300, 400), 1, 1, self.cost),
|
|
measure_cost((300, 400), (300, 400), 1, 2, self.cost),
|
|
clear_cost((300, 400), (300, 0), False, self.cost),
|
|
measure_cost((300, 0), (300, 0), 2, 2, self.cost),
|
|
None, # /exit
|
|
]
|
|
results = [virtual]
|
|
for breakdown in sequence:
|
|
virtual += 0 if breakdown is None else breakdown.total_s
|
|
results.append(round(virtual, 6))
|
|
self.assertEqual(results, [float(v) for v in expected])
|
|
|
|
|
|
class StateMachineTests(unittest.TestCase):
|
|
"""状态更新规则：accepted=false 不更新；/clear 不动频道。"""
|
|
|
|
@staticmethod
|
|
def _accepted_measure(virtual):
|
|
return {"accepted": True, "real_timestamp_ms": 0,
|
|
"virtual_time_s": virtual,
|
|
"measure_result": "no_signal"}
|
|
|
|
def test_clear_keeps_channel(self):
|
|
state = RobotState()
|
|
state.entered = True
|
|
measure_req = {"position": {"x": 300, "y": 400}, "channel": 2}
|
|
state.apply_measure(measure_req, self._accepted_measure(106))
|
|
self.assertEqual(state.channel, 2)
|
|
clear_req = {"position": {"x": 300, "y": 0}, "channel": 3}
|
|
response = {"accepted": True, "real_timestamp_ms": 0,
|
|
"virtual_time_s": 189,
|
|
"clear_result": CLEAR_NO_TARGET}
|
|
state.apply_clear(clear_req, response)
|
|
self.assertEqual(state.channel, 2, "/clear 不得改变测向机频道")
|
|
|
|
def test_rejected_response_never_updates(self):
|
|
state = RobotState(position=(10, 20), channel=5, virtual_time_s=77)
|
|
rejected = {"accepted": False, "real_timestamp_ms": 1, "virtual_time_s": 0}
|
|
request = {"position": {"x": 0, "y": 0}, "channel": 1}
|
|
with self.assertRaises(StateError):
|
|
state.apply_measure(request, rejected)
|
|
with self.assertRaises(StateError):
|
|
state.apply_clear(request, rejected)
|
|
with self.assertRaises(StateError):
|
|
state.apply_enter(rejected)
|
|
self.assertEqual((state.position, state.channel, state.virtual_time_s),
|
|
((10, 20), 5, 77), "accepted=false 不得更新任何状态")
|
|
|
|
def test_virtual_time_zero_of_rejection_is_not_a_clock(self):
|
|
"""accepted=false 的 virtual_time_s=0 不是当前时刻（回归附件2 4.1）。"""
|
|
state = RobotState(virtual_time_s=300)
|
|
rejected = {"accepted": False, "real_timestamp_ms": 9, "virtual_time_s": 0}
|
|
with self.assertRaises(StateError):
|
|
state.apply_measure({"position": {"x": 0, "y": 0}, "channel": 1}, rejected)
|
|
self.assertEqual(state.virtual_time_s, 300)
|
|
|
|
def test_strict_mode_flags_prediction_gap(self):
|
|
state = RobotState()
|
|
state.entered = True
|
|
request = {"position": {"x": 100, "y": 0}, "channel": 1}
|
|
bad = {"accepted": True, "real_timestamp_ms": 0, "virtual_time_s": 3,
|
|
"measure_result": "no_signal"} # 应为 25s，伪造 3s
|
|
with self.assertRaises(StateError):
|
|
state.apply_measure(request, bad, strict=True)
|
|
|
|
|
|
class FieldTableSimTests(unittest.TestCase):
|
|
"""字段表与错误语义（附件2 5.1/5.3/6.3/7.4）。"""
|
|
|
|
def setUp(self):
|
|
self.sim = LocalSimulator([], robot_id="T001")
|
|
self.base = {"arena_id": "default", "robot_id": "T001"}
|
|
|
|
def _measure(self, **override):
|
|
payload = {**self.base, "request_id": "m1",
|
|
"position": {"x": 0, "y": 0}, "channel": 1, **override}
|
|
return self.sim.handle("/measure", payload)
|
|
|
|
def test_measure_before_enter_rejected(self):
|
|
status, body = self._measure()
|
|
self.assertEqual((status, body["accepted"]), (200, False))
|
|
self.assertEqual(body["virtual_time_s"], 0)
|
|
self.assertEqual(set(body) >= {"accepted", "real_timestamp_ms", "virtual_time_s"}, True)
|
|
|
|
def test_unknown_field_rejected_but_id_not_consumed(self):
|
|
self.sim.handle("/enter", {**self.base, "request_id": "e1"})
|
|
status, body = self._measure(extra_field=1) # 未声明字段
|
|
self.assertEqual((status, body["accepted"]), (200, False))
|
|
status, body = self._measure() # 同 ID 修正后可复用
|
|
self.assertEqual((status, body["accepted"]), (200, True))
|
|
|
|
def test_wrong_robot_id_rejected(self):
|
|
status, body = self.sim.handle("/enter", {**self.base, "robot_id": "OTHER",
|
|
"request_id": "e1"})
|
|
self.assertEqual((status, body["accepted"]), (200, False))
|
|
|
|
def test_channel_and_coordinate_validation_400(self):
|
|
self.sim.handle("/enter", {**self.base, "request_id": "e1"})
|
|
for channel in (0, 21, 1.5):
|
|
status, _ = self._measure(request_id=f"c{channel}", channel=channel)
|
|
self.assertEqual(status, 400)
|
|
status, _ = self._measure(request_id="far",
|
|
position={"x": 2_000_001, "y": 0})
|
|
self.assertEqual(status, 400)
|
|
|
|
def test_position_unknown_field_rejected(self):
|
|
self.sim.handle("/enter", {**self.base, "request_id": "e1"})
|
|
status, body = self._measure(request_id="p1",
|
|
position={"x": 0, "y": 0, "z": 0})
|
|
self.assertEqual((status, body["accepted"]), (200, False))
|
|
|
|
def test_double_enter_rejected_then_measure_ok(self):
|
|
status, body = self.sim.handle("/enter", {**self.base, "request_id": "e1"})
|
|
self.assertEqual(body["accepted"], True)
|
|
status, body = self.sim.handle("/enter", {**self.base, "request_id": "e2"})
|
|
self.assertEqual((status, body["accepted"]), (200, False))
|
|
status, body = self._measure(request_id="m1")
|
|
self.assertEqual(body["accepted"], True)
|
|
|
|
|
|
class WorkedExampleEndToEndTests(unittest.TestCase):
|
|
"""完整链路：客户端 → 本地模拟器复现附件示例。"""
|
|
|
|
def test_attachment_example_reaches_199(self):
|
|
simulator = LocalSimulator([], robot_id="T001")
|
|
client = SimulatorClient("T001", transport=LocalSimTransport(simulator))
|
|
client.enter()
|
|
self.assertTrue(client.state.entered)
|
|
self.assertEqual((client.state.position, client.state.channel), ((0.0, 0.0), 1))
|
|
got = [client.measure((300, 400), 1).virtual_time_s,
|
|
client.measure((300, 400), 2).virtual_time_s,
|
|
client.clear((300, 0), 3).virtual_time_s,
|
|
client.measure((300, 0), 2).virtual_time_s]
|
|
self.assertEqual([round(v, 6) for v in got], [105, 111, 194, 199])
|
|
self.assertEqual(client.state.channel, 2)
|
|
info = client.exit()
|
|
self.assertEqual((info.reason, round(info.virtual_time_s, 6)),
|
|
("user_exit", 199))
|
|
self.assertEqual(client.state.mismatches, [], "预测与服务器计时应完全一致")
|
|
|
|
def test_rejected_action_keeps_client_state(self):
|
|
simulator = LocalSimulator([], robot_id="T001")
|
|
client = SimulatorClient("T001", transport=LocalSimTransport(simulator))
|
|
with self.assertRaises(ActionRejected):
|
|
client.measure((0, 0), 1) # 未 enter
|
|
self.assertEqual(client.state.virtual_time_s, 0)
|
|
self.assertFalse(client.state.entered)
|
|
client.enter()
|
|
self.assertEqual(client.state.virtual_time_s, 0)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|