304 lines
11 KiB
Python
304 lines
11 KiB
Python
from __future__ import annotations
|
||
|
||
import http.client
|
||
import json
|
||
import threading
|
||
from pathlib import Path
|
||
from typing import cast
|
||
|
||
import pytest
|
||
|
||
from mdpolish import DocumentSnapshot, Modifier, Pipeline, ProposedChange, TextEdit, TextSpan
|
||
from mdpolish.review import build_review_document, render_json_report
|
||
from mdpolish.reviewer import ReviewDataError, ReviewRepository, create_server, main
|
||
|
||
|
||
def _object(value: object) -> dict[str, object]:
|
||
assert isinstance(value, dict)
|
||
return cast(dict[str, object], value)
|
||
|
||
|
||
def _array(value: object) -> list[object]:
|
||
assert isinstance(value, list)
|
||
return cast(list[object], value)
|
||
|
||
|
||
def _full_review_json(source: str = "😀 old\r\nlast line\r\n") -> str:
|
||
"""生成包含 emoji、CRLF、一次修改和一个零修改阶段的虚构 full JSON。"""
|
||
|
||
def replace_old(snapshot: DocumentSnapshot) -> tuple[ProposedChange, ...]:
|
||
start = snapshot.markdown.find("old")
|
||
if start < 0:
|
||
return ()
|
||
return (
|
||
ProposedChange(
|
||
snapshot_sha256=snapshot.sha256,
|
||
reason="替换虚构测试词",
|
||
edits=(
|
||
TextEdit(
|
||
snapshot_sha256=snapshot.sha256,
|
||
span=TextSpan(start, start + 3),
|
||
expected_text="old",
|
||
replacement="new",
|
||
),
|
||
),
|
||
),
|
||
)
|
||
|
||
pipeline = Pipeline(
|
||
(
|
||
Modifier(
|
||
modifier_id="test.replace",
|
||
version="1.0.0",
|
||
parameters={"mode": "exact"},
|
||
applicability="只替换虚构测试词。",
|
||
propose=replace_old,
|
||
),
|
||
Modifier(
|
||
modifier_id="test.zero",
|
||
version="1.0.0",
|
||
parameters={},
|
||
applicability="用于验证零修改阶段。",
|
||
propose=lambda _snapshot: (),
|
||
),
|
||
)
|
||
)
|
||
result = pipeline.transform(source)
|
||
review = build_review_document(source, result)
|
||
return render_json_report(review, detail="full") + "\n"
|
||
|
||
|
||
def _review_directory(tmp_path: Path, content: str | None = None) -> Path:
|
||
review_dir = tmp_path / "reviews"
|
||
review_dir.mkdir(parents=True)
|
||
(review_dir / "paper.review.json").write_text(
|
||
_full_review_json() if content is None else content,
|
||
encoding="utf-8",
|
||
newline="",
|
||
)
|
||
return review_dir
|
||
|
||
|
||
def test_repository_exposes_collection_document_and_modifier_stages(tmp_path: Path) -> None:
|
||
repository = ReviewRepository(_review_directory(tmp_path))
|
||
|
||
collection = repository.collection_response()
|
||
document = repository.document("paper").document_response()
|
||
first_stage = repository.document("paper").stage_response(0)
|
||
zero_stage = repository.document("paper").stage_response(1)
|
||
|
||
assert collection["collection"] == {"label": "reviews", "status": "success"}
|
||
assert collection["summary"] == {
|
||
"document_count": 1,
|
||
"success_count": 1,
|
||
"failed_count": 0,
|
||
"unstable_count": 0,
|
||
"change_count": 1,
|
||
}
|
||
assert "input_markdown" not in json.dumps(collection)
|
||
assert document["input_markdown"] == "😀 old\r\nlast line\r\n"
|
||
assert document["current_markdown"] == "😀 new\r\nlast line\r\n"
|
||
modifiers = _array(document["modifiers"])
|
||
assert [_object(item)["modifier_id"] for item in modifiers] == ["test.replace", "test.zero"]
|
||
changes = _array(document["changes"])
|
||
# Python 码点 2 位于 emoji 后; CodeMirror UTF-16 下标因此是 3。
|
||
assert _object(changes[0])["span"] == {"start": 2, "end": 5}
|
||
assert _object(changes[0])["editor_range"] == {"start": 3, "end": 6}
|
||
assert first_stage["before_markdown"] == "😀 old\r\nlast line\r\n"
|
||
assert first_stage["after_markdown"] == "😀 new\r\nlast line\r\n"
|
||
assert zero_stage["before_markdown"] == zero_stage["after_markdown"]
|
||
assert zero_stage["changes"] == []
|
||
|
||
|
||
def test_repository_handles_unicode_empty_text_and_file_labels(tmp_path: Path) -> None:
|
||
source = "中e\u0301😀\r\nold\rend"
|
||
repository = ReviewRepository(_review_directory(tmp_path, _full_review_json(source)))
|
||
document = repository.document("paper").document_response()
|
||
changes = _array(document["changes"])
|
||
|
||
assert _object(changes[0])["location"] == {"line": 2, "column": 1}
|
||
assert _object(changes[0])["span"] == {"start": 6, "end": 9}
|
||
assert _object(changes[0])["editor_range"] == {"start": 7, "end": 10}
|
||
assert document["current_markdown"] == "中e\u0301😀\r\nnew\rend"
|
||
assert _object(document["document"])["source_label"] == "paper"
|
||
|
||
empty_directory = tmp_path / "empty"
|
||
empty_directory.mkdir()
|
||
(empty_directory / "empty.review.json").write_text(
|
||
_full_review_json(""),
|
||
encoding="utf-8",
|
||
newline="",
|
||
)
|
||
empty_document = ReviewRepository(empty_directory).document("empty").document_response()
|
||
assert empty_document["input_markdown"] == ""
|
||
assert empty_document["current_markdown"] == ""
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
"mutation",
|
||
(
|
||
lambda payload: payload.__setitem__("detail", "changes"),
|
||
lambda payload: payload["input"].__setitem__(
|
||
"markdown",
|
||
payload["input"]["markdown"].replace("old", "bad"),
|
||
),
|
||
lambda payload: payload["stages"][0]["changes"][0]["span"].__setitem__("start", 0),
|
||
),
|
||
)
|
||
def test_repository_rejects_untrusted_review_shapes(tmp_path: Path, mutation: object) -> None:
|
||
payload = json.loads(_full_review_json())
|
||
assert callable(mutation)
|
||
mutation(payload)
|
||
review_dir = _review_directory(tmp_path, json.dumps(payload, ensure_ascii=False) + "\n")
|
||
|
||
with pytest.raises(ReviewDataError) as raised:
|
||
ReviewRepository(review_dir)
|
||
|
||
assert raised.value.code == "invalid_review"
|
||
assert str(raised.value).startswith("paper.review.json：")
|
||
|
||
|
||
def test_repository_accepts_compatible_schema_minor(tmp_path: Path) -> None:
|
||
payload = json.loads(_full_review_json())
|
||
payload["schema_version"] = "1.4"
|
||
payload["future_summary"] = {"available": True}
|
||
|
||
repository = ReviewRepository(
|
||
_review_directory(tmp_path, json.dumps(payload, ensure_ascii=False) + "\n")
|
||
)
|
||
|
||
assert repository.document("paper").status == "success"
|
||
|
||
|
||
def test_repository_rejects_bom_invalid_utf8_and_symlinks(tmp_path: Path) -> None:
|
||
review_dir = _review_directory(tmp_path)
|
||
review_path = review_dir / "paper.review.json"
|
||
review_path.write_bytes(b"\xef\xbb\xbf" + review_path.read_bytes())
|
||
with pytest.raises(ReviewDataError, match="BOM"):
|
||
ReviewRepository(review_dir)
|
||
|
||
review_path.write_bytes(b'{"bad": "\xff"}')
|
||
with pytest.raises(ReviewDataError, match="UTF-8"):
|
||
ReviewRepository(review_dir)
|
||
|
||
review_path.unlink()
|
||
target = tmp_path / "outside.review.json"
|
||
target.write_text(_full_review_json(), encoding="utf-8", newline="")
|
||
review_path.symlink_to(target)
|
||
with pytest.raises(ReviewDataError) as raised:
|
||
ReviewRepository(review_dir)
|
||
assert raised.value.code == "unsafe_path"
|
||
|
||
review_path.unlink()
|
||
review_path.write_text(_full_review_json(), encoding="utf-8", newline="")
|
||
link = tmp_path / "review-link"
|
||
link.symlink_to(review_dir, target_is_directory=True)
|
||
with pytest.raises(ReviewDataError) as raised:
|
||
ReviewRepository(link)
|
||
assert raised.value.code == "unsafe_path"
|
||
|
||
|
||
def test_repository_rejects_empty_directory_and_empty_label(tmp_path: Path) -> None:
|
||
empty = tmp_path / "empty"
|
||
empty.mkdir()
|
||
with pytest.raises(ReviewDataError) as raised:
|
||
ReviewRepository(empty)
|
||
assert raised.value.code == "empty_review_directory"
|
||
|
||
(empty / ".review.json").write_text(_full_review_json(), encoding="utf-8", newline="")
|
||
with pytest.raises(ReviewDataError, match="缺少文档标签"):
|
||
ReviewRepository(empty)
|
||
|
||
|
||
def test_console_entry_reports_expected_startup_failure_without_traceback(tmp_path: Path) -> None:
|
||
empty = tmp_path / "empty"
|
||
empty.mkdir()
|
||
|
||
with pytest.raises(SystemExit) as raised:
|
||
main(["--review-dir", str(empty)])
|
||
|
||
assert str(raised.value) == "评审器启动失败：评审目录没有直属 full review JSON。"
|
||
|
||
|
||
def test_http_server_is_loopback_read_only_and_does_not_leak_text(tmp_path: Path) -> None:
|
||
repository = ReviewRepository(_review_directory(tmp_path))
|
||
static_root = tmp_path / "static"
|
||
static_root.mkdir()
|
||
(static_root / "index.html").write_text("<!doctype html><title>review</title>", encoding="utf-8")
|
||
server = create_server(repository, static_root)
|
||
thread = threading.Thread(target=server.serve_forever, daemon=True)
|
||
thread.start()
|
||
host = cast(str, server.server_address[0])
|
||
port = server.server_address[1]
|
||
|
||
try:
|
||
assert host == "127.0.0.1"
|
||
connection = http.client.HTTPConnection(host, port)
|
||
connection.request("GET", "/api/v1/collection")
|
||
response = connection.getresponse()
|
||
body = response.read().decode("utf-8")
|
||
assert response.status == 200
|
||
assert response.getheader("Cache-Control") == "no-store"
|
||
assert response.getheader("Access-Control-Allow-Origin") is None
|
||
assert "😀 old" not in body
|
||
|
||
connection.request("POST", "/api/v1/collection")
|
||
response = connection.getresponse()
|
||
response.read()
|
||
assert response.status == 405
|
||
|
||
connection.request("GET", "/api/v1/documents/missing")
|
||
response = connection.getresponse()
|
||
response.read()
|
||
assert response.status == 404
|
||
|
||
connection.request("GET", "/api/v1/documents/paper/modifiers/99")
|
||
response = connection.getresponse()
|
||
response.read()
|
||
assert response.status == 404
|
||
|
||
connection.request("HEAD", "/")
|
||
response = connection.getresponse()
|
||
assert response.status == 200
|
||
assert response.read() == b""
|
||
assert response.getheader("Content-Security-Policy") is not None
|
||
assert response.getheader("X-Content-Type-Options") == "nosniff"
|
||
|
||
connection.putrequest("GET", "/api/v1/collection", skip_host=True)
|
||
connection.putheader("Host", "example.com")
|
||
connection.endheaders()
|
||
response = connection.getresponse()
|
||
response.read()
|
||
assert response.status == 403
|
||
connection.close()
|
||
|
||
connection = http.client.HTTPConnection(host, port)
|
||
connection.putrequest("GET", "/api/v1/collection")
|
||
connection.putheader("Origin", "http://example.com")
|
||
connection.endheaders()
|
||
response = connection.getresponse()
|
||
response.read()
|
||
assert response.status == 403
|
||
connection.close()
|
||
finally:
|
||
server.shutdown()
|
||
server.server_close()
|
||
thread.join(timeout=5)
|
||
|
||
|
||
def test_server_rejects_missing_or_symlink_static_root(tmp_path: Path) -> None:
|
||
repository = ReviewRepository(_review_directory(tmp_path))
|
||
missing = tmp_path / "missing-static"
|
||
with pytest.raises(ReviewDataError) as raised:
|
||
create_server(repository, missing)
|
||
assert raised.value.code == "missing_build"
|
||
|
||
real = tmp_path / "real-static"
|
||
real.mkdir()
|
||
(real / "index.html").write_text("ok", encoding="utf-8")
|
||
link = tmp_path / "static-link"
|
||
link.symlink_to(real, target_is_directory=True)
|
||
with pytest.raises(ReviewDataError) as raised:
|
||
create_server(repository, link)
|
||
assert raised.value.code == "missing_build"
|