297 lines
9.8 KiB
Python
297 lines
9.8 KiB
Python
from __future__ import annotations
|
|
|
|
from typing import cast
|
|
|
|
import pytest
|
|
|
|
from mdpolish import (
|
|
DocumentSnapshot,
|
|
ErrorStage,
|
|
Modifier,
|
|
Pipeline,
|
|
ProposedChange,
|
|
RunStatus,
|
|
TextEdit,
|
|
TextSpan,
|
|
)
|
|
|
|
|
|
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(needle)),
|
|
expected_text=needle,
|
|
replacement=replacement,
|
|
)
|
|
return (
|
|
ProposedChange(
|
|
snapshot_sha256=snapshot.sha256,
|
|
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,
|
|
)
|
|
|
|
|
|
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 ()
|
|
|
|
return Modifier(
|
|
modifier_id=modifier_id,
|
|
version="1.0.0",
|
|
parameters={"trigger": trigger},
|
|
applicability="仅用于测试异常路径。",
|
|
propose=propose,
|
|
)
|
|
|
|
|
|
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,
|
|
)
|
|
|
|
|
|
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🙂"):
|
|
result = Pipeline([]).transform(markdown)
|
|
|
|
assert result.status is RunStatus.SUCCESS
|
|
assert result.output_markdown == markdown
|
|
assert result.partial_markdown is None
|
|
assert result.changes == ()
|
|
assert result.modifiers == ()
|
|
|
|
|
|
def test_later_modifier_reads_snapshot_produced_by_earlier_modifier() -> None:
|
|
pipeline = Pipeline(
|
|
[
|
|
replace_modifier("初", "中", modifier_id="test.first"),
|
|
replace_modifier("中", "终", modifier_id="test.second"),
|
|
]
|
|
)
|
|
|
|
result = pipeline.transform("初")
|
|
|
|
assert result.status is RunStatus.SUCCESS
|
|
assert result.output_markdown == "终"
|
|
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_modifiers_and_parameters_produce_same_ordered_result() -> None:
|
|
pipeline = Pipeline(
|
|
[
|
|
replace_modifier("a", "b", modifier_id="test.first"),
|
|
replace_modifier("b", "c", modifier_id="test.second"),
|
|
]
|
|
)
|
|
|
|
assert pipeline.transform("a") == pipeline.transform("a")
|
|
|
|
|
|
def test_successful_pipeline_is_idempotent_on_its_output() -> None:
|
|
pipeline = Pipeline([replace_modifier("old", "new", modifier_id="test.replace")])
|
|
|
|
first = pipeline.transform("old value")
|
|
assert first.status is RunStatus.SUCCESS
|
|
assert first.output_markdown is not None
|
|
|
|
second = pipeline.transform(first.output_markdown)
|
|
|
|
assert second.status is RunStatus.SUCCESS
|
|
assert second.output_markdown == "new value"
|
|
assert second.changes == ()
|
|
|
|
|
|
def test_transform_error_stops_later_modifiers_and_keeps_only_partial_text() -> None:
|
|
pipeline = Pipeline(
|
|
[
|
|
replace_modifier("a", "b", modifier_id="test.first"),
|
|
exploding_modifier(),
|
|
replace_modifier("b", "c", modifier_id="test.never-runs"),
|
|
]
|
|
)
|
|
|
|
result = pipeline.transform("a")
|
|
|
|
assert result.status is RunStatus.FAILED
|
|
assert result.output_markdown is None
|
|
assert result.partial_markdown == "b"
|
|
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_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"
|
|
assert "SECRET" not in result.errors[0].message
|
|
assert "private markdown" not in result.errors[0].message
|
|
|
|
|
|
def test_invalid_proposal_return_is_a_transform_contract_failure() -> None:
|
|
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 == "ModifierContractError"
|
|
assert result.errors[0].stage is ErrorStage.TRANSFORM
|
|
|
|
|
|
@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"
|
|
assert result.changes == ()
|
|
assert result.errors[0].error_type == "EditValidationError"
|
|
|
|
|
|
def test_duplicate_modifier_ids_fail_during_preflight_before_modification() -> None:
|
|
pipeline = Pipeline(
|
|
[
|
|
replace_modifier("a", "b", modifier_id="test.duplicate"),
|
|
replace_modifier("b", "c", modifier_id="test.duplicate"),
|
|
]
|
|
)
|
|
|
|
result = pipeline.transform("a")
|
|
|
|
assert result.status is RunStatus.FAILED
|
|
assert result.partial_markdown == "a"
|
|
assert result.changes == ()
|
|
assert result.errors[0].error_type == "PipelineContractError"
|
|
|
|
|
|
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].error_type == "ModifierContractError"
|
|
|
|
|
|
def test_cross_modifier_chain_is_reported_unstable_without_a_second_round() -> None:
|
|
pipeline = Pipeline(
|
|
[
|
|
replace_modifier("bad", "good", modifier_id="test.to-good"),
|
|
replace_modifier("good", "bad", modifier_id="test.to-bad"),
|
|
]
|
|
)
|
|
|
|
result = pipeline.transform("bad")
|
|
|
|
assert result.status is RunStatus.UNSTABLE
|
|
assert result.output_markdown is None
|
|
assert result.partial_markdown == "bad"
|
|
assert len(result.changes) == 2
|
|
assert len(result.residual_proposals) == 1
|
|
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(
|
|
[
|
|
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"),
|
|
]
|
|
)
|
|
|
|
result = pipeline.transform("start")
|
|
|
|
assert result.status is RunStatus.FAILED
|
|
assert result.output_markdown is None
|
|
assert result.partial_markdown == "done"
|
|
assert len(result.errors) == 1
|
|
assert result.errors[0].stage is ErrorStage.FINAL_REVIEW
|
|
assert result.errors[0].modifier_id == "test.review-error"
|
|
assert len(result.residual_proposals) == 1
|
|
assert result.residual_proposals[0].modifier_id == "test.residual"
|
|
assert result.residual_proposals[0].proposal_ref.snapshot_sha256 == result.current_sha256
|