feat: 增加评审文档机器投影与 JSON 报告

2026-08-28 14:23:30 +08:00
parent dd012f4905
commit 843952b194
7 changed files with 2096 additions and 16 deletions
@@ -0,0 +1,512 @@
+from __future__ import annotations
+
+from dataclasses import replace
+from json import loads
+from typing import cast
+
+import pytest
+
+from mdpolish import (
+    DocumentSnapshot,
+    Modifier,
+    ModifierInfo,
+    Pipeline,
+    ProposedChange,
+    RunStatus,
+    TextEdit,
+    TextSpan,
+)
+from mdpolish.review import (
+    JsonValue,
+    ReviewCurrentKind,
+    ReviewDetail,
+    ReviewDocument,
+    ReviewLocation,
+    ReviewProjection,
+    ReviewProjectionError,
+    build_review_document,
+    render_json_report,
+    review_document_to_dict,
+)
+
+_MAX_SAFE_JSON_INTEGER = 2**53 - 1
+
+
+def _replace_modifier(
+    needle: str,
+    replacement: str,
+    *,
+    modifier_id: str,
+    reason: str,
+    parameters: dict[str, object] | None = None,
+    applicability: str = "只处理虚构投影测试标记。",
+) -> Modifier:
+    def propose(snapshot: DocumentSnapshot) -> tuple[ProposedChange, ...]:
+        position = snapshot.markdown.find(needle)
+        if position < 0:
+            return ()
+        edit = TextEdit(
+            snapshot_sha256=snapshot.sha256,
+            span=TextSpan(position, position + len(needle)),
+            expected_text=needle,
+            replacement=replacement,
+        )
+        return (
+            ProposedChange(
+                snapshot_sha256=snapshot.sha256,
+                reason=reason,
+                edits=(edit,),
+            ),
+        )
+
+    return Modifier(
+        modifier_id=modifier_id,
+        version="1.2.3",
+        parameters=parameters or {"needle": needle, "replacement": replacement},
+        applicability=applicability,
+        propose=propose,
+    )
+
+
+def _rich_success_review() -> ReviewDocument:
+    modifier = _replace_modifier(
+        "BEFORE_SECRET",
+        "AFTER_SECRET",
+        modifier_id="test.machine-projection",
+        reason="REASON_SECRET",
+        parameters={
+            "enabled": True,
+            "finite": 1.25,
+            "mapping": {"alpha": 1},
+            "none": None,
+            "sequence": (("alpha", 1),),
+        },
+        applicability="APPLICABILITY_SECRET",
+    )
+    input_markdown = "中文_SOURCE_ONLY\r\nBEFORE_SECRET\tTAIL"
+    result = Pipeline((modifier,)).transform(input_markdown)
+    assert result.status is RunStatus.SUCCESS
+    return build_review_document(input_markdown, result)
+
+
+def _unstable_review() -> ReviewDocument:
+    def propose_residual(snapshot: DocumentSnapshot) -> tuple[ProposedChange, ...]:
+        needle = "RESIDUAL_EXPECTED_SECRET"
+        position = snapshot.markdown.find(needle)
+        if position < 0:
+            return ()
+        edit = TextEdit(
+            snapshot_sha256=snapshot.sha256,
+            span=TextSpan(position, position + len(needle)),
+            expected_text=needle,
+            replacement="RESIDUAL_REPLACEMENT_SECRET",
+        )
+        return (
+            ProposedChange(
+                snapshot_sha256=snapshot.sha256,
+                reason="RESIDUAL_REASON_SECRET",
+                edits=(edit,),
+            ),
+        )
+
+    residual = Modifier(
+        modifier_id="test.residual",
+        version="1.0.0",
+        parameters=(),
+        applicability="只在最终复查中产生虚构候选。",
+        propose=propose_residual,
+    )
+    producer = _replace_modifier(
+        "start",
+        "RESIDUAL_EXPECTED_SECRET",
+        modifier_id="test.producer",
+        reason="produce residual marker",
+    )
+    result = Pipeline((residual, producer)).transform("start")
+    assert result.status is RunStatus.UNSTABLE
+    return build_review_document("start", result)
+
+
+def _exploding_modifier(modifier_id: str, *, trigger: str | None = None) -> Modifier:
+    def propose(snapshot: DocumentSnapshot) -> tuple[ProposedChange, ...]:
+        if trigger is None or snapshot.markdown == trigger:
+            raise RuntimeError(f"ERROR_MESSAGE_SECRET: {snapshot.markdown}")
+        return ()
+
+    return Modifier(
+        modifier_id=modifier_id,
+        version="1.0.0",
+        parameters=(),
+        applicability="只测试机器错误投影。",
+        propose=propose,
+    )
+
+
+def _object(value: JsonValue) -> ReviewProjection:
+    assert isinstance(value, dict)
+    return value
+
+
+def _array(value: JsonValue) -> list[JsonValue]:
+    assert isinstance(value, list)
+    return value
+
+
+def _contains_key(value: JsonValue, key: str) -> bool:
+    if isinstance(value, dict):
+        return key in value or any(_contains_key(nested, key) for nested in value.values())
+    if isinstance(value, list):
+        return any(_contains_key(nested, key) for nested in value)
+    return False
+
+
+def test_summary_projection_has_stable_schema_without_body_content() -> None:
+    review = _rich_success_review()
+
+    projection = review_document_to_dict(review)
+
+    assert set(projection) == {
+        "schema_name",
+        "schema_version",
+        "detail",
+        "status",
+        "current_kind",
+        "stages_complete",
+        "hash_contract",
+        "coordinate_contract",
+        "input",
+        "current",
+        "counts",
+        "modifiers",
+        "stages",
+        "errors",
+    }
+    assert projection["schema_name"] == "mdpolish.review"
+    assert projection["schema_version"] == "1.0"
+    assert projection["detail"] == "summary"
+    assert projection["status"] == "success"
+    assert projection["current_kind"] == "success_output"
+    assert projection["stages_complete"] is True
+    assert projection["hash_contract"] == {
+        "algorithm": "sha256",
+        "encoding": "utf-8",
+        "normalization": "none",
+    }
+    assert projection["coordinate_contract"] == {
+        "offset_unit": "unicode_code_point",
+        "span_index_base": 0,
+        "span_end": "exclusive",
+        "location_index_base": 1,
+        "physical_line_endings": ["lf", "crlf", "cr"],
+    }
+    assert projection["counts"] == {
+        "modifier_count": 1,
+        "completed_stage_count": 1,
+        "change_count": 1,
+        "error_count": 0,
+        "residual_proposal_count": 0,
+    }
+
+    input_projection = _object(projection["input"])
+    current_projection = _object(projection["current"])
+    assert input_projection == {
+        "sha256": review.input_sha256,
+        "code_point_length": len(review.input_markdown),
+    }
+    assert current_projection == {
+        "sha256": review.current_sha256,
+        "code_point_length": len(review.current_markdown),
+    }
+
+    modifiers = _array(projection["modifiers"])
+    assert modifiers == [
+        {
+            "position": 0,
+            "modifier_id": "test.machine-projection",
+            "version": "1.2.3",
+        }
+    ]
+    stages = _array(projection["stages"])
+    stage = _object(stages[0])
+    assert set(stage) == {"modifier_position", "before", "after", "change_count"}
+    assert stage["modifier_position"] == 0
+    assert stage["change_count"] == 1
+    assert stage["before"] == {
+        "sha256": review.stages[0].before_sha256,
+        "code_point_length": len(review.stages[0].before_markdown),
+    }
+    assert stage["after"] == {
+        "sha256": review.stages[0].after_sha256,
+        "code_point_length": len(review.stages[0].after_markdown),
+    }
+
+    rendered = render_json_report(review)
+    for secret in (
+        "中文_SOURCE_ONLY",
+        "BEFORE_SECRET",
+        "AFTER_SECRET",
+        "REASON_SECRET",
+        "APPLICABILITY_SECRET",
+        "ERROR_MESSAGE_SECRET",
+    ):
+        assert secret not in rendered
+    for body_key in (
+        "markdown",
+        "expected_text",
+        "replacement",
+        "reason",
+        "message",
+        "parameters",
+        "applicability",
+    ):
+        assert not _contains_key(projection, body_key)
+
+
+def test_changes_and_full_details_add_content_monotonically() -> None:
+    review = _rich_success_review()
+
+    summary = review_document_to_dict(review, detail=ReviewDetail.SUMMARY)
+    changes = review_document_to_dict(review, detail="changes")
+    full = review_document_to_dict(review, detail=ReviewDetail.FULL)
+
+    assert set(summary) < set(changes)
+    assert set(changes) == set(full)
+    assert changes["detail"] == "changes"
+    assert full["detail"] == "full"
+    for key in set(summary) - {"detail", "input", "current", "modifiers", "stages"}:
+        assert summary[key] == changes[key] == full[key]
+
+    changes_modifier = _object(_array(changes["modifiers"])[0])
+    full_modifier = _object(_array(full["modifiers"])[0])
+    assert changes_modifier == full_modifier
+    assert changes_modifier["applicability"] == "APPLICABILITY_SECRET"
+    assert changes_modifier["parameters"] == [
+        ["enabled", True],
+        ["finite", 1.25],
+        ["mapping", [["alpha", 1]]],
+        ["none", None],
+        ["sequence", [["alpha", 1]]],
+    ]
+
+    changes_stage = _object(_array(changes["stages"])[0])
+    full_stage = _object(_array(full["stages"])[0])
+    assert "markdown" not in _object(changes_stage["before"])
+    assert "markdown" not in _object(changes_stage["after"])
+    assert not _contains_key(changes, "markdown")
+    assert _object(full_stage["before"])["markdown"] == review.input_markdown
+    assert _object(full_stage["after"])["markdown"] == review.current_markdown
+    assert _object(full["input"])["markdown"] == review.input_markdown
+    assert _object(full["current"])["markdown"] == review.current_markdown
+
+    change = _object(_array(changes_stage["changes"])[0])
+    assert change == {
+        "proposal_index": 0,
+        "edit_index": 0,
+        "reason": "REASON_SECRET",
+        "location": {"line": 2, "column": 1},
+        "span": {
+            "start": review.input_markdown.index("BEFORE_SECRET"),
+            "end": review.input_markdown.index("BEFORE_SECRET") + len("BEFORE_SECRET"),
+        },
+        "before": "BEFORE_SECRET",
+        "after": "AFTER_SECRET",
+        "before_sha256": review.stages[0].before_sha256,
+        "after_sha256": review.stages[0].after_sha256,
+    }
+    assert "中文_SOURCE_ONLY" not in render_json_report(review, detail="changes")
+
+
+def test_residual_content_requires_changes_or_full_detail() -> None:
+    review = _unstable_review()
+
+    summary = review_document_to_dict(review, detail="summary")
+    changes = review_document_to_dict(review, detail="changes")
+
+    assert summary["status"] == "unstable"
+    assert summary["current_kind"] == "partial_output"
+    assert _object(summary["counts"])["residual_proposal_count"] == 1
+    assert "residual_proposals" not in summary
+    assert "RESIDUAL_EXPECTED_SECRET" not in render_json_report(review, detail="summary")
+
+    residuals = _array(changes["residual_proposals"])
+    assert len(residuals) == 1
+    residual = _object(residuals[0])
+    assert residual["modifier_position"] == 0
+    assert residual["proposal_index"] == 0
+    assert residual["snapshot_sha256"] == review.current_sha256
+    assert residual["reason"] == "RESIDUAL_REASON_SECRET"
+    edit = _object(_array(residual["edits"])[0])
+    assert edit["edit_index"] == 0
+    assert edit["expected_text"] == "RESIDUAL_EXPECTED_SECRET"
+    assert edit["replacement"] == "RESIDUAL_REPLACEMENT_SECRET"
+
+
+def _error_reviews() -> tuple[tuple[ReviewDocument, str], ...]:
+    invalid_pipeline = Pipeline(cast(tuple[Modifier, ...], (object(),)))
+    preflight_result = invalid_pipeline.transform("PREFLIGHT_SOURCE_SECRET")
+    preflight = build_review_document("PREFLIGHT_SOURCE_SECRET", preflight_result)
+
+    transform_result = Pipeline((_exploding_modifier("test.transform-error"),)).transform(
+        "TRANSFORM_SOURCE_SECRET"
+    )
+    transform = build_review_document("TRANSFORM_SOURCE_SECRET", transform_result)
+
+    review_error = _exploding_modifier("test.final-review-error", trigger="done")
+    producer = _replace_modifier(
+        "start",
+        "done",
+        modifier_id="test.final-producer",
+        reason="produce final review trigger",
+    )
+    final_result = Pipeline((review_error, producer)).transform("start")
+    final_review = build_review_document("start", final_result)
+    return (
+        (preflight, "run.preflight_failed"),
+        (transform, "run.transform_failed"),
+        (final_review, "run.final_review_failed"),
+    )
+
+
+def test_error_projection_uses_stable_codes_and_hides_diagnostics_in_summary() -> None:
+    for review, expected_code in _error_reviews():
+        summary = review_document_to_dict(review, detail="summary")
+        changes = review_document_to_dict(review, detail="changes")
+
+        summary_error = _object(_array(summary["errors"])[0])
+        changes_error = _object(_array(changes["errors"])[0])
+        assert summary_error["code"] == expected_code
+        assert summary_error["stage"] == review.errors[0].stage.value
+        assert "diagnostic_type" not in summary_error
+        assert "message" not in summary_error
+        assert "ERROR_MESSAGE_SECRET" not in render_json_report(review, detail="summary")
+        assert changes_error["diagnostic_type"] == review.errors[0].error_type
+        assert changes_error["message"] == review.errors[0].message
+
+
+def test_projection_preserves_unicode_line_endings_and_exact_coordinates() -> None:
+    modifier = _replace_modifier(
+        "🙂Cafe\u0301",
+        "完成",
+        modifier_id="test.unicode",
+        reason="Unicode coordinate test",
+    )
+    input_markdown = "\ufeff首行\r\n🙂Cafe\u0301\r尾行\n"
+    result = Pipeline((modifier,)).transform(input_markdown)
+    review = build_review_document(input_markdown, result)
+
+    projection = review_document_to_dict(review, detail="full")
+    stage = _object(_array(projection["stages"])[0])
+    change = _object(_array(stage["changes"])[0])
+
+    assert _object(projection["input"])["code_point_length"] == len(input_markdown)
+    assert _object(projection["input"])["markdown"] == input_markdown
+    assert change["location"] == {"line": 2, "column": 1}
+    assert change["span"] == {
+        "start": input_markdown.index("🙂"),
+        "end": input_markdown.index("🙂") + len("🙂Cafe\u0301"),
+    }
+
+
+def test_json_report_is_deterministic_unicode_json_without_bom_or_final_newline() -> None:
+    review = _rich_success_review()
+
+    first = render_json_report(review, detail="full")
+    second = render_json_report(review, detail=ReviewDetail.FULL)
+
+    assert first == second
+    assert loads(first) == review_document_to_dict(review, detail="full")
+    assert not first.startswith("\ufeff")
+    assert not first.endswith("\n")
+    assert "中文_SOURCE_ONLY" in first
+    assert "\\u4e2d" not in first.lower()
+    assert "\\tTAIL" in first
+    assert "NaN" not in first
+    assert "Infinity" not in first
+
+
+def test_projection_returns_fresh_containers_without_mutating_review() -> None:
+    review = _rich_success_review()
+
+    first = review_document_to_dict(review, detail="full")
+    second = review_document_to_dict(review, detail="full")
+    first["schema_name"] = "changed"
+    _array(first["modifiers"]).clear()
+
+    assert second["schema_name"] == "mdpolish.review"
+    assert len(_array(second["modifiers"])) == 1
+    assert review.modifiers[0].modifier_id == "test.machine-projection"
+    assert review_document_to_dict(review, detail="full") == second
+
+
+@pytest.mark.parametrize("detail", ["", "SUMMARY", "unknown", 1, True, None])
+def test_projection_rejects_unknown_detail(detail: object) -> None:
+    review = _rich_success_review()
+
+    with pytest.raises(ReviewProjectionError, match="review projection failed at detail"):
+        review_document_to_dict(review, detail=detail)  # type: ignore[arg-type]
+
+
+def test_projection_rejects_wrong_review_and_nested_model_types() -> None:
+    with pytest.raises(ReviewProjectionError, match="ReviewDocument"):
+        review_document_to_dict(object())  # type: ignore[arg-type]
+
+    review = _rich_success_review()
+    malformed = replace(review, modifiers=cast(tuple[ModifierInfo, ...], (object(),)))
+    with pytest.raises(ReviewProjectionError, match=r"modifiers\[0\]"):
+        review_document_to_dict(malformed)
+
+
+def test_projection_rejects_unknown_enum_invalid_reference_and_large_integer() -> None:
+    review = _rich_success_review()
+    unknown_status = replace(review, status=cast(RunStatus, "future"))
+    bad_stage = replace(review.stages[0], modifier_position=1)
+    bad_reference = replace(review, stages=(bad_stage,))
+    review_change = review.stages[0].changes[0]
+    large_location = replace(
+        review_change,
+        location=ReviewLocation(line=_MAX_SAFE_JSON_INTEGER + 1, column=1),
+    )
+    large_stage = replace(review.stages[0], changes=(large_location,))
+    large_integer = replace(review, stages=(large_stage,))
+
+    for malformed in (unknown_status, bad_reference, large_integer):
+        with pytest.raises(ReviewProjectionError):
+            review_document_to_dict(malformed)
+
+
+def test_projection_rejects_nonfinite_parameter_and_invalid_unicode_without_leaking_values() -> None:
+    review = _rich_success_review()
+    modifier = review.modifiers[0]
+    nonfinite_modifier = replace(modifier, parameters=(("NONFINITE_SECRET", float("nan")),))
+    nonfinite_stage = replace(review.stages[0], modifier=nonfinite_modifier)
+    nonfinite_review = replace(
+        review,
+        modifiers=(nonfinite_modifier,),
+        stages=(nonfinite_stage,),
+    )
+    invalid_unicode = replace(review, input_markdown="UNICODE_SECRET\ud800")
+
+    with pytest.raises(ReviewProjectionError) as nonfinite_error:
+        review_document_to_dict(nonfinite_review)
+    assert "NONFINITE_SECRET" not in str(nonfinite_error.value)
+
+    with pytest.raises(ReviewProjectionError) as unicode_error:
+        review_document_to_dict(invalid_unicode)
+    assert "UNICODE_SECRET" not in str(unicode_error.value)
+
+
+def test_empty_document_projection_distinguishes_hidden_and_empty_markdown() -> None:
+    result = Pipeline(()).transform("")
+    review = build_review_document("", result)
+
+    summary = review_document_to_dict(review, detail="summary")
+    full = review_document_to_dict(review, detail="full")
+
+    assert "markdown" not in _object(summary["input"])
+    assert _object(full["input"])["markdown"] == ""
+    assert _object(full["current"])["markdown"] == ""
+    assert full["modifiers"] == []
+    assert full["stages"] == []
+    assert full["errors"] == []
+    assert full["residual_proposals"] == []
+    assert review.current_kind is ReviewCurrentKind.SUCCESS_OUTPUT