280 lines
10 KiB
Python
280 lines
10 KiB
Python
from app.models import Comment, ContentItem, Hotspot, Report, Task
|
|
from tests.helpers import make_test_client
|
|
|
|
|
|
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 'onclick="window.location.href=\'/\'"' 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_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 'onclick="window.location.reload()"' in response.text
|
|
assert "已处理 1 / 共 2 条内容" in response.text
|
|
assert "AI 成功率 100%" in response.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_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 "recover / interrupted：系统重启，任务被中断" 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)
|
|
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,"sentiment":{"positive":{"count":1,"pct":100}},"top_labels":[{"name":"认可","count":1}]}',
|
|
typical_comments_json='{"positive":[{"content":"评论内容","like_count":3}]}',
|
|
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}},"top_labels":[{"name":"认可","count":1}]}',
|
|
typical_comments_json='{"positive":[{"content":"评论内容","like_count":3}]}',
|
|
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 'onclick="window.location.reload()"' in task_response.text
|
|
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_return_csv_and_markdown():
|
|
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 == 200
|
|
assert csv_response.content.startswith(b"\xef\xbb\xbf")
|
|
assert "评论内容".encode("utf-8") in csv_response.content
|
|
assert hotspot_csv_response.status_code == 200
|
|
assert "评论内容".encode("utf-8") in hotspot_csv_response.content
|
|
assert hotspot_md.status_code == 200
|
|
assert "热点总结" in hotspot_md.text
|
|
assert item_md.status_code == 200
|
|
assert "内容总结" in item_md.text
|