298 lines
9.9 KiB
Python
298 lines
9.9 KiB
Python
from __future__ import annotations
|
|
|
|
from json import dumps, loads
|
|
from typing import cast
|
|
|
|
import pytest
|
|
|
|
from mdpolish import (
|
|
DocumentSnapshot,
|
|
Modifier,
|
|
Pipeline,
|
|
ProposedChange,
|
|
RunStatus,
|
|
TextEdit,
|
|
TextSpan,
|
|
)
|
|
from mdpolish.review import (
|
|
ReviewDetail,
|
|
ReviewDocument,
|
|
ReviewParseError,
|
|
build_review_document,
|
|
parse_json_report,
|
|
render_json_report,
|
|
)
|
|
|
|
|
|
def _replace_modifier(
|
|
needle: str,
|
|
replacement: str,
|
|
*,
|
|
modifier_id: str,
|
|
reason: str = "应用虚构测试替换",
|
|
) -> Modifier:
|
|
def propose(snapshot: DocumentSnapshot) -> tuple[ProposedChange, ...]:
|
|
start = snapshot.markdown.find(needle)
|
|
if start < 0:
|
|
return ()
|
|
return (
|
|
ProposedChange(
|
|
snapshot_sha256=snapshot.sha256,
|
|
reason=reason,
|
|
edits=(
|
|
TextEdit(
|
|
snapshot_sha256=snapshot.sha256,
|
|
span=TextSpan(start, start + len(needle)),
|
|
expected_text=needle,
|
|
replacement=replacement,
|
|
),
|
|
),
|
|
),
|
|
)
|
|
|
|
return Modifier(
|
|
modifier_id=modifier_id,
|
|
version="1.0.0",
|
|
parameters={"needle": needle, "replacement": replacement},
|
|
applicability="只处理虚构 JSON reader 测试标记。",
|
|
propose=propose,
|
|
)
|
|
|
|
|
|
def _noop_modifier(modifier_id: str = "test.noop") -> Modifier:
|
|
return Modifier(
|
|
modifier_id=modifier_id,
|
|
version="1.0.0",
|
|
parameters=(),
|
|
applicability="不修改任何文本。",
|
|
propose=lambda _snapshot: (),
|
|
)
|
|
|
|
|
|
def _success_review() -> ReviewDocument:
|
|
input_markdown = "\ufeff首行\r\n🙂Cafe\u0301\r末行"
|
|
result = Pipeline(
|
|
(
|
|
_noop_modifier(),
|
|
_replace_modifier("🙂Cafe\u0301", "完成", modifier_id="test.unicode"),
|
|
)
|
|
).transform(input_markdown)
|
|
assert result.status is RunStatus.SUCCESS
|
|
return build_review_document(input_markdown, result)
|
|
|
|
|
|
def _unstable_review() -> ReviewDocument:
|
|
residual_marker = "RESIDUAL_SECRET"
|
|
|
|
def residual(snapshot: DocumentSnapshot) -> tuple[ProposedChange, ...]:
|
|
start = snapshot.markdown.find(residual_marker)
|
|
if start < 0:
|
|
return ()
|
|
return (
|
|
ProposedChange(
|
|
snapshot_sha256=snapshot.sha256,
|
|
reason="残留虚构候选",
|
|
edits=(
|
|
TextEdit(
|
|
snapshot_sha256=snapshot.sha256,
|
|
span=TextSpan(start, start + len(residual_marker)),
|
|
expected_text=residual_marker,
|
|
replacement="resolved",
|
|
),
|
|
),
|
|
),
|
|
)
|
|
|
|
result = Pipeline(
|
|
(
|
|
Modifier(
|
|
modifier_id="test.residual",
|
|
version="1.0.0",
|
|
parameters=(),
|
|
applicability="只在最终复查产生候选。",
|
|
propose=residual,
|
|
),
|
|
_replace_modifier("start", residual_marker, modifier_id="test.producer"),
|
|
)
|
|
).transform("start")
|
|
assert result.status is RunStatus.UNSTABLE
|
|
return build_review_document("start", result)
|
|
|
|
|
|
def _error_reviews() -> tuple[ReviewDocument, ...]:
|
|
preflight_result = Pipeline(cast(tuple[Modifier, ...], (object(),))).transform("preflight")
|
|
|
|
def explode(_snapshot: DocumentSnapshot) -> tuple[ProposedChange, ...]:
|
|
raise RuntimeError("DIAGNOSTIC_SECRET")
|
|
|
|
transform_result = Pipeline(
|
|
(
|
|
Modifier(
|
|
modifier_id="test.transform-error",
|
|
version="1.0.0",
|
|
parameters=(),
|
|
applicability="只测试执行错误。",
|
|
propose=explode,
|
|
),
|
|
)
|
|
).transform("transform")
|
|
|
|
def explode_on_done(snapshot: DocumentSnapshot) -> tuple[ProposedChange, ...]:
|
|
if snapshot.markdown == "done":
|
|
raise RuntimeError("FINAL_DIAGNOSTIC_SECRET")
|
|
return ()
|
|
|
|
final_result = Pipeline(
|
|
(
|
|
Modifier(
|
|
modifier_id="test.final-error",
|
|
version="1.0.0",
|
|
parameters=(),
|
|
applicability="只测试最终复查错误。",
|
|
propose=explode_on_done,
|
|
),
|
|
_replace_modifier("start", "done", modifier_id="test.final-producer"),
|
|
)
|
|
).transform("start")
|
|
return (
|
|
build_review_document("preflight", preflight_result),
|
|
build_review_document("transform", transform_result),
|
|
build_review_document("start", final_result),
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize("detail", tuple(ReviewDetail))
|
|
def test_reader_accepts_every_official_detail(detail: ReviewDetail) -> None:
|
|
review = _success_review()
|
|
report = render_json_report(review, detail=detail)
|
|
|
|
parsed = parse_json_report(report, expected_detail=detail)
|
|
|
|
assert parsed == loads(report)
|
|
assert parsed["detail"] == detail.value
|
|
|
|
|
|
def test_reader_accepts_unstable_summary_without_materializing_residuals() -> None:
|
|
parsed = parse_json_report(render_json_report(_unstable_review(), detail="summary"))
|
|
|
|
assert parsed["status"] == "unstable"
|
|
assert "residual_proposals" not in parsed
|
|
|
|
|
|
@pytest.mark.parametrize("review", _error_reviews())
|
|
@pytest.mark.parametrize("detail", tuple(ReviewDetail))
|
|
def test_reader_accepts_each_failed_stage(review: ReviewDocument, detail: ReviewDetail) -> None:
|
|
parsed = parse_json_report(render_json_report(review, detail=detail))
|
|
|
|
assert parsed["status"] == "failed"
|
|
|
|
|
|
def test_reader_accepts_compatible_minor_and_unknown_object_field() -> None:
|
|
payload = loads(render_json_report(_success_review(), detail="full"))
|
|
payload["schema_version"] = "1.9"
|
|
payload["future_summary"] = {"available": True}
|
|
payload["hash_contract"]["future_hash_note"] = "compatible"
|
|
payload["coordinate_contract"]["future_coordinate_note"] = "compatible"
|
|
|
|
parsed = parse_json_report(dumps(payload, ensure_ascii=False))
|
|
|
|
assert parsed["schema_version"] == "1.9"
|
|
assert parsed["future_summary"] == {"available": True}
|
|
|
|
|
|
def test_reader_rejects_stage_after_that_was_not_produced_by_changes() -> None:
|
|
payload = loads(render_json_report(_success_review(), detail="full"))
|
|
stage = payload["stages"][1]
|
|
forged = "伪造结果"
|
|
forged_hash = DocumentSnapshot(forged).sha256
|
|
stage["after"] = {
|
|
"sha256": forged_hash,
|
|
"code_point_length": len(forged),
|
|
"markdown": forged,
|
|
}
|
|
stage["changes"][0]["after_sha256"] = forged_hash
|
|
payload["current"] = {
|
|
"sha256": forged_hash,
|
|
"code_point_length": len(forged),
|
|
"markdown": forged,
|
|
}
|
|
|
|
with pytest.raises(ReviewParseError, match="replayed changes do not produce"):
|
|
parse_json_report(dumps(payload, ensure_ascii=False), expected_detail="full")
|
|
|
|
|
|
def test_reader_rejects_changed_zero_change_stage() -> None:
|
|
payload = loads(render_json_report(_success_review(), detail="full"))
|
|
stage = payload["stages"][0]
|
|
forged = "改变了零修改阶段"
|
|
forged_hash = DocumentSnapshot(forged).sha256
|
|
stage["after"] = {
|
|
"sha256": forged_hash,
|
|
"code_point_length": len(forged),
|
|
"markdown": forged,
|
|
}
|
|
|
|
with pytest.raises(ReviewParseError, match="zero-change stage"):
|
|
parse_json_report(dumps(payload, ensure_ascii=False), expected_detail="full")
|
|
|
|
|
|
def test_reader_rejects_conflicting_change_batch() -> None:
|
|
payload = loads(render_json_report(_success_review(), detail="full"))
|
|
stage = payload["stages"][1]
|
|
duplicate = dict(stage["changes"][0])
|
|
duplicate["edit_index"] = 1
|
|
stage["changes"].append(duplicate)
|
|
stage["change_count"] = 2
|
|
payload["counts"]["change_count"] = 2
|
|
|
|
with pytest.raises(ReviewParseError, match="edit contract"):
|
|
parse_json_report(dumps(payload, ensure_ascii=False), expected_detail="full")
|
|
|
|
|
|
def test_reader_rejects_wrong_location_without_leaking_text() -> None:
|
|
payload = loads(render_json_report(_success_review(), detail="full"))
|
|
change = payload["stages"][1]["changes"][0]
|
|
change["location"] = {"line": 99, "column": 99}
|
|
|
|
with pytest.raises(ReviewParseError) as error:
|
|
parse_json_report(dumps(payload, ensure_ascii=False), expected_detail="full")
|
|
|
|
assert "🙂Cafe\u0301" not in str(error.value)
|
|
assert "location does not match" in str(error.value)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("report", "message"),
|
|
(
|
|
('{"schema_name":"a","schema_name":"b"}', "keys must be unique"),
|
|
('{"value":NaN}', "non-finite"),
|
|
('{"schema_name":"mdpolish.review","schema_version":"2.0"}', "schema major"),
|
|
(dumps({"schema_name": "\ud800"}), "valid Unicode"),
|
|
("not JSON", "malformed JSON"),
|
|
),
|
|
)
|
|
def test_reader_rejects_malformed_or_unsupported_json(report: str, message: str) -> None:
|
|
with pytest.raises(ReviewParseError, match=message):
|
|
parse_json_report(report)
|
|
|
|
|
|
def test_reader_rejects_detail_mismatch_and_invalid_expected_detail() -> None:
|
|
report = render_json_report(_success_review(), detail="summary")
|
|
|
|
with pytest.raises(ReviewParseError, match="does not match"):
|
|
parse_json_report(report, expected_detail="full")
|
|
with pytest.raises(ReviewParseError, match="expected_detail"):
|
|
parse_json_report(report, expected_detail="unknown")
|
|
|
|
|
|
def test_reader_enforces_body_exposure_boundaries() -> None:
|
|
summary = loads(render_json_report(_success_review(), detail="summary"))
|
|
summary["future"] = {"markdown": "SUMMARY_SECRET"}
|
|
with pytest.raises(ReviewParseError, match="body-bearing"):
|
|
parse_json_report(dumps(summary, ensure_ascii=False))
|
|
|
|
changes = loads(render_json_report(_success_review(), detail="changes"))
|
|
changes["future"] = {"markdown": "CHANGES_SECRET"}
|
|
with pytest.raises(ReviewParseError, match="full markdown"):
|
|
parse_json_report(dumps(changes, ensure_ascii=False))
|