mdpolish/tests/test_clindb_first_batch_pipeline.py

from __future__ import annotations

import ast
from pathlib import Path

from mdpolish import Pipeline, RunStatus
from mdpolish.components import (
    ArxivSubmissionStampComponent,
    HtmlTableDoubleEscapeComponent,
    HtmlTableLayoutComponent,
    ManuscriptLineNumberComponent,
    PageBreakWordJoinComponent,
    ReferenceSpacingComponent,
    RepeatedRunningHeaderComponent,
    WordReviewCommentComponent,
)

STAMP = "arXiv:2104.12345v2 [stat.ME] 31 Dec 2021"
HEADER = "## Repeated Paper Header"
MAPPINGS = (
    ("medi-", "cal", "medical"),
    ("possi-", "bly", "possibly"),
    ("cre-", "ated", "created"),
    ("SOFA-", "based", "SOFA-based"),
    ("life-", "threatening", "life-threatening"),
    ("threshold.", "olds", "thresholds"),
)


def _build_pipeline() -> Pipeline:
    return Pipeline(
        [
            WordReviewCommentComponent(),
            ManuscriptLineNumberComponent(),
            ArxivSubmissionStampComponent(),
            RepeatedRunningHeaderComponent(),
            PageBreakWordJoinComponent(MAPPINGS),
            HtmlTableDoubleEscapeComponent(),
            HtmlTableLayoutComponent(),
            ReferenceSpacingComponent(),
        ]
    )


def _numbered_manuscript() -> list[str]:
    return [
        f"## {number} Section {number}" if number in {5, 15} else f"{number} body {number}"
        for number in range(1, 21)
    ]


def _combined_markdown() -> str:
    return "\n".join(
        (
            "1 Affiliation",
            "## Abstract",
            *_numbered_manuscript(),
            "Commented [A1]: remove this",
            "",
            STAMP,
            "Sentence continues in",
            "",
            HEADER,
            "",
            "the next line.",
            "A word is possi-",
            "",
            "bly split.",
            "<table><tr><td>&amp;lt;5</td></tr><tr><td>B</td></tr></table>",
            "## References",
            "",
            "1. First",
            "",
            "2. Second",
            "",
            HEADER,
            "",
            "3. Third",
            "4. Fourth",
        )
    )


def test_script_builds_frozen_component_order_and_parameters() -> None:
    script_path = Path(__file__).parents[1] / "scripts" / "run_clindb_first_batch_experiment.py"
    module = ast.parse(script_path.read_text(encoding="utf-8"))
    build_function = next(
        node for node in module.body if isinstance(node, ast.FunctionDef) and node.name == "build_pipeline"
    )
    component_names = [
        call.func.id
        for node in ast.walk(build_function)
        if isinstance(node, ast.List)
        for call in node.elts
        if isinstance(call, ast.Call) and isinstance(call.func, ast.Name)
    ]
    mapping_assignment = next(
        node
        for node in module.body
        if isinstance(node, ast.Assign)
        and any(isinstance(target, ast.Name) and target.id == "CLINDB_WORD_JOIN_MAPPINGS" for target in node.targets)
    )
    assert component_names == [
        "WordReviewCommentComponent",
        "ManuscriptLineNumberComponent",
        "ArxivSubmissionStampComponent",
        "RepeatedRunningHeaderComponent",
        "PageBreakWordJoinComponent",
        "HtmlTableDoubleEscapeComponent",
        "HtmlTableLayoutComponent",
        "ReferenceSpacingComponent",
    ]
    assert ast.literal_eval(mapping_assignment.value) == MAPPINGS

    pipeline = _build_pipeline()
    result = pipeline.transform("")

    assert [component.component_id for component in result.components] == [
        "paper.word_review_comment",
        "paper.manuscript_line_number",
        "paper.arxiv_submission_stamp",
        "paper.repeated_running_header",
        "paper.page_break_word_join",
        "markdown.html_table_double_escape",
        "markdown.html_table_layout",
        "paper.reference_spacing",
    ]
    assert len(MAPPINGS) == 6


def test_full_pipeline_is_audited_stable_and_idempotent() -> None:
    pipeline = _build_pipeline()
    first = pipeline.transform(_combined_markdown())

    assert first.status is RunStatus.SUCCESS
    assert first.output_markdown is not None
    assert first.residual_proposals == ()
    counts: dict[str, int] = {}
    for change in first.changes:
        counts[change.component_id] = counts.get(change.component_id, 0) + 1
    assert counts == {
        "paper.word_review_comment": 1,
        "paper.manuscript_line_number": 20,
        "paper.arxiv_submission_stamp": 1,
        "paper.repeated_running_header": 2,
        "paper.page_break_word_join": 1,
        "markdown.html_table_double_escape": 1,
        "markdown.html_table_layout": 1,
        "paper.reference_spacing": 1,
    }

    second = pipeline.transform(first.output_markdown)
    assert second.status is RunStatus.SUCCESS
    assert second.output_markdown == first.output_markdown
    assert second.changes == ()


def test_business_components_are_not_exported_from_core_namespace() -> None:
    import mdpolish

    assert not hasattr(mdpolish, "WordReviewCommentComponent")
    assert not hasattr(mdpolish, "HtmlTableLayoutComponent")