# zm-rag 测试文档

## 一、测试概述

本项目采用 **pytest + pytest-asyncio + httpx** 进行自动化测试，测试分为两个粒度：

| 粒度 | 文件前缀 | 说明 | 基础设施要求 |
|------|----------|------|-------------|
| **接口粒度** (API Tests) | `test_api_*.py` | 针对每个 API 端点的独立测试 | Docker 服务运行即可 (ASGI 模式) |
| **业务流程粒度** (Flow Tests) | `test_flow_*.py` | 端到端的业务场景测试 | Docker + uvicorn + Celery Worker |

截至 **2026-03-13**：
- `python -m pytest --collect-only -q` 共收集 **303** 个用例
- 其中 `test_api_*.py` 共 **153** 个 API 用例
- `test_flow_*.py` 共 **7** 个流程用例
- 其余为 ACL、schema、pipeline、query planner、research engine 等单元测试

---

## 二、测试环境准备

### 2.1 基础设施 (Docker)

```bash
cd docker
docker compose -p zm-rag up -d
```

确保以下服务正常运行：
- **OpenSearch 2.19.4** — `localhost:9200`
- **OpenSearch Dashboards** — `localhost:5601`
- **Neo4j 5** — `localhost:7474` (Web) / `localhost:7687` (Bolt)
- **Redis** — `localhost:6379`

验证 OpenSearch 版本：
```bash
curl http://localhost:9200/
# 应返回 "number": "2.19.4", "distribution": "opensearch"
```

### 2.2 Python 依赖

```bash
pip install pytest pytest-asyncio "opensearch-py[async]" asgi-lifespan httpx
```

> **Windows TLS 代理问题**：如系统代理使用 `https://` scheme 导致 `check_hostname` 错误，
> 需要通过设置 `HTTPS_PROXY=http://127.0.0.1:PORT` (注意 http://) 来解决。

### 2.3 项目已有依赖

测试所需但项目已自带的依赖：`fastapi`, `uvicorn`, `python-jose`, `pydantic-settings`, `httpx`

---

## 三、测试运行方式

### 3.1 ASGI 模式 (推荐用于 API 接口测试)

无需启动 uvicorn 服务端，直接通过 `ASGITransport` + `asgi-lifespan` 在测试进程内运行 FastAPI 应用：

```bash
cd backend
python -m pytest tests/test_api_*.py -v
```

此模式下：
- 每个测试方法创建独立的 FastAPI app 实例
- `asgi-lifespan.LifespanManager` 确保 ES/Neo4j/Redis 客户端正确初始化和关闭
- 速度快，适合 CI/CD
- **不支持 Celery 异步任务**（ingest 仅能验证任务提交，无法验证任务完成）

### 3.2 实时服务模式 (用于业务流程测试)

启动后端服务 + Celery Worker，测试通过 HTTP 访问真实服务端：

```bash
# 终端 1: 启动 FastAPI 后端
cd backend
uvicorn app.main:app --host 0.0.0.0 --port 8900

# 终端 2: 启动 Celery Worker
cd backend
celery -A app.tasks.worker worker -l info

# 终端 3: 运行测试
cd backend
python -m pytest tests/ --base-url http://localhost:8900 -v
```

### 3.3 指定运行特定测试

```bash
# 仅运行健康检查测试
python -m pytest tests/test_api_health.py -v

# 仅运行搜索相关测试
python -m pytest tests/test_api_search.py -v

# 仅运行全部流程测试
python -m pytest tests/test_flow_*.py --base-url http://localhost:8900 -v

# 按关键字筛选
python -m pytest tests/ -k "auth" -v

# 聚焦验证 Research 深度研究链路
python -m pytest tests/test_research_engine_unit.py tests/test_api_research.py -q
# 2026-03-13 验证结果：16 passed
```

---

## 四、测试用例清单

### 4.1 接口粒度测试 (API Tests)

#### `test_api_health.py` — 系统健康检查 (2 用例)

| # | 测试方法 | 说明 | 预期结果 |
|---|---------|------|---------|
| 1 | `test_health_endpoint` | GET /health 健康检查 | 200, status="ok", app="zm-rag" |
| 2 | `test_not_found` | GET /nonexistent 不存在路径 | 404 |

#### `test_api_mock.py` — Mock OA 接口 (9 用例)

| # | 测试方法 | 说明 | 预期结果 |
|---|---------|------|---------|
| 1 | `test_generate_token_default` | POST /mock/token 默认参数生成 JWT | 200, 返回 access_token |
| 2 | `test_generate_token_with_claims` | POST /mock/token 自定义声明 | 200, 返回 access_token |
| 3 | `test_login_admin` | POST /mock/login admin 登录 | 200, user_id=user_admin, role=admin |
| 4 | `test_login_regular_user` | POST /mock/login zhang_san 登录 | 200, display_name=张三（财务科） |
| 5 | `test_login_wrong_password` | POST /mock/login 错误密码 | 401 |
| 6 | `test_login_unknown_user` | POST /mock/login 不存在用户 | 401 |
| 7 | `test_list_users` | GET /mock/users 用户列表 | 200, ≥4 用户 |

#### `test_api_search.py` — 搜索接口 (8 用例)

| # | 测试方法 | 说明 | 预期结果 |
|---|---------|------|---------|
| 1 | `test_search_requires_auth` | 搜索接口鉴权 | 401/403 |
| 2 | `test_search_basic` | 基础关键词搜索 "数字政府" | 200, 含 total/documents/page |
| 3 | `test_search_with_filters` | 带 doc_type 过滤搜索 | 200, 分页正确 |
| 4 | `test_search_pagination` | 分页参数 page=2, page_size=5 | 200, 分页参数被接受 |
| 5 | `test_search_empty_query_rejected` | 空查询被拒绝 | 422 |
| 6 | `test_search_aggregations` | 搜索返回聚合数据 | 200, 含 aggregations |
| 7 | `test_suggest_requires_auth` | 建议接口鉴权 | 401/403 |
| 8 | `test_suggest_basic` | 查询建议 "数字" | 200, 含 suggestions 列表 |

#### `test_api_ingest.py` — 文档导入接口 (11 用例)

| # | 测试方法 | 说明 | 预期结果 |
|---|---------|------|---------|
| 1 | `test_trigger_requires_auth` | 触发导入鉴权 | 401/403 |
| 2 | `test_trigger_missing_pdf` | 不存在 PDF 路径 | 400, "not found" |
| 3 | `test_trigger_valid_pdf` | 有效 PDF 触发导入 | 200, task_id + PENDING |
| 4 | `test_webhook_missing_metadata` | Webhook 缺少 metadata | 422 |
| 5 | `test_webhook_invalid_json` | Webhook 无效 JSON | 400 |
| 6 | `test_webhook_missing_doc_id` | Webhook 缺少 doc_id | 400 |
| 7 | `test_webhook_valid_upload` | Webhook 正常上传 | 200, status=queued |
| 8 | `test_status_requires_auth` | 状态查询鉴权 | 401/403 |
| 9 | `test_status_unknown_task` | 未知任务状态查询 | 200, PENDING |
| 10 | `test_permission_empty_acl` | 空 ACL 跳过 | 200, status=skipped |
| 11 | `test_permission_valid` | 有效 ACL 更新 | 200, status=queued |

#### `test_api_document.py` — 文档详情接口 (6 用例)

| # | 测试方法 | 说明 | 预期结果 |
|---|---------|------|---------|
| 1 | `test_detail_requires_auth` | 详情接口鉴权 | 401/403 |
| 2 | `test_detail_nonexistent` | 不存在文档详情 | 404 |
| 3 | `test_detail_valid_doc` | 已有文档详情 | 200 + doc_id/title (或 skip) |
| 4 | `test_graph_requires_auth` | 图谱接口鉴权 | 401/403 |
| 5 | `test_graph_nonexistent` | 不存在文档图谱 | 200(空) 或 404 |
| 6 | `test_delete_requires_auth` | 删除接口鉴权 | 401/403 |

#### `test_api_admin.py` — 管理接口 (8 用例)

| # | 测试方法 | 说明 | 预期结果 |
|---|---------|------|---------|
| 1 | `test_stats_requires_auth` | 统计接口鉴权 | 401/403 |
| 2 | `test_stats_returns_200` | 系统统计数据 | 200, 含 total_documents |
| 3 | `test_stats_contains_service_health` | 三项服务健康状态 | es/neo4j/redis_status |
| 4 | `test_ingest_logs_requires_auth` | 导入日志鉴权 | 401/403 |
| 5 | `test_ingest_logs_returns_200` | 导入日志列表 | 200, total + records |
| 6 | `test_ingest_logs_pagination` | 导入日志分页 | ≤5 条记录 |
| 7 | `test_delete_requires_auth` | 管理员删除鉴权 | 401/403 |
| 8 | `test_delete_nonexistent_doc` | 删除不存在文档 | 404 |

#### `test_api_research.py` — 深度研究接口 (7 用例)

| # | 测试方法 | 说明 | 预期结果 |
|---|---------|------|---------|
| 1 | `test_research_plan_requires_auth` | 研究计划接口鉴权 | 401/403 |
| 2 | `test_research_plan_basic` | 计划生成 | 200, 返回 plan.summary 和 section_outline |
| 3 | `test_research_requires_auth` | 兼容模式研究接口鉴权 | 401/403 |
| 4 | `test_research_basic` | 兼容模式基础研究 (SSE) | 200, text/event-stream |
| 5 | `test_research_with_session` | 兼容模式多轮对话 session_id | 200 |
| 6 | `test_research_run_basic` | 执行确认后的深度研究 (SSE) | 200, text/event-stream |
| 7 | `test_research_rerun_section_basic` | 章节局部重跑 (SSE) | 200, text/event-stream |

#### `test_api_graph.py` — 知识图谱接口 (6 用例)

| # | 测试方法 | 说明 | 预期结果 |
|---|---------|------|---------|
| 1 | `test_overview` | 图谱全局概览 | 200 |
| 2 | `test_search_requires_auth` | 实体搜索鉴权 | 401/403 |
| 3 | `test_search_basic` | 实体名称搜索 "广东" | 200, 返回列表 |
| 4 | `test_entity_nonexistent` | 不存在实体查询 | 200(空) 或 404 |
| 5 | `test_related_docs` | 关联文档查询 | 200, 返回列表 |
| 6 | `test_doc_entities` | 文档实体列表 | 200 或 404 |

**接口粒度合计: 153 个用例**（按 `python -m pytest --collect-only -q tests/test_api_*.py` 统计，2026-03-13）

---

### 4.2 业务流程粒度测试 (Flow Tests)

#### `test_flow_ingest_search.py` — 文档导入→搜索全流程

**流程 1: 完整导入搜索流程** (`test_full_ingest_search_flow`)

```
Login(admin) → Webhook 上传 PDF → 等待导入完成 → 管理日志验证
→ 搜索文档 → 文档详情 → 文档图谱 → 删除清理
```

| 步骤 | 操作 | 验证点 |
|------|------|--------|
| 1 | POST /mock/login 管理员登录 | 200, 获取 token |
| 2 | POST /ingest/webhook/document 上传 PDF | 200, 返回 task_id |
| 3 | 轮询 /ingest/status/{task_id} | 状态变为 COMPLETED |
| 4 | GET /admin/ingest-logs | doc_id 出现在日志中 |
| 5 | POST /search 搜索关键词 | 200, total ≥ 0 |
| 6 | GET /document/{doc_id} 文档详情 | 200, title 匹配 |
| 7 | GET /document/{doc_id}/graph | 200 |
| 8 | DELETE /admin/document/{doc_id} | 200, 文档被删除 |

**流程 2: 批量导入** (`test_batch_ingest`)

```
Login → 批量上传 3 个 PDF → 验证任务状态
```

#### `test_flow_permission.py` — ACL 权限控制流程

**流程: 权限可见性验证** (`test_acl_restricts_visibility`)

```
管理员导入文档(ACL=D_05) → zhang_san(D_05)可见 → li_si(D_08)不可见 → 清理
```

| 步骤 | 操作 | 验证点 |
|------|------|--------|
| 1 | 管理员通过 webhook 导入文档，acl_ids=["D_05"] | 200, 导入成功 |
| 2 | zhang_san(dept=D_05) 访问文档详情 | 200, 有权限 |
| 3 | li_si(dept=D_08) 访问文档详情 | 403/404, 无权限 |
| 4 | 管理员删除测试文档 | 清理 |

#### `test_flow_research.py` — 深度研究 Q&A 流程

| # | 测试方法 | 说明 | 验证点 |
|---|---------|------|--------|
| 1 | `test_research_sse_stream` | SSE 流式响应解析 | 包含 done 事件类型 |
| 2 | `test_research_multiturn` | 多轮对话 + session_id | 两轮均 200 |
| 3 | `test_research_empty_question_rejected` | 空问题拒绝 | 400/422 |

#### `test_flow_graph.py` — 知识图谱构建流程

**流程: 导入→图谱构建→查询** (`test_ingest_builds_graph`)

```
导入文档 → 等待完成 → 文档图谱子图 → 全局概览 → 实体搜索 → 清理
```

| 步骤 | 操作 | 验证点 |
|------|------|--------|
| 1 | Webhook 上传 PDF | 200, task_id |
| 2 | 等待导入完成 | COMPLETED |
| 3 | GET /document/{doc_id}/graph | 200, 返回图谱数据 |
| 4 | GET /graph/overview | 200 |
| 5 | POST /graph/search 实体搜索 | 200 |
| 6 | DELETE /admin/document/{doc_id} | 清理 |

**业务流程合计: 7 个用例**（其中 3 个需要 Celery Worker + LLM）

---

## 五、测试架构说明

### 5.1 目录结构

```
backend/tests/
├── conftest.py                    # 共享 fixture（client、auth、PDF 路径）
├── __init__.py
│
├── test_api_health.py             # 健康检查接口
├── test_api_mock.py               # Mock OA 接口
├── test_api_search.py             # 搜索接口
├── test_api_ingest.py             # 文档导入接口
├── test_api_document.py           # 文档详情接口
├── test_api_admin.py              # 管理接口
├── test_api_research.py           # 深度研究接口
├── test_api_graph.py              # 知识图谱接口
│
├── test_flow_ingest_search.py     # 流程: 导入→搜索全生命周期
├── test_flow_permission.py        # 流程: ACL 权限控制
├── test_flow_research.py          # 流程: 研究 Q&A 管道
└── test_flow_graph.py             # 流程: 图谱构建+查询
```

### 5.2 conftest.py 核心 Fixture

| Fixture | Scope | 说明 |
|---------|-------|------|
| `client` | function | AsyncClient, 自动选择 ASGI/HTTP 模式 |
| `auth_headers` | function | admin JWT token (user_admin, O_01, D_01, R_01) |
| `user_headers` | function | 普通用户 token (user_001, O_17, D_05) |
| `example_pdf_path` | session | 最小的 example-files/*.pdf 路径 |
| `all_example_pdfs` | session | 所有 PDF 列表 |
| `make_token` | function | JWT 生成工厂函数 |
| `api_prefix` | function | 返回 "/api/v1" |

### 5.3 测试模式切换

```python
# conftest.py 中的关键逻辑
if base_url:
    # 实时服务模式: 直接 HTTP 访问
    async with AsyncClient(base_url=base_url) as ac:
        yield ac
else:
    # ASGI 模式: 进程内运行 FastAPI + lifespan
    from asgi_lifespan import LifespanManager
    app = create_app()
    async with LifespanManager(app) as manager:
        transport = ASGITransport(app=manager.app)
        async with AsyncClient(transport=transport, base_url="http://testserver") as ac:
            yield ac
```

### 5.4 测试用户

| 用户名 | 密码 | user_id | 科室 | 部门 | 角色 |
|--------|------|---------|------|------|------|
| admin | admin123 | user_admin | O_01 | D_01 | R_01 (地区领导) |
| zhang_san | user123 | user_001 | O_17 | D_05 | — (普通用户) |
| li_si | user123 | user_002 | O_22 | D_08 | — (普通用户) |
| wang_wu | manager123 | user_003 | O_01 | D_05 | R_03 (部门领导) |

---

## 六、测试执行结果

### 6.1 最新运行 (2026-03-10)

**环境**: OpenSearch 2.19.4, Python 3.11, pytest 8.4.2

```
ASGI 模式 (test_api_*.py + test_flow_research.py):
  53 passed, 1 skipped, 0 failed  (26.48s)

跳过原因:
  - test_detail_valid_doc: 暂无已导入文档，skip
```

### 6.2 CI/CD 集成建议

```yaml
# GitHub Actions 示例
test:
  runs-on: ubuntu-latest
  services:
    opensearch:
      image: opensearchproject/opensearch:2.19.4
      ports: ["9200:9200"]
      env:
        discovery.type: single-node
        plugins.security.disabled: "true"
    neo4j:
      image: neo4j:5-community
      ports: ["7687:7687"]
      env:
        NEO4J_AUTH: neo4j/zm_rag_2024
    redis:
      image: redis:alpine
      ports: ["6379:6379"]

  steps:
    - uses: actions/checkout@v4
    - uses: actions/setup-python@v5
      with: { python-version: "3.11" }
    - run: pip install -e .[test]
    - run: pytest tests/test_api_*.py -v
```

---

## 七、常见问题

### Q1: pip 安装报 `check_hostname requires server_hostname`

**原因**: Windows 系统代理使用 `https://` scheme，pip 21.x 的 urllib3 不支持 TLS-in-TLS。

**解决**:
```python
# 通过 subprocess 设置 HTTPS_PROXY 为 http:// 前缀
import subprocess, os
env = os.environ.copy()
env['HTTPS_PROXY'] = 'http://127.0.0.1:7897'  # 注意 http:// 不是 https://
subprocess.run(['python', '-m', 'pip', 'install', 'pytest'], env=env)
```

### Q2: 流程测试超时

确保 Celery Worker 正在运行：
```bash
celery -A app.tasks.worker worker -l info
```

### Q3: OpenSearch 连接失败

检查 Docker 容器状态：
```bash
docker compose -p zm-rag ps
curl http://localhost:9200/_cluster/health
```