"""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()