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("review", 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"