test_crawler_job_dto.py
2.3 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
from __future__ import annotations
from datetime import datetime
from services.shared.dto import CrawlerJobDTO, CrawlerJobListDTO
from services.shared.models import CrawlerJob, CrawlerJobStatus, ErrorInfo, ProgressInfo
def test_crawler_job_dto_from_model_and_response_item():
job = CrawlerJob(
id="crawler-123",
research_task_id="task-123",
platform="xiaohongshu",
keywords=["museum", "coffee"],
status=CrawlerJobStatus.RUNNING,
progress=ProgressInfo(
current=3,
total=10,
percent=30.0,
stage="collecting",
message="collecting notes",
),
config={"region": "shanghai"},
result_summary={"items_collected": 12},
last_action="collecting notes",
error=ErrorInfo(code="RATE_LIMIT", message="slow down", retryable=True),
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={"platform_label": "Xiaohongshu"},
)
dto = CrawlerJobDTO.from_model(job)
payload = dto.to_response_item()
assert dto.id == "crawler-123"
assert dto.status == "running"
assert payload["job_id"] == "crawler-123"
assert payload["platform"] == "xiaohongshu"
assert payload["keywords"] == ["museum", "coffee"]
assert payload["config"]["region"] == "shanghai"
assert payload["result_summary"]["items_collected"] == 12
assert payload["legacy_payload"]["platform_label"] == "Xiaohongshu"
assert payload["platform_label"] == "Xiaohongshu"
assert payload["unified_job"]["research_task_id"] == "task-123"
def test_crawler_job_list_dto_builds_history_payload():
current = CrawlerJobDTO(
id="crawler-2",
research_task_id="task-1",
platform="xiaohongshu",
status="running",
)
previous = CrawlerJobDTO(
id="crawler-1",
research_task_id="task-1",
platform="dianping",
status="completed",
)
payload = CrawlerJobListDTO(current_job=current, jobs=[current, previous]).to_response_payload()
assert payload["current_job"]["id"] == "crawler-2"
assert [item["id"] for item in payload["jobs"]] == ["crawler-2", "crawler-1"]