jiang
b853b61a0b
Accumulate investigation paths and preserve failed attempts across rounds. Keep graph annotations separate from execution state, synchronize the API contract, and consolidate documentation.
46 lines
2.2 KiB
Python
46 lines
2.2 KiB
Python
from copy import deepcopy
|
|
from datetime import datetime, timezone
|
|
|
|
import pytest
|
|
|
|
from scripts.migrate_investigation_graphs import prepare_progress_document
|
|
from tests.support.operational_task_repository import _active_run, _context, _task_document
|
|
|
|
|
|
@pytest.mark.parametrize("node_refs", [[], [
|
|
{"node_id": "old", "label": "旧尝试", "status": "failed", "detail": "原查询超时", "evidence_refs": ["e1"]},
|
|
{"node_id": "next", "label": "后续核查", "status": "running"},
|
|
]])
|
|
def test_migration_keeps_source_paths_failures_and_original_publication_provenance(node_refs):
|
|
now = datetime(2026, 11, 10, tzinfo=timezone.utc)
|
|
task = _task_document(_context(), now)
|
|
run = _active_run(task, now)
|
|
document = {
|
|
"revision": 7, "run_id": run["run_id"], "execution_generation": 1,
|
|
"source_session_id": "session-1", "published_at": now.isoformat(),
|
|
"progress": {"summary": "原调查说明", "diagram": {"title": "旧图", "node_refs": node_refs,
|
|
"mermaid": "flowchart LR\nold{旧尝试} -->|失败后重试| next[后续核查]"}},
|
|
}
|
|
original = deepcopy(document)
|
|
migrated = prepare_progress_document(document)
|
|
assert document == original
|
|
assert migrated["revision"] == 8
|
|
assert {key: migrated[key] for key in original if key not in {"revision", "progress"}} == {
|
|
key: original[key] for key in original if key not in {"revision", "progress"}
|
|
}
|
|
assert migrated["progress"]["summary"] == original["progress"]["summary"]
|
|
graph = migrated["progress"]["diagram"]
|
|
assert graph["mermaid"] == graph["source_mermaid"] == original["progress"]["diagram"]["mermaid"]
|
|
assert graph["edges"] == []
|
|
if node_refs:
|
|
assert graph["node_refs"][0]["detail"] == "原查询超时"
|
|
assert graph["node_refs"][0]["evidence_refs"] == ["e1"]
|
|
assert [node["status"] for node in graph["node_refs"]] == ["failed", "reference"]
|
|
assert prepare_progress_document(migrated) is None
|
|
|
|
|
|
def test_migration_does_not_invent_a_graph_when_none_was_saved():
|
|
document = {"progress": {"summary": "仅说明", "diagram": None}}
|
|
assert prepare_progress_document(document) is None
|
|
assert document == {"progress": {"summary": "仅说明", "diagram": None}}
|