实现本地 Markdown 清洗评审器

tests/test_reviewer_server.py

@@ -0,0 +1,105 @@
+from __future__ import annotations
+
+import json
+import threading
+from collections.abc import Generator
+from http.client import HTTPConnection, HTTPResponse
+from pathlib import Path
+from typing import Any, cast
+
+import pytest
+
+from reviewer.server import ReviewArtifacts
+from reviewer.server.__main__ import create_server
+from tests.reviewer_fixture import ReviewFixture, create_review_run
+
+
+def request(
+    port: int,
+    method: str,
+    path: str,
+    *,
+    headers: dict[str, str] | None = None,
+) -> tuple[HTTPResponse, bytes]:
+    connection = HTTPConnection("127.0.0.1", port, timeout=3)
+    connection.request(method, path, headers=headers or {})
+    response = connection.getresponse()
+    content = response.read()
+    connection.close()
+    return response, content
+
+
+@pytest.fixture
+def running_server(tmp_path: Path) -> Generator[tuple[int, ReviewFixture], None, None]:
+    fixture = create_review_run(tmp_path)
+    static_root = tmp_path / "static"
+    (static_root / "assets").mkdir(parents=True)
+    (static_root / "index.html").write_text("<main>reviewer</main>\n", encoding="utf-8")
+    (static_root / "assets/app.js").write_text("export {};\n", encoding="utf-8")
+    repository = ReviewArtifacts(fixture["run_directory"])
+    server = create_server(repository, static_root)
+    thread = threading.Thread(target=server.serve_forever, daemon=True)
+    thread.start()
+    try:
+        yield server.server_address[1], fixture
+    finally:
+        server.shutdown()
+        server.server_close()
+        thread.join(timeout=3)
+
+
+def test_server_exposes_same_origin_api_and_static_build(
+    running_server: tuple[int, ReviewFixture],
+) -> None:
+    port, fixture = running_server
+
+    api_response, api_content = request(port, "GET", "/api/v1/run")
+    page_response, page_content = request(port, "GET", "/")
+    asset_response, _ = request(port, "HEAD", "/assets/app.js")
+
+    payload = cast(dict[str, Any], json.loads(api_content))
+    assert api_response.status == 200
+    assert payload["run"]["run_id"] == "review-run"
+    assert str(fixture["source_path"]) not in api_content.decode()
+    assert api_response.getheader("Access-Control-Allow-Origin") is None
+    assert api_response.getheader("Cache-Control") == "no-store"
+    assert "default-src 'self'" in cast(str, api_response.getheader("Content-Security-Policy"))
+    assert page_response.status == 200
+    assert page_content == b"<main>reviewer</main>\n"
+    assert asset_response.status == 200
+    assert asset_response.getheader("Content-Type") == "text/javascript; charset=utf-8"
+
+
+@pytest.mark.parametrize(
+    ("method", "path", "headers", "status", "code"),
+    [
+        ("POST", "/api/v1/run", None, 405, "method_not_allowed"),
+        ("GET", "/api/v1/run", {"Host": "example.com"}, 403, "invalid_origin"),
+        (
+            "GET",
+            "/api/v1/run",
+            {"Origin": "http://example.com"},
+            403,
+            "invalid_origin",
+        ),
+        ("GET", "/api/v1/documents/missing", None, 404, "unknown_document"),
+        ("GET", "/api/v1/documents/paper/components/99", None, 404, "unknown_component"),
+        ("GET", "/assets/missing.js", None, 404, "not_found"),
+        ("GET", "/..%2Fsecret.txt", None, 404, "not_found"),
+    ],
+)
+def test_server_rejects_unsafe_or_unknown_requests(
+    running_server: tuple[int, ReviewFixture],
+    method: str,
+    path: str,
+    headers: dict[str, str] | None,
+    status: int,
+    code: str,
+) -> None:
+    port, _ = running_server
+
+    response, content = request(port, method, path, headers=headers)
+    payload = cast(dict[str, Any], json.loads(content))
+
+    assert response.status == status
+    assert payload["error"]["code"] == code