mdpolish/tests/test_review.py

from __future__ import annotations

from collections.abc import Callable
from dataclasses import replace

import pytest

import mdpolish.pipeline as pipeline_module
from mdpolish import (
    AppliedBatch,
    Change,
    DocumentSnapshot,
    ErrorStage,
    Modifier,
    ModifierInfo,
    Pipeline,
    ProposedChange,
    RunError,
    RunStatus,
    TextEdit,
    TextSpan,
    TransformResult,
    apply_modifier_batch,
    markdown_sha256,
)
from mdpolish.review import (
    ReviewBuildError,
    ReviewCurrentKind,
    build_review_document,
    render_markdown_report,
)


def replace_first(
    needle: str,
    replacement: str,
    *,
    modifier_id: str,
    reason: str | None = None,
) -> 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 or f"replace token for {modifier_id}",
                edits=(edit,),
            ),
        )

    return Modifier(
        modifier_id=modifier_id,
        version="1.0.0",
        parameters={"needle": needle, "replacement": replacement},
        applicability="只处理虚构测试标记。",
        propose=propose,
    )


def empty_modifier(modifier_id: str) -> Modifier:
    return Modifier(
        modifier_id=modifier_id,
        version="1.0.0",
        parameters=(),
        applicability="测试零修改阶段。",
        propose=lambda snapshot: (),
    )


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"secret failure: {snapshot.markdown}")
        return ()

    return Modifier(
        modifier_id=modifier_id,
        version="1.0.0",
        parameters={"trigger": trigger},
        applicability="只测试失败路径。",
        propose=propose,
    )


def stale_modifier(modifier_id: str = "test.stale") -> Modifier:
    def propose(snapshot: DocumentSnapshot) -> tuple[ProposedChange, ...]:
        stale = DocumentSnapshot(snapshot.markdown + "!")
        edit = TextEdit(stale.sha256, TextSpan(0, 1), stale.markdown[0], "X")
        return (ProposedChange(stale.sha256, "stale proposal", (edit,)),)

    return Modifier(
        modifier_id=modifier_id,
        version="1.0.0",
        parameters=(),
        applicability="只测试批次验证失败。",
        propose=propose,
    )


def final_review_failure_result(*, error_count: int = 1) -> TransformResult:
    modifiers = [
        exploding_modifier(f"test.review-error-{index}", trigger="done") for index in range(error_count)
    ]
    modifiers.append(replace_first("start", "done", modifier_id="test.producer"))
    return Pipeline(modifiers).transform("start")


def residual_result() -> TransformResult:
    def propose(snapshot: DocumentSnapshot) -> tuple[ProposedChange, ...]:
        proposals: list[ProposedChange] = []
        for position, character in enumerate(snapshot.markdown):
            if character != "x":
                continue
            edit = TextEdit(
                snapshot_sha256=snapshot.sha256,
                span=TextSpan(position, position + 1),
                expected_text="x",
                replacement="DO-NOT-PRINT-RESIDUAL-TEXT",
            )
            proposals.append(ProposedChange(snapshot.sha256, "residual candidate", (edit,)))
        return tuple(proposals)

    residual = Modifier(
        modifier_id="test.residual",
        version="1.0.0",
        parameters=(),
        applicability="只测试残留候选。",
        propose=propose,
    )
    producer = replace_first("a", "xxx", modifier_id="test.producer")
    return Pipeline((residual, producer)).transform("a")


def test_empty_pipeline_builds_complete_unchanged_review() -> None:
    result = Pipeline(()).transform("")

    review = build_review_document("", result)

    assert review.status is RunStatus.SUCCESS
    assert review.current_kind is ReviewCurrentKind.SUCCESS_OUTPUT
    assert review.input_markdown == review.current_markdown == ""
    assert review.stages == ()
    assert review.stages_complete is True


def test_replay_preserves_zero_stages_and_exact_intermediate_snapshots() -> None:
    pipeline = Pipeline(
        (
            empty_modifier("test.zero-first"),
            replace_first("a", "b", modifier_id="test.first"),
            replace_first("b", "c", modifier_id="test.second"),
            empty_modifier("test.zero-last"),
        )
    )
    result = pipeline.transform("a")

    review = build_review_document("a", result)

    assert [stage.modifier_position for stage in review.stages] == [0, 1, 2, 3]
    assert [(stage.before_markdown, stage.after_markdown) for stage in review.stages] == [
        ("a", "a"),
        ("a", "b"),
        ("b", "c"),
        ("c", "c"),
    ]
    assert [len(stage.changes) for stage in review.stages] == [0, 1, 1, 0]
    assert [stage.before_sha256 for stage in review.stages[1:]] == [
        result.changes[0].before_sha256,
        result.changes[1].before_sha256,
        result.changes[1].after_sha256,
    ]
    assert review.current_markdown == "c"
    assert review.current_sha256 == markdown_sha256("c")


@pytest.mark.parametrize(
    ("markdown", "start", "end", "replacement", "expected"),
    [
        ("ab", 1, 1, "X", "aXb"),
        ("abc", 1, 2, "", "ac"),
        ("abc", 1, 2, "X", "aXc"),
    ],
)
def test_replay_covers_insert_delete_and_replace(
    markdown: str,
    start: int,
    end: int,
    replacement: str,
    expected: str,
) -> None:
    def propose(snapshot: DocumentSnapshot) -> tuple[ProposedChange, ...]:
        if snapshot.markdown != markdown:
            return ()
        edit = TextEdit(
            snapshot.sha256,
            TextSpan(start, end),
            snapshot.markdown[start:end],
            replacement,
        )
        return (ProposedChange(snapshot.sha256, "one exact edit", (edit,)),)

    modifier = Modifier(
        modifier_id="test.edit-kind",
        version="1.0.0",
        parameters=(),
        applicability="只测试插入、删除和替换。",
        propose=propose,
    )

    review = build_review_document(markdown, Pipeline((modifier,)).transform(markdown))

    assert review.stages[0].after_markdown == expected
    assert review.current_markdown == expected


def test_one_proposal_with_multiple_edits_keeps_references_and_core_order() -> None:
    def propose(snapshot: DocumentSnapshot) -> tuple[ProposedChange, ...]:
        if snapshot.markdown != "abcdef":
            return ()
        edits = (
            TextEdit(snapshot.sha256, TextSpan(4, 6), "ef", "E"),
            TextEdit(snapshot.sha256, TextSpan(0, 1), "a", "A"),
            TextEdit(snapshot.sha256, TextSpan(2, 3), "c", ""),
        )
        return (ProposedChange(snapshot.sha256, "three exact edits", edits),)

    modifier = Modifier(
        modifier_id="test.multi-edit",
        version="1.0.0",
        parameters=(),
        applicability="只测试一个候选中的多个编辑。",
        propose=propose,
    )
    result = Pipeline((modifier,)).transform("abcdef")

    review = build_review_document("abcdef", result)
    stage = review.stages[0]

    assert stage.after_markdown == "AbdE"
    assert [change.change.span.start for change in stage.changes] == [0, 2, 4]
    assert [change.change.edit_index for change in stage.changes] == [1, 2, 0]
    assert {change.change.proposal_ref for change in stage.changes} == {
        stage.changes[0].change.proposal_ref,
    }


def test_locations_use_python_code_points_and_physical_newlines() -> None:
    markdown = "\ufeffA\r\n中e\u0301🙂X\rY\nZ"
    positions = (1, 6, 8, 10, 12)

    def propose(snapshot: DocumentSnapshot) -> tuple[ProposedChange, ...]:
        if snapshot.markdown != markdown:
            return ()
        edits = tuple(
            TextEdit(
                snapshot.sha256,
                TextSpan(position, position + 1),
                snapshot.markdown[position],
                f"<{position}>",
            )
            for position in positions
        )
        return (ProposedChange(snapshot.sha256, "unicode locations", edits),)

    modifier = Modifier(
        modifier_id="test.locations",
        version="1.0.0",
        parameters=(),
        applicability="只测试 Python 码点位置。",
        propose=propose,
    )
    review = build_review_document(markdown, Pipeline((modifier,)).transform(markdown))

    assert [(change.location.line, change.location.column) for change in review.stages[0].changes] == [
        (1, 2),
        (2, 3),
        (2, 5),
        (3, 1),
        (4, 1),
    ]


def test_preflight_failure_has_no_stage_and_unchanged_partial_text() -> None:
    duplicate = replace_first("a", "b", modifier_id="test.duplicate")
    result = Pipeline((duplicate, duplicate)).transform("a")

    review = build_review_document("a", result)

    assert result.errors[0].stage is ErrorStage.PREFLIGHT
    assert review.current_markdown == "a"
    assert review.stages == ()
    assert review.stages_complete is False


def _assert_only_first_stage_completed(result: TransformResult) -> None:
    review = build_review_document("a", result)
    assert result.status is RunStatus.FAILED
    assert result.errors[0].stage is ErrorStage.TRANSFORM
    assert review.current_markdown == "b"
    assert [stage.modifier_position for stage in review.stages] == [0]
    assert review.stages[0].after_markdown == "b"
    assert review.stages_complete is False


def test_propose_failure_stops_before_failing_stage() -> None:
    result = Pipeline(
        (
            replace_first("a", "b", modifier_id="test.first"),
            exploding_modifier("test.exploding"),
            replace_first("b", "c", modifier_id="test.never"),
        )
    ).transform("a")

    _assert_only_first_stage_completed(result)


def test_metadata_change_after_proposal_stops_before_failing_stage() -> None:
    mutating: Modifier

    def propose(snapshot: DocumentSnapshot) -> tuple[ProposedChange, ...]:
        object.__setattr__(mutating, "version", "2.0.0")
        edit = TextEdit(snapshot.sha256, TextSpan(0, 1), "b", "B")
        return (ProposedChange(snapshot.sha256, "mutate metadata", (edit,)),)

    mutating = Modifier(
        modifier_id="test.mutating",
        version="1.0.0",
        parameters=(),
        applicability="只测试运行期元数据变化。",
        propose=propose,
    )
    result = Pipeline(
        (replace_first("a", "b", modifier_id="test.first"), mutating, empty_modifier("test.never"))
    ).transform("a")

    _assert_only_first_stage_completed(result)


def test_batch_validation_failure_does_not_create_a_zero_change_stage() -> None:
    result = Pipeline(
        (replace_first("a", "b", modifier_id="test.first"), stale_modifier(), empty_modifier("test.never"))
    ).transform("a")

    _assert_only_first_stage_completed(result)
    assert result.errors[0].error_type == "EditValidationError"


def test_batch_application_failure_does_not_create_a_stage(monkeypatch: pytest.MonkeyPatch) -> None:
    calls = 0

    def fail_second_application(
        snapshot: DocumentSnapshot,
        proposals: tuple[ProposedChange, ...],
        modifier: ModifierInfo,
        modifier_position: int,
    ) -> AppliedBatch:
        nonlocal calls
        calls += 1
        if calls == 2:
            raise RuntimeError("application failure")
        return apply_modifier_batch(snapshot, proposals, modifier, modifier_position)

    monkeypatch.setattr(pipeline_module, "apply_modifier_batch", fail_second_application)
    result = Pipeline(
        (
            replace_first("a", "b", modifier_id="test.first"),
            replace_first("b", "c", modifier_id="test.failing"),
            empty_modifier("test.never"),
        )
    ).transform("a")

    _assert_only_first_stage_completed(result)
    assert result.errors[0].error_type == "RuntimeError"


def test_final_review_failures_keep_all_transform_stages_complete() -> None:
    result = final_review_failure_result(error_count=2)

    review = build_review_document("start", result)

    assert result.status is RunStatus.FAILED
    assert [error.stage for error in result.errors] == [ErrorStage.FINAL_REVIEW, ErrorStage.FINAL_REVIEW]
    assert len(review.stages) == 3
    assert review.stages[-1].after_markdown == "done"
    assert review.stages_complete is True


def test_unstable_result_keeps_complete_stages_and_unapplied_residuals() -> None:
    result = residual_result()

    review = build_review_document("a", result)

    assert result.status is RunStatus.UNSTABLE
    assert review.current_kind is ReviewCurrentKind.PARTIAL_OUTPUT
    assert review.current_markdown == "xxx"
    assert review.stages_complete is True
    assert len(review.residual_proposals) == 3
    assert "DO-NOT-PRINT" not in review.current_markdown


def test_builder_does_not_run_modifiers_again() -> None:
    calls = 0

    def propose(snapshot: DocumentSnapshot) -> tuple[ProposedChange, ...]:
        nonlocal calls
        calls += 1
        return ()

    modifier = Modifier(
        modifier_id="test.call-count",
        version="1.0.0",
        parameters=(),
        applicability="只测试调用次数。",
        propose=propose,
    )
    result = Pipeline((modifier,)).transform("a")
    calls_after_pipeline = calls

    review = build_review_document("a", result)
    render_markdown_report(review)

    assert calls_after_pipeline == 2
    assert calls == calls_after_pipeline


def test_builder_rejects_input_and_replay_mismatches_without_source_text() -> None:
    result = Pipeline((replace_first("secret", "public", modifier_id="test.replace"),)).transform("secret")

    with pytest.raises(ReviewBuildError) as input_error:
        build_review_document("different secret", result)
    assert "secret" not in str(input_error.value)

    mismatched = replace(
        result,
        current_sha256=markdown_sha256("unrelated"),
        output_markdown="unrelated",
    )
    with pytest.raises(ReviewBuildError) as replay_error:
        build_review_document("secret", mismatched)
    assert "secret" not in str(replay_error.value)
    assert "public" not in str(replay_error.value)


def test_builder_rejects_mixed_error_stages() -> None:
    result = final_review_failure_result()
    transform_error = RunError(
        modifier_id=result.modifiers[0].modifier_id,
        modifier_version=result.modifiers[0].version,
        modifier_position=0,
        stage=ErrorStage.TRANSFORM,
        error_type="ExampleError",
        message="safe",
    )
    malformed = replace(result, errors=(*result.errors, transform_error))

    with pytest.raises(ReviewBuildError, match="mixes error stages"):
        build_review_document("start", malformed)


def test_builder_rejects_preflight_result_with_transform_output() -> None:
    duplicate = replace_first("a", "b", modifier_id="test.duplicate")
    preflight = Pipeline((duplicate, duplicate)).transform("a")
    successful = Pipeline((replace_first("a", "b", modifier_id="test.success"),)).transform("a")
    malformed = replace(
        preflight,
        current_sha256=markdown_sha256("b"),
        changes=successful.changes,
        partial_markdown="b",
    )

    with pytest.raises(ReviewBuildError, match="preflight failure contains transform output"):
        build_review_document("a", malformed)


def test_builder_rejects_change_at_transform_failure_position() -> None:
    successful = Pipeline((replace_first("a", "b", modifier_id="test.replace"),)).transform("a")
    error = RunError(
        modifier_id=successful.modifiers[0].modifier_id,
        modifier_version=successful.modifiers[0].version,
        modifier_position=0,
        stage=ErrorStage.TRANSFORM,
        error_type="ExampleError",
        message="safe",
    )
    malformed = TransformResult(
        status=RunStatus.FAILED,
        input_sha256=successful.input_sha256,
        current_sha256=successful.current_sha256,
        modifiers=successful.modifiers,
        changes=successful.changes,
        errors=(error,),
        partial_markdown=successful.output_markdown,
    )

    with pytest.raises(ReviewBuildError, match="incomplete modifier stage"):
        build_review_document("a", malformed)


@pytest.mark.parametrize(
    "tamper",
    [
        lambda change: replace(change, modifier_id="test.wrong"),
        lambda change: replace(change, before="z"),
        lambda change: replace(change, span=TextSpan(1, 2), before="z"),
        lambda change: replace(change, after_sha256=change.before_sha256),
        lambda change: replace(
            change,
            proposal_ref=replace(change.proposal_ref, snapshot_sha256=change.after_sha256),
        ),
        lambda change: replace(
            change,
            proposal_ref=replace(change.proposal_ref, proposal_index=1),
        ),
    ],
)
def test_builder_rejects_tampered_change_contract(tamper: Callable[[Change], Change]) -> None:
    result = Pipeline((replace_first("a", "b", modifier_id="test.replace"),)).transform("a")
    change = result.changes[0]
    tampered_change = tamper(change)
    malformed = replace(result, changes=(tampered_change,))

    with pytest.raises(ReviewBuildError):
        build_review_document("a", malformed)


def test_builder_rejects_duplicate_and_conflicting_changes() -> None:
    one_change = Pipeline((replace_first("a", "b", modifier_id="test.replace"),)).transform("a")
    duplicate = replace(one_change, changes=(one_change.changes[0], one_change.changes[0]))
    with pytest.raises(ReviewBuildError, match="repeats a proposal edit index"):
        build_review_document("a", duplicate)

    def propose(snapshot: DocumentSnapshot) -> tuple[ProposedChange, ...]:
        if snapshot.markdown != "abc":
            return ()
        edits = (
            TextEdit(snapshot.sha256, TextSpan(0, 1), "a", "A"),
            TextEdit(snapshot.sha256, TextSpan(2, 3), "c", "C"),
        )
        return (ProposedChange(snapshot.sha256, "two edits", edits),)

    modifier = Modifier(
        modifier_id="test.conflict",
        version="1.0.0",
        parameters=(),
        applicability="只测试篡改后的冲突范围。",
        propose=propose,
    )
    result = Pipeline((modifier,)).transform("abc")
    conflicting_second = replace(result.changes[1], span=TextSpan(0, 2), before="ab", after="X")
    conflicting = replace(result, changes=(result.changes[0], conflicting_second))

    with pytest.raises(ReviewBuildError, match="edit contract"):
        build_review_document("abc", conflicting)


def test_builder_rejects_noncanonical_change_order() -> None:
    def propose(snapshot: DocumentSnapshot) -> tuple[ProposedChange, ...]:
        if snapshot.markdown != "abc":
            return ()
        edits = (
            TextEdit(snapshot.sha256, TextSpan(0, 1), "a", "A"),
            TextEdit(snapshot.sha256, TextSpan(2, 3), "c", "C"),
        )
        return (ProposedChange(snapshot.sha256, "two edits", edits),)

    modifier = Modifier(
        modifier_id="test.order",
        version="1.0.0",
        parameters=(),
        applicability="只测试报告顺序。",
        propose=propose,
    )
    result = Pipeline((modifier,)).transform("abc")
    malformed = replace(result, changes=tuple(reversed(result.changes)))

    with pytest.raises(ReviewBuildError, match="canonical report order"):
        build_review_document("abc", malformed)


def test_builder_rejects_tampered_residual_snapshot_and_error_position() -> None:
    unstable = residual_result()
    first_residual = unstable.residual_proposals[0]
    stale_ref = replace(first_residual.proposal_ref, snapshot_sha256=unstable.input_sha256)
    malformed_residual = replace(
        unstable,
        residual_proposals=(replace(first_residual, proposal_ref=stale_ref), *unstable.residual_proposals[1:]),
    )
    with pytest.raises(ReviewBuildError, match="current snapshot"):
        build_review_document("a", malformed_residual)

    failed = final_review_failure_result()
    out_of_range_error = replace(
        failed.errors[0],
        modifier_id="<invalid>",
        modifier_version="<invalid>",
        modifier_position=len(failed.modifiers),
    )
    malformed_error = replace(failed, errors=(out_of_range_error,))
    with pytest.raises(ReviewBuildError, match="outside modifier metadata"):
        build_review_document("start", malformed_error)


def test_report_contains_complete_sources_stages_and_success_diff_label() -> None:
    markdown = "before\n```md\n<div>x</div>\n[link](https://example.test)\n![img](image.png)"
    result = Pipeline(
        (replace_first("before", "after", modifier_id="test.report", reason="render exact source"),)
    ).transform(markdown)
    review = build_review_document(markdown, result)

    report = render_markdown_report(review)

    assert "status: `success`" in report
    assert "current_kind: `success_output`" in report
    assert "--- a/input.md" in report
    assert "+++ b/output.md" in report
    assert "````markdown" in report
    assert markdown in report
    assert review.current_markdown in report
    assert "render exact source" in report
    assert "location: line `1`, column `1`" in report
    assert "<div>x</div>" in report


@pytest.mark.parametrize(
    ("result", "input_markdown", "expected_label", "expected_status"),
    [
        (
            Pipeline(
                (
                    replace_first("a", "b", modifier_id="test.first"),
                    exploding_modifier("test.error"),
                )
            ).transform("a"),
            "a",
            "+++ b/failed.partial.md",
            "status: `failed`",
        ),
        (residual_result(), "a", "+++ b/unstable.partial.md", "status: `unstable`"),
    ],
)
def test_partial_report_labels_cannot_be_mistaken_for_success(
    result: TransformResult,
    input_markdown: str,
    expected_label: str,
    expected_status: str,
) -> None:
    report = render_markdown_report(build_review_document(input_markdown, result))

    assert expected_label in report
    assert expected_status in report
    assert "current_kind: `partial_output`" in report


def test_unchanged_failed_diff_still_has_partial_label() -> None:
    duplicate = replace_first("a", "b", modifier_id="test.duplicate")
    review = build_review_document("a", Pipeline((duplicate, duplicate)).transform("a"))

    report = render_markdown_report(review)

    assert "+++ b/failed.partial.md" in report
    assert "(no textual difference)" in report


def test_diff_marks_physical_line_endings_and_missing_final_newline() -> None:
    result = Pipeline((replace_first("a", "A", modifier_id="test.replace"),)).transform("a\r\nb")
    report = render_markdown_report(build_review_document("a\r\nb", result))

    assert "-a ⟦CRLF⟧" in report
    assert "+A ⟦CRLF⟧" in report
    assert " b ⟦NO EOL⟧" in report
    assert "Final line ending: `none`" in report


def test_report_limits_residuals_without_printing_replacement_text() -> None:
    review = build_review_document("a", residual_result())

    report = render_markdown_report(review, residual_limit=1)
    count_only = render_markdown_report(review, residual_limit=0)

    assert "- total: `3`" in report
    assert report.count("### Residual ") == 1
    assert "Omitted residual proposals: `2`." in report
    assert "expected length `1`" in report
    assert "replacement length `26`" in report
    assert "DO-NOT-PRINT-RESIDUAL-TEXT" not in report
    assert "### Residual 1" not in count_only
    assert "Omitted residual proposals: `3`." in count_only


@pytest.mark.parametrize("value", [True, 1.5, "1", None])
def test_report_rejects_non_integer_residual_limit(value: object) -> None:
    review = build_review_document("", Pipeline(()).transform(""))

    with pytest.raises(TypeError, match="integer"):
        render_markdown_report(review, residual_limit=value)  # type: ignore[arg-type]


def test_report_rejects_negative_residual_limit() -> None:
    review = build_review_document("", Pipeline(()).transform(""))

    with pytest.raises(ValueError, match="non-negative"):
        render_markdown_report(review, residual_limit=-1)


def test_same_result_builds_equal_review_and_report() -> None:
    result = Pipeline((replace_first("a", "b", modifier_id="test.replace"),)).transform("a")

    first = build_review_document("a", result)
    second = build_review_document("a", result)

    assert first == second
    assert render_markdown_report(first) == render_markdown_report(second)