73 lines
3.6 KiB
Python
73 lines
3.6 KiB
Python
"""Presentation prose is validated on writes without rejecting historical scenes."""
|
||
from copy import deepcopy
|
||
from datetime import datetime, timezone
|
||
from uuid import uuid4
|
||
|
||
import pytest
|
||
|
||
from app.domain.schemas.operational_task import TaskPresentationEventCommand, TaskPresentationHistoryItem
|
||
|
||
|
||
def presentation():
|
||
return {
|
||
"request_id": uuid4(), "run_id": uuid4(), "source_session_id": "fixture-session",
|
||
"purpose": "核对压力变化与异常时刻的对应关系。",
|
||
"narration": {"stage": "evidence_verification", "summary": "压力已回升，恢复情况仍需结合现场反馈复核。",
|
||
"next_step": "核对完工后的恢复监测。", "evidence_refs": ["recovery:sha256:fixture"]},
|
||
"command": {"action": "open_chart", "chart_id": "pressure", "evidence_refs": ["recovery:sha256:fixture"],
|
||
"chart": {"title": "170486 压力恢复趋势", "chart_type": "line", "x_axis_name": "时间",
|
||
"y_axis_name": "压力（m）", "x_data": ["13:00", "13:15"],
|
||
"series": [{"name": "170486 当前压力", "data": [20.09, 26.12],
|
||
"sample_times": ["2026-07-16T05:00:00Z", "2026-07-16T05:15:00Z"]}]}},
|
||
}
|
||
|
||
|
||
@pytest.mark.parametrize("text", [
|
||
"status=passed，required_samples=1。", "Skill scada-anomaly v1.0.16 已执行。",
|
||
"已读取 recovery:sha256:01234567。", "恢复时刻 2026-07-16T05:15:00Z。",
|
||
"恢复时刻 13:15（UTC+8）。", "**恢复复核通过**。",
|
||
])
|
||
def test_presentation_rejects_execution_text_in_purpose_finding_and_follow_up(text):
|
||
value = presentation()
|
||
for field in ("purpose", "summary", "next_step"):
|
||
changed = deepcopy(value)
|
||
if field == "purpose":
|
||
changed[field] = text
|
||
else:
|
||
changed["narration"][field] = text
|
||
history = TaskPresentationHistoryItem(
|
||
event_seq=1, run_id=changed["run_id"], execution_generation=1,
|
||
occurred_at=datetime(2026, 7, 16, tzinfo=timezone.utc),
|
||
command=changed["command"], purpose=changed["purpose"], narration=changed["narration"],
|
||
)
|
||
assert history.purpose == changed["purpose"]
|
||
with pytest.raises(ValueError):
|
||
TaskPresentationEventCommand.model_validate(changed)
|
||
|
||
|
||
@pytest.mark.parametrize("field", ["title", "x_axis_name", "y_axis_name", "x_data", "series"])
|
||
def test_new_chart_visible_labels_follow_the_same_prose_contract(field):
|
||
value = presentation()
|
||
chart = value["command"]["chart"]
|
||
if field == "x_data":
|
||
chart[field][0] = "2026-07-16T05:00:00Z"
|
||
elif field == "series":
|
||
chart[field][0]["name"] = "execution_generation=3"
|
||
else:
|
||
chart[field] = "压力复核 status=passed"
|
||
with pytest.raises(ValueError):
|
||
TaskPresentationEventCommand.model_validate(value)
|
||
|
||
|
||
def test_display_validation_preserves_real_timestamps_identifiers_and_values():
|
||
value = presentation()
|
||
parsed = TaskPresentationEventCommand.model_validate(value).model_dump(mode="json")
|
||
assert parsed["command"] == value["command"]
|
||
assert parsed["narration"]["evidence_refs"] == value["narration"]["evidence_refs"]
|
||
value["command"] = {"action": "set_overlay", "overlay_id": "estimated-impact",
|
||
"title": "候选管段预计影响范围", "mode": "geometry", "evidence_id": "mcp:fixture"}
|
||
assert TaskPresentationEventCommand.model_validate(value).command.title == "候选管段预计影响范围"
|
||
value["command"]["title"] = "候选影响范围（UTC+8）"
|
||
with pytest.raises(ValueError):
|
||
TaskPresentationEventCommand.model_validate(value)
|