重构为函数式通用 Markdown 修改库
This commit is contained in:
@@ -1,212 +0,0 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from hashlib import sha256
|
||||
from pathlib import Path
|
||||
from typing import Any, TypedDict, cast
|
||||
|
||||
|
||||
class ReviewFixture(TypedDict):
|
||||
run_directory: Path
|
||||
source_path: Path
|
||||
manifest_path: Path
|
||||
locator_path: Path
|
||||
result_path: Path
|
||||
original: str
|
||||
cleaned: str
|
||||
|
||||
|
||||
def digest(text: str) -> str:
|
||||
return sha256(text.encode()).hexdigest()
|
||||
|
||||
|
||||
def write_json(path: Path, payload: object) -> None:
|
||||
path.write_text(json.dumps(payload, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
||||
|
||||
|
||||
def read_json(path: Path) -> dict[str, Any]:
|
||||
return cast(dict[str, Any], json.loads(path.read_text(encoding="utf-8")))
|
||||
|
||||
|
||||
def create_review_run(
|
||||
root: Path,
|
||||
*,
|
||||
status: str = "success",
|
||||
with_locator: bool = True,
|
||||
) -> ReviewFixture:
|
||||
run_directory = root / "artifacts/2026-08-23/runs/review-run"
|
||||
document_directory = run_directory / "documents/paper"
|
||||
document_directory.mkdir(parents=True)
|
||||
source_path = root / "paper.md"
|
||||
original = "\ufeff😀 old\r\nCafe\u0301\n"
|
||||
cleaned = "\ufeff😀 new\r\nCafe\u0301\n"
|
||||
source_path.write_text(original, encoding="utf-8")
|
||||
input_hash = digest(original)
|
||||
current_hash = digest(cleaned) if status == "success" else input_hash
|
||||
changes: list[dict[str, object]] = []
|
||||
if status == "success":
|
||||
changes.append(
|
||||
{
|
||||
"component_id": "paper.rule",
|
||||
"component_version": "1.0.0",
|
||||
"component_position": 0,
|
||||
"proposal_ref": {
|
||||
"component_position": 0,
|
||||
"snapshot_sha256": input_hash,
|
||||
"proposal_index": 0,
|
||||
},
|
||||
"edit_index": 0,
|
||||
"reason": "替换测试单词",
|
||||
"span": {"start": 3, "end": 6},
|
||||
"location": {"line": 1, "column": 4},
|
||||
"before": "old",
|
||||
"after": "new",
|
||||
"before_sha256": input_hash,
|
||||
"after_sha256": current_hash,
|
||||
}
|
||||
)
|
||||
errors: list[dict[str, object]] = []
|
||||
if status == "failed":
|
||||
errors.append(
|
||||
{
|
||||
"component_id": "paper.rule",
|
||||
"component_version": "1.0.0",
|
||||
"component_position": 0,
|
||||
"stage": "transform",
|
||||
"error_type": "SyntheticError",
|
||||
"message": "测试组件失败。",
|
||||
}
|
||||
)
|
||||
residuals: list[dict[str, object]] = []
|
||||
if status == "unstable":
|
||||
residuals.append(
|
||||
{
|
||||
"component_id": "paper.rule",
|
||||
"component_version": "1.0.0",
|
||||
"component_position": 0,
|
||||
"proposal_ref": {
|
||||
"component_position": 0,
|
||||
"snapshot_sha256": input_hash,
|
||||
"proposal_index": 0,
|
||||
},
|
||||
"proposal": {
|
||||
"snapshot_sha256": input_hash,
|
||||
"reason": "仍可替换测试单词",
|
||||
"edits": [
|
||||
{
|
||||
"snapshot_sha256": input_hash,
|
||||
"span": {"start": 3, "end": 6},
|
||||
"expected_text": "old",
|
||||
"replacement": "new",
|
||||
}
|
||||
],
|
||||
},
|
||||
}
|
||||
)
|
||||
result = {
|
||||
"schema_version": 1,
|
||||
"document": {"document_id": "paper", "source_label": "inputs/paper.md"},
|
||||
"status": status,
|
||||
"input_sha256": input_hash,
|
||||
"current_sha256": current_hash,
|
||||
"changes": changes,
|
||||
"errors": errors,
|
||||
"residual_proposals": residuals,
|
||||
"output": {
|
||||
"cleaned_path": "cleaned.md" if status == "success" else None,
|
||||
"diff_path": "changes.diff" if status == "success" else None,
|
||||
},
|
||||
}
|
||||
result_path = document_directory / "result.json"
|
||||
write_json(result_path, result)
|
||||
if status == "success":
|
||||
(document_directory / "cleaned.md").write_text(cleaned, encoding="utf-8")
|
||||
(document_directory / "changes.diff").write_text("synthetic diff\n", encoding="utf-8")
|
||||
|
||||
manifest = {
|
||||
"schema_version": 1,
|
||||
"run": {
|
||||
"run_id": "review-run",
|
||||
"run_date": "2026-08-23",
|
||||
"utc_offset": "+08:00",
|
||||
"status": status,
|
||||
"started_at_utc": "2026-08-23T01:00:00Z",
|
||||
"completed_at_utc": "2026-08-23T01:01:00Z",
|
||||
"retention_until": "2026-09-22T01:01:00Z",
|
||||
},
|
||||
"tool": {
|
||||
"name": "mdpolish",
|
||||
"package_version": "0.1.0",
|
||||
"python_version": "3.13.11",
|
||||
"platform": "linux-x86_64",
|
||||
"git_commit": None,
|
||||
"git_dirty": None,
|
||||
},
|
||||
"pipeline": {
|
||||
"components": [
|
||||
{
|
||||
"component_id": "paper.rule",
|
||||
"version": "1.0.0",
|
||||
"parameters": [],
|
||||
"applicability": "替换测试单词。",
|
||||
},
|
||||
{
|
||||
"component_id": "paper.zero",
|
||||
"version": "1.0.0",
|
||||
"parameters": [],
|
||||
"applicability": "不修改当前测试文档。",
|
||||
},
|
||||
]
|
||||
},
|
||||
"documents": [
|
||||
{
|
||||
"document_id": "paper",
|
||||
"source_label": "inputs/paper.md",
|
||||
"status": status,
|
||||
"input_sha256": input_hash,
|
||||
"current_sha256": current_hash,
|
||||
"change_count": len(changes),
|
||||
"result_path": "documents/paper/result.json",
|
||||
"cleaned_path": "documents/paper/cleaned.md" if status == "success" else None,
|
||||
"diff_path": "documents/paper/changes.diff" if status == "success" else None,
|
||||
}
|
||||
],
|
||||
"summary": {
|
||||
"document_count": 1,
|
||||
"success_count": int(status == "success"),
|
||||
"failed_count": int(status == "failed"),
|
||||
"unstable_count": int(status == "unstable"),
|
||||
"change_count": len(changes),
|
||||
},
|
||||
}
|
||||
manifest_path = run_directory / "manifest.json"
|
||||
write_json(manifest_path, manifest)
|
||||
locator_path = run_directory / "review-locator.json"
|
||||
if with_locator:
|
||||
write_json(
|
||||
locator_path,
|
||||
{
|
||||
"schema_version": 1,
|
||||
"run": {
|
||||
"run_id": "review-run",
|
||||
"run_directory": str(run_directory.resolve()),
|
||||
"manifest_path": "manifest.json",
|
||||
},
|
||||
"documents": [
|
||||
{
|
||||
"document_id": "paper",
|
||||
"source_path": str(source_path.resolve()),
|
||||
"input_sha256": input_hash,
|
||||
}
|
||||
],
|
||||
},
|
||||
)
|
||||
return {
|
||||
"run_directory": run_directory,
|
||||
"source_path": source_path,
|
||||
"manifest_path": manifest_path,
|
||||
"locator_path": locator_path,
|
||||
"result_path": result_path,
|
||||
"original": original,
|
||||
"cleaned": cleaned,
|
||||
}
|
||||
@@ -1,189 +0,0 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from collections.abc import Callable
|
||||
from dataclasses import replace
|
||||
|
||||
import pytest
|
||||
|
||||
from mdpolish._artifact_replay import (
|
||||
ReplayChange,
|
||||
ReplayComponent,
|
||||
ReplayError,
|
||||
replay_change_chain,
|
||||
)
|
||||
from mdpolish.models import markdown_sha256
|
||||
|
||||
|
||||
def change(
|
||||
*,
|
||||
component_position: int,
|
||||
before_text: str,
|
||||
after_text: str,
|
||||
start: int,
|
||||
end: int,
|
||||
before_sha256: str,
|
||||
after_sha256: str,
|
||||
proposal_index: int = 0,
|
||||
edit_index: int = 0,
|
||||
) -> ReplayChange:
|
||||
return ReplayChange(
|
||||
component_id=f"test.{component_position}",
|
||||
component_version="1.0.0",
|
||||
component_position=component_position,
|
||||
proposal_component_position=component_position,
|
||||
proposal_snapshot_sha256=before_sha256,
|
||||
proposal_index=proposal_index,
|
||||
edit_index=edit_index,
|
||||
start=start,
|
||||
end=end,
|
||||
before=before_text,
|
||||
after=after_text,
|
||||
before_sha256=before_sha256,
|
||||
after_sha256=after_sha256,
|
||||
)
|
||||
|
||||
|
||||
def test_replay_builds_zero_change_stages_and_utf16_editor_ranges() -> None:
|
||||
original = "\ufeff😀 old\r\nCafe\u0301"
|
||||
final = "\ufeff😀 new\r\nCafe\u0301"
|
||||
input_hash = markdown_sha256(original)
|
||||
final_hash = markdown_sha256(final)
|
||||
recorded = change(
|
||||
component_position=0,
|
||||
before_text="old",
|
||||
after_text="new",
|
||||
start=3,
|
||||
end=6,
|
||||
before_sha256=input_hash,
|
||||
after_sha256=final_hash,
|
||||
)
|
||||
|
||||
replayed = replay_change_chain(
|
||||
input_markdown=original,
|
||||
input_sha256=input_hash,
|
||||
components=(ReplayComponent("test.0", "1.0.0"), ReplayComponent("test.1", "1.0.0")),
|
||||
changes=(recorded,),
|
||||
current_sha256=final_hash,
|
||||
current_markdown=final,
|
||||
include_zero_change_stages=True,
|
||||
)
|
||||
|
||||
assert replayed.current_markdown == final
|
||||
assert len(replayed.stages) == 2
|
||||
assert replayed.stages[1].before_markdown == final
|
||||
assert replayed.stages[1].after_markdown == final
|
||||
assert replayed.stages[1].changes == ()
|
||||
assert (replayed.changes[0].line, replayed.changes[0].column) == (1, 4)
|
||||
assert (replayed.changes[0].editor_start, replayed.changes[0].editor_end) == (4, 7)
|
||||
|
||||
|
||||
def test_replay_uses_full_descending_application_key_not_record_order() -> None:
|
||||
original = "abcd"
|
||||
final = "aXXcYY"
|
||||
input_hash = markdown_sha256(original)
|
||||
final_hash = markdown_sha256(final)
|
||||
right = change(
|
||||
component_position=0,
|
||||
before_text="d",
|
||||
after_text="YY",
|
||||
start=3,
|
||||
end=4,
|
||||
before_sha256=input_hash,
|
||||
after_sha256=final_hash,
|
||||
proposal_index=1,
|
||||
)
|
||||
left = change(
|
||||
component_position=0,
|
||||
before_text="b",
|
||||
after_text="XX",
|
||||
start=1,
|
||||
end=2,
|
||||
before_sha256=input_hash,
|
||||
after_sha256=final_hash,
|
||||
)
|
||||
|
||||
replayed = replay_change_chain(
|
||||
input_markdown=original,
|
||||
input_sha256=input_hash,
|
||||
components=(ReplayComponent("test.0", "1.0.0"),),
|
||||
changes=(right, left),
|
||||
current_sha256=final_hash,
|
||||
current_markdown=final,
|
||||
include_zero_change_stages=True,
|
||||
)
|
||||
|
||||
assert replayed.current_markdown == final
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("mutate", "message"),
|
||||
[
|
||||
(lambda item: replace(item, component_position=2), "component position"),
|
||||
(lambda item: replace(item, proposal_component_position=1), "proposal reference"),
|
||||
(lambda item: replace(item, before="bad"), "recorded snapshot"),
|
||||
(lambda item: replace(item, after_sha256="0" * 64), "batch hash"),
|
||||
],
|
||||
)
|
||||
def test_replay_rejects_untrusted_change_chains(
|
||||
mutate: Callable[[ReplayChange], ReplayChange], message: str
|
||||
) -> None:
|
||||
original = "old"
|
||||
final = "new"
|
||||
input_hash = markdown_sha256(original)
|
||||
final_hash = markdown_sha256(final)
|
||||
valid = change(
|
||||
component_position=0,
|
||||
before_text="old",
|
||||
after_text="new",
|
||||
start=0,
|
||||
end=3,
|
||||
before_sha256=input_hash,
|
||||
after_sha256=final_hash,
|
||||
)
|
||||
tampered = mutate(valid)
|
||||
|
||||
with pytest.raises(ReplayError, match=message):
|
||||
replay_change_chain(
|
||||
input_markdown=original,
|
||||
input_sha256=input_hash,
|
||||
components=(ReplayComponent("test.0", "1.0.0"),),
|
||||
changes=(tampered,),
|
||||
current_sha256=final_hash,
|
||||
current_markdown=final,
|
||||
include_zero_change_stages=True,
|
||||
)
|
||||
|
||||
|
||||
def test_replay_rejects_conflicting_ranges() -> None:
|
||||
original = "abc"
|
||||
input_hash = markdown_sha256(original)
|
||||
first = change(
|
||||
component_position=0,
|
||||
before_text="ab",
|
||||
after_text="x",
|
||||
start=0,
|
||||
end=2,
|
||||
before_sha256=input_hash,
|
||||
after_sha256="0" * 64,
|
||||
)
|
||||
second = change(
|
||||
component_position=0,
|
||||
before_text="bc",
|
||||
after_text="y",
|
||||
start=1,
|
||||
end=3,
|
||||
before_sha256=input_hash,
|
||||
after_sha256="0" * 64,
|
||||
proposal_index=1,
|
||||
)
|
||||
|
||||
with pytest.raises(ReplayError, match="conflicting"):
|
||||
replay_change_chain(
|
||||
input_markdown=original,
|
||||
input_sha256=input_hash,
|
||||
components=(ReplayComponent("test.0", "1.0.0"),),
|
||||
changes=(first, second),
|
||||
current_sha256=input_hash,
|
||||
current_markdown=None,
|
||||
include_zero_change_stages=False,
|
||||
)
|
||||
@@ -1,398 +0,0 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import stat
|
||||
from hashlib import sha256
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
import mdpolish.artifact_store as artifact_store
|
||||
from mdpolish.artifact_store import ArtifactStoreError, StoredDocument, publish_run
|
||||
|
||||
_SOURCE_BYTES = b"source\n"
|
||||
|
||||
|
||||
def stored_document(document_id: str = "paper", output: bytes | None = b"cleaned\n") -> StoredDocument:
|
||||
status = "success" if output is not None else "failed"
|
||||
current_sha256 = sha256(output).hexdigest() if output is not None else "2" * 64
|
||||
payload = {
|
||||
"schema_version": 1,
|
||||
"document": {"document_id": document_id, "source_label": f"inputs/{document_id}.md"},
|
||||
"status": status,
|
||||
"input_sha256": sha256(_SOURCE_BYTES).hexdigest(),
|
||||
"current_sha256": current_sha256,
|
||||
"changes": [],
|
||||
"errors": [] if status == "success" else [{"error_type": "SyntheticError"}],
|
||||
"residual_proposals": [],
|
||||
"output": {
|
||||
"cleaned_path": "cleaned.md" if output is not None else None,
|
||||
"diff_path": "changes.diff" if output is not None else None,
|
||||
},
|
||||
}
|
||||
return StoredDocument(
|
||||
document_id=document_id,
|
||||
result_json=(json.dumps(payload, indent=2) + "\n").encode(),
|
||||
cleaned_markdown=output,
|
||||
diff=b"" if output is not None else None,
|
||||
output_sha256=sha256(output).hexdigest() if output is not None else None,
|
||||
)
|
||||
|
||||
|
||||
def manifest_json(
|
||||
run_date: str,
|
||||
run_id: str,
|
||||
documents: tuple[StoredDocument, ...],
|
||||
) -> bytes:
|
||||
indexes: list[dict[str, object]] = []
|
||||
statuses: list[str] = []
|
||||
change_count = 0
|
||||
for document in documents:
|
||||
report = json.loads(document.result_json)
|
||||
status = report["status"]
|
||||
statuses.append(status)
|
||||
change_count += len(report["changes"])
|
||||
base = f"documents/{document.document_id}"
|
||||
indexes.append(
|
||||
{
|
||||
"document_id": document.document_id,
|
||||
"source_label": report["document"]["source_label"],
|
||||
"status": status,
|
||||
"input_sha256": report["input_sha256"],
|
||||
"current_sha256": report["current_sha256"],
|
||||
"change_count": len(report["changes"]),
|
||||
"result_path": f"{base}/result.json",
|
||||
"cleaned_path": f"{base}/cleaned.md" if status == "success" else None,
|
||||
"diff_path": f"{base}/changes.diff" if status == "success" else None,
|
||||
}
|
||||
)
|
||||
failed_count = statuses.count("failed")
|
||||
unstable_count = statuses.count("unstable")
|
||||
overall_status = "failed" if failed_count else "unstable" if unstable_count else "success"
|
||||
payload = {
|
||||
"schema_version": 1,
|
||||
"run": {"run_id": run_id, "run_date": run_date, "status": overall_status},
|
||||
"documents": indexes,
|
||||
"summary": {
|
||||
"document_count": len(documents),
|
||||
"success_count": statuses.count("success"),
|
||||
"failed_count": failed_count,
|
||||
"unstable_count": unstable_count,
|
||||
"change_count": change_count,
|
||||
},
|
||||
}
|
||||
return (json.dumps(payload, indent=2) + "\n").encode()
|
||||
|
||||
|
||||
def review_locator_json(
|
||||
artifacts_root: Path,
|
||||
run_date: str,
|
||||
run_id: str,
|
||||
documents: tuple[StoredDocument, ...],
|
||||
) -> bytes:
|
||||
source_root = artifacts_root.parent / f"{artifacts_root.name}-sources"
|
||||
source_root.mkdir(exist_ok=True)
|
||||
locator_documents: list[dict[str, object]] = []
|
||||
for position, document in enumerate(documents):
|
||||
source_path = source_root / f"{position}-{document.document_id}.md"
|
||||
source_path.write_bytes(_SOURCE_BYTES)
|
||||
report = json.loads(document.result_json)
|
||||
locator_documents.append(
|
||||
{
|
||||
"document_id": document.document_id,
|
||||
"source_path": str(source_path.resolve()),
|
||||
"input_sha256": report["input_sha256"],
|
||||
}
|
||||
)
|
||||
payload = {
|
||||
"schema_version": 1,
|
||||
"run": {
|
||||
"run_id": run_id,
|
||||
"run_directory": str((artifacts_root / run_date / "runs" / run_id).resolve()),
|
||||
"manifest_path": "manifest.json",
|
||||
},
|
||||
"documents": locator_documents,
|
||||
}
|
||||
return (json.dumps(payload, indent=2) + "\n").encode()
|
||||
|
||||
|
||||
def test_publish_run_creates_private_date_layout_and_status_specific_files(tmp_path: Path) -> None:
|
||||
artifacts_root = tmp_path / "artifacts"
|
||||
documents = (stored_document("success"), stored_document("failed", None))
|
||||
locator_json = review_locator_json(artifacts_root, "2026-08-22", "example-run", documents)
|
||||
|
||||
run_directory = publish_run(
|
||||
artifacts_root=artifacts_root,
|
||||
run_date="2026-08-22",
|
||||
run_id="example-run",
|
||||
manifest_json=manifest_json("2026-08-22", "example-run", documents),
|
||||
review_locator_json=locator_json,
|
||||
documents=documents,
|
||||
)
|
||||
|
||||
assert run_directory == artifacts_root / "2026-08-22" / "runs" / "example-run"
|
||||
assert json.loads((run_directory / "manifest.json").read_bytes())["run"]["status"] == "failed"
|
||||
assert (run_directory / "documents/success/result.json").is_file()
|
||||
assert (run_directory / "documents/success/cleaned.md").read_bytes() == b"cleaned\n"
|
||||
assert (run_directory / "documents/success/changes.diff").read_bytes() == b""
|
||||
assert (run_directory / "documents/failed/result.json").is_file()
|
||||
assert (run_directory / "review-locator.json").read_bytes() == locator_json
|
||||
assert not (run_directory / "documents/failed/cleaned.md").exists()
|
||||
assert not (run_directory / "documents/failed/changes.diff").exists()
|
||||
|
||||
for directory in (
|
||||
artifacts_root,
|
||||
artifacts_root / "2026-08-22",
|
||||
artifacts_root / "2026-08-22/runs",
|
||||
run_directory,
|
||||
run_directory / "documents",
|
||||
run_directory / "documents/success",
|
||||
):
|
||||
assert stat.S_IMODE(directory.stat().st_mode) == 0o700
|
||||
for artifact_file in run_directory.rglob("*"):
|
||||
if artifact_file.is_file():
|
||||
assert stat.S_IMODE(artifact_file.stat().st_mode) == 0o600
|
||||
|
||||
|
||||
def test_publish_run_rejects_existing_target_without_overwriting(tmp_path: Path) -> None:
|
||||
artifacts_root = tmp_path / "artifacts"
|
||||
documents = (stored_document(),)
|
||||
first_manifest = manifest_json("2026-08-22", "same-run", documents)
|
||||
locator_json = review_locator_json(artifacts_root, "2026-08-22", "same-run", documents)
|
||||
run_directory = publish_run(
|
||||
artifacts_root=artifacts_root,
|
||||
run_date="2026-08-22",
|
||||
run_id="same-run",
|
||||
manifest_json=first_manifest,
|
||||
review_locator_json=locator_json,
|
||||
documents=documents,
|
||||
)
|
||||
|
||||
with pytest.raises(ArtifactStoreError, match="already exists"):
|
||||
publish_run(
|
||||
artifacts_root=artifacts_root,
|
||||
run_date="2026-08-22",
|
||||
run_id="same-run",
|
||||
manifest_json=first_manifest,
|
||||
review_locator_json=locator_json,
|
||||
documents=documents,
|
||||
)
|
||||
|
||||
assert (run_directory / "manifest.json").read_bytes() == first_manifest
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("run_date", "run_id"),
|
||||
[
|
||||
("2026-8-22", "valid"),
|
||||
("2026-02-30", "valid"),
|
||||
("2026-08-22", "Uppercase"),
|
||||
("2026-08-22", "../escape"),
|
||||
("2026-08-22", "two..dots"),
|
||||
],
|
||||
)
|
||||
def test_publish_run_rejects_unsafe_date_and_run_id(tmp_path: Path, run_date: str, run_id: str) -> None:
|
||||
artifacts_root = tmp_path / "artifacts"
|
||||
documents = (stored_document(),)
|
||||
with pytest.raises(ArtifactStoreError):
|
||||
publish_run(
|
||||
artifacts_root=artifacts_root,
|
||||
run_date=run_date,
|
||||
run_id=run_id,
|
||||
manifest_json=manifest_json(run_date, run_id, documents),
|
||||
review_locator_json=review_locator_json(artifacts_root, run_date, run_id, documents),
|
||||
documents=documents,
|
||||
)
|
||||
|
||||
|
||||
def test_publish_run_rejects_inconsistent_or_duplicate_document_artifacts(tmp_path: Path) -> None:
|
||||
valid = stored_document()
|
||||
bad_hash = StoredDocument("paper", valid.result_json, b"output", b"", "0" * 64)
|
||||
first_artifacts_root = tmp_path / "artifacts-a"
|
||||
with pytest.raises(ArtifactStoreError, match="output hash"):
|
||||
publish_run(
|
||||
artifacts_root=first_artifacts_root,
|
||||
run_date="2026-08-22",
|
||||
run_id="bad-hash",
|
||||
manifest_json=manifest_json("2026-08-22", "bad-hash", (bad_hash,)),
|
||||
review_locator_json=review_locator_json(
|
||||
first_artifacts_root, "2026-08-22", "bad-hash", (bad_hash,)
|
||||
),
|
||||
documents=(bad_hash,),
|
||||
)
|
||||
|
||||
duplicates = (stored_document(), stored_document())
|
||||
second_artifacts_root = tmp_path / "artifacts-b"
|
||||
with pytest.raises(ArtifactStoreError, match="unique"):
|
||||
publish_run(
|
||||
artifacts_root=second_artifacts_root,
|
||||
run_date="2026-08-22",
|
||||
run_id="duplicate",
|
||||
manifest_json=manifest_json("2026-08-22", "duplicate", duplicates),
|
||||
review_locator_json=review_locator_json(
|
||||
second_artifacts_root, "2026-08-22", "duplicate", duplicates
|
||||
),
|
||||
documents=duplicates,
|
||||
)
|
||||
|
||||
|
||||
def test_publish_run_rejects_manifest_path_or_document_mismatch(tmp_path: Path) -> None:
|
||||
documents = (stored_document(),)
|
||||
wrong_date = manifest_json("2026-08-21", "review", documents)
|
||||
first_artifacts_root = tmp_path / "artifacts-a"
|
||||
with pytest.raises(ArtifactStoreError, match="identity"):
|
||||
publish_run(
|
||||
artifacts_root=first_artifacts_root,
|
||||
run_date="2026-08-22",
|
||||
run_id="review",
|
||||
manifest_json=wrong_date,
|
||||
review_locator_json=review_locator_json(
|
||||
first_artifacts_root, "2026-08-22", "review", documents
|
||||
),
|
||||
documents=documents,
|
||||
)
|
||||
|
||||
payload = json.loads(manifest_json("2026-08-22", "review", documents))
|
||||
payload["documents"][0]["change_count"] = 99
|
||||
mismatched_index = (json.dumps(payload, indent=2) + "\n").encode()
|
||||
second_artifacts_root = tmp_path / "artifacts-b"
|
||||
with pytest.raises(ArtifactStoreError, match="index"):
|
||||
publish_run(
|
||||
artifacts_root=second_artifacts_root,
|
||||
run_date="2026-08-22",
|
||||
run_id="review",
|
||||
manifest_json=mismatched_index,
|
||||
review_locator_json=review_locator_json(
|
||||
second_artifacts_root, "2026-08-22", "review", documents
|
||||
),
|
||||
documents=documents,
|
||||
)
|
||||
|
||||
|
||||
def test_publish_run_rejects_inconsistent_review_locator(tmp_path: Path) -> None:
|
||||
artifacts_root = tmp_path / "artifacts"
|
||||
documents = (stored_document(),)
|
||||
locator = json.loads(review_locator_json(artifacts_root, "2026-08-22", "review", documents))
|
||||
locator["documents"][0]["input_sha256"] = "0" * 64
|
||||
inconsistent_locator = (json.dumps(locator, indent=2) + "\n").encode()
|
||||
|
||||
with pytest.raises(ArtifactStoreError, match="document identity"):
|
||||
publish_run(
|
||||
artifacts_root=artifacts_root,
|
||||
run_date="2026-08-22",
|
||||
run_id="review",
|
||||
manifest_json=manifest_json("2026-08-22", "review", documents),
|
||||
review_locator_json=inconsistent_locator,
|
||||
documents=documents,
|
||||
)
|
||||
|
||||
assert not artifacts_root.exists()
|
||||
|
||||
|
||||
def test_publish_run_rejects_source_changed_after_locator_creation(tmp_path: Path) -> None:
|
||||
artifacts_root = tmp_path / "artifacts"
|
||||
documents = (stored_document(),)
|
||||
locator_json = review_locator_json(artifacts_root, "2026-08-22", "review", documents)
|
||||
locator = json.loads(locator_json)
|
||||
Path(locator["documents"][0]["source_path"]).write_bytes(b"changed\n")
|
||||
|
||||
with pytest.raises(ArtifactStoreError, match="input hash"):
|
||||
publish_run(
|
||||
artifacts_root=artifacts_root,
|
||||
run_date="2026-08-22",
|
||||
run_id="review",
|
||||
manifest_json=manifest_json("2026-08-22", "review", documents),
|
||||
review_locator_json=locator_json,
|
||||
documents=documents,
|
||||
)
|
||||
|
||||
assert not artifacts_root.exists()
|
||||
|
||||
|
||||
def test_publish_race_does_not_replace_a_new_target(
|
||||
tmp_path: Path,
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
artifacts_root = tmp_path / "artifacts"
|
||||
documents = (stored_document(),)
|
||||
original_rename = artifact_store._rename_no_replace
|
||||
|
||||
def create_competing_target(source: Path, target: Path) -> None:
|
||||
target.mkdir(mode=0o700)
|
||||
(target / "keep").write_bytes(b"existing")
|
||||
original_rename(source, target)
|
||||
|
||||
monkeypatch.setattr(artifact_store, "_rename_no_replace", create_competing_target)
|
||||
with pytest.raises(ArtifactStoreError, match="already exists"):
|
||||
publish_run(
|
||||
artifacts_root=artifacts_root,
|
||||
run_date="2026-08-22",
|
||||
run_id="raced",
|
||||
manifest_json=manifest_json("2026-08-22", "raced", documents),
|
||||
review_locator_json=review_locator_json(artifacts_root, "2026-08-22", "raced", documents),
|
||||
documents=documents,
|
||||
)
|
||||
|
||||
target = tmp_path / "artifacts/2026-08-22/runs/raced"
|
||||
assert (target / "keep").read_bytes() == b"existing"
|
||||
assert not any(path.name.startswith(".raced.") for path in target.parent.iterdir())
|
||||
|
||||
|
||||
def test_write_failure_cleans_temporary_directory_and_does_not_publish(
|
||||
tmp_path: Path,
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
original_write = artifact_store._write_private_file
|
||||
call_count = 0
|
||||
|
||||
def fail_second_write(path: Path, content: bytes) -> None:
|
||||
nonlocal call_count
|
||||
call_count += 1
|
||||
if call_count == 2:
|
||||
raise OSError("synthetic write failure")
|
||||
original_write(path, content)
|
||||
|
||||
monkeypatch.setattr(artifact_store, "_write_private_file", fail_second_write)
|
||||
artifacts_root = tmp_path / "artifacts"
|
||||
documents = (stored_document(),)
|
||||
|
||||
with pytest.raises(OSError, match="synthetic"):
|
||||
publish_run(
|
||||
artifacts_root=artifacts_root,
|
||||
run_date="2026-08-22",
|
||||
run_id="broken",
|
||||
manifest_json=manifest_json("2026-08-22", "broken", documents),
|
||||
review_locator_json=review_locator_json(artifacts_root, "2026-08-22", "broken", documents),
|
||||
documents=documents,
|
||||
)
|
||||
|
||||
runs_directory = tmp_path / "artifacts/2026-08-22/runs"
|
||||
assert list(runs_directory.iterdir()) == []
|
||||
|
||||
|
||||
def test_locator_write_failure_does_not_publish(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
original_write = artifact_store._write_private_file
|
||||
|
||||
def fail_locator_write(path: Path, content: bytes) -> None:
|
||||
if path.name == "review-locator.json":
|
||||
raise OSError("synthetic locator write failure")
|
||||
original_write(path, content)
|
||||
|
||||
monkeypatch.setattr(artifact_store, "_write_private_file", fail_locator_write)
|
||||
artifacts_root = tmp_path / "artifacts"
|
||||
documents = (stored_document(),)
|
||||
|
||||
with pytest.raises(OSError, match="locator"):
|
||||
publish_run(
|
||||
artifacts_root=artifacts_root,
|
||||
run_date="2026-08-22",
|
||||
run_id="locator-failure",
|
||||
manifest_json=manifest_json("2026-08-22", "locator-failure", documents),
|
||||
review_locator_json=review_locator_json(
|
||||
artifacts_root, "2026-08-22", "locator-failure", documents
|
||||
),
|
||||
documents=documents,
|
||||
)
|
||||
|
||||
runs_directory = tmp_path / "artifacts/2026-08-22/runs"
|
||||
assert list(runs_directory.iterdir()) == []
|
||||
@@ -1,155 +0,0 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
|
||||
import mdpolish
|
||||
from mdpolish import Pipeline, RunStatus
|
||||
from mdpolish.components import ArxivSubmissionStampComponent
|
||||
|
||||
STAMP = "arXiv:2104.12345v2 [stat.ME] 31 Dec 2021"
|
||||
OTHER_STAMP = "arXiv:2301.7v1 [cs.AI] 1 Jan 2023"
|
||||
REASON = "删除完整匹配的 arXiv 提交边栏戳"
|
||||
|
||||
|
||||
def transform(markdown: str) -> mdpolish.TransformResult:
|
||||
return Pipeline([ArxivSubmissionStampComponent()]).transform(markdown)
|
||||
|
||||
|
||||
def test_component_metadata_and_package_export() -> None:
|
||||
result = transform("")
|
||||
|
||||
assert result.status is RunStatus.SUCCESS
|
||||
assert result.components[0].component_id == "paper.arxiv_submission_stamp"
|
||||
assert result.components[0].version == "1.0.0"
|
||||
assert result.components[0].parameters == ()
|
||||
assert result.components[0].applicability
|
||||
assert not hasattr(mdpolish, "ArxivSubmissionStampComponent")
|
||||
|
||||
|
||||
@pytest.mark.parametrize("markdown", ["", "普通正文", "中文\nCafe\u0301\n🙂\n"])
|
||||
def test_no_target_returns_unchanged_success(markdown: str) -> None:
|
||||
result = transform(markdown)
|
||||
|
||||
assert result.status is RunStatus.SUCCESS
|
||||
assert result.output_markdown == markdown
|
||||
assert result.changes == ()
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("markdown", "expected"),
|
||||
[
|
||||
(f"{STAMP}\n正文", "正文"),
|
||||
(f"正文\n{STAMP}\n后文", "正文\n后文"),
|
||||
(f"正文\n{STAMP}", "正文\n"),
|
||||
(STAMP, ""),
|
||||
(f"正文\r\n{STAMP}\r\n后文", "正文\r\n后文"),
|
||||
(f"正文\r{STAMP}\r后文", "正文\r后文"),
|
||||
],
|
||||
)
|
||||
def test_deletes_target_at_approved_line_boundaries(markdown: str, expected: str) -> None:
|
||||
result = transform(markdown)
|
||||
|
||||
assert result.status is RunStatus.SUCCESS
|
||||
assert result.output_markdown == expected
|
||||
assert len(result.changes) == 1
|
||||
|
||||
|
||||
def test_multiple_targets_are_reported_in_source_order_with_one_atomic_batch() -> None:
|
||||
markdown = f"{STAMP}\n保留\n{OTHER_STAMP}"
|
||||
|
||||
result = transform(markdown)
|
||||
|
||||
assert result.status is RunStatus.SUCCESS
|
||||
assert result.output_markdown == "保留\n"
|
||||
assert [change.before for change in result.changes] == [f"{STAMP}\n", OTHER_STAMP]
|
||||
assert [change.proposal_ref.proposal_index for change in result.changes] == [0, 1]
|
||||
assert [change.span.start for change in result.changes] == sorted(change.span.start for change in result.changes)
|
||||
assert {change.before_sha256 for change in result.changes} == {result.input_sha256}
|
||||
assert {change.after_sha256 for change in result.changes} == {result.current_sha256}
|
||||
|
||||
|
||||
def test_adjacent_targets_use_non_overlapping_delete_ranges() -> None:
|
||||
result = transform(f"{STAMP}\n{OTHER_STAMP}")
|
||||
|
||||
assert result.status is RunStatus.SUCCESS
|
||||
assert result.output_markdown == ""
|
||||
assert len(result.changes) == 2
|
||||
assert result.changes[0].span.end == result.changes[1].span.start
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"line",
|
||||
[
|
||||
f" {STAMP}",
|
||||
f"{STAMP} ",
|
||||
f"- {STAMP}",
|
||||
f"> {STAMP}",
|
||||
"1. Example. arXiv preprint arXiv:2104.12345v2 [stat.ME], 2021.",
|
||||
"See arXiv:2104.12345v2 for details.",
|
||||
"arXiv:2104.12345 [stat.ME] 31 Dec 2021",
|
||||
"arXiv:hep-ph/9901001 [hep-ph] 31 Dec 1999",
|
||||
"arXiv:2104.12345v2 31 Dec 2021",
|
||||
"arXiv:2104.12345v2 [stat.ME] 0 Dec 2021",
|
||||
"arXiv:2104.12345v2 [stat.ME] 32 Dec 2021",
|
||||
"arXiv:2104.12345v2 [stat.ME] 31 December 2021",
|
||||
"arXiv:2104.12345v2 [统计] 31 Dec 2021",
|
||||
],
|
||||
)
|
||||
def test_similar_arxiv_text_is_preserved(line: str) -> None:
|
||||
markdown = f"前文\n{line}\n后文"
|
||||
|
||||
result = transform(markdown)
|
||||
|
||||
assert result.status is RunStatus.SUCCESS
|
||||
assert result.output_markdown == markdown
|
||||
assert result.changes == ()
|
||||
|
||||
|
||||
def test_matching_line_inside_fenced_code_is_not_protected() -> None:
|
||||
markdown = f"```text\n{STAMP}\n```\n"
|
||||
|
||||
result = transform(markdown)
|
||||
|
||||
assert result.status is RunStatus.SUCCESS
|
||||
assert result.output_markdown == "```text\n```\n"
|
||||
assert len(result.changes) == 1
|
||||
|
||||
|
||||
def test_change_audit_records_identity_reason_source_and_batch_hashes() -> None:
|
||||
result = transform(f"{STAMP}\n正文")
|
||||
|
||||
assert result.status is RunStatus.SUCCESS
|
||||
change = result.changes[0]
|
||||
assert change.component_id == "paper.arxiv_submission_stamp"
|
||||
assert change.component_version == "1.0.0"
|
||||
assert change.component_position == 0
|
||||
assert change.proposal_ref.component_position == 0
|
||||
assert change.proposal_ref.proposal_index == 0
|
||||
assert change.edit_index == 0
|
||||
assert change.reason == REASON
|
||||
assert change.before == f"{STAMP}\n"
|
||||
assert change.after == ""
|
||||
assert change.before_sha256 == result.input_sha256
|
||||
assert change.after_sha256 == result.current_sha256
|
||||
|
||||
|
||||
def test_successful_output_is_stable_and_second_run_has_no_changes() -> None:
|
||||
pipeline = Pipeline([ArxivSubmissionStampComponent()])
|
||||
|
||||
first = pipeline.transform(f"{STAMP}\n正文")
|
||||
assert first.status is RunStatus.SUCCESS
|
||||
assert first.output_markdown == "正文"
|
||||
|
||||
second = pipeline.transform(first.output_markdown)
|
||||
|
||||
assert second.status is RunStatus.SUCCESS
|
||||
assert second.output_markdown == "正文"
|
||||
assert second.changes == ()
|
||||
assert second.residual_proposals == ()
|
||||
|
||||
|
||||
def test_same_input_produces_same_ordered_result() -> None:
|
||||
pipeline = Pipeline([ArxivSubmissionStampComponent()])
|
||||
markdown = f"{STAMP}\n正文\n{OTHER_STAMP}\n"
|
||||
|
||||
assert pipeline.transform(markdown) == pipeline.transform(markdown)
|
||||
@@ -1,162 +0,0 @@
|
||||
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>&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")
|
||||
@@ -1,121 +0,0 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from collections.abc import Mapping
|
||||
from typing import cast
|
||||
|
||||
import pytest
|
||||
|
||||
from mdpolish import Component, ComponentContractError, DocumentSnapshot, ProposedChange, TextEdit, TextSpan
|
||||
|
||||
|
||||
class ExampleComponent(Component):
|
||||
def __init__(
|
||||
self,
|
||||
*,
|
||||
component_id: object = "test.example",
|
||||
version: object = "1.2.3",
|
||||
parameters: object = None,
|
||||
applicability: object = "处理测试标记，要求精确匹配，排除所有其他内容。",
|
||||
) -> None:
|
||||
self._component_id = component_id
|
||||
self._version = version
|
||||
self._parameters = {} if parameters is None else parameters
|
||||
self._applicability = applicability
|
||||
|
||||
@property
|
||||
def component_id(self) -> str:
|
||||
return cast(str, self._component_id)
|
||||
|
||||
@property
|
||||
def version(self) -> str:
|
||||
return cast(str, self._version)
|
||||
|
||||
@property
|
||||
def parameters(self) -> Mapping[str, object]:
|
||||
return cast(Mapping[str, object], self._parameters)
|
||||
|
||||
@property
|
||||
def applicability(self) -> str:
|
||||
return cast(str, self._applicability)
|
||||
|
||||
def _propose_changes(self, snapshot: DocumentSnapshot) -> tuple[ProposedChange, ...]:
|
||||
if not snapshot.markdown:
|
||||
return ()
|
||||
edit = TextEdit(snapshot.sha256, TextSpan(0, 1), snapshot.markdown[0], "X")
|
||||
return (ProposedChange(snapshot.sha256, "replace first character", (edit,)),)
|
||||
|
||||
|
||||
class ListReturningComponent(ExampleComponent):
|
||||
def _propose_changes(self, snapshot: DocumentSnapshot) -> tuple[ProposedChange, ...]:
|
||||
return cast(tuple[ProposedChange, ...], [])
|
||||
|
||||
|
||||
class WrongValueComponent(ExampleComponent):
|
||||
def _propose_changes(self, snapshot: DocumentSnapshot) -> tuple[ProposedChange, ...]:
|
||||
return cast(tuple[ProposedChange, ...], ("wrong",))
|
||||
|
||||
|
||||
def test_component_metadata_is_validated_and_parameters_are_frozen_deterministically() -> None:
|
||||
component = ExampleComponent(
|
||||
parameters={
|
||||
"z": [1, {"b": False, "a": None}],
|
||||
"a": "value",
|
||||
}
|
||||
)
|
||||
|
||||
info = component._component_info()
|
||||
|
||||
assert info.component_id == "test.example"
|
||||
assert info.version == "1.2.3"
|
||||
assert info.parameters == (
|
||||
("a", "value"),
|
||||
("z", (1, (("a", None), ("b", False)))),
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("component_id", ["", "Uppercase", "has space", "two..dots", "_leading"])
|
||||
def test_invalid_component_id_is_a_contract_error(component_id: str) -> None:
|
||||
with pytest.raises(ComponentContractError, match="component_id"):
|
||||
ExampleComponent(component_id=component_id)._component_info()
|
||||
|
||||
|
||||
@pytest.mark.parametrize("version", ["1", "1.2", "v1.2.3", "01.2.3", "1.2.3-alpha"])
|
||||
def test_invalid_version_is_a_contract_error(version: str) -> None:
|
||||
with pytest.raises(ComponentContractError, match=r"MAJOR.MINOR.PATCH"):
|
||||
ExampleComponent(version=version)._component_info()
|
||||
|
||||
|
||||
def test_empty_applicability_is_a_contract_error() -> None:
|
||||
with pytest.raises(ComponentContractError, match="applicability"):
|
||||
ExampleComponent(applicability=" \n")._component_info()
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"parameters",
|
||||
[
|
||||
{"bad": {1, 2}},
|
||||
{"bad": float("inf")},
|
||||
{"bad": float("nan")},
|
||||
{1: "non-string key"},
|
||||
["not", "a", "mapping"],
|
||||
],
|
||||
)
|
||||
def test_unrepresentable_parameters_are_contract_errors(parameters: object) -> None:
|
||||
with pytest.raises(ComponentContractError, match="parameter"):
|
||||
ExampleComponent(parameters=parameters)._component_info()
|
||||
|
||||
|
||||
def test_collect_proposals_requires_a_tuple_of_proposed_changes() -> None:
|
||||
snapshot = DocumentSnapshot("abc")
|
||||
|
||||
with pytest.raises(ComponentContractError, match="return a tuple"):
|
||||
ListReturningComponent()._collect_proposals(snapshot)
|
||||
with pytest.raises(ComponentContractError, match="only ProposedChange"):
|
||||
WrongValueComponent()._collect_proposals(snapshot)
|
||||
|
||||
|
||||
def test_component_exposes_no_public_check_or_transform_shortcut() -> None:
|
||||
component = ExampleComponent()
|
||||
|
||||
assert not hasattr(component, "check")
|
||||
assert not hasattr(component, "transform")
|
||||
+25
-25
@@ -3,17 +3,17 @@ from __future__ import annotations
|
||||
import pytest
|
||||
|
||||
from mdpolish import (
|
||||
ComponentInfo,
|
||||
DocumentSnapshot,
|
||||
EditValidationError,
|
||||
ModifierInfo,
|
||||
ProposedChange,
|
||||
TextEdit,
|
||||
TextSpan,
|
||||
apply_component_batch,
|
||||
apply_modifier_batch,
|
||||
)
|
||||
|
||||
COMPONENT = ComponentInfo(
|
||||
component_id="test.component",
|
||||
MODIFIER = ModifierInfo(
|
||||
modifier_id="test.modifier",
|
||||
version="1.2.3",
|
||||
parameters=(),
|
||||
applicability="测试精确文本编辑，只处理测试字符串，排除其他输入。",
|
||||
@@ -38,22 +38,22 @@ def test_applies_insert_delete_and_replace() -> None:
|
||||
delete_snapshot = DocumentSnapshot("abc")
|
||||
replace_snapshot = DocumentSnapshot("abc")
|
||||
|
||||
inserted = apply_component_batch(
|
||||
inserted = apply_modifier_batch(
|
||||
insert_snapshot,
|
||||
(make_proposal(insert_snapshot, make_edit(insert_snapshot, 1, 1, "X")),),
|
||||
COMPONENT,
|
||||
MODIFIER,
|
||||
0,
|
||||
)
|
||||
deleted = apply_component_batch(
|
||||
deleted = apply_modifier_batch(
|
||||
delete_snapshot,
|
||||
(make_proposal(delete_snapshot, make_edit(delete_snapshot, 1, 2, "")),),
|
||||
COMPONENT,
|
||||
MODIFIER,
|
||||
0,
|
||||
)
|
||||
replaced = apply_component_batch(
|
||||
replaced = apply_modifier_batch(
|
||||
replace_snapshot,
|
||||
(make_proposal(replace_snapshot, make_edit(replace_snapshot, 1, 2, "X")),),
|
||||
COMPONENT,
|
||||
MODIFIER,
|
||||
0,
|
||||
)
|
||||
|
||||
@@ -71,7 +71,7 @@ def test_multiple_edits_apply_backwards_but_report_in_source_order() -> None:
|
||||
reason="normalize two locations",
|
||||
)
|
||||
|
||||
applied = apply_component_batch(snapshot, (proposal,), COMPONENT, 3)
|
||||
applied = apply_modifier_batch(snapshot, (proposal,), MODIFIER, 3)
|
||||
|
||||
assert applied.snapshot.markdown == "AbcdF"
|
||||
assert [change.span.start for change in applied.changes] == [0, 4]
|
||||
@@ -82,7 +82,7 @@ def test_multiple_edits_apply_backwards_but_report_in_source_order() -> None:
|
||||
}
|
||||
assert {change.reason for change in applied.changes} == {"normalize two locations"}
|
||||
assert [change.edit_index for change in applied.changes] == [1, 0]
|
||||
assert all(change.component_position == 3 for change in applied.changes)
|
||||
assert all(change.modifier_position == 3 for change in applied.changes)
|
||||
|
||||
|
||||
def test_adjacent_nonempty_ranges_are_allowed() -> None:
|
||||
@@ -93,7 +93,7 @@ def test_adjacent_nonempty_ranges_are_allowed() -> None:
|
||||
make_edit(snapshot, 2, 4, "D"),
|
||||
)
|
||||
|
||||
applied = apply_component_batch(snapshot, (proposal,), COMPONENT, 0)
|
||||
applied = apply_modifier_batch(snapshot, (proposal,), MODIFIER, 0)
|
||||
|
||||
assert applied.snapshot.markdown == "AD"
|
||||
|
||||
@@ -107,7 +107,7 @@ def test_overlapping_ranges_fail_without_changing_snapshot() -> None:
|
||||
)
|
||||
|
||||
with pytest.raises(EditValidationError, match="conflicting"):
|
||||
apply_component_batch(snapshot, (proposal,), COMPONENT, 0)
|
||||
apply_modifier_batch(snapshot, (proposal,), MODIFIER, 0)
|
||||
|
||||
assert snapshot.markdown == "abcdef"
|
||||
assert snapshot.sha256 == DocumentSnapshot("abcdef").sha256
|
||||
@@ -119,7 +119,7 @@ def test_duplicate_edits_fail_explicitly() -> None:
|
||||
proposal = make_proposal(snapshot, edit, edit)
|
||||
|
||||
with pytest.raises(EditValidationError, match="duplicate"):
|
||||
apply_component_batch(snapshot, (proposal,), COMPONENT, 0)
|
||||
apply_modifier_batch(snapshot, (proposal,), MODIFIER, 0)
|
||||
|
||||
|
||||
def test_distinct_insert_points_are_allowed() -> None:
|
||||
@@ -130,7 +130,7 @@ def test_distinct_insert_points_are_allowed() -> None:
|
||||
make_edit(snapshot, 3, 3, "Y"),
|
||||
)
|
||||
|
||||
applied = apply_component_batch(snapshot, (proposal,), COMPONENT, 0)
|
||||
applied = apply_modifier_batch(snapshot, (proposal,), MODIFIER, 0)
|
||||
|
||||
assert applied.snapshot.markdown == "aXbcYd"
|
||||
|
||||
@@ -144,7 +144,7 @@ def test_same_insert_point_conflicts() -> None:
|
||||
)
|
||||
|
||||
with pytest.raises(EditValidationError, match="conflicting"):
|
||||
apply_component_batch(snapshot, (proposal,), COMPONENT, 0)
|
||||
apply_modifier_batch(snapshot, (proposal,), MODIFIER, 0)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("insert_position", [1, 2, 3])
|
||||
@@ -157,7 +157,7 @@ def test_insert_at_start_inside_or_end_of_nonempty_range_conflicts(insert_positi
|
||||
)
|
||||
|
||||
with pytest.raises(EditValidationError, match="conflicting"):
|
||||
apply_component_batch(snapshot, (proposal,), COMPONENT, 0)
|
||||
apply_modifier_batch(snapshot, (proposal,), MODIFIER, 0)
|
||||
|
||||
|
||||
def test_stale_hash_out_of_range_and_expected_text_mismatch_fail() -> None:
|
||||
@@ -166,34 +166,34 @@ def test_stale_hash_out_of_range_and_expected_text_mismatch_fail() -> None:
|
||||
stale = make_proposal(original, make_edit(original, 0, 1, "A"))
|
||||
|
||||
with pytest.raises(EditValidationError, match="stale"):
|
||||
apply_component_batch(current, (stale,), COMPONENT, 0)
|
||||
apply_modifier_batch(current, (stale,), MODIFIER, 0)
|
||||
|
||||
out_of_range_edit = TextEdit(current.sha256, TextSpan(2, 5), "dxx", "D")
|
||||
out_of_range = make_proposal(current, out_of_range_edit)
|
||||
with pytest.raises(EditValidationError, match="outside"):
|
||||
apply_component_batch(current, (out_of_range,), COMPONENT, 0)
|
||||
apply_modifier_batch(current, (out_of_range,), MODIFIER, 0)
|
||||
|
||||
mismatch_edit = TextEdit(current.sha256, TextSpan(0, 1), "z", "A")
|
||||
mismatch = make_proposal(current, mismatch_edit)
|
||||
with pytest.raises(EditValidationError, match="expected_text"):
|
||||
apply_component_batch(current, (mismatch,), COMPONENT, 0)
|
||||
apply_modifier_batch(current, (mismatch,), MODIFIER, 0)
|
||||
|
||||
|
||||
def test_conflict_across_proposals_rejects_whole_component_batch() -> None:
|
||||
def test_conflict_across_proposals_rejects_whole_modifier_batch() -> None:
|
||||
snapshot = DocumentSnapshot("abcdef")
|
||||
first = make_proposal(snapshot, make_edit(snapshot, 0, 3, "X"), reason="first")
|
||||
second = make_proposal(snapshot, make_edit(snapshot, 2, 4, "Y"), reason="second")
|
||||
|
||||
with pytest.raises(EditValidationError, match="conflicting"):
|
||||
apply_component_batch(snapshot, (first, second), COMPONENT, 0)
|
||||
apply_modifier_batch(snapshot, (first, second), MODIFIER, 0)
|
||||
|
||||
assert snapshot.markdown == "abcdef"
|
||||
|
||||
|
||||
def test_empty_component_batch_keeps_same_snapshot_and_records_nothing() -> None:
|
||||
def test_empty_modifier_batch_keeps_same_snapshot_and_records_nothing() -> None:
|
||||
snapshot = DocumentSnapshot("abc")
|
||||
|
||||
applied = apply_component_batch(snapshot, (), COMPONENT, 0)
|
||||
applied = apply_modifier_batch(snapshot, (), MODIFIER, 0)
|
||||
|
||||
assert applied.snapshot is snapshot
|
||||
assert applied.changes == ()
|
||||
|
||||
@@ -1,356 +0,0 @@
|
||||
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())
|
||||
locator = json.loads((result.run_directory / "review-locator.json").read_bytes())
|
||||
assert locator == {
|
||||
"schema_version": 1,
|
||||
"run": {
|
||||
"run_id": "local-review",
|
||||
"run_directory": str(result.run_directory.resolve()),
|
||||
"manifest_path": "manifest.json",
|
||||
},
|
||||
"documents": [
|
||||
{
|
||||
"document_id": "first",
|
||||
"source_path": str(first_path.resolve()),
|
||||
"input_sha256": manifest["documents"][0]["input_sha256"],
|
||||
},
|
||||
{
|
||||
"document_id": "second",
|
||||
"source_path": str(second_path.resolve()),
|
||||
"input_sha256": manifest["documents"][1]["input_sha256"],
|
||||
},
|
||||
],
|
||||
}
|
||||
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"
|
||||
@@ -3,11 +3,11 @@ from __future__ import annotations
|
||||
import pytest
|
||||
|
||||
from mdpolish import Pipeline, RunStatus
|
||||
from mdpolish.components import HtmlTableDoubleEscapeComponent
|
||||
from mdpolish.modifiers import html_table_entity_unescape
|
||||
|
||||
|
||||
def transform(markdown: str): # type: ignore[no-untyped-def]
|
||||
return Pipeline([HtmlTableDoubleEscapeComponent()]).transform(markdown)
|
||||
return Pipeline([html_table_entity_unescape()]).transform(markdown)
|
||||
|
||||
|
||||
def test_unescapes_one_layer_only_in_strict_cell_text() -> None:
|
||||
@@ -47,13 +47,13 @@ def test_non_strict_or_outside_content_is_preserved(markdown: str) -> None:
|
||||
assert transform(markdown).output_markdown == markdown
|
||||
|
||||
|
||||
def test_fenced_table_is_not_protected() -> None:
|
||||
def test_fenced_table_is_not_protected_by_the_lexical_subset() -> None:
|
||||
markdown = "```html\n<table><tr><td>&lt;</td></tr></table>\n```"
|
||||
assert transform(markdown).output_markdown == "```html\n<table><tr><td><</td></tr></table>\n```"
|
||||
|
||||
|
||||
def test_successful_output_is_idempotent() -> None:
|
||||
pipeline = Pipeline([HtmlTableDoubleEscapeComponent()])
|
||||
pipeline = Pipeline([html_table_entity_unescape()])
|
||||
first = pipeline.transform("<table><tr><td>&lt;</td></tr></table>")
|
||||
assert first.output_markdown is not None
|
||||
assert pipeline.transform(first.output_markdown).changes == ()
|
||||
@@ -3,13 +3,13 @@ from __future__ import annotations
|
||||
import pytest
|
||||
|
||||
from mdpolish import Pipeline, RunStatus
|
||||
from mdpolish.components import HtmlTableLayoutComponent
|
||||
from mdpolish.modifiers import html_table_layout
|
||||
|
||||
TABLE = '<table class="x"><tr><td colspan="2">A</td></tr><tr><td>B</td><td>C</td></tr></table>'
|
||||
|
||||
|
||||
def transform(markdown: str): # type: ignore[no-untyped-def]
|
||||
return Pipeline([HtmlTableLayoutComponent()]).transform(markdown)
|
||||
return Pipeline([html_table_layout()]).transform(markdown)
|
||||
|
||||
|
||||
def test_expands_rows_without_changing_tags_attributes_or_cells() -> None:
|
||||
@@ -57,7 +57,7 @@ def test_mixed_multiline_or_non_strict_tables_are_preserved(markdown: str) -> No
|
||||
|
||||
|
||||
def test_successful_output_is_idempotent() -> None:
|
||||
pipeline = Pipeline([HtmlTableLayoutComponent()])
|
||||
pipeline = Pipeline([html_table_layout()])
|
||||
first = pipeline.transform(TABLE)
|
||||
assert first.output_markdown is not None
|
||||
assert pipeline.transform(first.output_markdown).changes == ()
|
||||
|
||||
@@ -1,85 +0,0 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
|
||||
from mdpolish import Pipeline, RunStatus
|
||||
from mdpolish.components import ManuscriptLineNumberComponent
|
||||
|
||||
|
||||
def _document(
|
||||
*,
|
||||
count: int = 20,
|
||||
heading_numbers: frozenset[int] = frozenset({5, 15}),
|
||||
numbers: tuple[int, ...] | None = None,
|
||||
line_ending: str = "\n",
|
||||
) -> str:
|
||||
values = numbers if numbers is not None else tuple(range(1, count + 1))
|
||||
body = [
|
||||
f"## {number} Section {number}" if number in heading_numbers else f"{number} body {number}"
|
||||
for number in values
|
||||
]
|
||||
return line_ending.join(("1 Affiliation", "2 Institute", "## Abstract", *body))
|
||||
|
||||
|
||||
def transform(markdown: str): # type: ignore[no-untyped-def]
|
||||
return Pipeline([ManuscriptLineNumberComponent()]).transform(markdown)
|
||||
|
||||
|
||||
def test_removes_long_monotonic_sequence_but_preserves_pre_abstract_affiliations() -> None:
|
||||
result = transform(_document())
|
||||
|
||||
assert result.status is RunStatus.SUCCESS
|
||||
assert result.output_markdown is not None
|
||||
assert result.output_markdown.startswith("1 Affiliation\n2 Institute\n## Abstract\nbody 1")
|
||||
assert "## Section 5" in result.output_markdown
|
||||
assert len(result.changes) == 20
|
||||
|
||||
|
||||
@pytest.mark.parametrize("line_ending", ["\n", "\r\n", "\r"])
|
||||
def test_preserves_all_supported_line_endings(line_ending: str) -> None:
|
||||
result = transform(_document(line_ending=line_ending))
|
||||
assert result.output_markdown is not None
|
||||
assert result.output_markdown.count(line_ending) == _document(line_ending=line_ending).count(line_ending)
|
||||
|
||||
|
||||
def test_allows_skipped_numbers_when_sequence_is_strictly_increasing() -> None:
|
||||
numbers = tuple(range(10, 30))
|
||||
result = transform(_document(numbers=numbers, heading_numbers=frozenset({14, 24})))
|
||||
|
||||
assert result.status is RunStatus.SUCCESS
|
||||
assert len(result.changes) == 20
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"markdown",
|
||||
[
|
||||
_document(count=19, heading_numbers=frozenset({5, 15})),
|
||||
_document(heading_numbers=frozenset({5})),
|
||||
_document(numbers=(*tuple(range(1, 20)), 10), heading_numbers=frozenset({5, 15})),
|
||||
_document().replace("## Abstract", "## ABSTRACT"),
|
||||
_document() + "\n## Abstract",
|
||||
],
|
||||
)
|
||||
def test_incomplete_or_ambiguous_evidence_preserves_the_document(markdown: str) -> None:
|
||||
result = transform(markdown)
|
||||
assert result.output_markdown == markdown
|
||||
assert result.changes == ()
|
||||
|
||||
|
||||
def test_lists_years_and_numbers_inside_body_are_not_candidates() -> None:
|
||||
markdown = _document() + "\n1. list\n1) list\n2024 report\nThe panel included 35 experts"
|
||||
result = transform(markdown)
|
||||
|
||||
assert result.output_markdown is not None
|
||||
assert result.output_markdown.endswith("1. list\n1) list\n2024 report\nThe panel included 35 experts")
|
||||
assert len(result.changes) == 20
|
||||
|
||||
|
||||
def test_successful_output_is_idempotent() -> None:
|
||||
pipeline = Pipeline([ManuscriptLineNumberComponent()])
|
||||
first = pipeline.transform(_document())
|
||||
assert first.output_markdown is not None
|
||||
second = pipeline.transform(first.output_markdown)
|
||||
|
||||
assert second.status is RunStatus.SUCCESS
|
||||
assert second.changes == ()
|
||||
@@ -0,0 +1,90 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
|
||||
from mdpolish import ModifierContractError, Pipeline, RunStatus
|
||||
from mdpolish.modifiers import mapped_line_join
|
||||
|
||||
MAPPINGS = (
|
||||
("exam-", "ple", "example"),
|
||||
("rule-", "based", "rule-based"),
|
||||
("value.", "ues", "values"),
|
||||
)
|
||||
|
||||
|
||||
def transform(markdown: str): # type: ignore[no-untyped-def]
|
||||
return Pipeline([mapped_line_join(MAPPINGS)]).transform(markdown)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("markdown", "expected"),
|
||||
[
|
||||
("an exam-\nple here", "an example here"),
|
||||
("an exam-\n\nple here", "an example here"),
|
||||
("a rule-\nbased method", "a rule-based method"),
|
||||
("the value.\n\nues differ", "the values differ"),
|
||||
("an exam-\r\n\r\nple here", "an example here"),
|
||||
("an exam-\r\rple here", "an example here"),
|
||||
],
|
||||
)
|
||||
def test_applies_exact_mapping_across_supported_line_shapes(markdown: str, expected: str) -> None:
|
||||
result = transform(markdown)
|
||||
assert result.status is RunStatus.SUCCESS
|
||||
assert result.output_markdown == expected
|
||||
assert len(result.changes) == 1
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"markdown",
|
||||
[
|
||||
"an unknown-\nword here",
|
||||
"an exam-\n\n\nple here",
|
||||
"an exam-\r\n\nple here",
|
||||
"an EXAM-\nple here",
|
||||
"an exam-\nplemore here",
|
||||
"an xrule-\nbased method",
|
||||
],
|
||||
)
|
||||
def test_unknown_or_unsafe_boundaries_are_preserved(markdown: str) -> None:
|
||||
assert transform(markdown).output_markdown == markdown
|
||||
|
||||
|
||||
def test_parameters_and_results_are_independent_of_mapping_order() -> None:
|
||||
first = mapped_line_join(MAPPINGS)
|
||||
second = mapped_line_join(reversed(MAPPINGS))
|
||||
markdown = "example becomes exam-\nple"
|
||||
|
||||
first_result = Pipeline([first]).transform(markdown)
|
||||
second_result = Pipeline([second]).transform(markdown)
|
||||
|
||||
assert first_result.modifiers == second_result.modifiers
|
||||
assert first_result.output_markdown == second_result.output_markdown
|
||||
|
||||
|
||||
def test_library_contains_no_default_mapping() -> None:
|
||||
modifier = mapped_line_join(())
|
||||
result = Pipeline([modifier]).transform("an exam-\nple here")
|
||||
|
||||
assert modifier.parameters == (("mappings", ()),)
|
||||
assert result.output_markdown == "an exam-\nple here"
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"mappings",
|
||||
[
|
||||
(("", "right", "word"),),
|
||||
(("left", "right", "two words"),),
|
||||
(("left", "right", "word"), ("left", "right", "other")),
|
||||
(("left", "right"),),
|
||||
],
|
||||
)
|
||||
def test_invalid_mappings_raise_contract_error(mappings: object) -> None:
|
||||
with pytest.raises(ModifierContractError):
|
||||
mapped_line_join(mappings) # type: ignore[arg-type]
|
||||
|
||||
|
||||
def test_successful_output_is_idempotent() -> None:
|
||||
pipeline = Pipeline([mapped_line_join(MAPPINGS)])
|
||||
first = pipeline.transform("an exam-\n\nple here")
|
||||
assert first.output_markdown is not None
|
||||
assert pipeline.transform(first.output_markdown).changes == ()
|
||||
@@ -95,13 +95,13 @@ def test_proposal_requires_reason_edits_and_one_matching_digest() -> None:
|
||||
|
||||
def test_transform_result_enforces_status_specific_output_fields() -> None:
|
||||
snapshot = DocumentSnapshot("abc")
|
||||
error = RunError("component", "1.0.0", 0, ErrorStage.TRANSFORM, "ExampleError", "safe")
|
||||
error = RunError("modifier", "1.0.0", 0, ErrorStage.TRANSFORM, "ExampleError", "safe")
|
||||
edit = TextEdit(snapshot.sha256, TextSpan(0, 1), "a", "A")
|
||||
proposal = ProposedChange(snapshot.sha256, "reason", (edit,))
|
||||
residual = ResidualProposal(
|
||||
component_id="component",
|
||||
component_version="1.0.0",
|
||||
component_position=0,
|
||||
modifier_id="modifier",
|
||||
modifier_version="1.0.0",
|
||||
modifier_position=0,
|
||||
proposal_ref=ProposalReference(0, snapshot.sha256, 0),
|
||||
proposal=proposal,
|
||||
)
|
||||
|
||||
@@ -0,0 +1,114 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import FrozenInstanceError
|
||||
from typing import cast
|
||||
|
||||
import pytest
|
||||
|
||||
from mdpolish import DocumentSnapshot, Modifier, ModifierContractError, ProposedChange
|
||||
|
||||
|
||||
def no_changes(snapshot: DocumentSnapshot) -> tuple[ProposedChange, ...]:
|
||||
return ()
|
||||
|
||||
|
||||
def test_plain_function_and_mapping_parameters_create_immutable_modifier() -> None:
|
||||
source_parameters: dict[str, object] = {
|
||||
"enabled": True,
|
||||
"nested": {"count": 2},
|
||||
"values": ["a", "b"],
|
||||
}
|
||||
|
||||
modifier = Modifier(
|
||||
modifier_id="example.rule",
|
||||
version="1.2.3",
|
||||
parameters=source_parameters,
|
||||
applicability="处理虚构测试输入；排除其他文本。",
|
||||
propose=no_changes,
|
||||
)
|
||||
source_parameters["enabled"] = False
|
||||
|
||||
assert modifier.propose is no_changes
|
||||
assert modifier.parameters == (
|
||||
("enabled", True),
|
||||
("nested", (("count", 2),)),
|
||||
("values", ("a", "b")),
|
||||
)
|
||||
with pytest.raises(FrozenInstanceError):
|
||||
modifier.version = "2.0.0" # type: ignore[misc]
|
||||
|
||||
|
||||
def test_normalized_parameters_are_sorted_and_validated() -> None:
|
||||
modifier = Modifier(
|
||||
modifier_id="example.normalized",
|
||||
version="1.0.0",
|
||||
parameters=(("z", 1), ("a", ("x", 2))),
|
||||
applicability="处理虚构测试输入。",
|
||||
propose=no_changes,
|
||||
)
|
||||
|
||||
assert modifier.parameters == (("a", ("x", 2)), ("z", 1))
|
||||
|
||||
|
||||
@pytest.mark.parametrize("modifier_id", ["", "Upper", "has space", ".leading", "trailing."])
|
||||
def test_invalid_modifier_id_is_rejected(modifier_id: str) -> None:
|
||||
with pytest.raises(ModifierContractError, match="modifier_id"):
|
||||
Modifier(
|
||||
modifier_id=modifier_id,
|
||||
version="1.0.0",
|
||||
parameters=(),
|
||||
applicability="测试。",
|
||||
propose=no_changes,
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("version", ["", "1", "1.0", "v1.0.0", "01.0.0", "1.0.0-beta"])
|
||||
def test_invalid_version_is_rejected(version: str) -> None:
|
||||
with pytest.raises(ModifierContractError, match="version"):
|
||||
Modifier(
|
||||
modifier_id="example.rule",
|
||||
version=version,
|
||||
parameters=(),
|
||||
applicability="测试。",
|
||||
propose=no_changes,
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"parameters",
|
||||
[
|
||||
{"bad": {1}},
|
||||
{"bad": float("inf")},
|
||||
(("duplicate", 1), ("duplicate", 2)),
|
||||
(("bad", object()),),
|
||||
("not-a-pair",),
|
||||
],
|
||||
)
|
||||
def test_invalid_parameters_are_rejected(parameters: object) -> None:
|
||||
with pytest.raises(ModifierContractError):
|
||||
Modifier(
|
||||
modifier_id="example.rule",
|
||||
version="1.0.0",
|
||||
parameters=parameters, # type: ignore[arg-type]
|
||||
applicability="测试。",
|
||||
propose=no_changes,
|
||||
)
|
||||
|
||||
|
||||
def test_empty_applicability_and_non_callable_proposal_are_rejected() -> None:
|
||||
with pytest.raises(ModifierContractError, match="applicability"):
|
||||
Modifier(
|
||||
modifier_id="example.rule",
|
||||
version="1.0.0",
|
||||
parameters=(),
|
||||
applicability=" ",
|
||||
propose=no_changes,
|
||||
)
|
||||
with pytest.raises(ModifierContractError, match="callable"):
|
||||
Modifier(
|
||||
modifier_id="example.rule",
|
||||
version="1.0.0",
|
||||
parameters=(),
|
||||
applicability="测试。",
|
||||
propose=cast(object, None), # type: ignore[arg-type]
|
||||
)
|
||||
@@ -1,83 +0,0 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
|
||||
from mdpolish import Pipeline, RunStatus
|
||||
from mdpolish.component import ComponentContractError
|
||||
from mdpolish.components import PageBreakWordJoinComponent
|
||||
|
||||
MAPPINGS = (
|
||||
("possi-", "bly", "possibly"),
|
||||
("SOFA-", "based", "SOFA-based"),
|
||||
("threshold.", "olds", "thresholds"),
|
||||
)
|
||||
|
||||
|
||||
def transform(markdown: str): # type: ignore[no-untyped-def]
|
||||
return Pipeline([PageBreakWordJoinComponent(MAPPINGS)]).transform(markdown)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("markdown", "expected"),
|
||||
[
|
||||
("except possi-\nbly through care", "except possibly through care"),
|
||||
("except possi-\n\nbly through care", "except possibly through care"),
|
||||
("use SOFA-\nbased criteria", "use SOFA-based criteria"),
|
||||
("at threshold.\n\nolds of eight", "at thresholds of eight"),
|
||||
("except possi-\r\n\r\nbly now", "except possibly now"),
|
||||
("except possi-\r\rbly now", "except possibly now"),
|
||||
],
|
||||
)
|
||||
def test_applies_exact_mapping_across_approved_line_shapes(markdown: str, expected: str) -> None:
|
||||
result = transform(markdown)
|
||||
assert result.status is RunStatus.SUCCESS
|
||||
assert result.output_markdown == expected
|
||||
assert len(result.changes) == 1
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"markdown",
|
||||
[
|
||||
"except unknown-\nword here",
|
||||
"except possi-\n\n\nbly here",
|
||||
"except possi-\r\n\nbly here",
|
||||
"except POSSI-\nbly here",
|
||||
"except possi-\nblymore here",
|
||||
"except xSOFA-\nbased here",
|
||||
],
|
||||
)
|
||||
def test_unknown_or_unsafe_boundaries_are_preserved(markdown: str) -> None:
|
||||
assert transform(markdown).output_markdown == markdown
|
||||
|
||||
|
||||
def test_parameters_and_results_are_independent_of_mapping_order() -> None:
|
||||
first = PageBreakWordJoinComponent(MAPPINGS)
|
||||
second = PageBreakWordJoinComponent(reversed(MAPPINGS))
|
||||
markdown = "possibly becomes possi-\nbly"
|
||||
|
||||
first_result = Pipeline([first]).transform(markdown)
|
||||
second_result = Pipeline([second]).transform(markdown)
|
||||
|
||||
assert first_result.components == second_result.components
|
||||
assert first_result.output_markdown == second_result.output_markdown
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"mappings",
|
||||
[
|
||||
(("", "right", "word"),),
|
||||
(("left", "right", "two words"),),
|
||||
(("left", "right", "word"), ("left", "right", "other")),
|
||||
(("left", "right"),),
|
||||
],
|
||||
)
|
||||
def test_invalid_mappings_raise_contract_error(mappings: object) -> None:
|
||||
with pytest.raises(ComponentContractError):
|
||||
PageBreakWordJoinComponent(mappings) # type: ignore[arg-type]
|
||||
|
||||
|
||||
def test_successful_output_is_idempotent() -> None:
|
||||
pipeline = Pipeline([PageBreakWordJoinComponent(MAPPINGS)])
|
||||
first = pipeline.transform("except possi-\n\nbly here")
|
||||
assert first.output_markdown is not None
|
||||
assert pipeline.transform(first.output_markdown).changes == ()
|
||||
+121
-118
@@ -1,14 +1,13 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from collections.abc import Mapping
|
||||
from typing import cast
|
||||
|
||||
import pytest
|
||||
|
||||
from mdpolish import (
|
||||
Component,
|
||||
DocumentSnapshot,
|
||||
ErrorStage,
|
||||
Modifier,
|
||||
Pipeline,
|
||||
ProposedChange,
|
||||
RunStatus,
|
||||
@@ -17,97 +16,84 @@ from mdpolish import (
|
||||
)
|
||||
|
||||
|
||||
class ReplaceComponent(Component):
|
||||
def __init__(
|
||||
self,
|
||||
needle: str,
|
||||
replacement: str,
|
||||
*,
|
||||
component_id: str,
|
||||
version: str = "1.0.0",
|
||||
parameters: object = None,
|
||||
applicability: str = "处理精确测试字符串，要求完整匹配，排除其他内容。",
|
||||
) -> None:
|
||||
self.needle = needle
|
||||
self.replacement = replacement
|
||||
self._component_id = component_id
|
||||
self._version = version
|
||||
self._parameters = {"needle": needle, "replacement": replacement} if parameters is None else parameters
|
||||
self._applicability = applicability
|
||||
|
||||
@property
|
||||
def component_id(self) -> str:
|
||||
return self._component_id
|
||||
|
||||
@property
|
||||
def version(self) -> str:
|
||||
return self._version
|
||||
|
||||
@property
|
||||
def parameters(self) -> Mapping[str, object]:
|
||||
return cast(Mapping[str, object], self._parameters)
|
||||
|
||||
@property
|
||||
def applicability(self) -> str:
|
||||
return self._applicability
|
||||
|
||||
def _propose_changes(self, snapshot: DocumentSnapshot) -> tuple[ProposedChange, ...]:
|
||||
position = snapshot.markdown.find(self.needle)
|
||||
def replace_modifier(
|
||||
needle: str,
|
||||
replacement: str,
|
||||
*,
|
||||
modifier_id: str,
|
||||
version: str = "1.0.0",
|
||||
) -> 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(self.needle)),
|
||||
expected_text=self.needle,
|
||||
replacement=self.replacement,
|
||||
span=TextSpan(position, position + len(needle)),
|
||||
expected_text=needle,
|
||||
replacement=replacement,
|
||||
)
|
||||
return (
|
||||
ProposedChange(
|
||||
snapshot_sha256=snapshot.sha256,
|
||||
reason=f"replace test token for {self.component_id}",
|
||||
reason=f"replace test token for {modifier_id}",
|
||||
edits=(edit,),
|
||||
),
|
||||
)
|
||||
|
||||
return Modifier(
|
||||
modifier_id=modifier_id,
|
||||
version=version,
|
||||
parameters={"needle": needle, "replacement": replacement},
|
||||
applicability="处理精确测试字符串；排除其他内容。",
|
||||
propose=propose,
|
||||
)
|
||||
|
||||
class ExplodingComponent(ReplaceComponent):
|
||||
def __init__(self, *, trigger: str | None = None, component_id: str = "test.exploding") -> None:
|
||||
super().__init__("unused", "unused-replacement", component_id=component_id)
|
||||
self.trigger = trigger
|
||||
|
||||
def _propose_changes(self, snapshot: DocumentSnapshot) -> tuple[ProposedChange, ...]:
|
||||
if self.trigger is None or snapshot.markdown == self.trigger:
|
||||
def exploding_modifier(*, trigger: str | None = None, modifier_id: str = "test.exploding") -> Modifier:
|
||||
def propose(snapshot: DocumentSnapshot) -> tuple[ProposedChange, ...]:
|
||||
if trigger is None or snapshot.markdown == trigger:
|
||||
raise RuntimeError(f"SECRET source: {snapshot.markdown}")
|
||||
return ()
|
||||
|
||||
|
||||
class InvalidReturnComponent(ReplaceComponent):
|
||||
def __init__(self) -> None:
|
||||
super().__init__("a", "A", component_id="test.invalid-return")
|
||||
|
||||
def _propose_changes(self, snapshot: DocumentSnapshot) -> tuple[ProposedChange, ...]:
|
||||
return cast(tuple[ProposedChange, ...], [])
|
||||
return Modifier(
|
||||
modifier_id=modifier_id,
|
||||
version="1.0.0",
|
||||
parameters={"trigger": trigger},
|
||||
applicability="仅用于测试异常路径。",
|
||||
propose=propose,
|
||||
)
|
||||
|
||||
|
||||
class StaleProposalComponent(ReplaceComponent):
|
||||
def __init__(self) -> None:
|
||||
super().__init__("a", "A", component_id="test.stale")
|
||||
|
||||
def _propose_changes(self, snapshot: DocumentSnapshot) -> tuple[ProposedChange, ...]:
|
||||
def stale_proposal_modifier() -> Modifier:
|
||||
def propose(snapshot: DocumentSnapshot) -> tuple[ProposedChange, ...]:
|
||||
stale_snapshot = DocumentSnapshot(snapshot.markdown + "!")
|
||||
edit = TextEdit(stale_snapshot.sha256, TextSpan(0, 1), stale_snapshot.markdown[0], "X")
|
||||
return (ProposedChange(stale_snapshot.sha256, "stale test proposal", (edit,)),)
|
||||
|
||||
return Modifier(
|
||||
modifier_id="test.stale",
|
||||
version="1.0.0",
|
||||
parameters=(),
|
||||
applicability="仅用于测试过期快照。",
|
||||
propose=propose,
|
||||
)
|
||||
|
||||
class OverlapComponent(ReplaceComponent):
|
||||
def __init__(self) -> None:
|
||||
super().__init__("a", "A", component_id="test.overlap")
|
||||
|
||||
def _propose_changes(self, snapshot: DocumentSnapshot) -> tuple[ProposedChange, ...]:
|
||||
def overlap_modifier() -> Modifier:
|
||||
def propose(snapshot: DocumentSnapshot) -> tuple[ProposedChange, ...]:
|
||||
first = TextEdit(snapshot.sha256, TextSpan(0, 3), snapshot.markdown[0:3], "X")
|
||||
second = TextEdit(snapshot.sha256, TextSpan(2, 4), snapshot.markdown[2:4], "Y")
|
||||
return (ProposedChange(snapshot.sha256, "overlapping test proposal", (first, second)),)
|
||||
|
||||
return Modifier(
|
||||
modifier_id="test.overlap",
|
||||
version="1.0.0",
|
||||
parameters=(),
|
||||
applicability="仅用于测试冲突范围。",
|
||||
propose=propose,
|
||||
)
|
||||
|
||||
|
||||
def test_empty_pipeline_returns_unchanged_success_for_empty_unicode_text() -> None:
|
||||
for markdown in ("", "中文\nCafe\u0301\n🙂"):
|
||||
@@ -117,14 +103,14 @@ def test_empty_pipeline_returns_unchanged_success_for_empty_unicode_text() -> No
|
||||
assert result.output_markdown == markdown
|
||||
assert result.partial_markdown is None
|
||||
assert result.changes == ()
|
||||
assert result.components == ()
|
||||
assert result.modifiers == ()
|
||||
|
||||
|
||||
def test_later_component_reads_snapshot_produced_by_earlier_component() -> None:
|
||||
def test_later_modifier_reads_snapshot_produced_by_earlier_modifier() -> None:
|
||||
pipeline = Pipeline(
|
||||
[
|
||||
ReplaceComponent("初", "中", component_id="test.first"),
|
||||
ReplaceComponent("中", "终", component_id="test.second"),
|
||||
replace_modifier("初", "中", modifier_id="test.first"),
|
||||
replace_modifier("中", "终", modifier_id="test.second"),
|
||||
]
|
||||
)
|
||||
|
||||
@@ -132,15 +118,15 @@ def test_later_component_reads_snapshot_produced_by_earlier_component() -> None:
|
||||
|
||||
assert result.status is RunStatus.SUCCESS
|
||||
assert result.output_markdown == "终"
|
||||
assert [change.component_id for change in result.changes] == ["test.first", "test.second"]
|
||||
assert [change.modifier_id for change in result.changes] == ["test.first", "test.second"]
|
||||
assert result.changes[1].before_sha256 == result.changes[0].after_sha256
|
||||
|
||||
|
||||
def test_same_input_components_and_parameters_produce_same_ordered_result() -> None:
|
||||
def test_same_input_modifiers_and_parameters_produce_same_ordered_result() -> None:
|
||||
pipeline = Pipeline(
|
||||
[
|
||||
ReplaceComponent("a", "b", component_id="test.first"),
|
||||
ReplaceComponent("b", "c", component_id="test.second"),
|
||||
replace_modifier("a", "b", modifier_id="test.first"),
|
||||
replace_modifier("b", "c", modifier_id="test.second"),
|
||||
]
|
||||
)
|
||||
|
||||
@@ -148,7 +134,7 @@ def test_same_input_components_and_parameters_produce_same_ordered_result() -> N
|
||||
|
||||
|
||||
def test_successful_pipeline_is_idempotent_on_its_output() -> None:
|
||||
pipeline = Pipeline([ReplaceComponent("old", "new", component_id="test.replace")])
|
||||
pipeline = Pipeline([replace_modifier("old", "new", modifier_id="test.replace")])
|
||||
|
||||
first = pipeline.transform("old value")
|
||||
assert first.status is RunStatus.SUCCESS
|
||||
@@ -161,22 +147,12 @@ def test_successful_pipeline_is_idempotent_on_its_output() -> None:
|
||||
assert second.changes == ()
|
||||
|
||||
|
||||
def test_single_component_runs_through_pipeline_without_shortcut() -> None:
|
||||
component = ReplaceComponent("a", "A", component_id="test.single")
|
||||
|
||||
result = Pipeline([component]).transform("a")
|
||||
|
||||
assert result.status is RunStatus.SUCCESS
|
||||
assert result.output_markdown == "A"
|
||||
assert len(result.changes) == 1
|
||||
|
||||
|
||||
def test_transform_error_stops_later_components_and_keeps_only_partial_text() -> None:
|
||||
def test_transform_error_stops_later_modifiers_and_keeps_only_partial_text() -> None:
|
||||
pipeline = Pipeline(
|
||||
[
|
||||
ReplaceComponent("a", "b", component_id="test.first"),
|
||||
ExplodingComponent(),
|
||||
ReplaceComponent("b", "c", component_id="test.never-runs"),
|
||||
replace_modifier("a", "b", modifier_id="test.first"),
|
||||
exploding_modifier(),
|
||||
replace_modifier("b", "c", modifier_id="test.never-runs"),
|
||||
]
|
||||
)
|
||||
|
||||
@@ -185,14 +161,14 @@ def test_transform_error_stops_later_components_and_keeps_only_partial_text() ->
|
||||
assert result.status is RunStatus.FAILED
|
||||
assert result.output_markdown is None
|
||||
assert result.partial_markdown == "b"
|
||||
assert [change.component_id for change in result.changes] == ["test.first"]
|
||||
assert [change.modifier_id for change in result.changes] == ["test.first"]
|
||||
assert len(result.errors) == 1
|
||||
assert result.errors[0].stage is ErrorStage.TRANSFORM
|
||||
assert result.residual_proposals == ()
|
||||
|
||||
|
||||
def test_unexpected_component_error_does_not_leak_source_or_exception_message() -> None:
|
||||
result = Pipeline([ExplodingComponent()]).transform("private markdown")
|
||||
def test_unexpected_modifier_error_does_not_leak_source_or_exception_message() -> None:
|
||||
result = Pipeline([exploding_modifier()]).transform("private markdown")
|
||||
|
||||
assert result.status is RunStatus.FAILED
|
||||
assert result.errors[0].error_type == "RuntimeError"
|
||||
@@ -201,17 +177,27 @@ def test_unexpected_component_error_does_not_leak_source_or_exception_message()
|
||||
|
||||
|
||||
def test_invalid_proposal_return_is_a_transform_contract_failure() -> None:
|
||||
result = Pipeline([InvalidReturnComponent()]).transform("abc")
|
||||
def propose(snapshot: DocumentSnapshot) -> tuple[ProposedChange, ...]:
|
||||
return cast(tuple[ProposedChange, ...], [])
|
||||
|
||||
modifier = Modifier(
|
||||
modifier_id="test.invalid-return",
|
||||
version="1.0.0",
|
||||
parameters=(),
|
||||
applicability="仅用于测试非法返回值。",
|
||||
propose=propose,
|
||||
)
|
||||
result = Pipeline([modifier]).transform("abc")
|
||||
|
||||
assert result.status is RunStatus.FAILED
|
||||
assert result.partial_markdown == "abc"
|
||||
assert result.errors[0].error_type == "ComponentContractError"
|
||||
assert result.errors[0].error_type == "ModifierContractError"
|
||||
assert result.errors[0].stage is ErrorStage.TRANSFORM
|
||||
|
||||
|
||||
@pytest.mark.parametrize("component", [StaleProposalComponent(), OverlapComponent()])
|
||||
def test_invalid_edit_batch_fails_atomically(component: Component) -> None:
|
||||
result = Pipeline([component]).transform("abcd")
|
||||
@pytest.mark.parametrize("modifier", [stale_proposal_modifier(), overlap_modifier()])
|
||||
def test_invalid_edit_batch_fails_atomically(modifier: Modifier) -> None:
|
||||
result = Pipeline([modifier]).transform("abcd")
|
||||
|
||||
assert result.status is RunStatus.FAILED
|
||||
assert result.partial_markdown == "abcd"
|
||||
@@ -219,11 +205,11 @@ def test_invalid_edit_batch_fails_atomically(component: Component) -> None:
|
||||
assert result.errors[0].error_type == "EditValidationError"
|
||||
|
||||
|
||||
def test_duplicate_component_ids_fail_during_preflight_before_modification() -> None:
|
||||
def test_duplicate_modifier_ids_fail_during_preflight_before_modification() -> None:
|
||||
pipeline = Pipeline(
|
||||
[
|
||||
ReplaceComponent("a", "b", component_id="test.duplicate"),
|
||||
ReplaceComponent("b", "c", component_id="test.duplicate"),
|
||||
replace_modifier("a", "b", modifier_id="test.duplicate"),
|
||||
replace_modifier("b", "c", modifier_id="test.duplicate"),
|
||||
]
|
||||
)
|
||||
|
||||
@@ -235,28 +221,45 @@ def test_duplicate_component_ids_fail_during_preflight_before_modification() ->
|
||||
assert result.errors[0].error_type == "PipelineContractError"
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"component",
|
||||
[
|
||||
ReplaceComponent("a", "b", component_id="test.bad-version", version="1.0"),
|
||||
ReplaceComponent("a", "b", component_id="test.bad-parameters", parameters={"bad": {1}}),
|
||||
ReplaceComponent("a", "b", component_id="test.bad-applicability", applicability=""),
|
||||
],
|
||||
)
|
||||
def test_invalid_component_metadata_fails_before_modification(component: Component) -> None:
|
||||
result = Pipeline([component]).transform("a")
|
||||
def test_non_modifier_entry_fails_during_preflight() -> None:
|
||||
pipeline = Pipeline(cast(list[Modifier], [object()]))
|
||||
|
||||
result = pipeline.transform("a")
|
||||
|
||||
assert result.status is RunStatus.FAILED
|
||||
assert result.partial_markdown == "a"
|
||||
assert result.errors[0].error_type == "ModifierContractError"
|
||||
|
||||
|
||||
def test_metadata_changed_by_proposal_function_fails_before_application() -> None:
|
||||
modifier: Modifier
|
||||
|
||||
def propose(snapshot: DocumentSnapshot) -> tuple[ProposedChange, ...]:
|
||||
object.__setattr__(modifier, "version", "2.0.0")
|
||||
edit = TextEdit(snapshot.sha256, TextSpan(0, 1), "a", "A")
|
||||
return (ProposedChange(snapshot.sha256, "mutating metadata", (edit,)),)
|
||||
|
||||
modifier = Modifier(
|
||||
modifier_id="test.mutating-metadata",
|
||||
version="1.0.0",
|
||||
parameters=(),
|
||||
applicability="仅用于测试运行期元数据变化。",
|
||||
propose=propose,
|
||||
)
|
||||
|
||||
result = Pipeline([modifier]).transform("a")
|
||||
|
||||
assert result.status is RunStatus.FAILED
|
||||
assert result.partial_markdown == "a"
|
||||
assert result.changes == ()
|
||||
assert result.errors[0].stage is ErrorStage.TRANSFORM
|
||||
assert result.errors[0].error_type == "ModifierContractError"
|
||||
|
||||
|
||||
def test_cross_component_chain_is_reported_unstable_without_a_second_round() -> None:
|
||||
def test_cross_modifier_chain_is_reported_unstable_without_a_second_round() -> None:
|
||||
pipeline = Pipeline(
|
||||
[
|
||||
ReplaceComponent("bad", "good", component_id="test.to-good"),
|
||||
ReplaceComponent("good", "bad", component_id="test.to-bad"),
|
||||
replace_modifier("bad", "good", modifier_id="test.to-good"),
|
||||
replace_modifier("good", "bad", modifier_id="test.to-bad"),
|
||||
]
|
||||
)
|
||||
|
||||
@@ -267,16 +270,16 @@ def test_cross_component_chain_is_reported_unstable_without_a_second_round() ->
|
||||
assert result.partial_markdown == "bad"
|
||||
assert len(result.changes) == 2
|
||||
assert len(result.residual_proposals) == 1
|
||||
assert result.residual_proposals[0].component_id == "test.to-good"
|
||||
assert result.residual_proposals[0].modifier_id == "test.to-good"
|
||||
assert result.errors == ()
|
||||
|
||||
|
||||
def test_final_review_continues_after_error_and_keeps_valid_residual_proposal() -> None:
|
||||
pipeline = Pipeline(
|
||||
[
|
||||
ExplodingComponent(trigger="done", component_id="test.review-error"),
|
||||
ReplaceComponent("done", "clean", component_id="test.residual"),
|
||||
ReplaceComponent("start", "done", component_id="test.producer"),
|
||||
exploding_modifier(trigger="done", modifier_id="test.review-error"),
|
||||
replace_modifier("done", "clean", modifier_id="test.residual"),
|
||||
replace_modifier("start", "done", modifier_id="test.producer"),
|
||||
]
|
||||
)
|
||||
|
||||
@@ -287,7 +290,7 @@ def test_final_review_continues_after_error_and_keeps_valid_residual_proposal()
|
||||
assert result.partial_markdown == "done"
|
||||
assert len(result.errors) == 1
|
||||
assert result.errors[0].stage is ErrorStage.FINAL_REVIEW
|
||||
assert result.errors[0].component_id == "test.review-error"
|
||||
assert result.errors[0].modifier_id == "test.review-error"
|
||||
assert len(result.residual_proposals) == 1
|
||||
assert result.residual_proposals[0].component_id == "test.residual"
|
||||
assert result.residual_proposals[0].modifier_id == "test.residual"
|
||||
assert result.residual_proposals[0].proposal_ref.snapshot_sha256 == result.current_sha256
|
||||
|
||||
@@ -1,65 +0,0 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
|
||||
from mdpolish import Pipeline, RunStatus
|
||||
from mdpolish.components import ReferenceSpacingComponent
|
||||
|
||||
|
||||
def transform(markdown: str): # type: ignore[no-untyped-def]
|
||||
return Pipeline([ReferenceSpacingComponent()]).transform(markdown)
|
||||
|
||||
|
||||
def test_normalizes_missing_and_extra_blank_lines_in_references_only() -> None:
|
||||
markdown = "1. Method\n2. Method\n\n## REFERENCES\n\n1. First\n2. Second\n\n\n3. Third"
|
||||
result = transform(markdown)
|
||||
|
||||
assert result.status is RunStatus.SUCCESS
|
||||
assert result.output_markdown == "1. Method\n2. Method\n\n## REFERENCES\n\n1. First\n\n2. Second\n\n3. Third"
|
||||
assert len(result.changes) == 2
|
||||
|
||||
|
||||
@pytest.mark.parametrize("heading", ["# References", "## REFERENCES", "### references", "#### ReFeReNcEs"])
|
||||
def test_accepts_exact_references_heading_with_ascii_case_folding(heading: str) -> None:
|
||||
markdown = f"{heading}\n\n1. First\n2. Second"
|
||||
assert transform(markdown).output_markdown == f"{heading}\n\n1. First\n\n2. Second"
|
||||
|
||||
|
||||
@pytest.mark.parametrize("line_ending", ["\n", "\r\n", "\r"])
|
||||
def test_preserves_line_ending_style(line_ending: str) -> None:
|
||||
markdown = line_ending.join(("## References", "", "1. First", "2. Second"))
|
||||
expected = line_ending.join(("## References", "", "1. First", "", "2. Second"))
|
||||
assert transform(markdown).output_markdown == expected
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"markdown",
|
||||
[
|
||||
"## Reference\n\n1. First\n2. Second",
|
||||
"## References\n\n1. First\n3. Third",
|
||||
"## References\n\n2. Second\n3. Third",
|
||||
"## References\n\n1. First\n### Subsection\n2. Second",
|
||||
"## References\r\n\r\n1. First\r\n\n2. Second",
|
||||
],
|
||||
)
|
||||
def test_ambiguous_or_mixed_sections_are_preserved(markdown: str) -> None:
|
||||
assert transform(markdown).output_markdown == markdown
|
||||
|
||||
|
||||
def test_same_or_higher_heading_ends_section() -> None:
|
||||
markdown = "## References\n\n1. First\n2. Second\n\n## Appendix\n\n1. Keep\n2. Keep"
|
||||
result = transform(markdown)
|
||||
assert result.output_markdown == "## References\n\n1. First\n\n2. Second\n\n## Appendix\n\n1. Keep\n2. Keep"
|
||||
|
||||
|
||||
def test_multiline_reference_uses_its_last_text_line_as_boundary() -> None:
|
||||
markdown = "## References\n\n1. First line\ncontinuation\n2. Second"
|
||||
result = transform(markdown)
|
||||
assert result.output_markdown == "## References\n\n1. First line\ncontinuation\n\n2. Second"
|
||||
|
||||
|
||||
def test_successful_output_is_idempotent() -> None:
|
||||
pipeline = Pipeline([ReferenceSpacingComponent()])
|
||||
first = pipeline.transform("## References\n\n1. First\n2. Second")
|
||||
assert first.output_markdown is not None
|
||||
assert pipeline.transform(first.output_markdown).changes == ()
|
||||
@@ -0,0 +1,108 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
|
||||
import pytest
|
||||
|
||||
from mdpolish import Modifier, ModifierContractError, Pipeline, RunStatus, regex_replace
|
||||
|
||||
|
||||
def test_factory_returns_modifier_and_expands_capture_groups() -> None:
|
||||
modifier = regex_replace(
|
||||
modifier_id="example.swap-date",
|
||||
version="1.0.0",
|
||||
pattern=r"(\d{4})-(\d{2})-(\d{2})",
|
||||
replacement=r"\3/\2/\1",
|
||||
applicability="处理 ISO 形状的虚构日期；不验证真实日历日期。",
|
||||
)
|
||||
|
||||
result = Pipeline([modifier]).transform("A 2026-08-26, B 2027-01-02")
|
||||
|
||||
assert isinstance(modifier, Modifier)
|
||||
assert result.status is RunStatus.SUCCESS
|
||||
assert result.output_markdown == "A 26/08/2026, B 02/01/2027"
|
||||
assert len(result.changes) == 2
|
||||
assert result.changes[0].before == "2026-08-26"
|
||||
assert result.changes[0].after == "26/08/2026"
|
||||
|
||||
|
||||
def test_flags_and_factory_parameters_are_recorded() -> None:
|
||||
modifier = regex_replace(
|
||||
modifier_id="example.case",
|
||||
version="2.0.0",
|
||||
pattern="token",
|
||||
replacement="value",
|
||||
flags=re.IGNORECASE | re.MULTILINE,
|
||||
)
|
||||
|
||||
result = Pipeline([modifier]).transform("TOKEN")
|
||||
|
||||
assert result.status is RunStatus.SUCCESS
|
||||
assert result.modifiers[0].modifier_id == modifier.modifier_id
|
||||
assert result.modifiers[0].parameters == modifier.parameters
|
||||
assert dict(modifier.parameters) == {
|
||||
"flags": int(re.IGNORECASE | re.MULTILINE),
|
||||
"pattern": "token",
|
||||
"replacement": "value",
|
||||
}
|
||||
|
||||
|
||||
def test_zero_match_and_no_op_replacement_are_stable() -> None:
|
||||
missing = Pipeline(
|
||||
[regex_replace(modifier_id="example.missing", version="1.0.0", pattern="missing", replacement="new")]
|
||||
).transform("text")
|
||||
no_op = Pipeline(
|
||||
[regex_replace(modifier_id="example.no-op", version="1.0.0", pattern="text", replacement="text")]
|
||||
).transform("text")
|
||||
|
||||
assert missing.output_markdown == "text"
|
||||
assert missing.changes == ()
|
||||
assert no_op.status is RunStatus.SUCCESS
|
||||
assert no_op.changes == ()
|
||||
|
||||
|
||||
@pytest.mark.parametrize("pattern", [r"", r"^", r"a*", r"(?=a)"])
|
||||
def test_zero_length_matches_are_rejected(pattern: str) -> None:
|
||||
if pattern == r"(?=a)":
|
||||
modifier = regex_replace(
|
||||
modifier_id="example.zero",
|
||||
version="1.0.0",
|
||||
pattern=pattern,
|
||||
replacement="x",
|
||||
)
|
||||
result = Pipeline([modifier]).transform("a")
|
||||
assert result.status is RunStatus.FAILED
|
||||
assert result.partial_markdown == "a"
|
||||
assert result.changes == ()
|
||||
return
|
||||
|
||||
with pytest.raises(ModifierContractError, match="zero-length"):
|
||||
regex_replace(
|
||||
modifier_id="example.zero",
|
||||
version="1.0.0",
|
||||
pattern=pattern,
|
||||
replacement="x",
|
||||
)
|
||||
|
||||
|
||||
def test_invalid_pattern_flags_and_input_types_are_rejected() -> None:
|
||||
with pytest.raises(ModifierContractError, match="compile"):
|
||||
regex_replace(modifier_id="example.invalid", version="1.0.0", pattern="(", replacement="x")
|
||||
with pytest.raises(ModifierContractError, match="flags"):
|
||||
regex_replace(
|
||||
modifier_id="example.flags",
|
||||
version="1.0.0",
|
||||
pattern="x",
|
||||
replacement="y",
|
||||
flags=True,
|
||||
)
|
||||
|
||||
|
||||
def test_successful_output_is_idempotent() -> None:
|
||||
pipeline = Pipeline(
|
||||
[regex_replace(modifier_id="example.space", version="1.0.0", pattern=r" {2,}", replacement=" ")]
|
||||
)
|
||||
first = pipeline.transform("a b")
|
||||
assert first.output_markdown == "a b"
|
||||
assert first.output_markdown is not None
|
||||
assert pipeline.transform(first.output_markdown).changes == ()
|
||||
@@ -1,64 +0,0 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
|
||||
from mdpolish import Pipeline, RunStatus
|
||||
from mdpolish.components import RepeatedRunningHeaderComponent
|
||||
|
||||
HEADER = "## Repeated Paper Header"
|
||||
|
||||
|
||||
def transform(markdown: str): # type: ignore[no-untyped-def]
|
||||
return Pipeline([RepeatedRunningHeaderComponent()]).transform(markdown)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("line_ending", ["\n", "\r\n", "\r"])
|
||||
def test_bridges_interrupted_sentence_and_deletes_other_occurrence(line_ending: str) -> None:
|
||||
markdown = line_ending.join(
|
||||
(
|
||||
"Sentence continues in",
|
||||
"",
|
||||
HEADER,
|
||||
"",
|
||||
"the next line.",
|
||||
"",
|
||||
"18. Reference",
|
||||
"",
|
||||
HEADER,
|
||||
"",
|
||||
"19. Reference",
|
||||
)
|
||||
)
|
||||
result = transform(markdown)
|
||||
|
||||
assert result.status is RunStatus.SUCCESS
|
||||
assert result.output_markdown == line_ending.join(
|
||||
("Sentence continues in the next line.", "", "18. Reference", "", "19. Reference")
|
||||
)
|
||||
assert len(result.changes) == 2
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"markdown",
|
||||
[
|
||||
f"before\n\n{HEADER}\n\nafter",
|
||||
f"Sentence ends.\n\n{HEADER}\n\nAfter\n\nAnother sentence.\n\n{HEADER}\n\nOther",
|
||||
f"Sentence continues\n\n{HEADER}\n\nAfter\n\nText ends.\n\n{HEADER}\n\nOther",
|
||||
f"Sentence continues\n\n{HEADER}\n\nafter\n{HEADER}\nnot blank",
|
||||
],
|
||||
)
|
||||
def test_missing_or_unsafe_group_evidence_preserves_document(markdown: str) -> None:
|
||||
assert transform(markdown).output_markdown == markdown
|
||||
|
||||
|
||||
def test_header_matching_is_exact_and_not_keyword_based() -> None:
|
||||
markdown = "continues\n\n## Any Header\n\nfrom here\n\n18. Ref\n\n## Any Header\n\n19. Ref"
|
||||
result = transform(markdown)
|
||||
assert result.output_markdown == "continues from here\n\n18. Ref\n\n19. Ref"
|
||||
|
||||
|
||||
def test_successful_output_is_idempotent() -> None:
|
||||
pipeline = Pipeline([RepeatedRunningHeaderComponent()])
|
||||
first = pipeline.transform(f"continues\n\n{HEADER}\n\nfrom here\n\ntext\n\n{HEADER}\n\nend")
|
||||
assert first.output_markdown is not None
|
||||
assert pipeline.transform(first.output_markdown).changes == ()
|
||||
@@ -1,177 +0,0 @@
|
||||
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,
|
||||
)
|
||||
@@ -1,132 +0,0 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
from typing import Any, cast
|
||||
|
||||
import pytest
|
||||
|
||||
from reviewer.server import ReviewArtifactError, ReviewArtifacts
|
||||
from tests.reviewer_fixture import create_review_run, read_json, write_json
|
||||
|
||||
|
||||
def test_reviewer_returns_full_comparison_and_all_component_stages(tmp_path: Path) -> None:
|
||||
fixture = create_review_run(tmp_path)
|
||||
repository = ReviewArtifacts(fixture["run_directory"])
|
||||
|
||||
run = cast(dict[str, Any], repository.run_summary())
|
||||
comparison = cast(dict[str, Any], repository.document_comparison("paper"))
|
||||
changed = cast(dict[str, Any], repository.component_stage("paper", 0))
|
||||
unchanged = cast(dict[str, Any], repository.component_stage("paper", 1))
|
||||
|
||||
assert run["summary"] == {
|
||||
"document_count": 1,
|
||||
"success_count": 1,
|
||||
"failed_count": 0,
|
||||
"unstable_count": 0,
|
||||
"change_count": 1,
|
||||
}
|
||||
assert run["documents"][0]["source_available"] is True
|
||||
assert comparison["original_markdown"] == fixture["original"]
|
||||
assert comparison["cleaned_markdown"] == fixture["cleaned"]
|
||||
assert comparison["changes"][0]["editor_range"] == {"start": 4, "end": 7}
|
||||
assert changed["before_markdown"] == fixture["original"]
|
||||
assert changed["after_markdown"] == fixture["cleaned"]
|
||||
assert unchanged["before_markdown"] == fixture["cleaned"]
|
||||
assert unchanged["after_markdown"] == fixture["cleaned"]
|
||||
assert unchanged["changes"] == []
|
||||
|
||||
|
||||
@pytest.mark.parametrize("status", ["failed", "unstable"])
|
||||
def test_non_success_documents_expose_audit_but_no_formal_output(tmp_path: Path, status: str) -> None:
|
||||
fixture = create_review_run(tmp_path, status=status)
|
||||
repository = ReviewArtifacts(fixture["run_directory"])
|
||||
|
||||
comparison = cast(dict[str, Any], repository.document_comparison("paper"))
|
||||
|
||||
assert comparison["original_markdown"] == fixture["original"]
|
||||
assert comparison["cleaned_markdown"] is None
|
||||
if status == "failed":
|
||||
assert comparison["errors"]
|
||||
assert comparison["residual_proposals"] == []
|
||||
else:
|
||||
assert comparison["errors"] == []
|
||||
assert comparison["residual_proposals"]
|
||||
with pytest.raises(ReviewArtifactError, match="success"):
|
||||
repository.component_stage("paper", 0)
|
||||
|
||||
|
||||
def test_historical_run_without_locator_reports_unavailable_source(tmp_path: Path) -> None:
|
||||
fixture = create_review_run(tmp_path, with_locator=False)
|
||||
repository = ReviewArtifacts(fixture["run_directory"])
|
||||
|
||||
summary = cast(dict[str, Any], repository.run_summary())
|
||||
|
||||
assert summary["documents"][0]["source_available"] is False
|
||||
assert "review-locator.json" in summary["documents"][0]["availability_error"]
|
||||
comparison = cast(dict[str, Any], repository.document_comparison("paper"))
|
||||
assert comparison["original_markdown"] is None
|
||||
assert comparison["cleaned_markdown"] == fixture["cleaned"]
|
||||
assert comparison["changes"][0]["editor_range"] is None
|
||||
with pytest.raises(ReviewArtifactError, match=r"review-locator\.json"):
|
||||
repository.component_stage("paper", 0)
|
||||
|
||||
|
||||
def test_source_hash_change_is_not_silently_displayed(tmp_path: Path) -> None:
|
||||
fixture = create_review_run(tmp_path)
|
||||
repository = ReviewArtifacts(fixture["run_directory"])
|
||||
source_path = fixture["source_path"]
|
||||
assert isinstance(source_path, Path)
|
||||
source_path.write_text("changed\n", encoding="utf-8")
|
||||
|
||||
summary = cast(dict[str, Any], repository.run_summary())
|
||||
|
||||
assert summary["documents"][0]["source_available"] is False
|
||||
assert "哈希" in summary["documents"][0]["availability_error"]
|
||||
comparison = cast(dict[str, Any], repository.document_comparison("paper"))
|
||||
assert comparison["original_markdown"] is None
|
||||
assert comparison["changes"][0]["editor_range"] is None
|
||||
with pytest.raises(ReviewArtifactError, match="哈希"):
|
||||
repository.component_stage("paper", 0)
|
||||
|
||||
|
||||
def test_reviewer_rejects_unknown_schema_and_artifact_path_escape(tmp_path: Path) -> None:
|
||||
fixture = create_review_run(tmp_path)
|
||||
manifest_path = fixture["manifest_path"]
|
||||
assert isinstance(manifest_path, Path)
|
||||
manifest = read_json(manifest_path)
|
||||
manifest["schema_version"] = 2
|
||||
write_json(manifest_path, manifest)
|
||||
|
||||
with pytest.raises(ReviewArtifactError) as unknown:
|
||||
ReviewArtifacts(fixture["run_directory"])
|
||||
assert unknown.value.code == "unsupported_schema"
|
||||
|
||||
second = create_review_run(tmp_path / "second")
|
||||
second_manifest_path = second["manifest_path"]
|
||||
assert isinstance(second_manifest_path, Path)
|
||||
second_manifest = read_json(second_manifest_path)
|
||||
second_manifest["documents"][0]["result_path"] = "../outside.json"
|
||||
write_json(second_manifest_path, second_manifest)
|
||||
with pytest.raises(ReviewArtifactError, match="路径"):
|
||||
ReviewArtifacts(second["run_directory"])
|
||||
|
||||
|
||||
def test_reviewer_rejects_tampered_snapshot_chain_and_unknown_identity(tmp_path: Path) -> None:
|
||||
fixture = create_review_run(tmp_path)
|
||||
result_path = fixture["result_path"]
|
||||
assert isinstance(result_path, Path)
|
||||
result = read_json(result_path)
|
||||
result["changes"][0]["after_sha256"] = "0" * 64
|
||||
write_json(result_path, result)
|
||||
repository = ReviewArtifacts(fixture["run_directory"])
|
||||
|
||||
with pytest.raises(ReviewArtifactError) as replay_error:
|
||||
repository.document_comparison("paper")
|
||||
assert replay_error.value.code == "untrusted_replay"
|
||||
|
||||
with pytest.raises(ReviewArtifactError) as document_error:
|
||||
repository.document_comparison("missing")
|
||||
assert document_error.value.http_status == 404
|
||||
with pytest.raises(ReviewArtifactError) as component_error:
|
||||
repository.component_stage("paper", 99)
|
||||
assert component_error.value.http_status == 404
|
||||
@@ -1,105 +0,0 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import threading
|
||||
from collections.abc import Generator
|
||||
from http.client import HTTPConnection, HTTPResponse
|
||||
from pathlib import Path
|
||||
from typing import Any, cast
|
||||
|
||||
import pytest
|
||||
|
||||
from reviewer.server import ReviewArtifacts
|
||||
from reviewer.server.__main__ import create_server
|
||||
from tests.reviewer_fixture import ReviewFixture, create_review_run
|
||||
|
||||
|
||||
def request(
|
||||
port: int,
|
||||
method: str,
|
||||
path: str,
|
||||
*,
|
||||
headers: dict[str, str] | None = None,
|
||||
) -> tuple[HTTPResponse, bytes]:
|
||||
connection = HTTPConnection("127.0.0.1", port, timeout=3)
|
||||
connection.request(method, path, headers=headers or {})
|
||||
response = connection.getresponse()
|
||||
content = response.read()
|
||||
connection.close()
|
||||
return response, content
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def running_server(tmp_path: Path) -> Generator[tuple[int, ReviewFixture], None, None]:
|
||||
fixture = create_review_run(tmp_path)
|
||||
static_root = tmp_path / "static"
|
||||
(static_root / "assets").mkdir(parents=True)
|
||||
(static_root / "index.html").write_text("<main>reviewer</main>\n", encoding="utf-8")
|
||||
(static_root / "assets/app.js").write_text("export {};\n", encoding="utf-8")
|
||||
repository = ReviewArtifacts(fixture["run_directory"])
|
||||
server = create_server(repository, static_root)
|
||||
thread = threading.Thread(target=server.serve_forever, daemon=True)
|
||||
thread.start()
|
||||
try:
|
||||
yield server.server_address[1], fixture
|
||||
finally:
|
||||
server.shutdown()
|
||||
server.server_close()
|
||||
thread.join(timeout=3)
|
||||
|
||||
|
||||
def test_server_exposes_same_origin_api_and_static_build(
|
||||
running_server: tuple[int, ReviewFixture],
|
||||
) -> None:
|
||||
port, fixture = running_server
|
||||
|
||||
api_response, api_content = request(port, "GET", "/api/v1/run")
|
||||
page_response, page_content = request(port, "GET", "/")
|
||||
asset_response, _ = request(port, "HEAD", "/assets/app.js")
|
||||
|
||||
payload = cast(dict[str, Any], json.loads(api_content))
|
||||
assert api_response.status == 200
|
||||
assert payload["run"]["run_id"] == "review-run"
|
||||
assert str(fixture["source_path"]) not in api_content.decode()
|
||||
assert api_response.getheader("Access-Control-Allow-Origin") is None
|
||||
assert api_response.getheader("Cache-Control") == "no-store"
|
||||
assert "default-src 'self'" in cast(str, api_response.getheader("Content-Security-Policy"))
|
||||
assert page_response.status == 200
|
||||
assert page_content == b"<main>reviewer</main>\n"
|
||||
assert asset_response.status == 200
|
||||
assert asset_response.getheader("Content-Type") == "text/javascript; charset=utf-8"
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("method", "path", "headers", "status", "code"),
|
||||
[
|
||||
("POST", "/api/v1/run", None, 405, "method_not_allowed"),
|
||||
("GET", "/api/v1/run", {"Host": "example.com"}, 403, "invalid_origin"),
|
||||
(
|
||||
"GET",
|
||||
"/api/v1/run",
|
||||
{"Origin": "http://example.com"},
|
||||
403,
|
||||
"invalid_origin",
|
||||
),
|
||||
("GET", "/api/v1/documents/missing", None, 404, "unknown_document"),
|
||||
("GET", "/api/v1/documents/paper/components/99", None, 404, "unknown_component"),
|
||||
("GET", "/assets/missing.js", None, 404, "not_found"),
|
||||
("GET", "/..%2Fsecret.txt", None, 404, "not_found"),
|
||||
],
|
||||
)
|
||||
def test_server_rejects_unsafe_or_unknown_requests(
|
||||
running_server: tuple[int, ReviewFixture],
|
||||
method: str,
|
||||
path: str,
|
||||
headers: dict[str, str] | None,
|
||||
status: int,
|
||||
code: str,
|
||||
) -> None:
|
||||
port, _ = running_server
|
||||
|
||||
response, content = request(port, method, path, headers=headers)
|
||||
payload = cast(dict[str, Any], json.loads(content))
|
||||
|
||||
assert response.status == status
|
||||
assert payload["error"]["code"] == code
|
||||
@@ -1,69 +0,0 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
|
||||
from mdpolish import Pipeline, RunStatus
|
||||
from mdpolish.components import WordReviewCommentComponent
|
||||
|
||||
|
||||
def transform(markdown: str): # type: ignore[no-untyped-def]
|
||||
return Pipeline([WordReviewCommentComponent()]).transform(markdown)
|
||||
|
||||
|
||||
def test_deletes_single_line_comment_and_one_following_blank() -> None:
|
||||
markdown = "before\nCommented [AB1]: review this\n\nafter"
|
||||
result = transform(markdown)
|
||||
|
||||
assert result.status is RunStatus.SUCCESS
|
||||
assert result.output_markdown == "before\nafter"
|
||||
assert len(result.changes) == 1
|
||||
assert result.changes[0].reason == "删除严格单行 Word 审阅批注及其后一个空行"
|
||||
|
||||
|
||||
def test_adjacent_comment_blocks_are_deleted_without_overlap() -> None:
|
||||
markdown = "before\nCommented [A1]: first\n\nCommented [B2R1]: second\n\nafter"
|
||||
result = transform(markdown)
|
||||
|
||||
assert result.output_markdown == "before\nafter"
|
||||
assert len(result.changes) == 2
|
||||
assert result.changes[0].span.end == result.changes[1].span.start
|
||||
|
||||
|
||||
@pytest.mark.parametrize("line_ending", ["\n", "\r\n", "\r"])
|
||||
def test_preserves_line_ending_style(line_ending: str) -> None:
|
||||
markdown = line_ending.join(("before", "Commented [A1]: note", "", "after"))
|
||||
assert transform(markdown).output_markdown == line_ending.join(("before", "after"))
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"comment",
|
||||
[
|
||||
"prefix Commented [A1]: note",
|
||||
" Commented [A1]: note",
|
||||
"Commented []: note",
|
||||
"Commented [A-1]: note",
|
||||
"Commented [A1]:",
|
||||
"Commented [A1]: ",
|
||||
],
|
||||
)
|
||||
def test_similar_lines_are_preserved(comment: str) -> None:
|
||||
markdown = f"before\n{comment}\n\nafter"
|
||||
assert transform(markdown).output_markdown == markdown
|
||||
|
||||
|
||||
def test_requires_a_following_blank_line_and_does_not_delete_extra_blanks() -> None:
|
||||
without_blank = "Commented [A1]: note\nafter"
|
||||
with_two_blanks = "before\nCommented [A1]: note\n\n\nafter"
|
||||
|
||||
assert transform(without_blank).output_markdown == without_blank
|
||||
assert transform(with_two_blanks).output_markdown == "before\n\nafter"
|
||||
|
||||
|
||||
def test_successful_output_is_idempotent() -> None:
|
||||
pipeline = Pipeline([WordReviewCommentComponent()])
|
||||
first = pipeline.transform("Commented [A1]: note\n\nafter")
|
||||
assert first.output_markdown is not None
|
||||
second = pipeline.transform(first.output_markdown)
|
||||
|
||||
assert second.status is RunStatus.SUCCESS
|
||||
assert second.changes == ()
|
||||
Reference in New Issue
Block a user