703 lines
27 KiB
Python
703 lines
27 KiB
Python
from datetime import UTC, datetime, timedelta
|
||
|
||
from bs4 import BeautifulSoup
|
||
|
||
from app.models import Comment, ContentItem, Hotspot, Report, Task
|
||
from tests.helpers import make_test_client
|
||
from sqlalchemy.exc import OperationalError
|
||
|
||
|
||
def test_index_page_renders_task_form_empty_state_and_default_scale():
|
||
with make_test_client() as (client, _engine):
|
||
response = client.get("/")
|
||
|
||
assert response.status_code == 200
|
||
assert "热榜评论分析工具" in response.text
|
||
assert 'name="platform"' in response.text
|
||
assert 'name="hotspot_limit"' in response.text
|
||
assert 'name="item_limit_per_hotspot"' in response.text
|
||
assert 'name="comment_limit_per_item"' in response.text
|
||
assert "1250" in response.text
|
||
assert "手动刷新" not in response.text
|
||
assert 'onclick="window.location.href=\'/\'"' not in response.text
|
||
assert "还没有任何任务" in response.text
|
||
|
||
|
||
def test_index_page_lists_existing_tasks():
|
||
with make_test_client() as (client, engine):
|
||
from sqlalchemy.orm import Session
|
||
|
||
with Session(engine) as session:
|
||
session.add(
|
||
Task(
|
||
id="task-visible",
|
||
platform="douyin",
|
||
status="running",
|
||
hotspot_limit=3,
|
||
item_limit_per_hotspot=4,
|
||
comment_limit_per_item=50,
|
||
total_items_count=4,
|
||
processed_items_count=2,
|
||
successful_items_count=1,
|
||
failed_items_count=1,
|
||
analysis_success_rate=0.75,
|
||
analysis_status="insufficient",
|
||
)
|
||
)
|
||
session.commit()
|
||
|
||
response = client.get("/")
|
||
|
||
assert response.status_code == 200
|
||
assert "task-visible" in response.text
|
||
assert "抖音" in response.text
|
||
assert "运行中" in response.text
|
||
assert "已处理 2 / 共 4 条内容" in response.text
|
||
assert "成功 1 / 失败 1" in response.text
|
||
assert 'style="width: 50%"' in response.text
|
||
assert "AI 成功率 75%" in response.text
|
||
assert "AI 样本不足" in response.text
|
||
assert 'data-task-list-auto-poll="true"' in response.text
|
||
assert "pollTaskListStatus" in response.text
|
||
assert "DOMContentLoaded" in response.text
|
||
|
||
|
||
def test_index_page_uses_short_task_numbers_compact_fields_and_friendly_error_copy():
|
||
first_id = "11111111-1111-4111-8111-111111111111"
|
||
second_id = "22222222-2222-4222-8222-222222222222"
|
||
with make_test_client() as (client, engine):
|
||
from sqlalchemy.orm import Session
|
||
|
||
with Session(engine) as session:
|
||
session.add(
|
||
Task(
|
||
id=first_id,
|
||
platform="xiaohongshu",
|
||
status="failed",
|
||
created_at=datetime(2026, 7, 3, 10, 30, tzinfo=UTC),
|
||
hotspot_limit=5,
|
||
item_limit_per_hotspot=5,
|
||
comment_limit_per_item=50,
|
||
error_stage="system",
|
||
error_type="unexpected_restart",
|
||
error_message="系统重启，任务被中断",
|
||
)
|
||
)
|
||
session.add(
|
||
Task(
|
||
id=second_id,
|
||
platform="douyin",
|
||
status="success",
|
||
created_at=datetime(2026, 7, 3, 10, 35, tzinfo=UTC),
|
||
hotspot_limit=1,
|
||
item_limit_per_hotspot=1,
|
||
comment_limit_per_item=10,
|
||
)
|
||
)
|
||
session.commit()
|
||
|
||
response = client.get("/")
|
||
|
||
assert response.status_code == 200
|
||
visible_text = BeautifulSoup(response.text, "html.parser").get_text(" ")
|
||
assert " 1 " in f" {visible_text} "
|
||
assert " 2 " in f" {visible_text} "
|
||
assert "#" not in visible_text
|
||
assert first_id not in visible_text
|
||
assert second_id not in visible_text
|
||
assert "2026-07-03 10:30" in visible_text
|
||
assert "2026-07-03 10:35" in visible_text
|
||
assert "5热点 × 5内容 × 50评论" in visible_text
|
||
assert "1热点 × 1内容 × 10评论" in visible_text
|
||
assert "小红书 · 5热点" not in visible_text
|
||
assert "抖音 · 1热点" not in visible_text
|
||
assert "系统重启，任务被中断" in visible_text
|
||
assert "system / unexpected_restart" not in visible_text
|
||
assert "unexpected_restart" not in visible_text
|
||
assert "内部演示工具 | 仅供学习参考" not in visible_text
|
||
|
||
|
||
def test_index_page_does_not_auto_poll_without_running_tasks():
|
||
with make_test_client() as (client, engine):
|
||
from sqlalchemy.orm import Session
|
||
|
||
with Session(engine) as session:
|
||
session.add(
|
||
Task(
|
||
id="task-done",
|
||
platform="xiaohongshu",
|
||
status="success",
|
||
total_items_count=1,
|
||
processed_items_count=1,
|
||
successful_items_count=1,
|
||
analysis_success_rate=1.0,
|
||
)
|
||
)
|
||
session.commit()
|
||
|
||
response = client.get("/")
|
||
|
||
assert response.status_code == 200
|
||
assert "task-done" in response.text
|
||
assert 'data-task-list-auto-poll="true"' not in response.text
|
||
assert "pollTaskListStatus" not in response.text
|
||
|
||
|
||
def test_running_task_detail_page_auto_polls_current_task():
|
||
with make_test_client() as (client, engine):
|
||
from sqlalchemy.orm import Session
|
||
|
||
with Session(engine) as session:
|
||
session.add(
|
||
Task(
|
||
id="task-running",
|
||
platform="xiaohongshu",
|
||
status="running",
|
||
hotspot_limit=1,
|
||
item_limit_per_hotspot=1,
|
||
comment_limit_per_item=10,
|
||
total_items_count=2,
|
||
processed_items_count=1,
|
||
successful_items_count=1,
|
||
failed_items_count=0,
|
||
analysis_success_rate=1.0,
|
||
)
|
||
)
|
||
session.commit()
|
||
|
||
response = client.get("/tasks/task-running")
|
||
|
||
assert response.status_code == 200
|
||
assert 'data-task-id="task-running"' in response.text
|
||
assert "pollTaskDetailStatus" in response.text
|
||
assert "手动刷新" not in response.text
|
||
assert 'onclick="window.location.reload()"' not in response.text
|
||
assert "已处理 1 / 共 2 条内容" in response.text
|
||
assert "AI 成功率 100%" in response.text
|
||
|
||
|
||
def test_task_detail_uses_short_number_title_and_hides_full_uuid():
|
||
task_id = "33333333-3333-4333-8333-333333333333"
|
||
with make_test_client() as (client, engine):
|
||
from sqlalchemy.orm import Session
|
||
|
||
with Session(engine) as session:
|
||
session.add(
|
||
Task(
|
||
id=task_id,
|
||
platform="xiaohongshu",
|
||
status="running",
|
||
created_at=datetime(2026, 7, 3, 10, 30, tzinfo=UTC),
|
||
hotspot_limit=5,
|
||
item_limit_per_hotspot=5,
|
||
comment_limit_per_item=50,
|
||
)
|
||
)
|
||
session.commit()
|
||
|
||
response = client.get(f"/tasks/{task_id}")
|
||
|
||
assert response.status_code == 200
|
||
visible_text = BeautifulSoup(response.text, "html.parser").get_text(" ")
|
||
assert "<title>任务 1 - 热榜评论分析工具</title>" in response.text
|
||
assert "任务 1" in visible_text
|
||
assert "完整 ID" not in visible_text
|
||
assert task_id not in visible_text
|
||
assert "任务 #33333333-3333-4333-8333-333333333333" not in visible_text
|
||
assert "#" not in visible_text
|
||
assert "创建时间 2026-07-03 10:30" in visible_text
|
||
|
||
|
||
def test_running_task_detail_page_keeps_polling_after_hotspots_exist():
|
||
with make_test_client() as (client, engine):
|
||
from sqlalchemy.orm import Session
|
||
|
||
with Session(engine) as session:
|
||
task = Task(
|
||
id="task-running-with-hotspot",
|
||
platform="xiaohongshu",
|
||
status="running",
|
||
total_items_count=2,
|
||
processed_items_count=1,
|
||
successful_items_count=1,
|
||
analysis_success_rate=1.0,
|
||
)
|
||
session.add(task)
|
||
session.add(
|
||
Hotspot(
|
||
id="hot-running",
|
||
task_id=task.id,
|
||
platform="xiaohongshu",
|
||
title="运行中的热点",
|
||
rank=1,
|
||
raw_data="{}",
|
||
)
|
||
)
|
||
session.commit()
|
||
|
||
response = client.get("/tasks/task-running-with-hotspot")
|
||
|
||
assert response.status_code == 200
|
||
assert "运行中的热点" in response.text
|
||
assert "正在抓取热点数据" not in response.text
|
||
assert "pollTaskDetailStatus" in response.text
|
||
|
||
|
||
def test_running_task_detail_page_shows_stage_runtime_and_stale_progress_warning(monkeypatch):
|
||
now = datetime(2026, 7, 3, 10, 30, tzinfo=UTC)
|
||
monkeypatch.setattr("app.services.task_service.utc_now", lambda: now)
|
||
|
||
with make_test_client() as (client, engine):
|
||
from sqlalchemy.orm import Session
|
||
|
||
with Session(engine) as session:
|
||
task = Task(
|
||
id="task-stale",
|
||
platform="xiaohongshu",
|
||
status="running",
|
||
current_stage="ai_analysis",
|
||
created_at=now - timedelta(hours=2, minutes=5),
|
||
last_progress_at=now - timedelta(minutes=12),
|
||
hotspot_limit=5,
|
||
item_limit_per_hotspot=5,
|
||
comment_limit_per_item=50,
|
||
total_items_count=10,
|
||
processed_items_count=6,
|
||
successful_items_count=6,
|
||
failed_items_count=0,
|
||
analysis_success_rate=0.6,
|
||
analysis_status="insufficient",
|
||
)
|
||
session.add(task)
|
||
hotspot = Hotspot(id="hot-stale", task_id=task.id, platform="xiaohongshu", title="运行热点", rank=1, raw_data="{}")
|
||
session.add(hotspot)
|
||
item = ContentItem(
|
||
id="item-stale",
|
||
task_id=task.id,
|
||
hotspot_id=hotspot.id,
|
||
platform="xiaohongshu",
|
||
source_item_id="note-stale",
|
||
item_type="note",
|
||
title="运行内容",
|
||
status="pending",
|
||
raw_data="{}",
|
||
)
|
||
session.add(item)
|
||
for index in range(3):
|
||
session.add(
|
||
Comment(
|
||
task_id=task.id,
|
||
hotspot_id=hotspot.id,
|
||
content_item_id=item.id,
|
||
platform="xiaohongshu",
|
||
source_comment_id=f"c{index}",
|
||
content=f"评论 {index}",
|
||
raw_data="{}",
|
||
)
|
||
)
|
||
session.commit()
|
||
|
||
response = client.get("/tasks/task-stale")
|
||
|
||
assert response.status_code == 200
|
||
assert "阶段 AI 分析中" in response.text
|
||
assert "运行时长" in response.text
|
||
assert "2小时5分钟" in response.text
|
||
assert "最近进度" in response.text
|
||
assert "12分钟前" in response.text
|
||
assert "已获取热点 1 / 目标 5" in response.text
|
||
assert "评论 3" in response.text
|
||
assert "超过 10 分钟没有进度更新" in response.text
|
||
assert "可能仍在等待外部接口或 AI 响应" in response.text
|
||
|
||
|
||
def test_failed_task_detail_page_shows_failure_reason_without_loading_copy():
|
||
with make_test_client() as (client, engine):
|
||
from sqlalchemy.orm import Session
|
||
|
||
with Session(engine) as session:
|
||
session.add(
|
||
Task(
|
||
id="task-failed",
|
||
platform="douyin",
|
||
status="failed",
|
||
error_stage="recover",
|
||
error_type="interrupted",
|
||
error_message="系统重启，任务被中断",
|
||
)
|
||
)
|
||
session.commit()
|
||
|
||
response = client.get("/tasks/task-failed")
|
||
|
||
assert response.status_code == 200
|
||
assert "失败" in response.text
|
||
assert "系统重启，任务被中断" in response.text
|
||
assert "recover / interrupted" not in response.text
|
||
assert "正在抓取热点数据" not in response.text
|
||
assert "pollTaskDetailStatus" not in response.text
|
||
|
||
|
||
def seed_result_data(engine):
|
||
from sqlalchemy.orm import Session
|
||
|
||
with Session(engine) as session:
|
||
task = Task(
|
||
id="task-result",
|
||
platform="douyin",
|
||
status="success",
|
||
total_items_count=1,
|
||
successful_items_count=1,
|
||
analysis_success_rate=1.0,
|
||
)
|
||
session.add(task)
|
||
hotspot = Hotspot(id="hot-result", task_id=task.id, platform="douyin", title="热点标题", rank=1, raw_data="{}")
|
||
session.add(hotspot)
|
||
item = ContentItem(
|
||
id="item-result",
|
||
task_id=task.id,
|
||
hotspot_id=hotspot.id,
|
||
platform="douyin",
|
||
source_item_id="v1",
|
||
item_type="video",
|
||
title="视频标题",
|
||
status="success",
|
||
raw_data="{}",
|
||
)
|
||
session.add(item)
|
||
session.add(
|
||
Comment(
|
||
id="comment-result",
|
||
task_id=task.id,
|
||
hotspot_id=hotspot.id,
|
||
content_item_id=item.id,
|
||
platform="douyin",
|
||
source_comment_id="c1",
|
||
content="评论内容",
|
||
sentiment="positive",
|
||
labels='["认可"]',
|
||
like_count=3,
|
||
raw_data="{}",
|
||
)
|
||
)
|
||
session.add(
|
||
Report(
|
||
task_id=task.id,
|
||
hotspot_id=hotspot.id,
|
||
report_type="hotspot",
|
||
title="热点标题",
|
||
metrics_json='{"sample_count":1,"item_count":1,"sentiment":{"positive":{"count":1,"pct":100},"neutral":{"count":0,"pct":0},"negative":{"count":0,"pct":0},"unknown":{"count":0,"pct":0}},"top_labels":[{"name":"认可","count":1}]}',
|
||
typical_comments_json='{"positive":[{"content":"评论内容","like_count":3}],"neutral":[],"negative":[]}',
|
||
summary="热点总结",
|
||
markdown_content="# 热点标题\n\n热点总结",
|
||
data="{}",
|
||
markdown="# 热点标题\n\n热点总结",
|
||
)
|
||
)
|
||
session.add(
|
||
Report(
|
||
task_id=task.id,
|
||
hotspot_id=hotspot.id,
|
||
content_item_id=item.id,
|
||
report_type="item",
|
||
title="视频标题",
|
||
metrics_json='{"sample_count":1,"sentiment":{"positive":{"count":1,"pct":100},"neutral":{"count":0,"pct":0},"negative":{"count":0,"pct":0},"unknown":{"count":0,"pct":0}},"top_labels":[{"name":"认可","count":1}]}',
|
||
typical_comments_json='{"positive":[{"content":"评论内容","like_count":3}],"neutral":[],"negative":[]}',
|
||
summary="内容总结",
|
||
markdown_content="# 视频标题\n\n内容总结",
|
||
data="{}",
|
||
markdown="# 视频标题\n\n内容总结",
|
||
)
|
||
)
|
||
session.commit()
|
||
|
||
|
||
def test_result_pages_render_seeded_data():
|
||
with make_test_client() as (client, engine):
|
||
seed_result_data(engine)
|
||
|
||
task_response = client.get("/tasks/task-result")
|
||
hotspot_response = client.get("/hotspots/hot-result/report")
|
||
item_response = client.get("/items/item-result")
|
||
|
||
assert task_response.status_code == 200
|
||
assert "热点标题" in task_response.text
|
||
assert "热点 1：热点标题" in BeautifulSoup(task_response.text, "html.parser").get_text(" ")
|
||
assert 'onclick="window.location.reload()"' not in task_response.text
|
||
assert "手动刷新" not in task_response.text
|
||
assert hotspot_response.status_code == 200
|
||
assert "热点总结" in hotspot_response.text
|
||
assert "downloadExport(" not in hotspot_response.text
|
||
assert "/api/export/" not in hotspot_response.text
|
||
assert "导出 Markdown" not in hotspot_response.text
|
||
assert "导出热点评论 CSV" not in hotspot_response.text
|
||
assert "评论样本" in hotspot_response.text
|
||
assert "关联内容" in hotspot_response.text
|
||
assert "正向" in hotspot_response.text
|
||
assert "1 条（100%）" in hotspot_response.text
|
||
assert "认可 (1)" in hotspot_response.text
|
||
assert "典型评论" in hotspot_response.text
|
||
assert "评论内容" in hotspot_response.text
|
||
assert "<pre" not in hotspot_response.text
|
||
assert item_response.status_code == 200
|
||
assert "内容总结" in item_response.text
|
||
item_visible_text = BeautifulSoup(item_response.text, "html.parser").get_text(" ")
|
||
assert "热点 1：热点标题" in item_visible_text
|
||
assert "#" not in item_visible_text
|
||
assert "downloadExport(" not in item_response.text
|
||
assert "/api/export/" not in item_response.text
|
||
assert "导出 Markdown" not in item_response.text
|
||
assert "导出评论 CSV" not in item_response.text
|
||
assert "评论样本" in item_response.text
|
||
assert "Top 标签" in item_response.text
|
||
assert "评论内容" in item_response.text
|
||
|
||
|
||
def test_report_pages_render_empty_state_when_report_missing():
|
||
with make_test_client() as (client, engine):
|
||
from sqlalchemy.orm import Session
|
||
|
||
with Session(engine) as session:
|
||
task = Task(id="task-no-report", platform="xiaohongshu", status="running")
|
||
session.add(task)
|
||
hotspot = Hotspot(id="hot-no-report", task_id=task.id, platform="xiaohongshu", title="缺报告热点", rank=1, raw_data="{}")
|
||
session.add(hotspot)
|
||
session.add(
|
||
ContentItem(
|
||
id="item-no-report",
|
||
task_id=task.id,
|
||
hotspot_id=hotspot.id,
|
||
platform="xiaohongshu",
|
||
source_item_id="n1",
|
||
item_type="note",
|
||
title="缺报告内容",
|
||
status="pending",
|
||
raw_data="{}",
|
||
)
|
||
)
|
||
session.commit()
|
||
|
||
hotspot_response = client.get("/hotspots/hot-no-report/report")
|
||
item_response = client.get("/items/item-no-report")
|
||
|
||
assert hotspot_response.status_code == 200
|
||
assert "报告生成中，请稍候" in hotspot_response.text
|
||
assert 'title="报告尚未生成"' not in hotspot_response.text
|
||
assert "Markdown 报告将在分析完成后开放下载" not in hotspot_response.text
|
||
assert "downloadExport(" not in hotspot_response.text
|
||
assert "/api/export/" not in hotspot_response.text
|
||
assert item_response.status_code == 200
|
||
assert "报告生成中，请稍候" in item_response.text
|
||
assert 'title="报告尚未生成"' not in item_response.text
|
||
assert "Markdown 报告将在分析完成后开放下载" not in item_response.text
|
||
assert "downloadExport(" not in item_response.text
|
||
assert "/api/export/" not in item_response.text
|
||
|
||
|
||
def test_report_pages_warn_when_ai_sample_is_insufficient():
|
||
with make_test_client() as (client, engine):
|
||
seed_result_data(engine)
|
||
from sqlalchemy.orm import Session
|
||
|
||
with Session(engine) as session:
|
||
task = session.get(Task, "task-result")
|
||
task.analysis_status = "insufficient"
|
||
task.analysis_success_rate = 0.5
|
||
session.commit()
|
||
|
||
hotspot_response = client.get("/hotspots/hot-result/report")
|
||
item_response = client.get("/items/item-result")
|
||
|
||
assert hotspot_response.status_code == 200
|
||
assert "当前有效评论样本不足" in hotspot_response.text
|
||
assert item_response.status_code == 200
|
||
assert "当前有效评论样本不足" in item_response.text
|
||
|
||
|
||
def test_export_routes_are_removed():
|
||
with make_test_client() as (client, engine):
|
||
seed_result_data(engine)
|
||
|
||
csv_response = client.get("/api/export/items/item-result/comments.csv")
|
||
hotspot_csv_response = client.get("/api/export/hotspots/hot-result/comments.csv")
|
||
hotspot_md = client.get("/api/export/hotspots/hot-result.md")
|
||
item_md = client.get("/api/export/items/item-result.md")
|
||
|
||
assert csv_response.status_code == 404
|
||
assert hotspot_csv_response.status_code == 404
|
||
assert hotspot_md.status_code == 404
|
||
assert item_md.status_code == 404
|
||
|
||
|
||
def test_api_tasks_returns_service_unavailable_when_database_has_io_error(monkeypatch):
|
||
def broken_list_tasks(_session):
|
||
raise OperationalError("SELECT 1", {}, Exception("disk I/O error"))
|
||
|
||
monkeypatch.setattr("app.main.list_tasks", broken_list_tasks)
|
||
|
||
with make_test_client() as (client, _engine):
|
||
response = client.get("/api/tasks")
|
||
|
||
assert response.status_code == 503
|
||
assert response.json()["detail"] == "数据库暂时不可用，请稍后重试或联系维护者恢复数据。"
|
||
|
||
|
||
def test_index_page_shows_database_error_state_when_database_has_io_error(monkeypatch):
|
||
def broken_list_tasks(_session):
|
||
raise OperationalError("SELECT 1", {}, Exception("disk I/O error"))
|
||
|
||
monkeypatch.setattr("app.main.list_tasks", broken_list_tasks)
|
||
|
||
with make_test_client() as (client, _engine):
|
||
response = client.get("/")
|
||
|
||
assert response.status_code == 503
|
||
assert "数据库暂时不可用" in response.text
|
||
assert "请先保留 data 目录" in response.text
|
||
|
||
|
||
def test_task_detail_page_shows_database_error_state_when_database_has_io_error(monkeypatch):
|
||
def broken_get_task(_session, _task_id):
|
||
raise OperationalError("SELECT 1", {}, Exception("disk I/O error"))
|
||
|
||
monkeypatch.setattr("app.main.get_task", broken_get_task)
|
||
|
||
with make_test_client() as (client, _engine):
|
||
response = client.get("/tasks/task-io-error")
|
||
|
||
assert response.status_code == 503
|
||
assert "数据库暂时不可用" in response.text
|
||
assert "请先保留 data 目录" in response.text
|
||
|
||
|
||
def test_hotspot_report_page_shows_database_error_state_when_database_has_io_error(monkeypatch):
|
||
def broken_get_task(_session, _task_id):
|
||
raise OperationalError("SELECT 1", {}, Exception("disk I/O error"))
|
||
|
||
monkeypatch.setattr("app.main.get_task", broken_get_task)
|
||
|
||
with make_test_client() as (client, engine):
|
||
from sqlalchemy.orm import Session
|
||
|
||
with Session(engine) as session:
|
||
session.add(Task(id="task-hot-io", platform="xiaohongshu", status="success"))
|
||
session.add(Hotspot(id="hot-io", task_id="task-hot-io", platform="xiaohongshu", title="热点", rank=1, raw_data="{}"))
|
||
session.commit()
|
||
|
||
response = client.get("/hotspots/hot-io/report")
|
||
|
||
assert response.status_code == 503
|
||
assert "数据库暂时不可用" in response.text
|
||
|
||
|
||
def test_item_detail_page_shows_database_error_state_when_database_has_io_error(monkeypatch):
|
||
def broken_get_task(_session, _task_id):
|
||
raise OperationalError("SELECT 1", {}, Exception("disk I/O error"))
|
||
|
||
monkeypatch.setattr("app.main.get_task", broken_get_task)
|
||
|
||
with make_test_client() as (client, engine):
|
||
from sqlalchemy.orm import Session
|
||
|
||
with Session(engine) as session:
|
||
session.add(Task(id="task-item-io", platform="xiaohongshu", status="success"))
|
||
session.add(Hotspot(id="hot-item-io", task_id="task-item-io", platform="xiaohongshu", title="热点", rank=1, raw_data="{}"))
|
||
session.add(
|
||
ContentItem(
|
||
id="item-io",
|
||
task_id="task-item-io",
|
||
hotspot_id="hot-item-io",
|
||
platform="xiaohongshu",
|
||
source_item_id="item-io-source",
|
||
item_type="note",
|
||
status="success",
|
||
raw_data="{}",
|
||
)
|
||
)
|
||
session.commit()
|
||
|
||
response = client.get("/items/item-io")
|
||
|
||
assert response.status_code == 503
|
||
assert "数据库暂时不可用" in response.text
|
||
|
||
|
||
|
||
def test_missing_html_pages_render_friendly_not_found_page():
|
||
with make_test_client() as (client, _engine):
|
||
task_response = client.get("/tasks/missing-task")
|
||
hotspot_response = client.get("/hotspots/missing-hotspot/report")
|
||
item_response = client.get("/items/missing-item")
|
||
|
||
assert task_response.status_code == 404
|
||
assert "任务不存在" in task_response.text
|
||
assert "Internal Server Error" not in task_response.text
|
||
assert hotspot_response.status_code == 404
|
||
assert "热点不存在" in hotspot_response.text
|
||
assert "Internal Server Error" not in hotspot_response.text
|
||
assert item_response.status_code == 404
|
||
assert "内容条目不存在" in item_response.text
|
||
assert "Internal Server Error" not in item_response.text
|
||
|
||
|
||
def test_task_api_includes_progress_counts_stage_and_demo_flag():
|
||
with make_test_client() as (client, engine):
|
||
seed_result_data(engine)
|
||
|
||
response = client.get("/api/tasks/task-result")
|
||
|
||
assert response.status_code == 200
|
||
data = response.json()
|
||
assert data["current_stage"] in (None, "success")
|
||
assert data["current_stage_label"] == "已完成"
|
||
assert data["comments_count"] == 1
|
||
assert data["reports_count"] == 2
|
||
assert data["is_demo"] is False
|
||
assert data["last_progress_at"] is not None
|
||
|
||
|
||
def test_task_detail_page_explains_target_and_actual_counts():
|
||
with make_test_client() as (client, engine):
|
||
seed_result_data(engine)
|
||
|
||
response = client.get("/tasks/task-result")
|
||
|
||
assert response.status_code == 200
|
||
assert "目标上限" in response.text
|
||
assert "实际结果" in response.text
|
||
assert "实际评论 1" in response.text
|
||
assert "少于理论上限通常是内容本身评论不足或平台返回不足" in response.text
|
||
|
||
|
||
def test_task_api_includes_runtime_and_stale_progress_fields(monkeypatch):
|
||
now = datetime(2026, 7, 3, 10, 30, tzinfo=UTC)
|
||
monkeypatch.setattr("app.services.task_service.utc_now", lambda: now)
|
||
|
||
with make_test_client() as (client, engine):
|
||
from sqlalchemy.orm import Session
|
||
|
||
with Session(engine) as session:
|
||
session.add(
|
||
Task(
|
||
id="task-api-stale",
|
||
platform="douyin",
|
||
status="running",
|
||
current_stage="crawl_comments",
|
||
created_at=now - timedelta(minutes=45),
|
||
last_progress_at=now - timedelta(minutes=11),
|
||
)
|
||
)
|
||
session.commit()
|
||
|
||
response = client.get("/api/tasks/task-api-stale")
|
||
|
||
assert response.status_code == 200
|
||
data = response.json()
|
||
assert data["current_stage_label"] == "抓取评论中"
|
||
assert data["running_seconds"] == 2700
|
||
assert data["seconds_since_last_progress"] == 660
|
||
assert data["running_duration_label"] == "45分钟"
|
||
assert data["last_progress_ago_label"] == "11分钟前"
|
||
assert data["is_progress_stale"] is True
|
||
assert data["stale_threshold_minutes"] == 10
|