junhong_cmp_fiber/flow_tests/AGENTS.md

流程测试规范文档

本文档定义了业务流程测试的编写规范。AI 助手根据用户描述的业务流程自动生成测试脚本。

快速开始

环境准备

cd flow_tests
python3 -m venv venv
source venv/bin/activate  # Windows: venv\Scripts\activate
pip install -r requirements.txt

运行测试

# 运行所有测试
pytest

# 运行指定模块
pytest tests/test_account_flow.py -v

# 运行指定流程
pytest tests/test_account_flow.py::test_create_agent_flow -v

# 查看详细输出
pytest -v --tb=short

---

用户如何描述流程

描述格式（自然语言即可）

用户只需用自然语言描述业务流程，AI 会自动：

1. 理解流程意图
2. 找到对应的后端接口
3. 生成测试脚本
4. 处理数据清理

描述示例

示例 1：简单流程

"测试创建代理商：平台管理员创建一个店铺，然后给这个店铺创建管理员账号，验证这个账号能登录"

示例 2：带条件的流程

"测试套餐分配：代理商只有被分配了套餐才能卖货。先创建店铺，不分配套餐时应该看不到任何可售套餐，分配后才能看到"

示例 3：异步流程

"测试设备导入：上传 CSV 文件导入设备，这是异步任务，需要等待任务完成后验证设备确实入库了"

示例 4：涉及第三方

"测试充值流程：用户下单充值，支付成功后卡片流量应该增加。支付回调可以直接模拟"

需要说明的要素

要素	必须	说明
操作角色	是	谁在操作（平台管理员/代理商/企业用户/普通用户）
操作步骤	是	按顺序描述要做什么
预期结果	是	每步操作后应该发生什么
前置条件	否	如果有特殊前置条件需要说明
异步等待	否	如果涉及异步任务需要说明

---

测试框架结构

flow_tests/
├── AGENTS.md              # 本规范文档
├── openapi.yaml           # OpenAPI 接口文档（AI 助手必读）
├── requirements.txt       # Python 依赖
├── pytest.ini            # pytest 配置
├── config/
│   ├── settings.py       # 配置加载
│   ├── local.yaml        # 本地环境配置
│   └── remote.yaml       # 远程环境配置
├── core/
│   ├── __init__.py
│   ├── client.py         # HTTP 客户端封装
│   ├── auth.py           # 认证管理
│   ├── database.py       # 数据库直连（验证用）
│   ├── cleanup.py        # 数据清理追踪器
│   ├── mock.py           # 第三方服务 Mock
│   └── wait.py           # 异步任务等待器
├── fixtures/
│   ├── __init__.py
│   └── common.py         # 通用 pytest fixtures
└── tests/
    ├── __init__.py
    ├── test_account_flow.py    # 账号管理流程
    ├── test_shop_flow.py       # 店铺管理流程
    └── ...                     # 其他模块流程

---

核心组件说明

1. HTTP 客户端 (core/client.py)

from core.client import APIClient

# 创建客户端
client = APIClient()

# 登录获取 token
client.login("admin", "password")

# 发起请求（自动带 token）
resp = client.post("/api/admin/accounts", json={...})
resp = client.get("/api/admin/accounts/1")

# 断言响应
assert resp.ok()                    # code == 0
assert resp.code == 0
assert resp.data["id"] == 1

2. 数据清理追踪器 (core/cleanup.py)

核心原则：只删除测试创建的数据，不影响原有数据

from core.cleanup import CleanupTracker

# 在 fixture 中初始化
tracker = CleanupTracker(db_connection)

# 记录创建的数据
account_id = create_account(...)
tracker.track("admin_accounts", account_id)

shop_id = create_shop(...)
tracker.track("shops", shop_id)

# 测试结束后自动清理（逆序删除，处理依赖）
tracker.cleanup()  # 先删 account，再删 shop

3. 认证管理 (core/auth.py)

from core.auth import AuthManager

auth = AuthManager(client)

# 预置角色快速登录
auth.as_super_admin()           # 超级管理员
auth.as_platform_admin()        # 平台管理员
auth.as_agent(shop_id)          # 代理商（需指定店铺）
auth.as_enterprise(ent_id)      # 企业用户

# 自定义账号登录
auth.login("custom_account", "password")

4. 异步任务等待器 (core/wait.py)

from core.wait import wait_for_task, wait_for_condition

# 等待异步任务完成
result = wait_for_task(
    task_type="device_import",
    task_id=task_id,
    timeout=60,
    poll_interval=2
)

# 等待条件满足
wait_for_condition(
    condition=lambda: db.query("SELECT count(*) FROM devices WHERE batch_id = %s", batch_id) > 0,
    timeout=30,
    message="等待设备入库"
)

5. Mock 服务 (core/mock.py)

from core.mock import MockService

mock = MockService(db_connection)

# 模拟支付回调
mock.payment_success(order_id, amount=100)

# 模拟短信验证码（直接写入数据库/Redis）
mock.sms_code(phone="13800138000", code="123456")

# 模拟第三方 API 响应（如果后端支持 mock 模式）
mock.external_api("carrier_recharge", response={"success": True})

---

测试编写规范

基本结构

"""
账号管理流程测试

测试场景：
1. 创建代理商账号流程
2. 账号权限验证流程
...
"""
import pytest
from core.client import APIClient
from core.auth import AuthManager
from core.cleanup import CleanupTracker


class TestAccountFlow:
    """账号管理流程"""

    def test_create_agent_account_flow(self, client, auth, tracker, db):
        """
        流程：创建代理商账号
        
        步骤：
        1. 平台管理员创建店铺
        2. 给店铺创建管理员账号
        3. 验证新账号能登录
        4. 验证只能看到自己店铺的数据
        """
        # === 1. 平台管理员创建店铺 ===
        auth.as_platform_admin()
        
        resp = client.post("/api/admin/shops", json={
            "name": "测试代理商",
            "contact": "张三",
            "phone": "13800138000"
        })
        assert resp.ok(), f"创建店铺失败: {resp.msg}"
        shop_id = resp.data["id"]
        tracker.track("shops", shop_id)

        # === 2. 创建店铺管理员账号 ===
        resp = client.post("/api/admin/accounts", json={
            "username": "test_agent_admin",
            "password": "Test123456",
            "shop_id": shop_id,
            "role_ids": [2]  # 假设 2 是店铺管理员角色
        })
        assert resp.ok(), f"创建账号失败: {resp.msg}"
        account_id = resp.data["id"]
        tracker.track("admin_accounts", account_id)

        # === 3. 验证新账号能登录 ===
        auth.login("test_agent_admin", "Test123456")
        
        resp = client.get("/api/admin/auth/me")
        assert resp.ok()
        assert resp.data["shop_id"] == shop_id

        # === 4. 验证只能看到自己店铺数据 ===
        resp = client.get("/api/admin/shops")
        assert resp.ok()
        # 代理商只能看到自己的店铺
        assert len(resp.data["list"]) == 1
        assert resp.data["list"][0]["id"] == shop_id

命名规范

类型	规范	示例
文件名	test_{模块}_flow.py	test_account_flow.py
类名	Test{模块}Flow	TestAccountFlow
方法名	test_{流程描述}	test_create_agent_account_flow
方法文档	必须包含流程步骤	见上方示例

Fixtures 使用

# fixtures/common.py 提供以下通用 fixtures

@pytest.fixture
def client():
    """HTTP 客户端"""
    return APIClient()

@pytest.fixture
def auth(client):
    """认证管理器"""
    return AuthManager(client)

@pytest.fixture
def db():
    """数据库连接（只读验证用）"""
    return get_db_connection()

@pytest.fixture
def tracker(db):
    """数据清理追踪器"""
    t = CleanupTracker(db)
    yield t
    t.cleanup()  # 测试结束自动清理

@pytest.fixture
def mock(db):
    """Mock 服务"""
    return MockService(db)

---

异步任务测试规范

导入类任务（设备导入、IoT卡导入）

def test_device_import_flow(self, client, auth, tracker, db):
    """
    流程：设备批量导入
    
    步骤：
    1. 上传 CSV 文件
    2. 创建导入任务
    3. 等待任务完成
    4. 验证设备入库
    """
    auth.as_platform_admin()
    
    # 1. 上传文件
    with open("fixtures/devices.csv", "rb") as f:
        resp = client.upload("/api/admin/storage/upload", file=f)
    file_url = resp.data["url"]
    
    # 2. 创建导入任务
    resp = client.post("/api/admin/device-imports", json={
        "file_url": file_url,
        "carrier_id": 1
    })
    assert resp.ok()
    task_id = resp.data["task_id"]
    
    # 3. 等待任务完成
    from core.wait import wait_for_task
    result = wait_for_task("device_import", task_id, timeout=60)
    assert result["status"] == "completed"
    
    # 4. 验证设备入库
    imported_count = db.scalar(
        "SELECT count(*) FROM devices WHERE import_task_id = %s", 
        task_id
    )
    assert imported_count == 10  # CSV 中有 10 条
    
    # 追踪清理
    tracker.track_by_query("devices", f"import_task_id = {task_id}")

支付回调类任务

def test_recharge_flow(self, client, auth, tracker, db, mock):
    """
    流程：充值支付
    
    步骤：
    1. 用户创建充值订单
    2. 模拟支付成功回调
    3. 验证卡片流量增加
    """
    auth.as_enterprise(enterprise_id=1)
    
    # 1. 创建充值订单
    resp = client.post("/api/h5/recharge/orders", json={
        "card_id": 123,
        "package_id": 456
    })
    assert resp.ok()
    order_id = resp.data["order_id"]
    tracker.track("orders", order_id)
    
    # 获取支付前流量
    before_data = db.scalar(
        "SELECT data_balance FROM iot_cards WHERE id = 123"
    )
    
    # 2. 模拟支付回调
    mock.payment_success(order_id, amount=50.00)
    
    # 3. 验证流量增加
    from core.wait import wait_for_condition
    wait_for_condition(
        condition=lambda: db.scalar(
            "SELECT data_balance FROM iot_cards WHERE id = 123"
        ) > before_data,
        timeout=10,
        message="等待流量到账"
    )

---

角色权限说明

测试时需要了解系统角色体系：

角色	说明	数据范围
超级管理员	系统最高权限	全部数据
平台管理员	平台运营人员	全部数据（受权限配置限制）
代理商管理员	店铺管理者	本店铺 + 下级店铺
代理商员工	店铺普通员工	本店铺
企业管理员	企业用户	本企业数据

---

环境配置

config/local.yaml

# 本地开发环境
api:
  base_url: "http://localhost:3000"
  timeout: 30

database:
  host: "localhost"
  port: 5432
  name: "junhong_dev"
  user: "postgres"
  password: "postgres"

redis:
  host: "localhost"
  port: 6379
  db: 0

# 预置测试账号
accounts:
  super_admin:
    username: "superadmin"
    password: "Admin123456"
  platform_admin:
    username: "platform"
    password: "Admin123456"

config/remote.yaml

# 远程测试环境
api:
  base_url: "https://test-api.example.com"
  timeout: 30

database:
  host: "test-db.example.com"
  port: 5432
  name: "junhong_test"
  user: "test_user"
  password: "test_password"

redis:
  host: "test-redis.example.com"
  port: 6379
  db: 0

切换环境

# 使用本地环境（默认）
pytest

# 使用远程环境
TEST_ENV=remote pytest

---

数据清理规则

清理原则

1. 只删除测试创建的数据 - 通过 tracker 追踪
2. 逆序删除 - 先删依赖方，再删被依赖方
3. 软删除优先 - 如果表支持软删除，使用软删除
4. 级联处理 - 自动处理关联数据

追踪方式

# 方式1：追踪单条记录
tracker.track("table_name", record_id)

# 方式2：追踪多条记录
tracker.track_many("table_name", [id1, id2, id3])

# 方式3：按条件追踪（用于批量导入等场景）
tracker.track_by_query("devices", "import_task_id = 123")

# 方式4：追踪关联数据（自动级联）
tracker.track_with_relations("shops", shop_id, relations=[
    ("admin_accounts", "shop_id"),
    ("shop_packages", "shop_id")
])

清理顺序配置

# core/cleanup.py 中定义表的依赖关系
TABLE_DEPENDENCIES = {
    "admin_accounts": ["shops"],           # accounts 依赖 shops
    "shop_packages": ["shops", "packages"], # shop_packages 依赖 shops 和 packages
    "orders": ["iot_cards", "packages"],
    # ... 根据实际表结构配置
}

---

常见问题

Q: 如何处理需要真实第三方服务的测试？

A: 两种方式：

1. 使用 Mock 模式（推荐）：直接模拟回调或写入预期数据
2. 如果必须真实调用：在配置中标记为 skip_in_mock_mode，仅在集成环境运行

Q: 测试数据影响了其他人怎么办？

A:

1. 本地环境：数据隔离，不影响他人
2. 远程环境：使用唯一标识（如 UUID）创建数据，确保清理

Q: 异步任务超时怎么办？

A:

1. 检查任务是否真的启动了
2. 检查 worker 是否在运行
3. 增加超时时间（最后手段）
4. 查看任务日志定位问题

---

接口文档（OpenAPI）

重要：生成测试时，AI 助手必须首先查阅 flow_tests/openapi.yaml 获取准确的接口信息。

文件位置

flow_tests/openapi.yaml    # OpenAPI 3.0 规范文档

文档结构

# openapi.yaml 结构
components:
  schemas:           # 数据模型定义（DTO）
    DtoCreateShopRequest:
      properties:
        shop_name: { type: string }
        shop_code: { type: string }
        # ...
      required: [shop_name, shop_code]

paths:               # API 路径定义
  /api/admin/shops:
    post:
      summary: 创建店铺
      requestBody:
        content:
          application/json:
            schema:
              $ref: '#/components/schemas/DtoCreateShopRequest'
      responses:
        '200':
          content:
            application/json:
              schema:
                $ref: '#/components/schemas/DtoShopResponse'

AI 助手使用方式

1. 查找接口：根据用户描述的流程，在 paths 中找到对应的 API 路径
2. 获取请求参数：从 requestBody.schema 获取请求体结构
3. 获取响应格式：从 responses.200.schema 获取响应体结构
4. 理解字段含义：从 components.schemas 中查看字段的 description

示例：查找创建店铺接口

用户说："创建一个店铺"

AI 助手应该：

1. 读取 openapi.yaml
2. 搜索 shops 相关路径 → 找到 POST /api/admin/shops
3. 查看 DtoCreateShopRequest 了解必填字段：shop_name, shop_code, init_username, init_phone, init_password
4. 生成正确的测试代码

resp = client.post("/api/admin/shops", json={
    "shop_name": "测试店铺",
    "shop_code": "TEST001",
    "init_username": "test_admin",
    "init_phone": "13800138000",
    "init_password": "Test123456",
})

---

AI 助手工作流程

当用户描述业务流程后，AI 助手按以下步骤工作：

1. 理解流程：分析用户描述，提取操作角色、步骤、预期结果
2. 查阅接口文档：读取 flow_tests/openapi.yaml 获取准确的接口路径、请求参数、响应格式
3. 生成测试：按照本规范生成测试代码，使用 OpenAPI 文档中的字段定义
4. 补充清理：添加数据追踪和清理逻辑
5. 运行验证：执行测试确保通过
6. 报告结果：告知用户测试结果，如发现问题则报告

接口查找优先级

优先级	来源	说明
1	flow_tests/openapi.yaml	首选，最准确的接口定义
2	项目代码 internal/handler/	OpenAPI 未覆盖时查找源码
3	项目代码 internal/router/	确认路由注册

---

版本记录

版本	日期	说明
1.0	2026-02-02	初始版本
1.1	2026-02-02	增加 OpenAPI 接口文档规范