93 lines
4.1 KiB
Python
93 lines
4.1 KiB
Python
"""接口边界（与传输无关）：handle_raw 承载附件2 §5 的全部语法层规则。
|
|
|
|
参赛队的内存传输与本包 HTTP 服务器共用这一实现，两条路径行为一致。
|
|
"""
|
|
|
|
import json
|
|
|
|
import pytest
|
|
|
|
from referee import LocalSimulator, SimulatorClosed
|
|
from referee.interface import handle_raw
|
|
|
|
|
|
def payload(request_id="e", **extra):
|
|
return dict(arena_id="default", robot_id="TEAM-0001", request_id=request_id, **extra)
|
|
|
|
|
|
def post(sim, path, body=None, headers=None, method="POST", target=None):
|
|
raw = json.dumps(body if body is not None else payload()).encode("utf-8")
|
|
return handle_raw(sim, method, target or path,
|
|
headers or [("Content-Type", "application/json")], raw)
|
|
|
|
|
|
def test_raw_json_format_errors():
|
|
sim = LocalSimulator([], robot_id="TEAM-0001")
|
|
cases = [
|
|
(b'{"request_id":"a","request_id":"b"}', 400), # 重复键
|
|
(b'{"position":{"x":1,"x":2}}', 400), # 嵌套重复键
|
|
(b'{}' + b' ' * 65535, 413), # 超 65536 字节
|
|
(b'\xef\xbb\xbf{}', 400), (b'\xff', 400), (b'{', 400), # BOM/坏编码/语法
|
|
(b'[]', 400), (b'{"x":NaN}', 400), (b'{"x":Infinity}', 400),
|
|
(b'{"a":' + b'[' * 16 + b'0' + b']' * 16 + b'}', 400), # 深度 17
|
|
]
|
|
for raw, expected in cases:
|
|
result = handle_raw(sim, "POST", "/enter", [("Content-Type", "application/json")], raw)
|
|
assert result.status == expected, (raw[:40], result.status)
|
|
body = json.loads(result.body)
|
|
assert body["accepted"] is False
|
|
assert set(body) >= {"accepted", "real_timestamp_ms", "virtual_time_s"}
|
|
assert not sim.entered
|
|
|
|
|
|
def test_raw_size_and_depth_boundaries_are_inclusive():
|
|
sim = LocalSimulator([], robot_id="TEAM-0001")
|
|
headers = [("Content-Type", "application/json")]
|
|
deep = b'{"extra":' + b'[' * 15 + b'0' + b']' * 15 + b'}' # 深度 16 合法
|
|
body = json.dumps({**payload(), **json.loads(deep)}).encode()
|
|
assert handle_raw(sim, "POST", "/enter", headers, body).status == 200
|
|
exact = json.dumps(payload()).encode().ljust(65536) # 恰好 65536 字节
|
|
result = handle_raw(sim, "POST", "/enter", headers, exact)
|
|
assert result.status == 200 and json.loads(result.body)["accepted"]
|
|
|
|
|
|
@pytest.mark.parametrize("method,path,headers,status", [
|
|
("POST", "/enter?x=1", None, 404),
|
|
("POST", "/enter?", None, 404),
|
|
("POST", "/enter/", None, 404),
|
|
("POST", "/measure/x", None, 404),
|
|
("GET", "/enter", None, 405),
|
|
("PUT", "/clear", None, 405),
|
|
("POST", "/enter", [("Content-Type", "text/plain")], 415),
|
|
("POST", "/enter", [("Content-Type", "application/json;foo=bar")], 415),
|
|
("POST", "/enter",
|
|
[("Content-Type", "application/json"), ("Content-Type", "application/json")], 415),
|
|
("POST", "/enter", [("Content-Type", "application/json"), ("Content-Encoding", "gzip")], 415),
|
|
("POST", "/enter", [("content-type", "Application/JSON; charset=UTF-8")], 200),
|
|
("POST", "/enter", [("content-type", 'application/json;charset="utf-8"')], 200),
|
|
])
|
|
def test_http_boundary(method, path, headers, status):
|
|
sim = LocalSimulator([], robot_id="TEAM-0001")
|
|
raw = json.dumps(payload()).encode()
|
|
result = handle_raw(sim, method, path, headers or [("Content-Type", "application/json")], raw)
|
|
assert result.status == status
|
|
|
|
|
|
def test_closed_simulator_raises_connection_level_error():
|
|
sim = LocalSimulator([], clock=lambda: 9999, window_started_at=0, robot_id="TEAM-0001")
|
|
with pytest.raises(SimulatorClosed):
|
|
handle_raw(sim, "POST", "/enter",
|
|
[("Content-Type", "application/json")], json.dumps(payload()).encode())
|
|
|
|
|
|
def test_on_request_callback_receives_parsed_payload():
|
|
sim = LocalSimulator([], robot_id="TEAM-0001")
|
|
seen = []
|
|
raw = json.dumps(payload("enter-1")).encode()
|
|
handle_raw(sim, "POST", "/enter", [("Content-Type", "application/json")], raw,
|
|
on_request=lambda path, payload_dict, result: seen.append(
|
|
(path, payload_dict, result)))
|
|
assert seen[0][0] == "/enter"
|
|
assert seen[0][1]["request_id"] == "enter-1"
|
|
assert seen[0][2].status == 200
|