test_task_model_mappers.py
2.83 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
from __future__ import annotations
from services.shared.models import AnalysisRunStatus, ReportJobStatus
from services.shared.models.mappers import map_analysis_run_record, map_legacy_research_task, map_report_task_record
def test_map_analysis_run_record_builds_unified_run_from_persisted_payload():
run = map_analysis_run_record(
{
"id": "analysis-1",
"research_task_id": "task-1",
"engines": ["query", "insight"],
"status": "partial",
"progress": {
"current": 1,
"total": 2,
"percent": 50,
"stage": "running",
"message": "query finished",
},
"partial_results": {"query": {"success": True}},
"metrics": {"success_count": 1, "failure_count": 1},
"last_action": "query finished",
"created_at": "2026-04-20T10:00:00",
"updated_at": "2026-04-20T10:05:00",
}
)
assert run.id == "analysis-1"
assert run.research_task_id == "task-1"
assert run.status == AnalysisRunStatus.PARTIAL
assert run.progress.percent == 50
assert run.partial_results["query"]["success"] is True
assert run.legacy_payload["status"] == "partial"
def test_map_report_task_record_builds_unified_job_from_runtime_payload():
job = map_report_task_record(
{
"task_id": "report-1",
"research_task_id": "task-1",
"query": "museum report",
"status": "running",
"progress": 42,
"report_file_name": "report-1.html",
"report_file_path": "reports/report-1.html",
"state_file_path": "reports/report-1.state.json",
"ir_file_path": "reports/report-1.ir.json",
"markdown_file_path": "reports/report-1.md",
"markdown_file_name": "report-1.md",
"created_at": "2026-04-20T11:00:00",
"updated_at": "2026-04-20T11:05:00",
}
)
assert job.id == "report-1"
assert job.research_task_id == "task-1"
assert job.status == ReportJobStatus.RENDERING
assert job.progress.percent == 42
assert job.artifacts["report_file_ready"] is True
assert job.artifacts["markdown_file_name"] == "report-1.md"
assert job.output_path == "reports/report-1.html"
def test_map_legacy_research_task_moves_crawler_keywords_text_into_metadata_extras():
task = map_legacy_research_task(
{
"task_id": "task-1",
"status": "ready",
"crawler_defaults": {"crawler_type": "search"},
"crawler_keywords_text": "Museum Task review\nMuseum Task service",
}
)
assert task.metadata.extras["crawler_defaults"] == {"crawler_type": "search"}
assert task.metadata.extras["crawler_keywords_text"] == "Museum Task review\nMuseum Task service"