meijiali/hot_comment_radar

Fork 0

Files

T

meijiali 7eea937f28 docs: add TDD and task planning documents

2026-07-01 20:31:11 +08:00

28 KiB

Raw Blame History

TDD.md：小红书 / 抖音热榜评论抓取 + AI 分析报告工具

1. 文档信息

文档阶段：TDD（Test-Driven Development，测试驱动开发计划）
需求依据：docs/PRD.md、docs/FeatureSummary.md
技术依据：docs/DevelopmentPlan.md
UI 依据：docs/UIDesign.md
当前目标：定义 MVP 开发前必须先写的测试范围、测试顺序、Mock 策略、验收命令与端到端测试清单
测试原则：先写失败测试，再写最小实现；任何业务代码变更前必须先有对应测试

2. TDD 总原则

2.1 红绿重构流程

每个功能点按以下顺序执行：

Red：先写一个最小失败测试。
Verify Red：运行测试，确认失败原因是目标功能缺失，而不是测试代码错误。
Green：写最小实现让测试通过。
Verify Green：运行该测试和相关测试，确认通过。
Refactor：只在测试通过后清理代码。

禁止事项：

禁止先实现后补测试。
禁止为了让测试通过而删除关键断言。
禁止用真实 TikHub / AI API 作为单元测试依赖。
禁止在模板中硬编码中文状态文案，状态展示必须通过映射或 Macro 测试覆盖。

2.2 测试分层

层级	工具	目标
单元测试	pytest	字段映射、配置校验、分页、统计、导出、AI schema
服务集成测试	pytest + TestClient + mock httpx	任务创建、后台流程、平台链路、报告生成
模板测试	pytest + Jinja2 渲染	状态 Badge、空状态、导出按钮、标签渲染
端到端测试	Playwright	首页创建任务、查看报告、导出按钮、手动刷新
Docker 验收	docker compose + curl	启动、`/health`、页面可访问

2.3 测试目录建议

tests/
  conftest.py
  fixtures/
    xhs_hot_list.json
    xhs_search_notes.json
    xhs_comments_page_1.json
    xhs_comments_page_2_empty.json
    douyin_hot_list.json
    douyin_search_videos.json
    douyin_comments_page_1.json
    ai_comments_success.json
    ai_comments_invalid_json.txt
  unit/
    test_config.py
    test_models.py
    test_xiaohongshu_mapping.py
    test_douyin_mapping.py
    test_comment_pagination.py
    test_ai_schema.py
    test_report_stats.py
    test_export.py
    test_template_filters.py
  integration/
    test_task_creation.py
    test_task_flow_xiaohongshu.py
    test_task_flow_douyin.py
    test_failure_tolerance.py
    test_routes.py
  e2e/
    test_mvp_flow.py

3. 测试环境与命令

3.1 推荐依赖

pytest
pytest-cov
pytest-mock
respx
httpx
beautifulsoup4
playwright

说明：

respx 用于 mock httpx.Client 外部 HTTP 调用。
beautifulsoup4 用于断言 HTML 片段内容。
Playwright 仅覆盖关键页面流程，不替代单元测试。

工具使用原则

respx：用于 mock httpx.Client 发出的外部 HTTP 请求（TikHub API、AI API）。
pytest-mock：用于 mock Python 函数、类方法、time.sleep 调用和时间函数。
两者不混用于同一外部请求的 mock。即：若某个测试需要拦截一个 HTTP 请求，必须使用 respx，不得用 pytest-mock.patch("httpx.Client.get") 替代。

3.2 测试数据库隔离与环境变量

测试数据库隔离策略

所有单元测试和集成测试统一使用内存数据库 sqlite:///:memory:，不依赖物理 test.db 文件，避免测试用例间状态污染（Flaky Tests）。

在 tests/conftest.py 中定义函数级别（scope="function"）的数据库 fixture：

@pytest.fixture(scope="function")
def db_session():
    engine = create_engine(
        "sqlite:///:memory:",
        connect_args={"check_same_thread": False},
    )
    Base.metadata.create_all(engine)
    session = Session(engine)
    yield session
    session.close()
    Base.metadata.drop_all(engine)

每个测试用例运行前自动创建全部表结构，运行后自动销毁，确保测试在绝对隔离的环境中执行。

禁止在集成测试中复用同一 engine 实例跨用例写入数据而不做 teardown。

环境变量

APP_ENV=test
DATABASE_URL=sqlite:///:memory:
TIKHUB_API_KEY=test-token
TIKHUB_BASE_URL=https://api.tikhub.test
AI_PROVIDER=openai-compatible
AI_BASE_URL=https://ai.test
AI_API_KEY=test-ai-key
AI_MODEL=test-model
AI_BATCH_SIZE=20
AI_CONCURRENCY=2
AI_MAX_RETRIES=3
AI_TIMEOUT_SECONDS=30

3.3 常用命令

pytest tests/unit -q
pytest tests/integration -q
pytest tests/unit tests/integration \
  --cov=app \
  --cov-branch \
  --cov-report=term-missing
pytest tests/e2e -q
docker compose up --build
curl -f http://localhost:8000/health

覆盖率豁免配置

在 pyproject.toml 的 [tool.coverage.report] 节中添加以下配置，将模板层和静态资源排除在覆盖率统计之外：

[tool.coverage.report]
omit = [
    "app/templates/*",
    "app/static/*",
    "tests/*",
]

全局覆盖率目标维持 80%，app/services/ 和 app/platforms/ 下的核心域覆盖率不低于 90%（通过 CI 脚本或 code review 人工把关，不在 pytest 命令层面强制差异化）。

4. Mock 数据规范

4.1 小红书 Mock 样例

xhs_hot_list.json 必须覆盖：

data.data.items[] 是真实热榜条目；
外层 data.data.title 不应被误用为热点标题；
热点字段包含 id、title、score。

xhs_search_notes.json 必须覆盖：

至少 2 条 comments_count > 0 的笔记；
至少 1 条 comments_count = 0 的笔记；
字段包含 note.id、note.title、note.desc、note.comments_count。

xhs_comments_page_1.json 必须覆盖：

评论 ID 同时存在 comment_id 和 id 时，优先 comment_id；
评论正文可来自 content 或 text；
包含 like_count、create_time。

4.2 抖音 Mock 样例

douyin_hot_list.json 必须覆盖：

热点字段包含 query_id、title、rank、hot_score。

douyin_search_videos.json 必须覆盖：

视频字段包含 aweme_info.aweme_id、aweme_info.desc、aweme_info.author、aweme_info.statistics。

douyin_comments_page_1.json 必须覆盖：

评论 ID 同时存在 comment_id 和 cid 时，优先 comment_id；
评论正文来自 text；
包含 digg_count、create_time。

4.3 AI Mock 样例

成功响应：

[
  {
    "comment_id": "c1",
    "sentiment": "positive",
    "labels": ["价格实惠", "质量好"],
    "reason": "评论表达了明确认可"
  }
]

失败响应至少覆盖：

非 JSON 文本；
JSON Object 而不是 JSON Array；
缺失 comment_id；
comment_id 与输入不匹配；
sentiment 不在枚举值内；
labels 超过 3 个；
混入 Markdown 或自然语言前缀。

4.4 HTTP 429 Mock Fixture

在 tests/conftest.py 中提供可复用的 429 响应 fixture：

@pytest.fixture
def mock_429_response():
    return httpx.Response(
        status_code=429,
        headers={"Retry-After": "1"},
        json={"message": "Too Many Requests"},
    )

§8.2（分页限流重试）和 §13.3（任务级限流容错）均使用此 fixture，不各自重复定义。

4.5 评论字段缺失 Mock 数据

在 tests/fixtures/ 中新增以下两个文件：

xhs_comments_missing_fields.json：小红书评论列表，其中 like_count 和 create_time 字段均不存在（非 null，而是 key 完全缺失）。
douyin_comments_missing_fields.json：抖音评论列表，其中 digg_count 和 create_time 字段均不存在。

这两个 fixture 用于 §7.2 和 §7.5 的字段缺失测试用例。

4.6 AI 全情绪值 Mock 数据

在 tests/fixtures/ai_comments_success.json 中确保包含四种 sentiment 值各至少一条评论，用于统计逻辑（analysis_success_rate、情绪分布）的测试：

[
  {"comment_id": "c001", "sentiment": "positive", "labels": ["品质好"], "reason": ""},
  {"comment_id": "c002", "sentiment": "negative", "labels": ["物流慢"], "reason": ""},
  {"comment_id": "c003", "sentiment": "neutral", "labels": [], "reason": ""},
  {"comment_id": "c004", "sentiment": "unknown", "labels": [], "reason": ""}
]

若 ai_comments_success.json 已有其他用途，新增 ai_comments_all_sentiments.json 作为统计测试专用 fixture。

5. 配置与数据模型测试

5.1 配置校验

测试文件：tests/unit/test_config.py

用例：

默认配置读取成功。
hot_limit 范围为 1–10。
item_limit_per_hot 范围为 1–10。
comment_limit_per_item 范围为 10–100。
AI_CONCURRENCY 默认值为 2。
AI_MAX_RETRIES 默认值为 3。

5.2 SQLite 初始化

测试文件：tests/unit/test_models.py

用例：

应用启动时调用 Base.metadata.create_all(engine) 可创建所有表。
SQLite engine 配置包含 check_same_thread=False 和 timeout=10。
连接建立后启用 PRAGMA journal_mode=WAL。

5.3 任务状态模型

测试文件：tests/unit/test_models.py

用例：

Task.status 仅允许 running、success、failed。
不允许写入 partial_success 或 partial_failed。
analysis_status 允许 normal、insufficient。
任务 status=success 时，analysis_success_rate < 0.8 只更新 analysis_status='insufficient'，不改变 status='success'。
任务 status=failed 时，analysis_success_rate 字段写入实际计算值，status 保持 'failed' 不变，不因 AI 成功率判断被覆盖为其他值。

6. 任务创建与恢复测试

6.1 创建任务 API

测试文件：tests/integration/test_task_creation.py

用例：

POST /api/tasks 使用合法参数创建任务，返回 task_id 和 status=running。
创建任务保存平台和配置字段：
- hot_limit
- item_limit_per_hot
- comment_limit_per_item
platform 非法时返回 422。
配置值越界时返回 422。
同一平台可重复创建任务，任务彼此独立。

6.2 ThreadPoolExecutor 单任务约束

测试文件：tests/unit/test_task_executor.py

用例：

任务执行器初始化为 ThreadPoolExecutor(max_workers=1)。
同时提交两个任务时，第二个任务必须等待第一个任务完成后才开始。

httpx 使用方式断言

测试文件：tests/unit/test_task_executor.py

用例：

后台任务执行函数不是协程：使用 inspect.iscoroutinefunction(run_task) 断言返回 False，确保其在 ThreadPoolExecutor 中以同步方式运行。
小红书平台客户端实例为 httpx.Client 同步类型：assert isinstance(client._http_client, httpx.Client)。
抖音平台客户端实例为 httpx.Client 同步类型，同上。
AI 服务客户端发出的请求使用 httpx.Client，不存在 httpx.AsyncClient 的实例化调用。

6.3 僵尸任务恢复

测试文件：tests/integration/test_task_recovery.py

用例：

数据库存在 status=running 的任务。
应用 lifespan 启动恢复逻辑执行后，该任务变为 status=failed。
error_stage=system。
error_type=unexpected_restart。
error_message=系统重启，任务被中断。

7. 平台抓取测试

7.1 小红书字段映射

测试文件：tests/unit/test_xiaohongshu_mapping.py

用例：

热点标题从 data.data.items[].title 读取。
不使用外层 data.data.title 作为热点标题。
热点 id 映射到 source_hot_id。
score 映射到 heat_value。
笔记 note.id 映射到 source_item_id。
优先选择 comments_count > 0 的笔记。
有评论笔记不足目标数时，补充 comments_count = 0 的笔记。
平台返回笔记总数不足目标数时，不标记任务失败。

笔记搜索分页限制

测试文件：tests/unit/test_xiaohongshu_mapping.py

用例：

搜索笔记接口只请求第一页（page=1），不发起后续翻页请求。
首页返回 comments_count > 0 的笔记已满足目标数量时，不发起第二次搜索请求。
首页返回结果不足目标数量时，执行降级策略（补充 comments_count = 0 的笔记），而非翻页搜索。

7.2 小红书评论字段映射

测试文件：tests/unit/test_xiaohongshu_mapping.py

用例：

source_comment_id 优先取 data.get("comment_id")。
comment_id 不存在时回退到 data.get("id")。
content 不存在时回退到 text。
like_count 和 create_time 正确映射。
原始评论 JSON 保存到 raw_data。
like_count 字段缺失时，入库值为 null，不抛出异常，不阻断评论抓取流程。
comment_time（create_time / create_time_str）字段缺失时，入库值为 null，不抛出异常，不阻断评论抓取流程。

7.4 抖音字段映射

测试文件：tests/unit/test_douyin_mapping.py

用例：

query_id 映射到 source_hot_id。
title 映射到热点标题。
rank 映射到热点排名。
hot_score 映射到 heat_value。
aweme_info.aweme_id 映射到 source_item_id。
aweme_info.desc 映射到标题或摘要。

7.5 抖音评论字段映射

测试文件：tests/unit/test_douyin_mapping.py

用例：

source_comment_id 优先取 data.get("comment_id")。
comment_id 不存在时回退到 data.get("cid")。
text 映射到 content。
digg_count 映射到 like_count。
create_time 映射到 comment_time。
digg_count 字段缺失时，入库值为 null，不抛出异常，不阻断评论抓取流程。
comment_time（create_time / create_time_str）字段缺失时，入库值为 null，不抛出异常，不阻断评论抓取流程。

8. 评论分页、限流与去重测试

8.1 评论分页终止条件

测试文件：tests/unit/test_comment_pagination.py

用例：

抓取评论数达到 comment_limit_per_item 后停止。
API 返回空评论列表后停止。
达到最大翻页轮次 5 后停止。
API 未返回下一页游标时停止。

8.2 分页请求间隔

测试文件：tests/unit/test_comment_pagination.py

用例：

每次评论分页请求之间调用 time.sleep，间隔值在 1.0 到 2.0 秒范围内（含边界值）。断言方式：assert 1.0 <= mock_sleep.call_args[0][0] <= 2.0。
收到 HTTP 429 后使用指数退避 1s → 2s → 4s。
429 退避完成后恢复基础分页间隔。

备注：若后续将间隔提取为配置项 CRAWL_PAGE_INTERVAL_SECONDS，改为读取配置值后断言，不硬编码数值。

8.3 评论去重

测试文件：tests/unit/test_comment_pagination.py

用例：

同一任务、同一内容条目、同一评论 ID 不重复入库。
重复抓取时更新已有记录或跳过重复记录。
不同任务下相同评论 ID 可分别保存。
同一内容出现在不同热点下时，按 task_id + hotspot_id + source_item_id 保留重复内容条目。