mdpolish/tests/test_experiment.py

from __future__ import annotations

import json
from collections.abc import Mapping
from datetime import UTC, datetime, timedelta, timezone
from pathlib import Path

import pytest

from mdpolish import Component, DocumentSnapshot, Pipeline, ProposedChange, RunStatus, TextEdit, TextSpan
from mdpolish.experiment import ExperimentError, InputDocument, ToolMetadata, run_experiment


class ConditionalComponent(Component):
    @property
    def component_id(self) -> str:
        return "test.conditional"

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

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

    @property
    def applicability(self) -> str:
        return "替换测试标记 old，遇到 boom 时模拟组件失败，排除其他内容。"

    def _propose_changes(self, snapshot: DocumentSnapshot) -> tuple[ProposedChange, ...]:
        if snapshot.markdown == "boom":
            raise RuntimeError("private source")
        position = snapshot.markdown.find("old")
        if position < 0:
            return ()
        edit = TextEdit(snapshot.sha256, TextSpan(position, position + 3), "old", "new")
        return (ProposedChange(snapshot.sha256, "replace old test marker", (edit,)),)


class NonIdempotentComponent(Component):
    @property
    def component_id(self) -> str:
        return "test.non-idempotent"

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

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

    @property
    def applicability(self) -> str:
        return "在测试文本末尾反复插入标记，只用于验证 unstable 产物边界。"

    def _propose_changes(self, snapshot: DocumentSnapshot) -> tuple[ProposedChange, ...]:
        position = len(snapshot.markdown)
        edit = TextEdit(snapshot.sha256, TextSpan(position, position), "", "!")
        return (ProposedChange(snapshot.sha256, "append synthetic marker", (edit,)),)


def tool_metadata() -> ToolMetadata:
    return ToolMetadata(
        name="mdpolish",
        package_version="0.1.0",
        python_version="3.13.11",
        platform="linux-x86_64",
        git_commit="a" * 40,
        git_dirty=False,
    )


def input_document(path: Path, document_id: str) -> InputDocument:
    return InputDocument(document_id, path, f"inputs/{path.name}")


def test_run_experiment_publishes_manifest_reports_diff_and_preserves_inputs(tmp_path: Path) -> None:
    first_path = tmp_path / "first.md"
    second_path = tmp_path / "second.md"
    first_path.write_bytes(b"before old\r\nafter\r\n")
    second_path.write_text("unchanged\n", encoding="utf-8")
    original_first = first_path.read_bytes()
    original_second = second_path.read_bytes()
    local_timezone = timezone(timedelta(hours=8))
    started_at = datetime(2026, 8, 22, 10, 30, tzinfo=local_timezone)
    completed_at = datetime(2026, 8, 22, 2, 31, tzinfo=UTC)

    result = run_experiment(
        pipeline=Pipeline([ConditionalComponent()]),
        documents=(input_document(first_path, "first"), input_document(second_path, "second")),
        run_id="local-review",
        artifacts_root=tmp_path / "artifacts",
        started_at=started_at,
        completed_at=completed_at,
        tool=tool_metadata(),
    )

    assert result.status is RunStatus.SUCCESS
    assert result.run_directory == tmp_path / "artifacts/2026-08-22/runs/local-review"
    assert result.document_count == 2
    assert result.success_count == 2
    assert result.change_count == 1
    assert first_path.read_bytes() == original_first
    assert second_path.read_bytes() == original_second
    assert (result.run_directory / "documents/first/cleaned.md").read_bytes() == b"before new\r\nafter\r\n"
    assert (result.run_directory / "documents/second/cleaned.md").read_bytes() == original_second
    assert (result.run_directory / "documents/second/changes.diff").read_bytes() == b""

    manifest = json.loads((result.run_directory / "manifest.json").read_bytes())
    assert manifest["run"]["run_date"] == "2026-08-22"
    assert manifest["run"]["utc_offset"] == "+08:00"
    assert manifest["run"]["started_at_utc"] == "2026-08-22T02:30:00Z"
    assert manifest["run"]["completed_at_utc"] == "2026-08-22T02:31:00Z"
    assert manifest["run"]["retention_until"] == "2026-09-21T02:31:00Z"
    assert manifest["pipeline"]["components"][0]["component_id"] == "test.conditional"
    assert manifest["pipeline"]["components"][0]["parameters"] == [
        ["needle", "old"],
        ["replacement", "new"],
    ]
    assert manifest["summary"] == {
        "document_count": 2,
        "success_count": 2,
        "failed_count": 0,
        "unstable_count": 0,
        "change_count": 1,
    }
    first_report = json.loads((result.run_directory / "documents/first/result.json").read_bytes())
    assert first_report["changes"][0]["location"] == {"line": 1, "column": 8}
    assert first_report["changes"][0]["before"] == "old"
    assert b"--- a/first.md\n+++ b/first.md\n" in (
        result.run_directory / "documents/first/changes.diff"
    ).read_bytes()


def test_document_failure_is_isolated_and_never_writes_partial_markdown(tmp_path: Path) -> None:
    good_path = tmp_path / "good.md"
    bad_path = tmp_path / "bad.md"
    good_path.write_text("old", encoding="utf-8")
    bad_path.write_text("boom", encoding="utf-8")

    result = run_experiment(
        pipeline=Pipeline([ConditionalComponent()]),
        documents=(input_document(good_path, "good"), input_document(bad_path, "bad")),
        run_id="mixed",
        artifacts_root=tmp_path / "artifacts",
        started_at=datetime(2026, 8, 22, 10, tzinfo=UTC),
        completed_at=datetime(2026, 8, 22, 10, 1, tzinfo=UTC),
        tool=tool_metadata(),
    )

    assert result.status is RunStatus.FAILED
    assert result.success_count == 1
    assert result.failed_count == 1
    assert (result.run_directory / "documents/good/cleaned.md").read_text() == "new"
    assert (result.run_directory / "documents/bad/result.json").is_file()
    assert not (result.run_directory / "documents/bad/cleaned.md").exists()
    assert not (result.run_directory / "documents/bad/changes.diff").exists()
    bad_report = json.loads((result.run_directory / "documents/bad/result.json").read_bytes())
    assert bad_report["status"] == "failed"
    assert "partial_markdown" not in bad_report


def test_preflight_rejects_invalid_utf8_without_running_or_creating_artifacts(tmp_path: Path) -> None:
    invalid_path = tmp_path / "invalid.md"
    invalid_path.write_bytes(b"\xff")
    artifacts_root = tmp_path / "artifacts"

    with pytest.raises(ExperimentError, match="strict UTF-8"):
        run_experiment(
            pipeline=Pipeline([ConditionalComponent()]),
            documents=(input_document(invalid_path, "invalid"),),
            run_id="invalid-input",
            artifacts_root=artifacts_root,
            started_at=datetime(2026, 8, 22, tzinfo=UTC),
            tool=tool_metadata(),
        )

    assert not artifacts_root.exists()


def test_preflight_rejects_invalid_document_manifests_before_creating_artifacts(tmp_path: Path) -> None:
    source_path = tmp_path / "paper.md"
    source_path.write_text("content", encoding="utf-8")
    directory_path = tmp_path / "directory"
    directory_path.mkdir()
    missing_path = tmp_path / "missing.md"
    cases = (
        (input_document(source_path, "duplicate"), input_document(source_path, "duplicate")),
        (input_document(source_path, "first"), input_document(source_path, "second")),
        (input_document(missing_path, "missing"),),
        (input_document(directory_path, "directory"),),
        (input_document(source_path, "two..dots"),),
    )

    for index, documents in enumerate(cases):
        artifacts_root = tmp_path / f"artifacts-{index}"
        with pytest.raises(ExperimentError):
            run_experiment(
                pipeline=Pipeline([]),
                documents=documents,
                run_id="preflight",
                artifacts_root=artifacts_root,
                started_at=datetime(2026, 8, 22, tzinfo=UTC),
                tool=tool_metadata(),
            )
        assert not artifacts_root.exists()


def test_preflight_rejects_an_output_path_nested_under_an_input_file(tmp_path: Path) -> None:
    source_path = tmp_path / "paper.md"
    source_path.write_text("content", encoding="utf-8")

    with pytest.raises(ExperimentError, match="cannot overlap"):
        run_experiment(
            pipeline=Pipeline([]),
            documents=(input_document(source_path, "paper"),),
            run_id="overlap",
            artifacts_root=source_path,
            started_at=datetime(2026, 8, 22, tzinfo=UTC),
            tool=tool_metadata(),
        )

    assert source_path.read_text(encoding="utf-8") == "content"


def test_utf8_bom_crlf_and_missing_final_newline_are_preserved_exactly(tmp_path: Path) -> None:
    source_path = tmp_path / "bom.md"
    empty_path = tmp_path / "empty.md"
    source_bytes = b"\xef\xbb\xbfhead\r\nlast"
    source_path.write_bytes(source_bytes)
    empty_path.write_bytes(b"")

    result = run_experiment(
        pipeline=Pipeline([]),
        documents=(input_document(source_path, "bom"), input_document(empty_path, "empty")),
        run_id="byte-preservation",
        artifacts_root=tmp_path / "artifacts",
        started_at=datetime(2026, 8, 22, tzinfo=UTC),
        completed_at=datetime(2026, 8, 22, 0, 1, tzinfo=UTC),
        tool=tool_metadata(),
    )

    assert source_path.read_bytes() == source_bytes
    assert (result.run_directory / "documents/bom/cleaned.md").read_bytes() == source_bytes
    assert (result.run_directory / "documents/bom/changes.diff").read_bytes() == b""
    assert empty_path.read_bytes() == b""
    assert (result.run_directory / "documents/empty/cleaned.md").read_bytes() == b""
    assert (result.run_directory / "documents/empty/changes.diff").read_bytes() == b""


def test_unstable_document_only_publishes_a_result_json(tmp_path: Path) -> None:
    source_path = tmp_path / "paper.md"
    source_path.write_text("content", encoding="utf-8")

    result = run_experiment(
        pipeline=Pipeline([NonIdempotentComponent()]),
        documents=(input_document(source_path, "paper"),),
        run_id="unstable",
        artifacts_root=tmp_path / "artifacts",
        started_at=datetime(2026, 8, 22, tzinfo=UTC),
        completed_at=datetime(2026, 8, 22, 0, 1, tzinfo=UTC),
        tool=tool_metadata(),
    )

    assert result.status is RunStatus.UNSTABLE
    assert result.unstable_count == 1
    assert (result.run_directory / "documents/paper/result.json").is_file()
    assert not (result.run_directory / "documents/paper/cleaned.md").exists()
    assert not (result.run_directory / "documents/paper/changes.diff").exists()
    manifest = json.loads((result.run_directory / "manifest.json").read_bytes())
    assert manifest["run"]["status"] == "unstable"


def test_existing_same_date_run_is_rejected_without_overwriting(tmp_path: Path) -> None:
    source_path = tmp_path / "paper.md"
    source_path.write_text("old", encoding="utf-8")
    artifacts_root = tmp_path / "artifacts"
    pipeline = Pipeline([ConditionalComponent()])
    documents = (input_document(source_path, "paper"),)
    started_at = datetime(2026, 8, 22, tzinfo=UTC)
    completed_at = datetime(2026, 8, 22, 0, 1, tzinfo=UTC)
    tool = tool_metadata()
    first = run_experiment(
        pipeline=pipeline,
        documents=documents,
        run_id="same",
        artifacts_root=artifacts_root,
        started_at=started_at,
        completed_at=completed_at,
        tool=tool,
    )
    original_manifest = (first.run_directory / "manifest.json").read_bytes()

    with pytest.raises(ExperimentError, match="already exists"):
        run_experiment(
            pipeline=pipeline,
            documents=documents,
            run_id="same",
            artifacts_root=artifacts_root,
            started_at=started_at,
            completed_at=completed_at,
            tool=tool,
        )

    assert (first.run_directory / "manifest.json").read_bytes() == original_manifest


def test_same_run_id_on_another_local_date_uses_a_separate_directory(tmp_path: Path) -> None:
    source_path = tmp_path / "paper.md"
    source_path.write_text("unchanged", encoding="utf-8")
    artifacts_root = tmp_path / "artifacts"

    first = run_experiment(
        pipeline=Pipeline([]),
        documents=(input_document(source_path, "paper"),),
        run_id="daily",
        artifacts_root=artifacts_root,
        started_at=datetime(2026, 8, 22, 23, tzinfo=UTC),
        completed_at=datetime(2026, 8, 22, 23, 1, tzinfo=UTC),
        tool=tool_metadata(),
    )
    second = run_experiment(
        pipeline=Pipeline([]),
        documents=(input_document(source_path, "paper"),),
        run_id="daily",
        artifacts_root=artifacts_root,
        started_at=datetime(2026, 8, 23, 0, tzinfo=UTC),
        completed_at=datetime(2026, 8, 23, 0, 1, tzinfo=UTC),
        tool=tool_metadata(),
    )

    assert first.run_directory.parent.parent.name == "2026-08-22"
    assert second.run_directory.parent.parent.name == "2026-08-23"