"""P4 全向闭环端到端验收：10/16/0 源、边界源、预算耗尽与故障注入。 对应 docs/B题建模分析与实施计划.md P4 行验收门槛（自建部分）： - 10/16 源场景全部清除；16 源触发数量上界停止； - 近圆域边界源不漏搜、可清除； - 持续复扫 [用户决策 2026-09-11]：找全后只复扫未清除频道直到虚拟上限； - 每步计时（实机 + 比赛虚拟耗时分项）逐步留档可汇总； - 虚拟时刻逐步复算与服务器响应一致； - 现实预算耗尽（找全前）退出记为未完成，不与成功混记； - 网络故障重试（幂等回放）不污染闭环结果。 SMALL_CAP 把虚拟限时压到 60k 秒：持续复扫在数轮内触及安全余量， 避免测试扫满 360k 秒上限（等价行为，仅缩短复扫轮数）。 """ import unittest from dataclasses import replace from itertools import count from mcm26b.config import ProblemConfig from mcm26b.coverage import coverage_points from mcm26b.evaluation import recompute_virtual_time from mcm26b.planner import OmnidirectionalLoop from mcm26b.protocol.client import SimulatorClient from mcm26b.sim import (LocalSimTransport, LocalSimulator, ScriptedFaultTransport, Source) from mcm26b.sim.scenes import random_omni_scene CONFIG = replace(ProblemConfig(), search=replace(ProblemConfig().search, bearing_rounding_margin_deg=0.005)) SMALL_CAP = replace(CONFIG, field=replace(CONFIG.field, max_virtual_duration_s=60_000.0)) def _run_scene(sources, fault_script=None, clock=None, config=SMALL_CAP): simulator = LocalSimulator(sources, robot_id="T001", config=config) transport = LocalSimTransport(simulator) if fault_script is not None: transport = ScriptedFaultTransport(transport, fault_script) kwargs = {"clock": clock} if clock is not None else {} client = SimulatorClient("T001", config=config, transport=transport, **kwargs) loop = OmnidirectionalLoop(client, config) summary = loop.run() report = recompute_virtual_time(client.log.action_events()) return summary, report, simulator class LoopAcceptanceTests(unittest.TestCase): def test_ten_sources_all_cleared(self): sources = random_omni_scene(seed=5, count=10) summary, report, simulator = _run_scene(sources) truth = sorted(s.channel for s in sources) self.assertTrue(summary["completed"]) self.assertEqual(summary["stop_reason"], "virtual_time_cap") self.assertEqual(sorted(summary["cleared_channels"]), truth) self.assertEqual(len(summary["absent_channels"]), 10) self.assertEqual(summary["unresolved_channels"], []) self.assertIsNotNone(summary["accounted_virtual_time_s"], "找全证明时刻必须留档") self.assertLessEqual(summary["accounted_virtual_time_s"], summary["virtual_time_s"]) self.assertGreater(summary["measures"], 49 * 20, "找全后应继续复扫未清除频道（扫到限时为止）") self.assertEqual(summary["anomalies"], [], "守恒性或证书不应被打破") self.assertEqual(report["mismatches"], []) self.assertEqual(summary["virtual_recompute_mismatches"], 0) self.assertEqual(sorted(simulator.cleared), truth) self.assertEqual(summary["exit_reason"], "user_exit") self.assertLess(summary["virtual_time_s"], CONFIG.field.max_virtual_duration_s) def test_sixteen_sources_stop_by_count_bound(self): sources = random_omni_scene(seed=6, count=16) summary, report, simulator = _run_scene(sources) truth = sorted(s.channel for s in sources) self.assertTrue(summary["completed"]) self.assertEqual(summary["stop_reason"], "sixteen_cleared") self.assertEqual(sorted(summary["cleared_channels"]), truth) self.assertEqual(report["mismatches"], []) self.assertEqual(sorted(simulator.cleared), truth) def test_empty_field_proves_absence_everywhere(self): summary, report, _ = _run_scene([]) self.assertTrue(summary["completed"]) self.assertEqual(summary["stop_reason"], "virtual_time_cap") self.assertEqual(summary["cleared_channels"], []) self.assertEqual(summary["absent_channels"], list(range(1, 21))) self.assertGreaterEqual(summary["measures"], 49 * 20, "缺席证明要求全网格逐频道检测") self.assertIsNotNone(summary["accounted_virtual_time_s"]) self.assertEqual(report["mismatches"], []) def test_boundary_source_close_to_domain_edge(self): """半径 1790m、接收半径 1000m 的源：覆盖保证的最近网格点必须发现它。""" source = Source(channel=13, x=1790.0, y=0.0, r_eff_m=1000.0) summary, _, simulator = _run_scene([source]) self.assertTrue(summary["completed"]) self.assertEqual(sorted(summary["cleared_channels"]), [13]) self.assertEqual(sorted(simulator.cleared), [13]) def test_source_on_grid_vertex(self): source = Source(channel=20, x=-1400.0, y=700.0, r_eff_m=1500.0) summary, _, _ = _run_scene([source]) self.assertTrue(summary["completed"]) self.assertEqual(sorted(summary["cleared_channels"]), [20]) def test_real_budget_exhaustion_exits_cleanly(self): """现实预算在找全前耗尽：立即退出并记为未证明完成。""" ticks = count(0.0, 200.0) # 每次查询时钟前进 200s，迅速耗尽 1200s 预算 sources = random_omni_scene(seed=8, count=4) summary, _, _ = _run_scene(sources, clock=lambda: next(ticks)) self.assertFalse(summary["completed"]) self.assertEqual(summary["stop_reason"], "real_budget_exhausted") self.assertEqual(summary["exit_reason"], "user_exit") def test_fault_injection_during_coverage(self): """覆盖中途注入读超时与断连：幂等回放保证闭环结果不变。""" sources = random_omni_scene(seed=9, count=3) script = [None, None, None, "timeout_after", None, "drop_before", None, ("status", 500), None, None] summary, report, simulator = _run_scene(sources, fault_script=script) truth = sorted(s.channel for s in sources) self.assertTrue(summary["completed"]) self.assertEqual(sorted(summary["cleared_channels"]), truth) self.assertEqual(report["mismatches"], []) self.assertEqual(sorted(simulator.cleared), truth) class StepTimingTests(unittest.TestCase): """每步计时 [用户决策 2026-09-11]：实机耗时 + 比赛虚拟耗时分项。""" def test_steps_recorded_with_virtual_breakdown(self): sources = random_omni_scene(seed=5, count=2) summary, report, _ = _run_scene(sources) steps = summary["steps"] self.assertEqual(steps[0]["action"], "enter") self.assertEqual(steps[-1]["action"], "exit") self.assertTrue(all(step["real_elapsed_ms"] >= 0 for step in steps)) first_measure = next(step for step in steps if step["action"] == "measure") self.assertEqual(first_measure["step_virtual_s"], first_measure["move_s"] + first_measure["switch_s"] + first_measure["action_s"]) totals = summary["step_totals"] self.assertEqual(totals["count"], len(steps)) # 虚拟分项之和 = 虚拟总时长（比赛口径：移动+切频道+检测+清除） self.assertAlmostEqual( totals["virtual_move_s"] + totals["virtual_switch_s"] + totals["virtual_measure_action_s"] + totals["virtual_clear_action_s"], totals["virtual_total_s"], delta=0.5) self.assertAlmostEqual(totals["virtual_total_s"], summary["virtual_time_s"], delta=0.5) self.assertGreaterEqual(totals["real_elapsed_total_s"], 0.0) # 清除成功一步 5s、未发现 3s：步级记录可复核 clear_step = next(step for step in steps if step["action"] == "clear") self.assertIn(clear_step["action_s"], (3.0, 5.0)) self.assertEqual(clear_step["switch_s"], 0.0, "/clear 不切频道") def test_avg_locate_clear_time_metric(self): """题面口径：平均定位清除时间 = 虚拟总时长 / 被清除个数。""" sources = random_omni_scene(seed=5, count=2) summary, _, _ = _run_scene(sources) cleared = len(summary["cleared_channels"]) self.assertEqual(summary["avg_locate_clear_time_s"], round(summary["virtual_time_s"] / cleared, 3)) class PersistentSearchTests(unittest.TestCase): def test_persistence_disabled_stops_when_resolved(self): """persistent_search=False：回到"证明找全即退出"的基线行为。""" config = replace(CONFIG, search=replace( CONFIG.search, persistent_search=False, bearing_rounding_margin_deg=0.005)) summary, _, _ = _run_scene(random_omni_scene(seed=5, count=2), config=config) self.assertTrue(summary["completed"]) self.assertEqual(summary["stop_reason"], "all_resolved") self.assertLessEqual(summary["measures"], 49 * 20 + 60) def test_persistence_resweeps_only_uncleared_channels(self): """复扫只测未清除频道：已清除频道不再消耗检测。""" sources = random_omni_scene(seed=5, count=2) summary, _, _ = _run_scene(sources) cleared = set(summary["cleared_channels"]) resweep_channels = {step["channel"] for step in summary["steps"] if step["index"] > 0 and step["virtual_time_s"] > summary["accounted_virtual_time_s"] and step["action"] == "measure"} self.assertTrue(resweep_channels) self.assertFalse(resweep_channels & cleared, "已清除频道不得在找全后被复扫") def test_virtual_budget_stops_before_simulator_cap(self): """计划层在虚拟安全余量处停下，不触发模拟器 accepted=false 兜底。""" sources = random_omni_scene(seed=5, count=2) summary, report, _ = _run_scene(sources) self.assertEqual(summary["stop_reason"], "virtual_time_cap") field = SMALL_CAP.field self.assertLessEqual(summary["virtual_time_s"], field.max_virtual_duration_s) self.assertGreaterEqual(summary["virtual_time_s"], field.max_virtual_duration_s - SMALL_CAP.search.virtual_reserve_s - 5_000) self.assertEqual(summary["anomalies"], []) self.assertEqual(report["mismatches"], []) class SimulatorVirtualCapTests(unittest.TestCase): def test_simulator_rejects_actions_beyond_virtual_cap(self): """附件2 §4.5 兜底：虚拟时长达限后本地模拟器不再执行检测。""" simulator = LocalSimulator([], robot_id="T001", config=CONFIG) simulator.entered = True simulator.virtual_time_s = CONFIG.field.max_virtual_duration_s from mcm26b.sim import SimulatorClosed with self.assertRaises(SimulatorClosed): simulator.handle( "/measure", {"arena_id": "default", "robot_id": "T001", "request_id": "x1", "position": {"x": 0.0, "y": 0.0}, "channel": 1}) class AbortFinalizationTests(unittest.TestCase): """断连超过重试上限：仍须 /exit、产出未完成 summary，不得静默崩溃。""" def test_transport_failure_aborts_with_summary_and_exit(self): """首个检测动作三次断连：run() 正常返回，/exit 成功，stop_reason 留档。""" summary, _, _ = _run_scene( random_omni_scene(seed=5, count=2), fault_script=[None, "drop_before", "drop_before", "drop_before"]) self.assertFalse(summary["completed"]) self.assertEqual(summary["stop_reason"], "transport_failure") self.assertEqual(summary["exit_reason"], "user_exit", "接口仍可用时必须尝试退出") self.assertEqual(summary["abort_detail"]["type"], "TransportError") self.assertLessEqual(summary["measures"] + summary["clears"], summary["total_requests"]) self.assertEqual(len(summary["channel_status"]), 20, "未完成状态须有逐频道快照") def test_exit_failure_still_produces_summary(self): """检测与 /exit 均三次断连：产出 exit_unreachable 的未完成 summary。""" summary, _, _ = _run_scene( random_omni_scene(seed=7, count=2), fault_script=[None, "drop_before", "drop_before", "drop_before", "drop_before", "drop_before", "drop_before"]) self.assertFalse(summary["completed"]) self.assertEqual(summary["stop_reason"], "transport_failure") self.assertEqual(summary["exit_reason"], "exit_unreachable") self.assertEqual(summary["exit_error"]["type"], "TransportError") self.assertIn("channel_status", summary) class CoverageLedgerIntegrationTests(unittest.TestCase): def test_grid_matches_plan_derivation(self): points = coverage_points(CONFIG) self.assertEqual(len(points), 49) self.assertEqual(min(p[0] for p in points), -2100.0) self.assertEqual(max(p[1] for p in points), 2100.0) # 单元内任意点到最近网格点 <= 700*sqrt(2) <= 1000 的覆盖保证 worst = 0.0 for gx in range(-20, 21): for gy in range(-20, 21): g = (gx * 90.0, gy * 90.0) if (g[0] ** 2 + g[1] ** 2) ** 0.5 <= 1800.0: nearest = min(math_dist(g, p) for p in points) worst = max(worst, nearest) self.assertLessEqual(worst, 700.0 * 2 ** 0.5 + 1e-9) class SceneGenerationTests(unittest.TestCase): """场景生成 [题面]：默认 10~16 源、Ω 内均匀、频道互异、r_eff∈[1000,1500]。""" def test_default_count_within_problem_range(self): for seed in range(20): sources = random_omni_scene(seed=seed) self.assertTrue(10 <= len(sources) <= 16, f"seed={seed}") def test_scene_respects_problem_constraints(self): for seed in range(10): sources = random_omni_scene(seed=seed) channels = [s.channel for s in sources] self.assertEqual(len(channels), len(set(channels))) self.assertTrue(all(1 <= c <= 20 for c in channels)) for s in sources: self.assertLessEqual((s.x ** 2 + s.y ** 2) ** 0.5, 1800.0) self.assertGreaterEqual(s.r_eff_m, 1000.0) self.assertLessEqual(s.r_eff_m, 1500.0) def math_dist(a, b): return ((a[0] - b[0]) ** 2 + (a[1] - b[1]) ** 2) ** 0.5 if __name__ == "__main__": unittest.main()