|
|
|
@@ -0,0 +1,720 @@
|
|
|
|
|
from __future__ import annotations
|
|
|
|
|
|
|
|
|
|
from collections.abc import Callable
|
|
|
|
|
from dataclasses import replace
|
|
|
|
|
|
|
|
|
|
import pytest
|
|
|
|
|
|
|
|
|
|
import mdpolish.pipeline as pipeline_module
|
|
|
|
|
from mdpolish import (
|
|
|
|
|
AppliedBatch,
|
|
|
|
|
Change,
|
|
|
|
|
DocumentSnapshot,
|
|
|
|
|
ErrorStage,
|
|
|
|
|
Modifier,
|
|
|
|
|
ModifierInfo,
|
|
|
|
|
Pipeline,
|
|
|
|
|
ProposedChange,
|
|
|
|
|
RunError,
|
|
|
|
|
RunStatus,
|
|
|
|
|
TextEdit,
|
|
|
|
|
TextSpan,
|
|
|
|
|
TransformResult,
|
|
|
|
|
apply_modifier_batch,
|
|
|
|
|
markdown_sha256,
|
|
|
|
|
)
|
|
|
|
|
from mdpolish.review import (
|
|
|
|
|
ReviewBuildError,
|
|
|
|
|
ReviewCurrentKind,
|
|
|
|
|
build_review_document,
|
|
|
|
|
render_markdown_report,
|
|
|
|
|
)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def replace_first(
|
|
|
|
|
needle: str,
|
|
|
|
|
replacement: str,
|
|
|
|
|
*,
|
|
|
|
|
modifier_id: str,
|
|
|
|
|
reason: str | None = None,
|
|
|
|
|
) -> 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=reason or f"replace token for {modifier_id}",
|
|
|
|
|
edits=(edit,),
|
|
|
|
|
),
|
|
|
|
|
)
|
|
|
|
|
|
|
|
|
|
return Modifier(
|
|
|
|
|
modifier_id=modifier_id,
|
|
|
|
|
version="1.0.0",
|
|
|
|
|
parameters={"needle": needle, "replacement": replacement},
|
|
|
|
|
applicability="只处理虚构测试标记。",
|
|
|
|
|
propose=propose,
|
|
|
|
|
)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def empty_modifier(modifier_id: str) -> Modifier:
|
|
|
|
|
return Modifier(
|
|
|
|
|
modifier_id=modifier_id,
|
|
|
|
|
version="1.0.0",
|
|
|
|
|
parameters=(),
|
|
|
|
|
applicability="测试零修改阶段。",
|
|
|
|
|
propose=lambda snapshot: (),
|
|
|
|
|
)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def exploding_modifier(modifier_id: str, *, trigger: str | None = None) -> Modifier:
|
|
|
|
|
def propose(snapshot: DocumentSnapshot) -> tuple[ProposedChange, ...]:
|
|
|
|
|
if trigger is None or snapshot.markdown == trigger:
|
|
|
|
|
raise RuntimeError(f"secret failure: {snapshot.markdown}")
|
|
|
|
|
return ()
|
|
|
|
|
|
|
|
|
|
return Modifier(
|
|
|
|
|
modifier_id=modifier_id,
|
|
|
|
|
version="1.0.0",
|
|
|
|
|
parameters={"trigger": trigger},
|
|
|
|
|
applicability="只测试失败路径。",
|
|
|
|
|
propose=propose,
|
|
|
|
|
)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def stale_modifier(modifier_id: str = "test.stale") -> Modifier:
|
|
|
|
|
def propose(snapshot: DocumentSnapshot) -> tuple[ProposedChange, ...]:
|
|
|
|
|
stale = DocumentSnapshot(snapshot.markdown + "!")
|
|
|
|
|
edit = TextEdit(stale.sha256, TextSpan(0, 1), stale.markdown[0], "X")
|
|
|
|
|
return (ProposedChange(stale.sha256, "stale proposal", (edit,)),)
|
|
|
|
|
|
|
|
|
|
return Modifier(
|
|
|
|
|
modifier_id=modifier_id,
|
|
|
|
|
version="1.0.0",
|
|
|
|
|
parameters=(),
|
|
|
|
|
applicability="只测试批次验证失败。",
|
|
|
|
|
propose=propose,
|
|
|
|
|
)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def final_review_failure_result(*, error_count: int = 1) -> TransformResult:
|
|
|
|
|
modifiers = [
|
|
|
|
|
exploding_modifier(f"test.review-error-{index}", trigger="done") for index in range(error_count)
|
|
|
|
|
]
|
|
|
|
|
modifiers.append(replace_first("start", "done", modifier_id="test.producer"))
|
|
|
|
|
return Pipeline(modifiers).transform("start")
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def residual_result() -> TransformResult:
|
|
|
|
|
def propose(snapshot: DocumentSnapshot) -> tuple[ProposedChange, ...]:
|
|
|
|
|
proposals: list[ProposedChange] = []
|
|
|
|
|
for position, character in enumerate(snapshot.markdown):
|
|
|
|
|
if character != "x":
|
|
|
|
|
continue
|
|
|
|
|
edit = TextEdit(
|
|
|
|
|
snapshot_sha256=snapshot.sha256,
|
|
|
|
|
span=TextSpan(position, position + 1),
|
|
|
|
|
expected_text="x",
|
|
|
|
|
replacement="DO-NOT-PRINT-RESIDUAL-TEXT",
|
|
|
|
|
)
|
|
|
|
|
proposals.append(ProposedChange(snapshot.sha256, "residual candidate", (edit,)))
|
|
|
|
|
return tuple(proposals)
|
|
|
|
|
|
|
|
|
|
residual = Modifier(
|
|
|
|
|
modifier_id="test.residual",
|
|
|
|
|
version="1.0.0",
|
|
|
|
|
parameters=(),
|
|
|
|
|
applicability="只测试残留候选。",
|
|
|
|
|
propose=propose,
|
|
|
|
|
)
|
|
|
|
|
producer = replace_first("a", "xxx", modifier_id="test.producer")
|
|
|
|
|
return Pipeline((residual, producer)).transform("a")
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def test_empty_pipeline_builds_complete_unchanged_review() -> None:
|
|
|
|
|
result = Pipeline(()).transform("")
|
|
|
|
|
|
|
|
|
|
review = build_review_document("", result)
|
|
|
|
|
|
|
|
|
|
assert review.status is RunStatus.SUCCESS
|
|
|
|
|
assert review.current_kind is ReviewCurrentKind.SUCCESS_OUTPUT
|
|
|
|
|
assert review.input_markdown == review.current_markdown == ""
|
|
|
|
|
assert review.stages == ()
|
|
|
|
|
assert review.stages_complete is True
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def test_replay_preserves_zero_stages_and_exact_intermediate_snapshots() -> None:
|
|
|
|
|
pipeline = Pipeline(
|
|
|
|
|
(
|
|
|
|
|
empty_modifier("test.zero-first"),
|
|
|
|
|
replace_first("a", "b", modifier_id="test.first"),
|
|
|
|
|
replace_first("b", "c", modifier_id="test.second"),
|
|
|
|
|
empty_modifier("test.zero-last"),
|
|
|
|
|
)
|
|
|
|
|
)
|
|
|
|
|
result = pipeline.transform("a")
|
|
|
|
|
|
|
|
|
|
review = build_review_document("a", result)
|
|
|
|
|
|
|
|
|
|
assert [stage.modifier_position for stage in review.stages] == [0, 1, 2, 3]
|
|
|
|
|
assert [(stage.before_markdown, stage.after_markdown) for stage in review.stages] == [
|
|
|
|
|
("a", "a"),
|
|
|
|
|
("a", "b"),
|
|
|
|
|
("b", "c"),
|
|
|
|
|
("c", "c"),
|
|
|
|
|
]
|
|
|
|
|
assert [len(stage.changes) for stage in review.stages] == [0, 1, 1, 0]
|
|
|
|
|
assert [stage.before_sha256 for stage in review.stages[1:]] == [
|
|
|
|
|
result.changes[0].before_sha256,
|
|
|
|
|
result.changes[1].before_sha256,
|
|
|
|
|
result.changes[1].after_sha256,
|
|
|
|
|
]
|
|
|
|
|
assert review.current_markdown == "c"
|
|
|
|
|
assert review.current_sha256 == markdown_sha256("c")
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
|
|
|
("markdown", "start", "end", "replacement", "expected"),
|
|
|
|
|
[
|
|
|
|
|
("ab", 1, 1, "X", "aXb"),
|
|
|
|
|
("abc", 1, 2, "", "ac"),
|
|
|
|
|
("abc", 1, 2, "X", "aXc"),
|
|
|
|
|
],
|
|
|
|
|
)
|
|
|
|
|
def test_replay_covers_insert_delete_and_replace(
|
|
|
|
|
markdown: str,
|
|
|
|
|
start: int,
|
|
|
|
|
end: int,
|
|
|
|
|
replacement: str,
|
|
|
|
|
expected: str,
|
|
|
|
|
) -> None:
|
|
|
|
|
def propose(snapshot: DocumentSnapshot) -> tuple[ProposedChange, ...]:
|
|
|
|
|
if snapshot.markdown != markdown:
|
|
|
|
|
return ()
|
|
|
|
|
edit = TextEdit(
|
|
|
|
|
snapshot.sha256,
|
|
|
|
|
TextSpan(start, end),
|
|
|
|
|
snapshot.markdown[start:end],
|
|
|
|
|
replacement,
|
|
|
|
|
)
|
|
|
|
|
return (ProposedChange(snapshot.sha256, "one exact edit", (edit,)),)
|
|
|
|
|
|
|
|
|
|
modifier = Modifier(
|
|
|
|
|
modifier_id="test.edit-kind",
|
|
|
|
|
version="1.0.0",
|
|
|
|
|
parameters=(),
|
|
|
|
|
applicability="只测试插入、删除和替换。",
|
|
|
|
|
propose=propose,
|
|
|
|
|
)
|
|
|
|
|
|
|
|
|
|
review = build_review_document(markdown, Pipeline((modifier,)).transform(markdown))
|
|
|
|
|
|
|
|
|
|
assert review.stages[0].after_markdown == expected
|
|
|
|
|
assert review.current_markdown == expected
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def test_one_proposal_with_multiple_edits_keeps_references_and_core_order() -> None:
|
|
|
|
|
def propose(snapshot: DocumentSnapshot) -> tuple[ProposedChange, ...]:
|
|
|
|
|
if snapshot.markdown != "abcdef":
|
|
|
|
|
return ()
|
|
|
|
|
edits = (
|
|
|
|
|
TextEdit(snapshot.sha256, TextSpan(4, 6), "ef", "E"),
|
|
|
|
|
TextEdit(snapshot.sha256, TextSpan(0, 1), "a", "A"),
|
|
|
|
|
TextEdit(snapshot.sha256, TextSpan(2, 3), "c", ""),
|
|
|
|
|
)
|
|
|
|
|
return (ProposedChange(snapshot.sha256, "three exact edits", edits),)
|
|
|
|
|
|
|
|
|
|
modifier = Modifier(
|
|
|
|
|
modifier_id="test.multi-edit",
|
|
|
|
|
version="1.0.0",
|
|
|
|
|
parameters=(),
|
|
|
|
|
applicability="只测试一个候选中的多个编辑。",
|
|
|
|
|
propose=propose,
|
|
|
|
|
)
|
|
|
|
|
result = Pipeline((modifier,)).transform("abcdef")
|
|
|
|
|
|
|
|
|
|
review = build_review_document("abcdef", result)
|
|
|
|
|
stage = review.stages[0]
|
|
|
|
|
|
|
|
|
|
assert stage.after_markdown == "AbdE"
|
|
|
|
|
assert [change.change.span.start for change in stage.changes] == [0, 2, 4]
|
|
|
|
|
assert [change.change.edit_index for change in stage.changes] == [1, 2, 0]
|
|
|
|
|
assert {change.change.proposal_ref for change in stage.changes} == {
|
|
|
|
|
stage.changes[0].change.proposal_ref,
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def test_locations_use_python_code_points_and_physical_newlines() -> None:
|
|
|
|
|
markdown = "\ufeffA\r\n中e\u0301🙂X\rY\nZ"
|
|
|
|
|
positions = (1, 6, 8, 10, 12)
|
|
|
|
|
|
|
|
|
|
def propose(snapshot: DocumentSnapshot) -> tuple[ProposedChange, ...]:
|
|
|
|
|
if snapshot.markdown != markdown:
|
|
|
|
|
return ()
|
|
|
|
|
edits = tuple(
|
|
|
|
|
TextEdit(
|
|
|
|
|
snapshot.sha256,
|
|
|
|
|
TextSpan(position, position + 1),
|
|
|
|
|
snapshot.markdown[position],
|
|
|
|
|
f"<{position}>",
|
|
|
|
|
)
|
|
|
|
|
for position in positions
|
|
|
|
|
)
|
|
|
|
|
return (ProposedChange(snapshot.sha256, "unicode locations", edits),)
|
|
|
|
|
|
|
|
|
|
modifier = Modifier(
|
|
|
|
|
modifier_id="test.locations",
|
|
|
|
|
version="1.0.0",
|
|
|
|
|
parameters=(),
|
|
|
|
|
applicability="只测试 Python 码点位置。",
|
|
|
|
|
propose=propose,
|
|
|
|
|
)
|
|
|
|
|
review = build_review_document(markdown, Pipeline((modifier,)).transform(markdown))
|
|
|
|
|
|
|
|
|
|
assert [(change.location.line, change.location.column) for change in review.stages[0].changes] == [
|
|
|
|
|
(1, 2),
|
|
|
|
|
(2, 3),
|
|
|
|
|
(2, 5),
|
|
|
|
|
(3, 1),
|
|
|
|
|
(4, 1),
|
|
|
|
|
]
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def test_preflight_failure_has_no_stage_and_unchanged_partial_text() -> None:
|
|
|
|
|
duplicate = replace_first("a", "b", modifier_id="test.duplicate")
|
|
|
|
|
result = Pipeline((duplicate, duplicate)).transform("a")
|
|
|
|
|
|
|
|
|
|
review = build_review_document("a", result)
|
|
|
|
|
|
|
|
|
|
assert result.errors[0].stage is ErrorStage.PREFLIGHT
|
|
|
|
|
assert review.current_markdown == "a"
|
|
|
|
|
assert review.stages == ()
|
|
|
|
|
assert review.stages_complete is False
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def _assert_only_first_stage_completed(result: TransformResult) -> None:
|
|
|
|
|
review = build_review_document("a", result)
|
|
|
|
|
assert result.status is RunStatus.FAILED
|
|
|
|
|
assert result.errors[0].stage is ErrorStage.TRANSFORM
|
|
|
|
|
assert review.current_markdown == "b"
|
|
|
|
|
assert [stage.modifier_position for stage in review.stages] == [0]
|
|
|
|
|
assert review.stages[0].after_markdown == "b"
|
|
|
|
|
assert review.stages_complete is False
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def test_propose_failure_stops_before_failing_stage() -> None:
|
|
|
|
|
result = Pipeline(
|
|
|
|
|
(
|
|
|
|
|
replace_first("a", "b", modifier_id="test.first"),
|
|
|
|
|
exploding_modifier("test.exploding"),
|
|
|
|
|
replace_first("b", "c", modifier_id="test.never"),
|
|
|
|
|
)
|
|
|
|
|
).transform("a")
|
|
|
|
|
|
|
|
|
|
_assert_only_first_stage_completed(result)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def test_metadata_change_after_proposal_stops_before_failing_stage() -> None:
|
|
|
|
|
mutating: Modifier
|
|
|
|
|
|
|
|
|
|
def propose(snapshot: DocumentSnapshot) -> tuple[ProposedChange, ...]:
|
|
|
|
|
object.__setattr__(mutating, "version", "2.0.0")
|
|
|
|
|
edit = TextEdit(snapshot.sha256, TextSpan(0, 1), "b", "B")
|
|
|
|
|
return (ProposedChange(snapshot.sha256, "mutate metadata", (edit,)),)
|
|
|
|
|
|
|
|
|
|
mutating = Modifier(
|
|
|
|
|
modifier_id="test.mutating",
|
|
|
|
|
version="1.0.0",
|
|
|
|
|
parameters=(),
|
|
|
|
|
applicability="只测试运行期元数据变化。",
|
|
|
|
|
propose=propose,
|
|
|
|
|
)
|
|
|
|
|
result = Pipeline(
|
|
|
|
|
(replace_first("a", "b", modifier_id="test.first"), mutating, empty_modifier("test.never"))
|
|
|
|
|
).transform("a")
|
|
|
|
|
|
|
|
|
|
_assert_only_first_stage_completed(result)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def test_batch_validation_failure_does_not_create_a_zero_change_stage() -> None:
|
|
|
|
|
result = Pipeline(
|
|
|
|
|
(replace_first("a", "b", modifier_id="test.first"), stale_modifier(), empty_modifier("test.never"))
|
|
|
|
|
).transform("a")
|
|
|
|
|
|
|
|
|
|
_assert_only_first_stage_completed(result)
|
|
|
|
|
assert result.errors[0].error_type == "EditValidationError"
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def test_batch_application_failure_does_not_create_a_stage(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
|
|
|
calls = 0
|
|
|
|
|
|
|
|
|
|
def fail_second_application(
|
|
|
|
|
snapshot: DocumentSnapshot,
|
|
|
|
|
proposals: tuple[ProposedChange, ...],
|
|
|
|
|
modifier: ModifierInfo,
|
|
|
|
|
modifier_position: int,
|
|
|
|
|
) -> AppliedBatch:
|
|
|
|
|
nonlocal calls
|
|
|
|
|
calls += 1
|
|
|
|
|
if calls == 2:
|
|
|
|
|
raise RuntimeError("application failure")
|
|
|
|
|
return apply_modifier_batch(snapshot, proposals, modifier, modifier_position)
|
|
|
|
|
|
|
|
|
|
monkeypatch.setattr(pipeline_module, "apply_modifier_batch", fail_second_application)
|
|
|
|
|
result = Pipeline(
|
|
|
|
|
(
|
|
|
|
|
replace_first("a", "b", modifier_id="test.first"),
|
|
|
|
|
replace_first("b", "c", modifier_id="test.failing"),
|
|
|
|
|
empty_modifier("test.never"),
|
|
|
|
|
)
|
|
|
|
|
).transform("a")
|
|
|
|
|
|
|
|
|
|
_assert_only_first_stage_completed(result)
|
|
|
|
|
assert result.errors[0].error_type == "RuntimeError"
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def test_final_review_failures_keep_all_transform_stages_complete() -> None:
|
|
|
|
|
result = final_review_failure_result(error_count=2)
|
|
|
|
|
|
|
|
|
|
review = build_review_document("start", result)
|
|
|
|
|
|
|
|
|
|
assert result.status is RunStatus.FAILED
|
|
|
|
|
assert [error.stage for error in result.errors] == [ErrorStage.FINAL_REVIEW, ErrorStage.FINAL_REVIEW]
|
|
|
|
|
assert len(review.stages) == 3
|
|
|
|
|
assert review.stages[-1].after_markdown == "done"
|
|
|
|
|
assert review.stages_complete is True
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def test_unstable_result_keeps_complete_stages_and_unapplied_residuals() -> None:
|
|
|
|
|
result = residual_result()
|
|
|
|
|
|
|
|
|
|
review = build_review_document("a", result)
|
|
|
|
|
|
|
|
|
|
assert result.status is RunStatus.UNSTABLE
|
|
|
|
|
assert review.current_kind is ReviewCurrentKind.PARTIAL_OUTPUT
|
|
|
|
|
assert review.current_markdown == "xxx"
|
|
|
|
|
assert review.stages_complete is True
|
|
|
|
|
assert len(review.residual_proposals) == 3
|
|
|
|
|
assert "DO-NOT-PRINT" not in review.current_markdown
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def test_builder_does_not_run_modifiers_again() -> None:
|
|
|
|
|
calls = 0
|
|
|
|
|
|
|
|
|
|
def propose(snapshot: DocumentSnapshot) -> tuple[ProposedChange, ...]:
|
|
|
|
|
nonlocal calls
|
|
|
|
|
calls += 1
|
|
|
|
|
return ()
|
|
|
|
|
|
|
|
|
|
modifier = Modifier(
|
|
|
|
|
modifier_id="test.call-count",
|
|
|
|
|
version="1.0.0",
|
|
|
|
|
parameters=(),
|
|
|
|
|
applicability="只测试调用次数。",
|
|
|
|
|
propose=propose,
|
|
|
|
|
)
|
|
|
|
|
result = Pipeline((modifier,)).transform("a")
|
|
|
|
|
calls_after_pipeline = calls
|
|
|
|
|
|
|
|
|
|
review = build_review_document("a", result)
|
|
|
|
|
render_markdown_report(review)
|
|
|
|
|
|
|
|
|
|
assert calls_after_pipeline == 2
|
|
|
|
|
assert calls == calls_after_pipeline
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def test_builder_rejects_input_and_replay_mismatches_without_source_text() -> None:
|
|
|
|
|
result = Pipeline((replace_first("secret", "public", modifier_id="test.replace"),)).transform("secret")
|
|
|
|
|
|
|
|
|
|
with pytest.raises(ReviewBuildError) as input_error:
|
|
|
|
|
build_review_document("different secret", result)
|
|
|
|
|
assert "secret" not in str(input_error.value)
|
|
|
|
|
|
|
|
|
|
mismatched = replace(
|
|
|
|
|
result,
|
|
|
|
|
current_sha256=markdown_sha256("unrelated"),
|
|
|
|
|
output_markdown="unrelated",
|
|
|
|
|
)
|
|
|
|
|
with pytest.raises(ReviewBuildError) as replay_error:
|
|
|
|
|
build_review_document("secret", mismatched)
|
|
|
|
|
assert "secret" not in str(replay_error.value)
|
|
|
|
|
assert "public" not in str(replay_error.value)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def test_builder_rejects_mixed_error_stages() -> None:
|
|
|
|
|
result = final_review_failure_result()
|
|
|
|
|
transform_error = RunError(
|
|
|
|
|
modifier_id=result.modifiers[0].modifier_id,
|
|
|
|
|
modifier_version=result.modifiers[0].version,
|
|
|
|
|
modifier_position=0,
|
|
|
|
|
stage=ErrorStage.TRANSFORM,
|
|
|
|
|
error_type="ExampleError",
|
|
|
|
|
message="safe",
|
|
|
|
|
)
|
|
|
|
|
malformed = replace(result, errors=(*result.errors, transform_error))
|
|
|
|
|
|
|
|
|
|
with pytest.raises(ReviewBuildError, match="mixes error stages"):
|
|
|
|
|
build_review_document("start", malformed)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def test_builder_rejects_preflight_result_with_transform_output() -> None:
|
|
|
|
|
duplicate = replace_first("a", "b", modifier_id="test.duplicate")
|
|
|
|
|
preflight = Pipeline((duplicate, duplicate)).transform("a")
|
|
|
|
|
successful = Pipeline((replace_first("a", "b", modifier_id="test.success"),)).transform("a")
|
|
|
|
|
malformed = replace(
|
|
|
|
|
preflight,
|
|
|
|
|
current_sha256=markdown_sha256("b"),
|
|
|
|
|
changes=successful.changes,
|
|
|
|
|
partial_markdown="b",
|
|
|
|
|
)
|
|
|
|
|
|
|
|
|
|
with pytest.raises(ReviewBuildError, match="preflight failure contains transform output"):
|
|
|
|
|
build_review_document("a", malformed)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def test_builder_rejects_change_at_transform_failure_position() -> None:
|
|
|
|
|
successful = Pipeline((replace_first("a", "b", modifier_id="test.replace"),)).transform("a")
|
|
|
|
|
error = RunError(
|
|
|
|
|
modifier_id=successful.modifiers[0].modifier_id,
|
|
|
|
|
modifier_version=successful.modifiers[0].version,
|
|
|
|
|
modifier_position=0,
|
|
|
|
|
stage=ErrorStage.TRANSFORM,
|
|
|
|
|
error_type="ExampleError",
|
|
|
|
|
message="safe",
|
|
|
|
|
)
|
|
|
|
|
malformed = TransformResult(
|
|
|
|
|
status=RunStatus.FAILED,
|
|
|
|
|
input_sha256=successful.input_sha256,
|
|
|
|
|
current_sha256=successful.current_sha256,
|
|
|
|
|
modifiers=successful.modifiers,
|
|
|
|
|
changes=successful.changes,
|
|
|
|
|
errors=(error,),
|
|
|
|
|
partial_markdown=successful.output_markdown,
|
|
|
|
|
)
|
|
|
|
|
|
|
|
|
|
with pytest.raises(ReviewBuildError, match="incomplete modifier stage"):
|
|
|
|
|
build_review_document("a", malformed)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
|
|
|
"tamper",
|
|
|
|
|
[
|
|
|
|
|
lambda change: replace(change, modifier_id="test.wrong"),
|
|
|
|
|
lambda change: replace(change, before="z"),
|
|
|
|
|
lambda change: replace(change, span=TextSpan(1, 2), before="z"),
|
|
|
|
|
lambda change: replace(change, after_sha256=change.before_sha256),
|
|
|
|
|
lambda change: replace(
|
|
|
|
|
change,
|
|
|
|
|
proposal_ref=replace(change.proposal_ref, snapshot_sha256=change.after_sha256),
|
|
|
|
|
),
|
|
|
|
|
lambda change: replace(
|
|
|
|
|
change,
|
|
|
|
|
proposal_ref=replace(change.proposal_ref, proposal_index=1),
|
|
|
|
|
),
|
|
|
|
|
],
|
|
|
|
|
)
|
|
|
|
|
def test_builder_rejects_tampered_change_contract(tamper: Callable[[Change], Change]) -> None:
|
|
|
|
|
result = Pipeline((replace_first("a", "b", modifier_id="test.replace"),)).transform("a")
|
|
|
|
|
change = result.changes[0]
|
|
|
|
|
tampered_change = tamper(change)
|
|
|
|
|
malformed = replace(result, changes=(tampered_change,))
|
|
|
|
|
|
|
|
|
|
with pytest.raises(ReviewBuildError):
|
|
|
|
|
build_review_document("a", malformed)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def test_builder_rejects_duplicate_and_conflicting_changes() -> None:
|
|
|
|
|
one_change = Pipeline((replace_first("a", "b", modifier_id="test.replace"),)).transform("a")
|
|
|
|
|
duplicate = replace(one_change, changes=(one_change.changes[0], one_change.changes[0]))
|
|
|
|
|
with pytest.raises(ReviewBuildError, match="repeats a proposal edit index"):
|
|
|
|
|
build_review_document("a", duplicate)
|
|
|
|
|
|
|
|
|
|
def propose(snapshot: DocumentSnapshot) -> tuple[ProposedChange, ...]:
|
|
|
|
|
if snapshot.markdown != "abc":
|
|
|
|
|
return ()
|
|
|
|
|
edits = (
|
|
|
|
|
TextEdit(snapshot.sha256, TextSpan(0, 1), "a", "A"),
|
|
|
|
|
TextEdit(snapshot.sha256, TextSpan(2, 3), "c", "C"),
|
|
|
|
|
)
|
|
|
|
|
return (ProposedChange(snapshot.sha256, "two edits", edits),)
|
|
|
|
|
|
|
|
|
|
modifier = Modifier(
|
|
|
|
|
modifier_id="test.conflict",
|
|
|
|
|
version="1.0.0",
|
|
|
|
|
parameters=(),
|
|
|
|
|
applicability="只测试篡改后的冲突范围。",
|
|
|
|
|
propose=propose,
|
|
|
|
|
)
|
|
|
|
|
result = Pipeline((modifier,)).transform("abc")
|
|
|
|
|
conflicting_second = replace(result.changes[1], span=TextSpan(0, 2), before="ab", after="X")
|
|
|
|
|
conflicting = replace(result, changes=(result.changes[0], conflicting_second))
|
|
|
|
|
|
|
|
|
|
with pytest.raises(ReviewBuildError, match="edit contract"):
|
|
|
|
|
build_review_document("abc", conflicting)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def test_builder_rejects_noncanonical_change_order() -> None:
|
|
|
|
|
def propose(snapshot: DocumentSnapshot) -> tuple[ProposedChange, ...]:
|
|
|
|
|
if snapshot.markdown != "abc":
|
|
|
|
|
return ()
|
|
|
|
|
edits = (
|
|
|
|
|
TextEdit(snapshot.sha256, TextSpan(0, 1), "a", "A"),
|
|
|
|
|
TextEdit(snapshot.sha256, TextSpan(2, 3), "c", "C"),
|
|
|
|
|
)
|
|
|
|
|
return (ProposedChange(snapshot.sha256, "two edits", edits),)
|
|
|
|
|
|
|
|
|
|
modifier = Modifier(
|
|
|
|
|
modifier_id="test.order",
|
|
|
|
|
version="1.0.0",
|
|
|
|
|
parameters=(),
|
|
|
|
|
applicability="只测试报告顺序。",
|
|
|
|
|
propose=propose,
|
|
|
|
|
)
|
|
|
|
|
result = Pipeline((modifier,)).transform("abc")
|
|
|
|
|
malformed = replace(result, changes=tuple(reversed(result.changes)))
|
|
|
|
|
|
|
|
|
|
with pytest.raises(ReviewBuildError, match="canonical report order"):
|
|
|
|
|
build_review_document("abc", malformed)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def test_builder_rejects_tampered_residual_snapshot_and_error_position() -> None:
|
|
|
|
|
unstable = residual_result()
|
|
|
|
|
first_residual = unstable.residual_proposals[0]
|
|
|
|
|
stale_ref = replace(first_residual.proposal_ref, snapshot_sha256=unstable.input_sha256)
|
|
|
|
|
malformed_residual = replace(
|
|
|
|
|
unstable,
|
|
|
|
|
residual_proposals=(replace(first_residual, proposal_ref=stale_ref), *unstable.residual_proposals[1:]),
|
|
|
|
|
)
|
|
|
|
|
with pytest.raises(ReviewBuildError, match="current snapshot"):
|
|
|
|
|
build_review_document("a", malformed_residual)
|
|
|
|
|
|
|
|
|
|
failed = final_review_failure_result()
|
|
|
|
|
out_of_range_error = replace(
|
|
|
|
|
failed.errors[0],
|
|
|
|
|
modifier_id="<invalid>",
|
|
|
|
|
modifier_version="<invalid>",
|
|
|
|
|
modifier_position=len(failed.modifiers),
|
|
|
|
|
)
|
|
|
|
|
malformed_error = replace(failed, errors=(out_of_range_error,))
|
|
|
|
|
with pytest.raises(ReviewBuildError, match="outside modifier metadata"):
|
|
|
|
|
build_review_document("start", malformed_error)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def test_report_contains_complete_sources_stages_and_success_diff_label() -> None:
|
|
|
|
|
markdown = "before\n```md\n<div>x</div>\n[link](https://example.test)\n"
|
|
|
|
|
result = Pipeline(
|
|
|
|
|
(replace_first("before", "after", modifier_id="test.report", reason="render exact source"),)
|
|
|
|
|
).transform(markdown)
|
|
|
|
|
review = build_review_document(markdown, result)
|
|
|
|
|
|
|
|
|
|
report = render_markdown_report(review)
|
|
|
|
|
|
|
|
|
|
assert "status: `success`" in report
|
|
|
|
|
assert "current_kind: `success_output`" in report
|
|
|
|
|
assert "--- a/input.md" in report
|
|
|
|
|
assert "+++ b/output.md" in report
|
|
|
|
|
assert "````markdown" in report
|
|
|
|
|
assert markdown in report
|
|
|
|
|
assert review.current_markdown in report
|
|
|
|
|
assert "render exact source" in report
|
|
|
|
|
assert "location: line `1`, column `1`" in report
|
|
|
|
|
assert "<div>x</div>" in report
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
|
|
|
("result", "input_markdown", "expected_label", "expected_status"),
|
|
|
|
|
[
|
|
|
|
|
(
|
|
|
|
|
Pipeline(
|
|
|
|
|
(
|
|
|
|
|
replace_first("a", "b", modifier_id="test.first"),
|
|
|
|
|
exploding_modifier("test.error"),
|
|
|
|
|
)
|
|
|
|
|
).transform("a"),
|
|
|
|
|
"a",
|
|
|
|
|
"+++ b/failed.partial.md",
|
|
|
|
|
"status: `failed`",
|
|
|
|
|
),
|
|
|
|
|
(residual_result(), "a", "+++ b/unstable.partial.md", "status: `unstable`"),
|
|
|
|
|
],
|
|
|
|
|
)
|
|
|
|
|
def test_partial_report_labels_cannot_be_mistaken_for_success(
|
|
|
|
|
result: TransformResult,
|
|
|
|
|
input_markdown: str,
|
|
|
|
|
expected_label: str,
|
|
|
|
|
expected_status: str,
|
|
|
|
|
) -> None:
|
|
|
|
|
report = render_markdown_report(build_review_document(input_markdown, result))
|
|
|
|
|
|
|
|
|
|
assert expected_label in report
|
|
|
|
|
assert expected_status in report
|
|
|
|
|
assert "current_kind: `partial_output`" in report
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def test_unchanged_failed_diff_still_has_partial_label() -> None:
|
|
|
|
|
duplicate = replace_first("a", "b", modifier_id="test.duplicate")
|
|
|
|
|
review = build_review_document("a", Pipeline((duplicate, duplicate)).transform("a"))
|
|
|
|
|
|
|
|
|
|
report = render_markdown_report(review)
|
|
|
|
|
|
|
|
|
|
assert "+++ b/failed.partial.md" in report
|
|
|
|
|
assert "(no textual difference)" in report
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def test_diff_marks_physical_line_endings_and_missing_final_newline() -> None:
|
|
|
|
|
result = Pipeline((replace_first("a", "A", modifier_id="test.replace"),)).transform("a\r\nb")
|
|
|
|
|
report = render_markdown_report(build_review_document("a\r\nb", result))
|
|
|
|
|
|
|
|
|
|
assert "-a ⟦CRLF⟧" in report
|
|
|
|
|
assert "+A ⟦CRLF⟧" in report
|
|
|
|
|
assert " b ⟦NO EOL⟧" in report
|
|
|
|
|
assert "Final line ending: `none`" in report
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def test_report_limits_residuals_without_printing_replacement_text() -> None:
|
|
|
|
|
review = build_review_document("a", residual_result())
|
|
|
|
|
|
|
|
|
|
report = render_markdown_report(review, residual_limit=1)
|
|
|
|
|
count_only = render_markdown_report(review, residual_limit=0)
|
|
|
|
|
|
|
|
|
|
assert "- total: `3`" in report
|
|
|
|
|
assert report.count("### Residual ") == 1
|
|
|
|
|
assert "Omitted residual proposals: `2`." in report
|
|
|
|
|
assert "expected length `1`" in report
|
|
|
|
|
assert "replacement length `26`" in report
|
|
|
|
|
assert "DO-NOT-PRINT-RESIDUAL-TEXT" not in report
|
|
|
|
|
assert "### Residual 1" not in count_only
|
|
|
|
|
assert "Omitted residual proposals: `3`." in count_only
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
@pytest.mark.parametrize("value", [True, 1.5, "1", None])
|
|
|
|
|
def test_report_rejects_non_integer_residual_limit(value: object) -> None:
|
|
|
|
|
review = build_review_document("", Pipeline(()).transform(""))
|
|
|
|
|
|
|
|
|
|
with pytest.raises(TypeError, match="integer"):
|
|
|
|
|
render_markdown_report(review, residual_limit=value) # type: ignore[arg-type]
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def test_report_rejects_negative_residual_limit() -> None:
|
|
|
|
|
review = build_review_document("", Pipeline(()).transform(""))
|
|
|
|
|
|
|
|
|
|
with pytest.raises(ValueError, match="non-negative"):
|
|
|
|
|
render_markdown_report(review, residual_limit=-1)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def test_same_result_builds_equal_review_and_report() -> None:
|
|
|
|
|
result = Pipeline((replace_first("a", "b", modifier_id="test.replace"),)).transform("a")
|
|
|
|
|
|
|
|
|
|
first = build_review_document("a", result)
|
|
|
|
|
second = build_review_document("a", result)
|
|
|
|
|
|
|
|
|
|
assert first == second
|
|
|
|
|
assert render_markdown_report(first) == render_markdown_report(second)
|
