294 lines
10 KiB
Python
294 lines
10 KiB
Python
from __future__ import annotations
|
|
|
|
from collections.abc import Mapping
|
|
from typing import cast
|
|
|
|
import pytest
|
|
|
|
from mdpolish import (
|
|
Component,
|
|
DocumentSnapshot,
|
|
ErrorStage,
|
|
Pipeline,
|
|
ProposedChange,
|
|
RunStatus,
|
|
TextEdit,
|
|
TextSpan,
|
|
)
|
|
|
|
|
|
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)
|
|
if position < 0:
|
|
return ()
|
|
edit = TextEdit(
|
|
snapshot_sha256=snapshot.sha256,
|
|
span=TextSpan(position, position + len(self.needle)),
|
|
expected_text=self.needle,
|
|
replacement=self.replacement,
|
|
)
|
|
return (
|
|
ProposedChange(
|
|
snapshot_sha256=snapshot.sha256,
|
|
reason=f"replace test token for {self.component_id}",
|
|
edits=(edit,),
|
|
),
|
|
)
|
|
|
|
|
|
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:
|
|
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, ...], [])
|
|
|
|
|
|
class StaleProposalComponent(ReplaceComponent):
|
|
def __init__(self) -> None:
|
|
super().__init__("a", "A", component_id="test.stale")
|
|
|
|
def _propose_changes(self, 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,)),)
|
|
|
|
|
|
class OverlapComponent(ReplaceComponent):
|
|
def __init__(self) -> None:
|
|
super().__init__("a", "A", component_id="test.overlap")
|
|
|
|
def _propose_changes(self, 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)),)
|
|
|
|
|
|
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.components == ()
|
|
|
|
|
|
def test_later_component_reads_snapshot_produced_by_earlier_component() -> None:
|
|
pipeline = Pipeline(
|
|
[
|
|
ReplaceComponent("初", "中", component_id="test.first"),
|
|
ReplaceComponent("中", "终", component_id="test.second"),
|
|
]
|
|
)
|
|
|
|
result = pipeline.transform("初")
|
|
|
|
assert result.status is RunStatus.SUCCESS
|
|
assert result.output_markdown == "终"
|
|
assert [change.component_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:
|
|
pipeline = Pipeline(
|
|
[
|
|
ReplaceComponent("a", "b", component_id="test.first"),
|
|
ReplaceComponent("b", "c", component_id="test.second"),
|
|
]
|
|
)
|
|
|
|
assert pipeline.transform("a") == pipeline.transform("a")
|
|
|
|
|
|
def test_successful_pipeline_is_idempotent_on_its_output() -> None:
|
|
pipeline = Pipeline([ReplaceComponent("old", "new", component_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_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:
|
|
pipeline = Pipeline(
|
|
[
|
|
ReplaceComponent("a", "b", component_id="test.first"),
|
|
ExplodingComponent(),
|
|
ReplaceComponent("b", "c", component_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.component_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")
|
|
|
|
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:
|
|
result = Pipeline([InvalidReturnComponent()]).transform("abc")
|
|
|
|
assert result.status is RunStatus.FAILED
|
|
assert result.partial_markdown == "abc"
|
|
assert result.errors[0].error_type == "ComponentContractError"
|
|
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")
|
|
|
|
assert result.status is RunStatus.FAILED
|
|
assert result.partial_markdown == "abcd"
|
|
assert result.changes == ()
|
|
assert result.errors[0].error_type == "EditValidationError"
|
|
|
|
|
|
def test_duplicate_component_ids_fail_during_preflight_before_modification() -> None:
|
|
pipeline = Pipeline(
|
|
[
|
|
ReplaceComponent("a", "b", component_id="test.duplicate"),
|
|
ReplaceComponent("b", "c", component_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"
|
|
|
|
|
|
@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")
|
|
|
|
assert result.status is RunStatus.FAILED
|
|
assert result.partial_markdown == "a"
|
|
assert result.changes == ()
|
|
assert result.errors[0].stage is ErrorStage.TRANSFORM
|
|
|
|
|
|
def test_cross_component_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"),
|
|
]
|
|
)
|
|
|
|
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].component_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"),
|
|
]
|
|
)
|
|
|
|
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].component_id == "test.review-error"
|
|
assert len(result.residual_proposals) == 1
|
|
assert result.residual_proposals[0].component_id == "test.residual"
|
|
assert result.residual_proposals[0].proposal_ref.snapshot_sha256 == result.current_sha256
|