test_task_analysis_resource_dto.py
4.03 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
from __future__ import annotations
from services.shared.dto import AnalysisRunDTO, ResearchTaskDTO, TaskAnalysisResourceDTO
def test_task_analysis_resource_dto_builds_summary_history_and_stats():
current_run = AnalysisRunDTO(
id="analysis-current",
research_task_id="task-1",
engines=["insight", "query"],
status="partial",
partial_results={
"insight": {
"success": True,
"status": "completed",
"summary": "insight done",
"artifacts": {"themes": ["queueing"]},
"metrics": {"duration_seconds": 2.5},
},
"query": "timeout while waiting for upstream reviews",
},
last_action="insight done",
updated_at="2026-04-16T10:05:00",
finished_at="2026-04-16T10:06:00",
)
older_run = AnalysisRunDTO(
id="analysis-older",
research_task_id="task-1",
engines=["query"],
status="failed",
error={"message": "engine failed"},
)
payload = TaskAnalysisResourceDTO.from_parts(
task_id="task-1",
analysis_run_id="analysis-current",
current_run=current_run,
analysis_runs=[current_run, older_run],
).to_response_payload()
assert payload["task_id"] == "task-1"
assert payload["analysis"]["analysis_run_id"] == "analysis-current"
assert payload["analysis"]["status"] == "partial"
assert payload["analysis"]["summary"]["current_run_id"] == "analysis-current"
assert payload["analysis"]["summary"]["latest_run_id"] == "analysis-current"
assert payload["analysis"]["summary"]["current_result_count"] == 2
assert payload["analysis"]["history"]["has_history"] is True
assert payload["analysis"]["history"]["linked_run_available"] is True
assert payload["analysis"]["history"]["linked_run_is_current"] is True
assert payload["analysis"]["stats"]["status_counts"]["partial"] == 1
assert payload["analysis"]["stats"]["status_counts"]["failed"] == 1
assert payload["analysis"]["stats"]["runs_with_results"] == 1
assert payload["analysis"]["stats"]["runs_with_errors"] == 2
assert payload["analysis"]["stats"]["unique_engines"] == ["insight", "query"]
assert payload["analysis"]["stats"]["current_result_engine_count"] == 2
assert payload["analysis"]["stats"]["current_success_result_count"] == 1
assert payload["analysis"]["stats"]["current_failed_result_count"] == 1
assert payload["analysis"]["current_run"]["partial_results"]["insight"]["engine_name"] == "insight"
assert payload["analysis"]["current_run"]["partial_results"]["insight"]["artifacts"]["themes"] == ["queueing"]
assert payload["analysis"]["current_run"]["partial_results"]["query"]["engine_name"] == "query"
assert payload["analysis"]["current_run"]["partial_results"]["query"]["success"] is False
def test_task_analysis_resource_dto_uses_task_fallbacks_when_no_runs_exist():
task = ResearchTaskDTO(
id="task-2",
status="draft",
last_action="waiting for analysis",
error={"message": "not started"},
updated_at="2026-04-16T11:00:00",
)
payload = TaskAnalysisResourceDTO.from_parts(
task_id="task-2",
analysis_run_id="analysis-missing",
task=task,
current_run=None,
analysis_runs=[],
).to_response_payload()
assert payload["analysis"]["status"] == "idle"
assert payload["analysis"]["task_status"] == "draft"
assert payload["analysis"]["current_run"] is None
assert payload["analysis"]["runs"] == []
assert payload["analysis"]["summary"]["last_action"] == "waiting for analysis"
assert payload["analysis"]["summary"]["error"]["message"] == "not started"
assert payload["analysis"]["summary"]["updated_at"] == "2026-04-16T11:00:00"
assert payload["analysis"]["history"]["has_runs"] is False
assert payload["analysis"]["history"]["linked_run_available"] is False
assert payload["analysis"]["stats"]["total_runs"] == 0
assert payload["analysis"]["stats"]["active_run_count"] == 0