513 lines
18 KiB
Python
513 lines
18 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import replace
|
|
from json import loads
|
|
from typing import cast
|
|
|
|
import pytest
|
|
|
|
from mdpolish import (
|
|
DocumentSnapshot,
|
|
Modifier,
|
|
ModifierInfo,
|
|
Pipeline,
|
|
ProposedChange,
|
|
RunStatus,
|
|
TextEdit,
|
|
TextSpan,
|
|
)
|
|
from mdpolish.review import (
|
|
JsonValue,
|
|
ReviewCurrentKind,
|
|
ReviewDetail,
|
|
ReviewDocument,
|
|
ReviewLocation,
|
|
ReviewProjection,
|
|
ReviewProjectionError,
|
|
build_review_document,
|
|
render_json_report,
|
|
review_document_to_dict,
|
|
)
|
|
|
|
_MAX_SAFE_JSON_INTEGER = 2**53 - 1
|
|
|
|
|
|
def _replace_modifier(
|
|
needle: str,
|
|
replacement: str,
|
|
*,
|
|
modifier_id: str,
|
|
reason: str,
|
|
parameters: dict[str, object] | None = None,
|
|
applicability: str = "只处理虚构投影测试标记。",
|
|
) -> 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,
|
|
edits=(edit,),
|
|
),
|
|
)
|
|
|
|
return Modifier(
|
|
modifier_id=modifier_id,
|
|
version="1.2.3",
|
|
parameters=parameters or {"needle": needle, "replacement": replacement},
|
|
applicability=applicability,
|
|
propose=propose,
|
|
)
|
|
|
|
|
|
def _rich_success_review() -> ReviewDocument:
|
|
modifier = _replace_modifier(
|
|
"BEFORE_SECRET",
|
|
"AFTER_SECRET",
|
|
modifier_id="test.machine-projection",
|
|
reason="REASON_SECRET",
|
|
parameters={
|
|
"enabled": True,
|
|
"finite": 1.25,
|
|
"mapping": {"alpha": 1},
|
|
"none": None,
|
|
"sequence": (("alpha", 1),),
|
|
},
|
|
applicability="APPLICABILITY_SECRET",
|
|
)
|
|
input_markdown = "中文_SOURCE_ONLY\r\nBEFORE_SECRET\tTAIL"
|
|
result = Pipeline((modifier,)).transform(input_markdown)
|
|
assert result.status is RunStatus.SUCCESS
|
|
return build_review_document(input_markdown, result)
|
|
|
|
|
|
def _unstable_review() -> ReviewDocument:
|
|
def propose_residual(snapshot: DocumentSnapshot) -> tuple[ProposedChange, ...]:
|
|
needle = "RESIDUAL_EXPECTED_SECRET"
|
|
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="RESIDUAL_REPLACEMENT_SECRET",
|
|
)
|
|
return (
|
|
ProposedChange(
|
|
snapshot_sha256=snapshot.sha256,
|
|
reason="RESIDUAL_REASON_SECRET",
|
|
edits=(edit,),
|
|
),
|
|
)
|
|
|
|
residual = Modifier(
|
|
modifier_id="test.residual",
|
|
version="1.0.0",
|
|
parameters=(),
|
|
applicability="只在最终复查中产生虚构候选。",
|
|
propose=propose_residual,
|
|
)
|
|
producer = _replace_modifier(
|
|
"start",
|
|
"RESIDUAL_EXPECTED_SECRET",
|
|
modifier_id="test.producer",
|
|
reason="produce residual marker",
|
|
)
|
|
result = Pipeline((residual, producer)).transform("start")
|
|
assert result.status is RunStatus.UNSTABLE
|
|
return build_review_document("start", result)
|
|
|
|
|
|
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"ERROR_MESSAGE_SECRET: {snapshot.markdown}")
|
|
return ()
|
|
|
|
return Modifier(
|
|
modifier_id=modifier_id,
|
|
version="1.0.0",
|
|
parameters=(),
|
|
applicability="只测试机器错误投影。",
|
|
propose=propose,
|
|
)
|
|
|
|
|
|
def _object(value: JsonValue) -> ReviewProjection:
|
|
assert isinstance(value, dict)
|
|
return value
|
|
|
|
|
|
def _array(value: JsonValue) -> list[JsonValue]:
|
|
assert isinstance(value, list)
|
|
return value
|
|
|
|
|
|
def _contains_key(value: JsonValue, key: str) -> bool:
|
|
if isinstance(value, dict):
|
|
return key in value or any(_contains_key(nested, key) for nested in value.values())
|
|
if isinstance(value, list):
|
|
return any(_contains_key(nested, key) for nested in value)
|
|
return False
|
|
|
|
|
|
def test_summary_projection_has_stable_schema_without_body_content() -> None:
|
|
review = _rich_success_review()
|
|
|
|
projection = review_document_to_dict(review)
|
|
|
|
assert set(projection) == {
|
|
"schema_name",
|
|
"schema_version",
|
|
"detail",
|
|
"status",
|
|
"current_kind",
|
|
"stages_complete",
|
|
"hash_contract",
|
|
"coordinate_contract",
|
|
"input",
|
|
"current",
|
|
"counts",
|
|
"modifiers",
|
|
"stages",
|
|
"errors",
|
|
}
|
|
assert projection["schema_name"] == "mdpolish.review"
|
|
assert projection["schema_version"] == "1.0"
|
|
assert projection["detail"] == "summary"
|
|
assert projection["status"] == "success"
|
|
assert projection["current_kind"] == "success_output"
|
|
assert projection["stages_complete"] is True
|
|
assert projection["hash_contract"] == {
|
|
"algorithm": "sha256",
|
|
"encoding": "utf-8",
|
|
"normalization": "none",
|
|
}
|
|
assert projection["coordinate_contract"] == {
|
|
"offset_unit": "unicode_code_point",
|
|
"span_index_base": 0,
|
|
"span_end": "exclusive",
|
|
"location_index_base": 1,
|
|
"physical_line_endings": ["lf", "crlf", "cr"],
|
|
}
|
|
assert projection["counts"] == {
|
|
"modifier_count": 1,
|
|
"completed_stage_count": 1,
|
|
"change_count": 1,
|
|
"error_count": 0,
|
|
"residual_proposal_count": 0,
|
|
}
|
|
|
|
input_projection = _object(projection["input"])
|
|
current_projection = _object(projection["current"])
|
|
assert input_projection == {
|
|
"sha256": review.input_sha256,
|
|
"code_point_length": len(review.input_markdown),
|
|
}
|
|
assert current_projection == {
|
|
"sha256": review.current_sha256,
|
|
"code_point_length": len(review.current_markdown),
|
|
}
|
|
|
|
modifiers = _array(projection["modifiers"])
|
|
assert modifiers == [
|
|
{
|
|
"position": 0,
|
|
"modifier_id": "test.machine-projection",
|
|
"version": "1.2.3",
|
|
}
|
|
]
|
|
stages = _array(projection["stages"])
|
|
stage = _object(stages[0])
|
|
assert set(stage) == {"modifier_position", "before", "after", "change_count"}
|
|
assert stage["modifier_position"] == 0
|
|
assert stage["change_count"] == 1
|
|
assert stage["before"] == {
|
|
"sha256": review.stages[0].before_sha256,
|
|
"code_point_length": len(review.stages[0].before_markdown),
|
|
}
|
|
assert stage["after"] == {
|
|
"sha256": review.stages[0].after_sha256,
|
|
"code_point_length": len(review.stages[0].after_markdown),
|
|
}
|
|
|
|
rendered = render_json_report(review)
|
|
for secret in (
|
|
"中文_SOURCE_ONLY",
|
|
"BEFORE_SECRET",
|
|
"AFTER_SECRET",
|
|
"REASON_SECRET",
|
|
"APPLICABILITY_SECRET",
|
|
"ERROR_MESSAGE_SECRET",
|
|
):
|
|
assert secret not in rendered
|
|
for body_key in (
|
|
"markdown",
|
|
"expected_text",
|
|
"replacement",
|
|
"reason",
|
|
"message",
|
|
"parameters",
|
|
"applicability",
|
|
):
|
|
assert not _contains_key(projection, body_key)
|
|
|
|
|
|
def test_changes_and_full_details_add_content_monotonically() -> None:
|
|
review = _rich_success_review()
|
|
|
|
summary = review_document_to_dict(review, detail=ReviewDetail.SUMMARY)
|
|
changes = review_document_to_dict(review, detail="changes")
|
|
full = review_document_to_dict(review, detail=ReviewDetail.FULL)
|
|
|
|
assert set(summary) < set(changes)
|
|
assert set(changes) == set(full)
|
|
assert changes["detail"] == "changes"
|
|
assert full["detail"] == "full"
|
|
for key in set(summary) - {"detail", "input", "current", "modifiers", "stages"}:
|
|
assert summary[key] == changes[key] == full[key]
|
|
|
|
changes_modifier = _object(_array(changes["modifiers"])[0])
|
|
full_modifier = _object(_array(full["modifiers"])[0])
|
|
assert changes_modifier == full_modifier
|
|
assert changes_modifier["applicability"] == "APPLICABILITY_SECRET"
|
|
assert changes_modifier["parameters"] == [
|
|
["enabled", True],
|
|
["finite", 1.25],
|
|
["mapping", [["alpha", 1]]],
|
|
["none", None],
|
|
["sequence", [["alpha", 1]]],
|
|
]
|
|
|
|
changes_stage = _object(_array(changes["stages"])[0])
|
|
full_stage = _object(_array(full["stages"])[0])
|
|
assert "markdown" not in _object(changes_stage["before"])
|
|
assert "markdown" not in _object(changes_stage["after"])
|
|
assert not _contains_key(changes, "markdown")
|
|
assert _object(full_stage["before"])["markdown"] == review.input_markdown
|
|
assert _object(full_stage["after"])["markdown"] == review.current_markdown
|
|
assert _object(full["input"])["markdown"] == review.input_markdown
|
|
assert _object(full["current"])["markdown"] == review.current_markdown
|
|
|
|
change = _object(_array(changes_stage["changes"])[0])
|
|
assert change == {
|
|
"proposal_index": 0,
|
|
"edit_index": 0,
|
|
"reason": "REASON_SECRET",
|
|
"location": {"line": 2, "column": 1},
|
|
"span": {
|
|
"start": review.input_markdown.index("BEFORE_SECRET"),
|
|
"end": review.input_markdown.index("BEFORE_SECRET") + len("BEFORE_SECRET"),
|
|
},
|
|
"before": "BEFORE_SECRET",
|
|
"after": "AFTER_SECRET",
|
|
"before_sha256": review.stages[0].before_sha256,
|
|
"after_sha256": review.stages[0].after_sha256,
|
|
}
|
|
assert "中文_SOURCE_ONLY" not in render_json_report(review, detail="changes")
|
|
|
|
|
|
def test_residual_content_requires_changes_or_full_detail() -> None:
|
|
review = _unstable_review()
|
|
|
|
summary = review_document_to_dict(review, detail="summary")
|
|
changes = review_document_to_dict(review, detail="changes")
|
|
|
|
assert summary["status"] == "unstable"
|
|
assert summary["current_kind"] == "partial_output"
|
|
assert _object(summary["counts"])["residual_proposal_count"] == 1
|
|
assert "residual_proposals" not in summary
|
|
assert "RESIDUAL_EXPECTED_SECRET" not in render_json_report(review, detail="summary")
|
|
|
|
residuals = _array(changes["residual_proposals"])
|
|
assert len(residuals) == 1
|
|
residual = _object(residuals[0])
|
|
assert residual["modifier_position"] == 0
|
|
assert residual["proposal_index"] == 0
|
|
assert residual["snapshot_sha256"] == review.current_sha256
|
|
assert residual["reason"] == "RESIDUAL_REASON_SECRET"
|
|
edit = _object(_array(residual["edits"])[0])
|
|
assert edit["edit_index"] == 0
|
|
assert edit["expected_text"] == "RESIDUAL_EXPECTED_SECRET"
|
|
assert edit["replacement"] == "RESIDUAL_REPLACEMENT_SECRET"
|
|
|
|
|
|
def _error_reviews() -> tuple[tuple[ReviewDocument, str], ...]:
|
|
invalid_pipeline = Pipeline(cast(tuple[Modifier, ...], (object(),)))
|
|
preflight_result = invalid_pipeline.transform("PREFLIGHT_SOURCE_SECRET")
|
|
preflight = build_review_document("PREFLIGHT_SOURCE_SECRET", preflight_result)
|
|
|
|
transform_result = Pipeline((_exploding_modifier("test.transform-error"),)).transform(
|
|
"TRANSFORM_SOURCE_SECRET"
|
|
)
|
|
transform = build_review_document("TRANSFORM_SOURCE_SECRET", transform_result)
|
|
|
|
review_error = _exploding_modifier("test.final-review-error", trigger="done")
|
|
producer = _replace_modifier(
|
|
"start",
|
|
"done",
|
|
modifier_id="test.final-producer",
|
|
reason="produce final review trigger",
|
|
)
|
|
final_result = Pipeline((review_error, producer)).transform("start")
|
|
final_review = build_review_document("start", final_result)
|
|
return (
|
|
(preflight, "run.preflight_failed"),
|
|
(transform, "run.transform_failed"),
|
|
(final_review, "run.final_review_failed"),
|
|
)
|
|
|
|
|
|
def test_error_projection_uses_stable_codes_and_hides_diagnostics_in_summary() -> None:
|
|
for review, expected_code in _error_reviews():
|
|
summary = review_document_to_dict(review, detail="summary")
|
|
changes = review_document_to_dict(review, detail="changes")
|
|
|
|
summary_error = _object(_array(summary["errors"])[0])
|
|
changes_error = _object(_array(changes["errors"])[0])
|
|
assert summary_error["code"] == expected_code
|
|
assert summary_error["stage"] == review.errors[0].stage.value
|
|
assert "diagnostic_type" not in summary_error
|
|
assert "message" not in summary_error
|
|
assert "ERROR_MESSAGE_SECRET" not in render_json_report(review, detail="summary")
|
|
assert changes_error["diagnostic_type"] == review.errors[0].error_type
|
|
assert changes_error["message"] == review.errors[0].message
|
|
|
|
|
|
def test_projection_preserves_unicode_line_endings_and_exact_coordinates() -> None:
|
|
modifier = _replace_modifier(
|
|
"🙂Cafe\u0301",
|
|
"完成",
|
|
modifier_id="test.unicode",
|
|
reason="Unicode coordinate test",
|
|
)
|
|
input_markdown = "\ufeff首行\r\n🙂Cafe\u0301\r尾行\n"
|
|
result = Pipeline((modifier,)).transform(input_markdown)
|
|
review = build_review_document(input_markdown, result)
|
|
|
|
projection = review_document_to_dict(review, detail="full")
|
|
stage = _object(_array(projection["stages"])[0])
|
|
change = _object(_array(stage["changes"])[0])
|
|
|
|
assert _object(projection["input"])["code_point_length"] == len(input_markdown)
|
|
assert _object(projection["input"])["markdown"] == input_markdown
|
|
assert change["location"] == {"line": 2, "column": 1}
|
|
assert change["span"] == {
|
|
"start": input_markdown.index("🙂"),
|
|
"end": input_markdown.index("🙂") + len("🙂Cafe\u0301"),
|
|
}
|
|
|
|
|
|
def test_json_report_is_deterministic_unicode_json_without_bom_or_final_newline() -> None:
|
|
review = _rich_success_review()
|
|
|
|
first = render_json_report(review, detail="full")
|
|
second = render_json_report(review, detail=ReviewDetail.FULL)
|
|
|
|
assert first == second
|
|
assert loads(first) == review_document_to_dict(review, detail="full")
|
|
assert not first.startswith("\ufeff")
|
|
assert not first.endswith("\n")
|
|
assert "中文_SOURCE_ONLY" in first
|
|
assert "\\u4e2d" not in first.lower()
|
|
assert "\\tTAIL" in first
|
|
assert "NaN" not in first
|
|
assert "Infinity" not in first
|
|
|
|
|
|
def test_projection_returns_fresh_containers_without_mutating_review() -> None:
|
|
review = _rich_success_review()
|
|
|
|
first = review_document_to_dict(review, detail="full")
|
|
second = review_document_to_dict(review, detail="full")
|
|
first["schema_name"] = "changed"
|
|
_array(first["modifiers"]).clear()
|
|
|
|
assert second["schema_name"] == "mdpolish.review"
|
|
assert len(_array(second["modifiers"])) == 1
|
|
assert review.modifiers[0].modifier_id == "test.machine-projection"
|
|
assert review_document_to_dict(review, detail="full") == second
|
|
|
|
|
|
@pytest.mark.parametrize("detail", ["", "SUMMARY", "unknown", 1, True, None])
|
|
def test_projection_rejects_unknown_detail(detail: object) -> None:
|
|
review = _rich_success_review()
|
|
|
|
with pytest.raises(ReviewProjectionError, match="review projection failed at detail"):
|
|
review_document_to_dict(review, detail=detail) # type: ignore[arg-type]
|
|
|
|
|
|
def test_projection_rejects_wrong_review_and_nested_model_types() -> None:
|
|
with pytest.raises(ReviewProjectionError, match="ReviewDocument"):
|
|
review_document_to_dict(object()) # type: ignore[arg-type]
|
|
|
|
review = _rich_success_review()
|
|
malformed = replace(review, modifiers=cast(tuple[ModifierInfo, ...], (object(),)))
|
|
with pytest.raises(ReviewProjectionError, match=r"modifiers\[0\]"):
|
|
review_document_to_dict(malformed)
|
|
|
|
|
|
def test_projection_rejects_unknown_enum_invalid_reference_and_large_integer() -> None:
|
|
review = _rich_success_review()
|
|
unknown_status = replace(review, status=cast(RunStatus, "future"))
|
|
bad_stage = replace(review.stages[0], modifier_position=1)
|
|
bad_reference = replace(review, stages=(bad_stage,))
|
|
review_change = review.stages[0].changes[0]
|
|
large_location = replace(
|
|
review_change,
|
|
location=ReviewLocation(line=_MAX_SAFE_JSON_INTEGER + 1, column=1),
|
|
)
|
|
large_stage = replace(review.stages[0], changes=(large_location,))
|
|
large_integer = replace(review, stages=(large_stage,))
|
|
|
|
for malformed in (unknown_status, bad_reference, large_integer):
|
|
with pytest.raises(ReviewProjectionError):
|
|
review_document_to_dict(malformed)
|
|
|
|
|
|
def test_projection_rejects_nonfinite_parameter_and_invalid_unicode_without_leaking_values() -> None:
|
|
review = _rich_success_review()
|
|
modifier = review.modifiers[0]
|
|
nonfinite_modifier = replace(modifier, parameters=(("NONFINITE_SECRET", float("nan")),))
|
|
nonfinite_stage = replace(review.stages[0], modifier=nonfinite_modifier)
|
|
nonfinite_review = replace(
|
|
review,
|
|
modifiers=(nonfinite_modifier,),
|
|
stages=(nonfinite_stage,),
|
|
)
|
|
invalid_unicode = replace(review, input_markdown="UNICODE_SECRET\ud800")
|
|
|
|
with pytest.raises(ReviewProjectionError) as nonfinite_error:
|
|
review_document_to_dict(nonfinite_review)
|
|
assert "NONFINITE_SECRET" not in str(nonfinite_error.value)
|
|
|
|
with pytest.raises(ReviewProjectionError) as unicode_error:
|
|
review_document_to_dict(invalid_unicode)
|
|
assert "UNICODE_SECRET" not in str(unicode_error.value)
|
|
|
|
|
|
def test_empty_document_projection_distinguishes_hidden_and_empty_markdown() -> None:
|
|
result = Pipeline(()).transform("")
|
|
review = build_review_document("", result)
|
|
|
|
summary = review_document_to_dict(review, detail="summary")
|
|
full = review_document_to_dict(review, detail="full")
|
|
|
|
assert "markdown" not in _object(summary["input"])
|
|
assert _object(full["input"])["markdown"] == ""
|
|
assert _object(full["current"])["markdown"] == ""
|
|
assert full["modifiers"] == []
|
|
assert full["stages"] == []
|
|
assert full["errors"] == []
|
|
assert full["residual_proposals"] == []
|
|
assert review.current_kind is ReviewCurrentKind.SUCCESS_OUTPUT
|