test_analysis_run_dto.py
3.21 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
from __future__ import annotations
from datetime import datetime
from services.shared.dto import AnalysisRunDTO, AnalysisRunListDTO
from services.shared.models import AnalysisRun, AnalysisRunStatus, ProgressInfo
def test_analysis_run_dto_from_model_and_response_item():
run = AnalysisRun(
id="analysis-123",
research_task_id="task-123",
engines=["query", "insight"],
status=AnalysisRunStatus.PARTIAL,
progress=ProgressInfo(
current=1,
total=2,
percent=50.0,
stage="running",
message="query finished",
),
partial_results={
"query": {
"success": True,
"status": "completed",
"summary": "query finished",
"artifacts": {"report_preview": "query preview"},
"metrics": {"duration_seconds": 1.2},
}
},
metrics={"success_count": 1, "failure_count": 1},
last_action="query finished",
created_at=datetime(2026, 4, 16, 10, 0, 0),
updated_at=datetime(2026, 4, 16, 10, 5, 0),
started_at=datetime(2026, 4, 16, 10, 1, 0),
finished_at=datetime(2026, 4, 16, 10, 6, 0),
legacy_payload={"query": "museum research"},
)
dto = AnalysisRunDTO.from_model(run)
payload = dto.to_response_item()
assert dto.id == "analysis-123"
assert dto.status == "partial"
assert payload["run_id"] == "analysis-123"
assert payload["engines"] == ["query", "insight"]
assert payload["partial_results"]["query"]["success"] is True
assert payload["partial_results"]["query"]["engine_name"] == "query"
assert payload["partial_results"]["query"]["artifacts"]["report_preview"] == "query preview"
assert payload["unified_run"]["research_task_id"] == "task-123"
assert payload["query"] == "museum research"
def test_analysis_run_list_dto_builds_history_payload():
first = AnalysisRunDTO(
id="analysis-1",
research_task_id="task-1",
status="completed",
)
second = AnalysisRunDTO(
id="analysis-2",
research_task_id="task-1",
status="failed",
)
payload = AnalysisRunListDTO(analysis_runs=[first, second]).to_response_payload()
assert [item["id"] for item in payload["analysis_runs"]] == ["analysis-1", "analysis-2"]
def test_analysis_run_dto_normalizes_legacy_partial_result_payloads_to_engine_results():
dto = AnalysisRunDTO(
id="analysis-legacy",
research_task_id="task-legacy",
partial_results={
"query": "timeout",
"insight": {
"success": True,
"summary": "insight done",
},
},
)
engine_results = dto.get_engine_results()
payload = dto.to_response_item()
assert engine_results["query"].success is False
assert engine_results["query"].error is not None
assert engine_results["query"].error.message == "timeout"
assert engine_results["insight"].success is True
assert payload["partial_results"]["query"]["status"] == "failed"
assert payload["partial_results"]["query"]["engine_name"] == "query"
assert payload["partial_results"]["insight"]["engine_name"] == "insight"