mdpolish/tests/test_reporting.py

from __future__ import annotations

import json
from collections.abc import Mapping
from dataclasses import replace

import pytest

from mdpolish import Component, DocumentSnapshot, Pipeline, ProposedChange, RunStatus, TextEdit, TextSpan
from mdpolish.reporting import ReportingError, build_document_report, build_unified_diff


class ReplaceComponent(Component):
    def __init__(self, needle: str, replacement: str, component_id: str) -> None:
        self.needle = needle
        self.replacement = replacement
        self._component_id = component_id

    @property
    def component_id(self) -> str:
        return self._component_id

    @property
    def version(self) -> str:
        return "1.0.0"

    @property
    def parameters(self) -> Mapping[str, object]:
        return {"needle": self.needle, "replacement": self.replacement}

    @property
    def applicability(self) -> str:
        return "处理精确测试标记，要求完整匹配，排除其他文本。"

    def _propose_changes(self, snapshot: DocumentSnapshot) -> tuple[ProposedChange, ...]:
        position = snapshot.markdown.find(self.needle)
        if position < 0:
            return ()
        edit = TextEdit(
            snapshot.sha256,
            TextSpan(position, position + len(self.needle)),
            self.needle,
            self.replacement,
        )
        return (ProposedChange(snapshot.sha256, f"replace {self.needle}", (edit,)),)


class ExplodingComponent(ReplaceComponent):
    def __init__(self) -> None:
        super().__init__("unused", "unused-replacement", "test.exploding")

    def _propose_changes(self, snapshot: DocumentSnapshot) -> tuple[ProposedChange, ...]:
        raise RuntimeError(f"private: {snapshot.markdown}")


def test_success_report_replays_multiple_component_snapshots_and_derives_locations() -> None:
    markdown = "x\r\ntarget\n"
    pipeline = Pipeline(
        [
            ReplaceComponent("x", "XX", "test.expand"),
            ReplaceComponent("target", "done", "test.target"),
        ]
    )
    result = pipeline.transform(markdown)

    report = build_document_report(
        document_id="paper",
        source_label="data/md/paper.md",
        input_markdown=markdown,
        result=result,
    )
    payload = json.loads(report.result_json)

    assert result.status is RunStatus.SUCCESS
    assert report.cleaned_markdown == b"XX\r\ndone\n"
    assert payload["changes"][0]["location"] == {"line": 1, "column": 1}
    assert payload["changes"][1]["location"] == {"line": 2, "column": 1}
    assert payload["changes"][1]["before"] == "target"
    assert payload["changes"][1]["after"] == "done"
    assert payload["changes"][1]["before_sha256"] == result.changes[1].before_sha256
    assert payload["changes"][1]["after_sha256"] == result.changes[1].after_sha256
    assert report.diff is not None
    assert b"--- a/paper.md\n+++ b/paper.md\n" in report.diff
    assert b"data/md/paper.md" not in report.diff


def test_zero_change_success_still_has_cleaned_markdown_and_empty_diff() -> None:
    markdown = "中文\nCafe\u0301\n"
    result = Pipeline([]).transform(markdown)

    report = build_document_report(
        document_id="unchanged",
        source_label="论文.md",
        input_markdown=markdown,
        result=result,
    )

    assert report.cleaned_markdown == markdown.encode()
    assert report.diff == b""
    assert report.change_count == 0
    assert report.result_json.endswith(b"\n")
    assert "论文".encode() in report.result_json
    assert b"\\u4e2d" not in report.result_json


def test_diff_preserves_final_newline_changes_and_uses_report_newlines() -> None:
    diff = build_unified_diff("paper", "same\r\n", "same")

    assert b"--- a/paper.md\n+++ b/paper.md\n" in diff
    assert b"-same\n+same\n\\ No newline at end of file\n" in diff
    assert b"\r" not in diff


def test_failed_report_keeps_change_audit_but_does_not_render_partial_markdown() -> None:
    pipeline = Pipeline(
        [
            ReplaceComponent("a", "b", "test.first"),
            ExplodingComponent(),
        ]
    )
    result = pipeline.transform("a")

    report = build_document_report(
        document_id="failed",
        source_label="failed.md",
        input_markdown="a",
        result=result,
    )
    payload = json.loads(report.result_json)

    assert result.status is RunStatus.FAILED
    assert report.cleaned_markdown is None
    assert report.diff is None
    assert payload["output"] == {"cleaned_path": None, "diff_path": None}
    assert len(payload["changes"]) == 1
    assert payload["errors"][0]["error_type"] == "RuntimeError"
    assert "private" not in payload["errors"][0]["message"]
    assert "partial_markdown" not in payload


def test_unstable_report_serializes_residual_proposals_without_partial_markdown() -> None:
    pipeline = Pipeline(
        [
            ReplaceComponent("bad", "good", "test.to-good"),
            ReplaceComponent("good", "bad", "test.to-bad"),
        ]
    )
    result = pipeline.transform("bad")

    report = build_document_report(
        document_id="unstable",
        source_label="unstable.md",
        input_markdown="bad",
        result=result,
    )
    payload = json.loads(report.result_json)

    assert result.status is RunStatus.UNSTABLE
    assert report.cleaned_markdown is None
    assert report.diff is None
    assert payload["residual_proposals"][0]["proposal"]["reason"] == "replace bad"
    assert payload["residual_proposals"][0]["proposal"]["edits"][0]["expected_text"] == "bad"
    assert "partial_markdown" not in payload


def test_report_rejects_a_tampered_change_hash_chain() -> None:
    result = Pipeline([ReplaceComponent("a", "b", "test.replace")]).transform("a")
    tampered_change = replace(result.changes[0], after_sha256="0" * 64)
    tampered_result = replace(result, changes=(tampered_change,))

    with pytest.raises(ReportingError, match="batch hash"):
        build_document_report(
            document_id="tampered",
            source_label="tampered.md",
            input_markdown="a",
            result=tampered_result,
        )