模拟器细节特征对齐

2026-09-11 08:29:56 +08:00
parent 7930eb65bf
commit 52c69b1276
13 changed files with 509 additions and 46 deletions
@@ -0,0 +1,200 @@
+"""审核报告的独立反例：时限、原始协议、并发和量化。"""
+import json
+from concurrent.futures import ThreadPoolExecutor
+from dataclasses import replace
+from threading import Event
+
+import pytest
+
+from mcm26b.config import ProblemConfig
+from mcm26b.sim import LocalSimulator, LocalSimTransport, SimulatorClosed, Source
+
+
+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):
+    return payload(id, position=dict(x=x, y=y), channel=1)
+
+
+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
+    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():
+    config = ProblemConfig()
+    config = replace(config, field=replace(config.field, max_real_duration_s=1800))
+    official = LocalSimulator([], config=config, clock=lambda: 0)
+    stress = LocalSimulator([], config=config, 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=1800000))[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_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
+
+
+@pytest.mark.parametrize("bad", ["e\n", "e\u200b", "e\u0085", "e\ud800"])
+def test_identifier_rejection_does_not_consume_id(bad):
+    sim = LocalSimulator([])
+    assert sim.handle("/enter", payload(bad))[0] == 400
+    assert sim._idempotency == {}
+    assert sim.handle("/enter", payload())[1]["accepted"]
+
+
+@pytest.mark.parametrize("body,status", [
+    (b'{"request_id":"a","request_id":"b"}', 400),
+    (b'{"position":{"x":1,"x":2}}', 400),
+    (b'{}' + b' ' * 65535, 413),
+    (b'\xef\xbb\xbf{}', 400), (b'\xff', 400), (b'{', 400),
+    (b'[]', 400), (b'{"x":NaN}', 400), (b'{"x":Infinity}', 400),
+    (b'{"a":' + b'[' * 16 + b'0' + b']' * 16 + b'}', 400),
+], ids=["duplicate-root", "duplicate-nested", "oversize", "bom", "utf8",
+        "syntax", "non-object", "nan", "infinity", "depth"])
+def test_raw_json(body, status):
+    sim = LocalSimulator([])
+    response = LocalSimTransport(sim).post("http://local/enter", body,
+                                           {"Content-Type": "application/json"}, None)
+    assert response.status == status
+    assert json.loads(response.body)["accepted"] is False
+    assert not sim.entered
+
+
+@pytest.mark.parametrize("method,path,headers,status", [
+    ("POST", "/enter?x=1", {"Content-Type": "application/json"}, 404),
+    ("POST", "/enter?", {"Content-Type": "application/json"}, 404),
+    ("POST", "/enter/", {"Content-Type": "application/json"}, 404),
+    ("GET", "/enter", {"Content-Type": "application/json"}, 405),
+    ("POST", "/enter", {"Content-Type": "text/plain"}, 415),
+    ("POST", "/enter", {"Content-Type": "application/json;foo=bar"}, 415),
+    ("POST", "/enter", {"Content-Type": "application/json", "Content-Encoding": "gzip"}, 415),
+    ("POST", "/enter", {"content-type": "Application/JSON; charset=UTF-8"}, 200),
+])
+def test_http_boundary(method, path, headers, status):
+    transport = LocalSimTransport(LocalSimulator([]))
+    response = transport.request(method, "http://local" + path,
+                                 json.dumps(payload()).encode(), headers)
+    assert response.status == status
+
+
+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_raw_size_and_depth_boundaries_are_inclusive():
+    sim = LocalSimulator([])
+    transport = LocalSimTransport(sim)
+    headers = {"Content-Type": "application/json"}
+    deep = b'{"extra":' + b'[' * 15 + b'0' + b']' * 15 + b'}'
+    # 深度16合法，因缺少公共字段得到400，补公共字段后应为业务拒绝200。
+    body = json.dumps({**payload(), **json.loads(deep)}).encode()
+    assert transport.post("http://local/enter", body, headers, None).status == 200
+    body = json.dumps(payload()).encode()
+    response = transport.post("http://local/enter", body.ljust(65536), headers, None)
+    assert json.loads(response.body)["accepted"]
+
+
+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"]
+
+
+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())