"""P1 验收测试：幂等重试、故障恢复、现实预算与真实演练日志回放。 对应验收标准（docs/B题建模分析与实施计划.md P1 行）： - accepted=false 和超时重试不污染状态； - 失败重试复用原 request_id 与原内容，服务器端动作恰好生效一次； - 用 /enter 的 remaining_real_duration_s 建立本地截止时钟； - 官方演练日志可完整回放，本地重建状态与服务器记录逐步吻合。 """ import json import math import tempfile import unittest from dataclasses import replace from pathlib import Path from mcm26b.config import ProblemConfig from mcm26b.evaluation import recompute_virtual_time, summarize from mcm26b.protocol.client import (DeadlineExceeded, ProtocolError, SimulatorClient) from mcm26b.protocol.log import SessionLog from mcm26b.protocol.transport import TransportError from mcm26b.harness import (LocalSimTransport, LocalSimulator, PlaybackTransport, ScriptedFaultTransport) from referee import Source ROOT = Path(__file__).resolve().parents[1] class FaultRecoveryTests(unittest.TestCase): """故障注入：失败重试不污染状态、不重复生效。""" def _client(self, script, sources=None): simulator = LocalSimulator(sources or [], robot_id="T001") inner = LocalSimTransport(simulator) faults = ScriptedFaultTransport(inner, script) client = SimulatorClient("T001", transport=faults) return client, faults, simulator def test_timeout_after_apply_replays_cached_response(self): """读超时（服务器已生效）后重试：同一响应回放，动作不二次生效。""" # 脚本按请求顺序消耗：enter 放行，measure 第一次注入"已生效但超时"。 client, faults, simulator = self._client([None, "timeout_after", None, None]) client.enter() outcome = client.measure((0, 0), 1) self.assertEqual(outcome.virtual_time_s, 5.0) # measure 只被服务器执行了一次：请求发出两次，但虚拟时刻只推进一次。 measure_bodies = list(faults.sent_requests[1:]) self.assertEqual(len(measure_bodies), 2, "应发出原始请求+一次重试") first, retry = (json.loads(b) for b in measure_bodies) self.assertEqual(first["request_id"], retry["request_id"], "重试必须复用原 request_id") self.assertEqual(first, retry, "重试请求体必须逐字段一致") self.assertEqual(simulator.virtual_time_s, 5.0, "动作只生效一次") self.assertEqual(client.state.virtual_time_s, 5.0) def test_drop_before_apply_is_safe_to_retry(self): """连接失败（未送达）后重试成功，无残留效果。""" client, _, simulator = self._client(["drop_before", None]) client.enter() client.measure((100, 0), 1) self.assertEqual(simulator.virtual_time_s, 25.0) # 100/5 移动 + 5 检测 self.assertEqual(client.state.virtual_time_s, 25.0) def test_server_500_retried_with_same_id(self): client, _, _ = self._client([None, ("status", 500), None, None]) client.enter() client.measure((0, 0), 1) self.assertEqual(client.state.virtual_time_s, 5.0) def test_exhausted_retries_raise_and_leave_state_clean(self): # enter 放行后，measure 连续注入断连直至超过 max_attempts=3。 client, _, simulator = self._client([None, "drop_before", "drop_before", "drop_before"]) client.enter() with self.assertRaises(TransportError): client.measure((0, 0), 1) self.assertEqual(client.state.virtual_time_s, 0.0, "失败动作不得污染本地状态") self.assertTrue(simulator.entered) # enter 已成功，会话仍然有效 # 后续新动作用新 request_id 正常执行（脚本已耗尽，全部放行）。 outcome = client.measure((0, 0), 1) self.assertEqual(outcome.virtual_time_s, 5.0) def test_accepted_false_raises_and_state_untouched(self): client, _, _ = self._client([None]) with self.assertRaises(ProtocolError) as ctx: client.measure((0, 0), 1) # 未 enter -> accepted=false self.assertEqual(ctx.exception.status, 200) self.assertEqual(client.state.virtual_time_s, 0.0) class DeadlineBudgetTests(unittest.TestCase): """现实预算：本地单调时钟、预留余量、/exit 豁免。""" def test_deadline_from_enter_remaining(self): now = [1000.0] simulator = LocalSimulator([], robot_id="T001") client = SimulatorClient("T001", transport=LocalSimTransport(simulator), clock=lambda: now[0]) client.enter() self.assertAlmostEqual(client.real_remaining_s, 1200.0) now[0] += 300.0 self.assertAlmostEqual(client.real_remaining_s, 900.0) def test_action_blocked_when_reserve_margin_low(self): now = [0.0] simulator = LocalSimulator([], robot_id="T001") client = SimulatorClient("T001", transport=LocalSimTransport(simulator), clock=lambda: now[0]) client.enter() # remaining = 1200s now[0] += 1145.0 # 只剩 55s < 60s 预留 with self.assertRaises(DeadlineExceeded): client.measure((0, 0), 1) info = client.exit() # /exit 不受预算拦截 self.assertEqual(info.reason, "user_exit") def test_custom_reserve_margin(self): now = [0.0] simulator = LocalSimulator([], robot_id="T001") config = replace(ProblemConfig(), session=replace(ProblemConfig().session, reserve_exit_s=10.0)) client = SimulatorClient("T001", config=config, transport=LocalSimTransport(simulator), clock=lambda: now[0]) client.enter() now[0] += 1195.0 # 剩 5s < 新预留 10s with self.assertRaises(DeadlineExceeded): client.measure((0, 0), 1) class PracticeReplayTests(unittest.TestCase): """官方演练日志经 PlaybackTransport 驱动完整客户端。""" def _events(self): files = sorted((ROOT / "output" / "q1" / "practice").glob("*_actions.json")) self.assertTrue(files, "缺少演练采集证据") events = [] for path in files: events.extend(json.loads(path.read_text(encoding="utf-8"))) return events def test_client_state_matches_recorded_server_clock(self): events = self._events() client = SimulatorClient("任意队号", transport=PlaybackTransport(events)) for event in events: path, request = event["path"], event["request"] if path == "/enter": client.enter() elif path == "/measure": client.measure((request["position"]["x"], request["position"]["y"]), request["channel"]) elif path == "/exit": client.exit() self.assertAlmostEqual( client.state.virtual_time_s, event["response"]["virtual_time_s"], places=5, msg=f"{path} 后本地虚拟时刻应与服务器记录一致") self.assertTrue(client.state.exited) def test_evaluation_recompute_agrees_with_recording(self): events = [dict(path=e["path"], request=e["request"], response=e["response"]) for e in self._events() if e["response"].get("accepted") is True] report = recompute_virtual_time(events) self.assertEqual(report["mismatches"], []) self.assertAlmostEqual(report["final_virtual_time_s"], events[-1]["response"]["virtual_time_s"], places=5) def test_summarize_counts(self): events = [dict(path=e["path"], request=e["request"], response=e["response"]) for e in self._events()] stats = summarize(events) _directions = [e for e in events if e["response"].get("measure_result") == "direction"] self.assertGreater(stats["measures"], 0) self.assertEqual(stats["clears"], 0, "该次演练只采集观测，未调用 /clear") class LocalSimObservationTests(unittest.TestCase): """观测语义：no_signal / near / direction 与定向覆盖角。""" def test_distance_and_coverage_gate(self): source = Source(channel=1, x=500, y=0, r_eff_m=1000) simulator = LocalSimulator([source], robot_id="T001") client = SimulatorClient("T001", transport=LocalSimTransport(simulator)) client.enter() self.assertEqual(client.measure((1505, 0), 1).result, "no_signal") # 距源1005>1000 self.assertEqual(client.measure((500, 998), 1).result, "direction") # 1000内 self.assertEqual(client.measure((500, 0), 1).result, "near") # <=5m def test_directional_source_backside_invisible(self): source = Source(channel=1, x=500, y=0, r_eff_m=1500, directional=True, heading_deg=90.0) # 朝北 simulator = LocalSimulator([source], robot_id="T001") client = SimulatorClient("T001", transport=LocalSimTransport(simulator)) client.enter() self.assertEqual(client.measure((500, 600), 1).result, "direction") # 正北 self.assertEqual(client.measure((500, -300), 1).result, "no_signal") # 正南背面 self.assertEqual(client.measure((500, 998), 1).result, "direction") # 边界含90度 def test_clear_ignores_directional_facing(self): """清除只看距离：定向源背面同样可清除 [题面]。""" source = Source(channel=7, x=0, y=0, r_eff_m=1500, directional=True, heading_deg=180.0) simulator = LocalSimulator([source], robot_id="T001") client = SimulatorClient("T001", transport=LocalSimTransport(simulator)) client.enter() self.assertEqual(client.state.channel, 1) outcome = client.clear((15, 0), 7) # 距离15m<=20，位于背面 self.assertEqual(outcome.result, "success") outcome = client.clear((0, 0), 7) # 同源只能清除一次 self.assertEqual(outcome.result, "no_target_in_range") # 两次 /clear 后测向机频道仍是初始 1：/clear 不切换频道 [题面]。 self.assertEqual(client.state.channel, 1) # 清除后同频道无信号；/measure 会把频道更新为 7。 self.assertEqual(client.measure((100, 0), 7).result, "no_signal") self.assertEqual(client.state.channel, 7) def test_bearing_error_within_bound(self): source = Source(channel=1, x=1000, y=300, r_eff_m=1500) simulator = LocalSimulator([source], error_half_deg=1.0, robot_id="T001") client = SimulatorClient("T001", transport=LocalSimTransport(simulator)) client.enter() outcome = client.measure((0, 0), 1) true_bearing = math.degrees(math.atan2(300, 1000)) # 误差在 ±1° 内均匀分布，返回值再保留两位小数。 self.assertLessEqual(abs(outcome.svd_deg - true_bearing), 1.005) def test_same_position_error_is_fixed(self): """同一（源，位置）重复测量读数不变 [题面/官方演练实测]。""" source = Source(channel=1, x=1000, y=300, r_eff_m=1500) simulator = LocalSimulator([source], robot_id="T001") client = SimulatorClient("T001", transport=LocalSimTransport(simulator)) client.enter() first = client.measure((0, 0), 1) second = client.measure((0, 0), 1) # 新 request_id，同位置 self.assertEqual(first.result, "direction") self.assertEqual(first.svd_deg, second.svd_deg, "同位置重复测量的示向度必须相同") def test_error_field_deterministic_per_seed(self): """同种子误差场逐点一致；不同种子得到不同误差场（逐点白噪声）。""" source = Source(channel=1, x=1000, y=0, r_eff_m=1500) readings = {} for label, seed in (("a", 1), ("b", 1), ("c", 2)): simulator = LocalSimulator([source], error_seed=seed, robot_id="T001") client = SimulatorClient("T001", transport=LocalSimTransport(simulator)) client.enter() readings[label] = [client.measure((float(x), 0.0), 1).svd_deg for x in (0, 100, 200, 300, 400)] self.assertEqual(readings["a"], readings["b"]) self.assertNotEqual(readings["a"], readings["c"], "不同种子应在至少一个测点产生不同读数") class SessionLogTests(unittest.TestCase): def test_log_records_attempts_and_accepted(self): with tempfile.TemporaryDirectory() as folder: path = Path(folder) / "session.jsonl" simulator = LocalSimulator([], robot_id="T001") faults = ScriptedFaultTransport(LocalSimTransport(simulator), [None, "drop_before", None, None]) client = SimulatorClient("T001", transport=faults, log=SessionLog(path=path)) client.enter() client.measure((0, 0), 1) client.exit() rows = [json.loads(line) for line in path.read_text(encoding="utf-8").splitlines()] accepted = [row for row in rows if row["event"] == "accepted"] self.assertEqual(len(accepted), 3) self.assertEqual(accepted[1]["attempts"], 2, "重试后成功应记录尝试次数") self.assertEqual(len({row["request_id"] for row in accepted}), 3, "每个被接受动作各占一个 request_id") if __name__ == "__main__": unittest.main()