实现第一版内存清洗核心
落实不可变数据契约、组件基类、原子修改执行器与顺序流水线。补充稳定性复查、审计记录、测试和当前机制文档。
This commit is contained in:
@@ -0,0 +1,121 @@
|
||||
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")
|
||||
@@ -0,0 +1,199 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
|
||||
from mdpolish import (
|
||||
ComponentInfo,
|
||||
DocumentSnapshot,
|
||||
EditValidationError,
|
||||
ProposedChange,
|
||||
TextEdit,
|
||||
TextSpan,
|
||||
apply_component_batch,
|
||||
)
|
||||
|
||||
COMPONENT = ComponentInfo(
|
||||
component_id="test.component",
|
||||
version="1.2.3",
|
||||
parameters=(),
|
||||
applicability="测试精确文本编辑，只处理测试字符串，排除其他输入。",
|
||||
)
|
||||
|
||||
|
||||
def make_edit(snapshot: DocumentSnapshot, start: int, end: int, replacement: str) -> TextEdit:
|
||||
return TextEdit(
|
||||
snapshot_sha256=snapshot.sha256,
|
||||
span=TextSpan(start, end),
|
||||
expected_text=snapshot.markdown[start:end],
|
||||
replacement=replacement,
|
||||
)
|
||||
|
||||
|
||||
def make_proposal(snapshot: DocumentSnapshot, *edits: TextEdit, reason: str = "test reason") -> ProposedChange:
|
||||
return ProposedChange(snapshot_sha256=snapshot.sha256, reason=reason, edits=edits)
|
||||
|
||||
|
||||
def test_applies_insert_delete_and_replace() -> None:
|
||||
insert_snapshot = DocumentSnapshot("ab")
|
||||
delete_snapshot = DocumentSnapshot("abc")
|
||||
replace_snapshot = DocumentSnapshot("abc")
|
||||
|
||||
inserted = apply_component_batch(
|
||||
insert_snapshot,
|
||||
(make_proposal(insert_snapshot, make_edit(insert_snapshot, 1, 1, "X")),),
|
||||
COMPONENT,
|
||||
0,
|
||||
)
|
||||
deleted = apply_component_batch(
|
||||
delete_snapshot,
|
||||
(make_proposal(delete_snapshot, make_edit(delete_snapshot, 1, 2, "")),),
|
||||
COMPONENT,
|
||||
0,
|
||||
)
|
||||
replaced = apply_component_batch(
|
||||
replace_snapshot,
|
||||
(make_proposal(replace_snapshot, make_edit(replace_snapshot, 1, 2, "X")),),
|
||||
COMPONENT,
|
||||
0,
|
||||
)
|
||||
|
||||
assert inserted.snapshot.markdown == "aXb"
|
||||
assert deleted.snapshot.markdown == "ac"
|
||||
assert replaced.snapshot.markdown == "aXc"
|
||||
|
||||
|
||||
def test_multiple_edits_apply_backwards_but_report_in_source_order() -> None:
|
||||
snapshot = DocumentSnapshot("abcdef")
|
||||
proposal = make_proposal(
|
||||
snapshot,
|
||||
make_edit(snapshot, 4, 6, "F"),
|
||||
make_edit(snapshot, 0, 1, "A"),
|
||||
reason="normalize two locations",
|
||||
)
|
||||
|
||||
applied = apply_component_batch(snapshot, (proposal,), COMPONENT, 3)
|
||||
|
||||
assert applied.snapshot.markdown == "AbcdF"
|
||||
assert [change.span.start for change in applied.changes] == [0, 4]
|
||||
assert {change.before_sha256 for change in applied.changes} == {snapshot.sha256}
|
||||
assert {change.after_sha256 for change in applied.changes} == {applied.snapshot.sha256}
|
||||
assert {change.proposal_ref for change in applied.changes} == {
|
||||
applied.changes[0].proposal_ref,
|
||||
}
|
||||
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)
|
||||
|
||||
|
||||
def test_adjacent_nonempty_ranges_are_allowed() -> None:
|
||||
snapshot = DocumentSnapshot("abcd")
|
||||
proposal = make_proposal(
|
||||
snapshot,
|
||||
make_edit(snapshot, 0, 2, "A"),
|
||||
make_edit(snapshot, 2, 4, "D"),
|
||||
)
|
||||
|
||||
applied = apply_component_batch(snapshot, (proposal,), COMPONENT, 0)
|
||||
|
||||
assert applied.snapshot.markdown == "AD"
|
||||
|
||||
|
||||
def test_overlapping_ranges_fail_without_changing_snapshot() -> None:
|
||||
snapshot = DocumentSnapshot("abcdef")
|
||||
proposal = make_proposal(
|
||||
snapshot,
|
||||
make_edit(snapshot, 1, 4, "X"),
|
||||
make_edit(snapshot, 3, 5, "Y"),
|
||||
)
|
||||
|
||||
with pytest.raises(EditValidationError, match="conflicting"):
|
||||
apply_component_batch(snapshot, (proposal,), COMPONENT, 0)
|
||||
|
||||
assert snapshot.markdown == "abcdef"
|
||||
assert snapshot.sha256 == DocumentSnapshot("abcdef").sha256
|
||||
|
||||
|
||||
def test_duplicate_edits_fail_explicitly() -> None:
|
||||
snapshot = DocumentSnapshot("abc")
|
||||
edit = make_edit(snapshot, 0, 1, "A")
|
||||
proposal = make_proposal(snapshot, edit, edit)
|
||||
|
||||
with pytest.raises(EditValidationError, match="duplicate"):
|
||||
apply_component_batch(snapshot, (proposal,), COMPONENT, 0)
|
||||
|
||||
|
||||
def test_distinct_insert_points_are_allowed() -> None:
|
||||
snapshot = DocumentSnapshot("abcd")
|
||||
proposal = make_proposal(
|
||||
snapshot,
|
||||
make_edit(snapshot, 1, 1, "X"),
|
||||
make_edit(snapshot, 3, 3, "Y"),
|
||||
)
|
||||
|
||||
applied = apply_component_batch(snapshot, (proposal,), COMPONENT, 0)
|
||||
|
||||
assert applied.snapshot.markdown == "aXbcYd"
|
||||
|
||||
|
||||
def test_same_insert_point_conflicts() -> None:
|
||||
snapshot = DocumentSnapshot("abc")
|
||||
proposal = make_proposal(
|
||||
snapshot,
|
||||
make_edit(snapshot, 1, 1, "X"),
|
||||
make_edit(snapshot, 1, 1, "Y"),
|
||||
)
|
||||
|
||||
with pytest.raises(EditValidationError, match="conflicting"):
|
||||
apply_component_batch(snapshot, (proposal,), COMPONENT, 0)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("insert_position", [1, 2, 3])
|
||||
def test_insert_at_start_inside_or_end_of_nonempty_range_conflicts(insert_position: int) -> None:
|
||||
snapshot = DocumentSnapshot("abcd")
|
||||
proposal = make_proposal(
|
||||
snapshot,
|
||||
make_edit(snapshot, 1, 3, "X"),
|
||||
make_edit(snapshot, insert_position, insert_position, "Y"),
|
||||
)
|
||||
|
||||
with pytest.raises(EditValidationError, match="conflicting"):
|
||||
apply_component_batch(snapshot, (proposal,), COMPONENT, 0)
|
||||
|
||||
|
||||
def test_stale_hash_out_of_range_and_expected_text_mismatch_fail() -> None:
|
||||
original = DocumentSnapshot("abc")
|
||||
current = DocumentSnapshot("abd")
|
||||
stale = make_proposal(original, make_edit(original, 0, 1, "A"))
|
||||
|
||||
with pytest.raises(EditValidationError, match="stale"):
|
||||
apply_component_batch(current, (stale,), COMPONENT, 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)
|
||||
|
||||
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)
|
||||
|
||||
|
||||
def test_conflict_across_proposals_rejects_whole_component_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)
|
||||
|
||||
assert snapshot.markdown == "abcdef"
|
||||
|
||||
|
||||
def test_empty_component_batch_keeps_same_snapshot_and_records_nothing() -> None:
|
||||
snapshot = DocumentSnapshot("abc")
|
||||
|
||||
applied = apply_component_batch(snapshot, (), COMPONENT, 0)
|
||||
|
||||
assert applied.snapshot is snapshot
|
||||
assert applied.changes == ()
|
||||
@@ -0,0 +1,148 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import FrozenInstanceError
|
||||
from hashlib import sha256
|
||||
|
||||
import pytest
|
||||
|
||||
from mdpolish import (
|
||||
DocumentSnapshot,
|
||||
ErrorStage,
|
||||
ProposedChange,
|
||||
ResidualProposal,
|
||||
RunError,
|
||||
RunStatus,
|
||||
TextEdit,
|
||||
TextSpan,
|
||||
TransformResult,
|
||||
)
|
||||
from mdpolish.models import ProposalReference
|
||||
|
||||
|
||||
def test_snapshot_preserves_exact_markdown_and_computes_hash() -> None:
|
||||
markdown = "标题\r\nCafe\u0301\n🙂\n"
|
||||
|
||||
snapshot = DocumentSnapshot(markdown)
|
||||
|
||||
assert snapshot.markdown == markdown
|
||||
assert snapshot.sha256 == sha256(markdown.encode("utf-8")).hexdigest()
|
||||
|
||||
|
||||
def test_empty_snapshot_is_valid_and_hash_cannot_be_supplied() -> None:
|
||||
snapshot = DocumentSnapshot("")
|
||||
|
||||
assert snapshot.sha256 == sha256(b"").hexdigest()
|
||||
with pytest.raises(TypeError):
|
||||
DocumentSnapshot("", sha256="0" * 64) # type: ignore[call-arg]
|
||||
|
||||
|
||||
def test_snapshot_is_frozen() -> None:
|
||||
snapshot = DocumentSnapshot("original")
|
||||
|
||||
with pytest.raises(FrozenInstanceError):
|
||||
snapshot.markdown = "changed" # type: ignore[misc]
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("start", "end", "error_type"),
|
||||
[
|
||||
(-1, 0, ValueError),
|
||||
(2, 1, ValueError),
|
||||
(True, 1, TypeError),
|
||||
],
|
||||
)
|
||||
def test_span_rejects_invalid_indexes(start: int, end: int, error_type: type[Exception]) -> None:
|
||||
with pytest.raises(error_type):
|
||||
TextSpan(start, end)
|
||||
|
||||
|
||||
def test_text_edit_supports_insert_delete_and_replace() -> None:
|
||||
snapshot = DocumentSnapshot("中文abc")
|
||||
|
||||
insertion = TextEdit(snapshot.sha256, TextSpan(2, 2), "", "!")
|
||||
deletion = TextEdit(snapshot.sha256, TextSpan(2, 3), "a", "")
|
||||
replacement = TextEdit(snapshot.sha256, TextSpan(3, 5), "bc", "BC")
|
||||
|
||||
assert insertion.span.is_empty
|
||||
assert deletion.replacement == ""
|
||||
assert replacement.expected_text == "bc"
|
||||
|
||||
|
||||
def test_text_edit_rejects_bad_digest_length_mismatch_and_no_op() -> None:
|
||||
digest = DocumentSnapshot("abc").sha256
|
||||
|
||||
with pytest.raises(ValueError, match="SHA-256"):
|
||||
TextEdit("bad", TextSpan(0, 1), "a", "b")
|
||||
with pytest.raises(ValueError, match="length"):
|
||||
TextEdit(digest, TextSpan(0, 2), "a", "b")
|
||||
with pytest.raises(ValueError, match="must change"):
|
||||
TextEdit(digest, TextSpan(0, 1), "a", "a")
|
||||
|
||||
|
||||
def test_proposal_requires_reason_edits_and_one_matching_digest() -> None:
|
||||
snapshot = DocumentSnapshot("abc")
|
||||
other = DocumentSnapshot("xyz")
|
||||
edit = TextEdit(snapshot.sha256, TextSpan(0, 1), "a", "A")
|
||||
other_edit = TextEdit(other.sha256, TextSpan(0, 1), "x", "X")
|
||||
|
||||
with pytest.raises(ValueError, match="reason"):
|
||||
ProposedChange(snapshot.sha256, " ", (edit,))
|
||||
with pytest.raises(ValueError, match="non-empty tuple"):
|
||||
ProposedChange(snapshot.sha256, "reason", ())
|
||||
with pytest.raises(ValueError, match="proposal digest"):
|
||||
ProposedChange(snapshot.sha256, "reason", (other_edit,))
|
||||
|
||||
|
||||
def test_transform_result_enforces_status_specific_output_fields() -> None:
|
||||
snapshot = DocumentSnapshot("abc")
|
||||
error = RunError("component", "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,
|
||||
proposal_ref=ProposalReference(0, snapshot.sha256, 0),
|
||||
proposal=proposal,
|
||||
)
|
||||
|
||||
success = TransformResult(
|
||||
status=RunStatus.SUCCESS,
|
||||
input_sha256=snapshot.sha256,
|
||||
current_sha256=snapshot.sha256,
|
||||
output_markdown=snapshot.markdown,
|
||||
)
|
||||
failed = TransformResult(
|
||||
status=RunStatus.FAILED,
|
||||
input_sha256=snapshot.sha256,
|
||||
current_sha256=snapshot.sha256,
|
||||
errors=(error,),
|
||||
partial_markdown=snapshot.markdown,
|
||||
)
|
||||
unstable = TransformResult(
|
||||
status=RunStatus.UNSTABLE,
|
||||
input_sha256=snapshot.sha256,
|
||||
current_sha256=snapshot.sha256,
|
||||
residual_proposals=(residual,),
|
||||
partial_markdown=snapshot.markdown,
|
||||
)
|
||||
|
||||
assert success.output_markdown == "abc"
|
||||
assert failed.partial_markdown == "abc"
|
||||
assert unstable.residual_proposals == (residual,)
|
||||
|
||||
with pytest.raises(ValueError, match="successful result"):
|
||||
TransformResult(
|
||||
status=RunStatus.SUCCESS,
|
||||
input_sha256=snapshot.sha256,
|
||||
current_sha256=snapshot.sha256,
|
||||
errors=(error,),
|
||||
output_markdown=snapshot.markdown,
|
||||
)
|
||||
with pytest.raises(ValueError, match="failed result"):
|
||||
TransformResult(
|
||||
status=RunStatus.FAILED,
|
||||
input_sha256=snapshot.sha256,
|
||||
current_sha256=snapshot.sha256,
|
||||
partial_markdown=snapshot.markdown,
|
||||
)
|
||||
@@ -0,0 +1,293 @@
|
||||
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
|
||||
Reference in New Issue
Block a user