"""审核报告的独立反例：时限、原始协议、并发和量化。""" 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())