save code

backend/tests/TEST_GUIDE.md

@@ -0,0 +1,339 @@
+# 后端测试文档
+
+## 1. 测试概述
+
+### 1.1 测试目标
+- 确保API接口功能正确
+- 验证业务逻辑准确性
+- 保证代码质量和稳定性
+- 测试覆盖率达到80%以上
+
+### 1.2 测试范围
+- 单元测试：测试各个模块的独立功能
+- 集成测试：测试API接口的端到端功能
+- 性能测试：测试API响应时间（可选）
+
+### 1.3 测试技术栈
+| 技术 | 说明 |
+|------|------|
+| 测试框架 | pytest |
+| 异步测试 | pytest-asyncio |
+| HTTP测试 | httpx / TestClient |
+| 数据库测试 | pytest-postgresql（可选） |
+| 覆盖率 | pytest-cov |
+| Mock | pytest-mock |
+
+### 1.4 测试目录结构
+```
+backend/tests/
+├── __init__.py
+├── conftest.py              # pytest配置和fixture
+├── test_auth.py             # 认证模块测试
+├── test_users.py            # 用户管理模块测试
+├── test_projects.py         # 项目管理模块测试
+├── test_statistics.py       # 统计功能测试
+└── fixtures/                # 测试数据fixture
+    ├── __init__.py
+    └── test_data.py
+```
+
+## 2. 测试环境配置
+
+### 2.1 依赖安装
+```bash
+pip install pytest pytest-asyncio pytest-cov pytest-mock httpx
+```
+
+### 2.2 测试配置文件 pytest.ini
+```ini
+[pytest]
+testpaths = tests
+python_files = test_*.py
+python_classes = Test*
+python_functions = test_*
+addopts =
+    -v
+    --strict-markers
+    --cov=src
+    --cov-report=term-missing
+    --cov-report=html
+asyncio_mode = auto
+markers =
+    unit: 单元测试
+    integration: 集成测试
+    slow: 慢速测试
+```
+
+### 2.3 环境变量
+测试环境使用独立数据库配置：
+```env
+TEST_DB_HOST=localhost
+TEST_DB_NAME=project_manager_test
+TEST_DB_USER=test_user
+TEST_DB_PASSWORD=test_password
+```
+
+## 3. 测试策略
+
+### 3.1 单元测试
+- 测试各个Service层方法
+- 测试工具函数
+- 不涉及数据库和网络调用
+- 使用Mock隔离外部依赖
+
+### 3.2 集成测试
+- 测试完整的API端点
+- 使用测试数据库
+- 验证请求/响应格式
+- 测试权限控制
+
+### 3.3 测试数据管理
+- 使用pytest fixture创建测试数据
+- 每个测试独立运行，互不影响
+- 测试后清理数据
+
+## 4. 测试用例设计
+
+### 4.1 认证模块测试 (test_auth.py)
+
+#### 4.1.1 用户登录测试
+| 测试用例 | 描述 | 预期结果 |
+|---------|------|---------|
+| test_login_success | 正确的用户名和密码 | 返回token和用户信息 |
+| test_login_wrong_username | 错误的用户名 | 返回401错误 |
+| test_login_wrong_password | 错误的密码 | 返回401错误 |
+| test_login_missing_fields | 缺少必填字段 | 返回400错误 |
+
+#### 4.1.2 Token验证测试
+| 测试用例 | 描述 | 预期结果 |
+|---------|------|---------|
+| test_get_current_user_success | 有效token | 返回用户信息 |
+| test_get_current_user_invalid_token | 无效token | 返回401错误 |
+| test_get_current_user_expired_token | 过期token | 返回401错误 |
+| test_logout_success | 有效token登出 | 返回成功 |
+
+### 4.2 用户管理模块测试 (test_users.py)
+
+#### 4.2.1 创建用户测试
+| 测试用例 | 描述 | 预期结果 |
+|---------|------|---------|
+| test_create_user_admin_success | 管理员创建用户 | 返回创建的用户 |
+| test_create_user_market_forbidden | 非管理员创建用户 | 返回403错误 |
+| test_create_user_duplicate_username | 重复的用户名 | 返回409错误 |
+| test_create_user_invalid_role | 无效的角色 | 返回400错误 |
+
+#### 4.2.2 获取用户列表测试
+| 测试用例 | 描述 | 预期结果 |
+|---------|------|---------|
+| test_get_users_admin_success | 管理员获取用户列表 | 返回用户列表 |
+| test_get_users_market_forbidden | 非管理员获取用户列表 | 返回403错误 |
+| test_get_users_with_pagination | 测试分页 | 返回正确的分页数据 |
+| test_get_users_with_filter | 测试筛选 | 返回筛选后的数据 |
+
+#### 4.2.3 更新用户测试
+| 测试用例 | 描述 | 预期结果 |
+|---------|------|---------|
+| test_update_user_admin_success | 管理员更新用户 | 返回更新后的用户 |
+| test_update_user_market_forbidden | 非管理员更新用户 | 返回403错误 |
+| test_update_user_not_found | 更新不存在的用户 | 返回404错误 |
+
+#### 4.2.4 删除用户测试
+| 测试用例 | 描述 | 预期结果 |
+|---------|------|---------|
+| test_delete_user_admin_success | 管理员删除用户 | 返回成功 |
+| test_delete_user_market_forbidden | 非管理员删除用户 | 返回403错误 |
+| test_delete_user_not_found | 删除不存在的用户 | 返回404错误 |
+
+#### 4.2.5 重置密码测试
+| 测试用例 | 描述 | 预期结果 |
+|---------|------|---------|
+| test_reset_password_admin_success | 管理员重置密码 | 返回成功 |
+| test_reset_password_market_forbidden | 非管理员重置密码 | 返回403错误 |
+| test_reset_password_invalid_password | 无效的密码格式 | 返回400错误 |
+
+### 4.3 项目管理模块测试 (test_projects.py)
+
+#### 4.3.1 获取项目列表测试
+| 测试用例 | 描述 | 预期结果 |
+|---------|------|---------|
+| test_get_projects_all_users_success | 所有用户获取项目列表 | 返回项目列表 |
+| test_get_projects_with_pagination | 测试分页 | 返回正确的分页数据 |
+| test_get_projects_with_filters | 测试组合筛选 | 返回筛选后的数据 |
+| test_get_projects_sort_by_contract_amount | 按合同金额排序 | 返回排序后的数据 |
+
+#### 4.3.2 创建项目测试
+| 测试用例 | 描述 | 预期结果 |
+|---------|------|---------|
+| test_create_project_admin_success | 管理员创建项目 | 返回创建的项目 |
+| test_create_project_market_success | 市场部用户创建项目 | 返回创建的项目 |
+| test_create_project_other_forbidden | 其他部门用户创建项目 | 返回403错误 |
+| test_create_project_duplicate_project_no | 重复的项目编号 | 返回409错误 |
+| test_create_project_missing_required_fields | 缺少必填字段 | 返回400错误 |
+
+#### 4.3.3 更新项目测试
+| 测试用例 | 描述 | 预期结果 |
+|---------|------|---------|
+| test_update_project_admin_success | 管理员更新项目 | 返回更新后的项目 |
+| test_update_project_market_own_project | 市场部更新自己的项目 | 返回更新后的项目 |
+| test_update_project_market_other_project | 市场部更新其他人的项目 | 返回403错误 |
+| test_update_project_other_financial_info | 其他部门更新财务信息 | 返回更新后的项目 |
+| test_update_project_other_basic_info | 其他部门更新基础信息 | 返回403错误 |
+| test_update_project_not_found | 更新不存在的项目 | 返回404错误 |
+
+#### 4.3.4 删除项目测试
+| 测试用例 | 描述 | 预期结果 |
+|---------|------|---------|
+| test_delete_project_admin_success | 管理员删除项目 | 返回成功 |
+| test_delete_project_market_own_project | 市场部删除自己的项目 | 返回成功 |
+| test_delete_project_market_other_project | 市场部删除其他人的项目 | 返回403错误 |
+| test_delete_project_other_forbidden | 其他部门删除项目 | 返回403错误 |
+| test_delete_project_not_found | 删除不存在的项目 | 返回404错误 |
+
+### 4.4 统计功能测试 (test_statistics.py)
+
+#### 4.4.1 基础统计测试
+| 测试用例 | 描述 | 预期结果 |
+|---------|------|---------|
+| test_get_statistics_all_users_success | 所有用户获取统计 | 返回统计数据 |
+| test_get_statistics_with_filters | 测试筛选条件 | 返回筛选后的统计 |
+| test_get_statistics_empty_data | 无数据时统计 | 返回零值统计 |
+
+#### 4.4.2 分组统计测试
+| 测试用例 | 描述 | 预期结果 |
+|---------|------|---------|
+| test_get_group_statistics_by_engineering_type | 按工程类别分组 | 返回分组统计 |
+| test_get_group_statistics_by_department | 按项目部分组 | 返回分组统计 |
+| test_get_group_statistics_with_date_filter | 带日期筛选的分组统计 | 返回分组统计 |
+
+#### 4.4.3 时间维度统计测试
+| 测试用例 | 描述 | 预期结果 |
+|---------|------|---------|
+| test_get_timeline_statistics_by_month | 按月统计 | 返回时间维度统计 |
+| test_get_timeline_statistics_by_year | 按年统计 | 返回时间维度统计 |
+| test_get_timeline_statistics_with_filter | 带筛选的时间统计 | 返回时间维度统计 |
+
+## 5. 测试Fixture
+
+### 5.1 数据库Fixture
+```python
+@pytest.fixture
+async def db_session():
+    # 创建测试数据库会话
+    # 测试结束后回滚
+    yield session
+    # 清理数据
+```
+
+### 5.2 用户Fixture
+```python
+@pytest.fixture
+async def admin_user(db_session):
+    # 创建管理员用户
+    return user
+
+@pytest.fixture
+async def market_user(db_session):
+    # 创建市场部用户
+    return user
+
+@pytest.fixture
+async def other_user(db_session):
+    # 创建其他部门用户
+    return user
+```
+
+### 5.3 项目Fixture
+```python
+@pytest.fixture
+async def test_project(db_session, admin_user):
+    # 创建测试项目
+    return project
+```
+
+### 5.4 Token Fixture
+```python
+@pytest.fixture
+async def admin_token(client, admin_user):
+    # 获取管理员token
+    return token
+```
+
+## 6. 运行测试
+
+### 6.1 运行所有测试
+```bash
+cd backend
+pytest tests/
+```
+
+### 6.2 运行特定测试文件
+```bash
+pytest tests/test_auth.py
+```
+
+### 6.3 运行特定测试用例
+```bash
+pytest tests/test_auth.py::test_login_success
+```
+
+### 6.4 生成覆盖率报告
+```bash
+pytest tests/ --cov=src --cov-report=html
+```
+
+### 6.5 运行集成测试
+```bash
+pytest tests/ -m integration
+```
+
+### 6.6 运行单元测试
+```bash
+pytest tests/ -m unit
+```
+
+## 7. 持续集成
+
+### 7.1 测试流程
+1. 提交代码前运行单元测试
+2. CI/CD流水线运行完整测试套件
+3. 生成覆盖率报告
+4. 测试通过才能合并代码
+
+### 7.2 测试覆盖率要求
+- 单元测试覆盖率 > 80%
+- 关键业务逻辑覆盖率 > 90%
+- API接口覆盖率 100%
+
+## 8. 测试最佳实践
+
+### 8.1 测试命名规范
+- 测试文件: `test_<模块名>.py`
+- 测试类: `Test<功能名>`
+- 测试方法: `test_<功能>_<场景>`
+
+### 8.2 测试编写原则
+- 每个测试只验证一个功能点
+- 测试之间相互独立
+- 使用描述性的测试名称
+- 遵循AAA模式：Arrange, Act, Assert
+
+### 8.3 Mock使用
+- Mock外部依赖（数据库、网络等）
+- 不要Mock被测试的代码
+- 保持Mock的行为真实
+
+## 9. 常见问题
+
+### 9.1 数据库连接失败
+检查测试数据库配置，确保测试数据库已创建。
+
+### 9.2 测试数据冲突
+使用pytest fixture和scope控制测试数据的生命周期。
+
+### 9.3 异步测试失败
+使用`pytest-asyncio`，确保测试函数使用`async def`定义。
+
+---
+
+**文档维护**: 后端程序员
+**更新时间**: 2026-01-25

backend/tests/__init__.py

@@ -0,0 +1 @@
+# Backend Tests Package

backend/tests/conftest.py

@@ -0,0 +1,165 @@
+import pytest
+import asyncio
+from typing import AsyncGenerator, Generator
+from httpx import AsyncClient
+from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, async_sessionmaker
+from sqlalchemy.pool import StaticPool
+
+from src.main import app
+from src.config.database import Base, get_db
+from src.models.user import User
+from src.utils.password import hash_password
+
+
+# 测试数据库配置
+TEST_DATABASE_URL = "sqlite+aiosqlite:///:memory:"
+
+# 创建测试引擎
+engine = create_async_engine(
+    TEST_DATABASE_URL,
+    connect_args={"check_same_thread": False},
+    poolclass=StaticPool,
+)
+
+# 创建测试会话工厂
+TestingSessionLocal = async_sessionmaker(
+    engine, expire_on_commit=False, class_=AsyncSession
+)
+
+
+@pytest.fixture(scope="function")
+async def db_session() -> AsyncGenerator[AsyncSession, None]:
+    """创建测试数据库会话"""
+    async with engine.begin() as conn:
+        await conn.run_sync(Base.metadata.create_all)
+
+    async with TestingSessionLocal() as session:
+        yield session
+        await session.rollback()
+
+    async with engine.begin() as conn:
+        await conn.run_sync(Base.metadata.drop_all)
+
+
+@pytest.fixture(scope="function")
+async def client(db_session: AsyncSession) -> AsyncGenerator[AsyncClient, None]:
+    """创建测试客户端"""
+
+    async def override_get_db():
+        yield db_session
+
+    app.dependency_overrides[get_db] = override_get_db
+
+    async with AsyncClient(app=app, base_url="http://test") as ac:
+        yield ac
+
+    app.dependency_overrides.clear()
+
+
+@pytest.fixture
+async def admin_user(db_session: AsyncSession) -> User:
+    """创建管理员用户"""
+    user = User(
+        username="admin",
+        password_hash=hash_password("admin123"),
+        real_name="系统管理员",
+        department="管理部",
+        role="admin",
+        email="admin@test.com",
+        phone="13800000001",
+        is_active=True,
+    )
+    db_session.add(user)
+    await db_session.commit()
+    await db_session.refresh(user)
+    return user
+
+
+@pytest.fixture
+async def market_user(db_session: AsyncSession) -> User:
+    """创建市场部用户"""
+    user = User(
+        username="market",
+        password_hash=hash_password("market123"),
+        real_name="市场部用户",
+        department="市场部",
+        role="market",
+        email="market@test.com",
+        phone="13800000002",
+        is_active=True,
+    )
+    db_session.add(user)
+    await db_session.commit()
+    await db_session.refresh(user)
+    return user
+
+
+@pytest.fixture
+async def other_user(db_session: AsyncSession) -> User:
+    """创建其他部门用户"""
+    user = User(
+        username="other",
+        password_hash=hash_password("other123"),
+        real_name="其他部门用户",
+        department="技术部",
+        role="other",
+        email="other@test.com",
+        phone="13800000003",
+        is_active=True,
+    )
+    db_session.add(user)
+    await db_session.commit()
+    await db_session.refresh(user)
+    return user
+
+
+@pytest.fixture
+async def inactive_user(db_session: AsyncSession) -> User:
+    """创建未激活用户"""
+    user = User(
+        username="inactive",
+        password_hash=hash_password("inactive123"),
+        real_name="未激活用户",
+        department="市场部",
+        role="market",
+        email="inactive@test.com",
+        phone="13800000004",
+        is_active=False,
+    )
+    db_session.add(user)
+    await db_session.commit()
+    await db_session.refresh(user)
+    return user
+
+
+@pytest.fixture
+async def admin_token(client: AsyncClient, admin_user: User) -> str:
+    """获取管理员token"""
+    response = await client.post(
+        "/api/v1/auth/login",
+        json={"username": "admin", "password": "admin123"},
+    )
+    data = response.json()
+    return data["data"]["token"]
+
+
+@pytest.fixture
+async def market_token(client: AsyncClient, market_user: User) -> str:
+    """获取市场部用户token"""
+    response = await client.post(
+        "/api/v1/auth/login",
+        json={"username": "market", "password": "market123"},
+    )
+    data = response.json()
+    return data["data"]["token"]
+
+
+@pytest.fixture
+async def other_token(client: AsyncClient, other_user: User) -> str:
+    """获取其他部门用户token"""
+    response = await client.post(
+        "/api/v1/auth/login",
+        json={"username": "other", "password": "other123"},
+    )
+    data = response.json()
+    return data["data"]["token"]

backend/tests/test_auth.py

@@ -0,0 +1,151 @@
+import pytest
+from httpx import AsyncClient
+
+
+@pytest.mark.asyncio
+@pytest.mark.unit
+async def test_login_success(client: AsyncClient, admin_user):
+    """测试用户登录成功"""
+    response = await client.post(
+        "/api/v1/auth/login",
+        json={"username": "admin", "password": "admin123"},
+    )
+
+    assert response.status_code == 200
+    data = response.json()
+    assert data["success"] is True
+    assert data["message"] == "登录成功"
+    assert "token" in data["data"]
+    assert data["data"]["user"]["username"] == "admin"
+    assert data["data"]["user"]["role"] == "admin"
+
+
+@pytest.mark.asyncio
+@pytest.mark.unit
+async def test_login_wrong_username(client: AsyncClient, admin_user):
+    """测试错误的用户名"""
+    response = await client.post(
+        "/api/v1/auth/login",
+        json={"username": "wrong", "password": "admin123"},
+    )
+
+    assert response.status_code == 401
+    data = response.json()
+    assert data["success"] is False
+    assert data["error_code"] == "1002"
+
+
+@pytest.mark.asyncio
+@pytest.mark.unit
+async def test_login_wrong_password(client: AsyncClient, admin_user):
+    """测试错误的密码"""
+    response = await client.post(
+        "/api/v1/auth/login",
+        json={"username": "admin", "password": "wrong"},
+    )
+
+    assert response.status_code == 401
+    data = response.json()
+    assert data["success"] is False
+    assert data["error_code"] == "1002"
+
+
+@pytest.mark.asyncio
+@pytest.mark.unit
+async def test_login_missing_username(client: AsyncClient, admin_user):
+    """测试缺少用户名"""
+    response = await client.post(
+        "/api/v1/auth/login",
+        json={"password": "admin123"},
+    )
+
+    assert response.status_code == 400
+    data = response.json()
+    assert data["success"] is False
+    assert data["error_code"] == "1001"
+
+
+@pytest.mark.asyncio
+@pytest.mark.unit
+async def test_login_missing_password(client: AsyncClient, admin_user):
+    """测试缺少密码"""
+    response = await client.post(
+        "/api/v1/auth/login",
+        json={"username": "admin"},
+    )
+
+    assert response.status_code == 400
+    data = response.json()
+    assert data["success"] is False
+    assert data["error_code"] == "1001"
+
+
+@pytest.mark.asyncio
+@pytest.mark.unit
+async def test_login_inactive_user(client: AsyncClient, inactive_user):
+    """测试未激活用户登录"""
+    response = await client.post(
+        "/api/v1/auth/login",
+        json={"username": "inactive", "password": "inactive123"},
+    )
+
+    assert response.status_code == 401
+    data = response.json()
+    assert data["success"] is False
+
+
+@pytest.mark.asyncio
+@pytest.mark.unit
+async def test_get_current_user_success(client: AsyncClient, admin_token: str):
+    """测试获取当前用户信息成功"""
+    response = await client.get(
+        "/api/v1/auth/me",
+        headers={"Authorization": f"Bearer {admin_token}"},
+    )
+
+    assert response.status_code == 200
+    data = response.json()
+    assert data["success"] is True
+    assert data["data"]["username"] == "admin"
+    assert data["data"]["role"] == "admin"
+
+
+@pytest.mark.asyncio
+@pytest.mark.unit
+async def test_get_current_user_no_token(client: AsyncClient):
+    """测试未提供token获取用户信息"""
+    response = await client.get("/api/v1/auth/me")
+
+    assert response.status_code == 401
+    data = response.json()
+    assert data["success"] is False
+
+
+@pytest.mark.asyncio
+@pytest.mark.unit
+async def test_get_current_user_invalid_token(client: AsyncClient):
+    """测试无效token获取用户信息"""
+    response = await client.get(
+        "/api/v1/auth/me",
+        headers={"Authorization": "Bearer invalid_token"},
+    )
+
+    assert response.status_code == 401
+    data = response.json()
+    assert data["success"] is False
+    assert data["error_code"] == "1003"
+
+
+@pytest.mark.asyncio
+@pytest.mark.unit
+async def test_logout_success(client: AsyncClient, admin_token: str):
+    """测试登出成功"""
+    response = await client.post(
+        "/api/v1/auth/logout",
+        headers={"Authorization": f"Bearer {admin_token}"},
+    )
+
+    assert response.status_code == 200
+    data = response.json()
+    assert data["success"] is True
+    assert data["message"] == "登出成功"

backend/tests/test_projects.py

@@ -0,0 +1,544 @@
+import pytest
+from httpx import AsyncClient
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+async def test_get_projects_all_users_success(
+    client: AsyncClient, admin_token: str, db_session
+):
+    """测试所有用户获取项目列表成功"""
+    # 创建测试项目
+    from src.models.project import Project
+    from datetime import date
+
+    project = Project(
+        project_no="PRJ001",
+        name="测试项目",
+        engineering_type="基建",
+        contract_amount=100.0,
+        signing_date=date(2026, 1, 1),
+        created_by=1,
+    )
+    db_session.add(project)
+    await db_session.commit()
+
+    response = await client.get(
+        "/api/v1/projects",
+        headers={"Authorization": f"Bearer {admin_token}"},
+    )
+
+    assert response.status_code == 200
+    data = response.json()
+    assert data["success"] is True
+    assert "items" in data["data"]
+    assert "total" in data["data"]
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+async def test_get_projects_with_pagination(
+    client: AsyncClient, admin_token: str, db_session
+):
+    """测试项目列表分页"""
+    from src.models.project import Project
+    from datetime import date
+
+    # 创建5个测试项目
+    for i in range(5):
+        project = Project(
+            project_no=f"PRJ{i:03d}",
+            name=f"测试项目{i}",
+            engineering_type="基建",
+            contract_amount=100.0,
+            signing_date=date(2026, 1, 1),
+            created_by=1,
+        )
+        db_session.add(project)
+    await db_session.commit()
+
+    response = await client.get(
+        "/api/v1/projects?page=1&page_size=3",
+        headers={"Authorization": f"Bearer {admin_token}"},
+    )
+
+    assert response.status_code == 200
+    data = response.json()
+    assert data["success"] is True
+    assert data["data"]["page"] == 1
+    assert data["data"]["page_size"] == 3
+    assert len(data["data"]["items"]) == 3
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+async def test_get_projects_with_filters(
+    client: AsyncClient, admin_token: str, db_session
+):
+    """测试项目列表组合筛选"""
+    from src.models.project import Project
+    from datetime import date
+
+    # 创建不同类型的项目
+    project1 = Project(
+        project_no="PRJ001",
+        name="基建项目",
+        engineering_type="基建",
+        contract_amount=100.0,
+        signing_date=date(2026, 1, 1),
+        created_by=1,
+    )
+    project2 = Project(
+        project_no="PRJ002",
+        name="业扩项目",
+        engineering_type="业扩",
+        contract_amount=200.0,
+        signing_date=date(2026, 2, 1),
+        created_by=1,
+    )
+    db_session.add(project1)
+    db_session.add(project2)
+    await db_session.commit()
+
+    # 筛选：工程类别=基建 AND 签订日期>=2026-01-01
+    response = await client.get(
+        "/api/v1/projects?engineering_type=基建&signing_date_start=2026-01-01",
+        headers={"Authorization": f"Bearer {admin_token}"},
+    )
+
+    assert response.status_code == 200
+    data = response.json()
+    assert data["success"] is True
+    assert len(data["data"]["items"]) == 1
+    assert data["data"]["items"][0]["engineering_type"] == "基建"
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+async def test_get_projects_sort_by_contract_amount(
+    client: AsyncClient, admin_token: str, db_session
+):
+    """测试按合同金额排序"""
+    from src.models.project import Project
+    from datetime import date
+
+    # 创建不同金额的项目
+    project1 = Project(
+        project_no="PRJ001",
+        name="项目1",
+        engineering_type="基建",
+        contract_amount=100.0,
+        signing_date=date(2026, 1, 1),
+        created_by=1,
+    )
+    project2 = Project(
+        project_no="PRJ002",
+        name="项目2",
+        engineering_type="基建",
+        contract_amount=200.0,
+        signing_date=date(2026, 1, 1),
+        created_by=1,
+    )
+    db_session.add(project1)
+    db_session.add(project2)
+    await db_session.commit()
+
+    response = await client.get(
+        "/api/v1/projects?sort_by=contract_amount&sort_order=desc",
+        headers={"Authorization": f"Bearer {admin_token}"},
+    )
+
+    assert response.status_code == 200
+    data = response.json()
+    assert data["success"] is True
+    assert data["data"]["items"][0]["contract_amount"] == 200.0
+    assert data["data"]["items"][1]["contract_amount"] == 100.0
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+async def test_create_project_admin_success(
+    client: AsyncClient, admin_token: str, admin_user
+):
+    """测试管理员创建项目成功"""
+    response = await client.post(
+        "/api/v1/projects",
+        headers={"Authorization": f"Bearer {admin_token}"},
+        json={
+            "project_no": "PRJ001",
+            "name": "新项目",
+            "engineering_type": "基建",
+            "contract_amount": 100.0,
+            "signing_date": "2026-01-01",
+        },
+    )
+
+    assert response.status_code == 200
+    data = response.json()
+    assert data["success"] is True
+    assert data["message"] == "项目创建成功"
+    assert data["data"]["project_no"] == "PRJ001"
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+async def test_create_project_market_success(
+    client: AsyncClient, market_token: str, market_user
+):
+    """测试市场部用户创建项目成功"""
+    response = await client.post(
+        "/api/v1/projects",
+        headers={"Authorization": f"Bearer {market_token}"},
+        json={
+            "project_no": "PRJ002",
+            "name": "市场部项目",
+            "engineering_type": "业扩",
+            "contract_amount": 200.0,
+            "signing_date": "2026-01-01",
+        },
+    )
+
+    assert response.status_code == 200
+    data = response.json()
+    assert data["success"] is True
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+async def test_create_project_other_forbidden(
+    client: AsyncClient, other_token: str, other_user
+):
+    """测试其他部门用户创建项目被禁止"""
+    response = await client.post(
+        "/api/v1/projects",
+        headers={"Authorization": f"Bearer {other_token}"},
+        json={
+            "project_no": "PRJ003",
+            "name": "其他部门项目",
+            "engineering_type": "基建",
+            "contract_amount": 100.0,
+        },
+    )
+
+    assert response.status_code == 403
+    data = response.json()
+    assert data["success"] is False
+    assert data["error_code"] == "3001"
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+async def test_create_project_duplicate_project_no(
+    client: AsyncClient, admin_token: str, db_session, admin_user
+):
+    """测试创建重复项目编号"""
+    from src.models.project import Project
+    from datetime import date
+
+    project = Project(
+        project_no="PRJ001",
+        name="已存在项目",
+        engineering_type="基建",
+        contract_amount=100.0,
+        signing_date=date(2026, 1, 1),
+        created_by=admin_user.id,
+    )
+    db_session.add(project)
+    await db_session.commit()
+
+    response = await client.post(
+        "/api/v1/projects",
+        headers={"Authorization": f"Bearer {admin_token}"},
+        json={
+            "project_no": "PRJ001",
+            "name": "新项目",
+            "engineering_type": "基建",
+            "contract_amount": 100.0,
+        },
+    )
+
+    assert response.status_code == 409
+    data = response.json()
+    assert data["success"] is False
+    assert data["error_code"] == "2002"
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+async def test_create_project_missing_required_fields(
+    client: AsyncClient, admin_token: str
+):
+    """测试创建项目时缺少必填字段"""
+    response = await client.post(
+        "/api/v1/projects",
+        headers={"Authorization": f"Bearer {admin_token}"},
+        json={
+            "project_no": "PRJ001",
+            # 缺少 name
+            "engineering_type": "基建",
+            "contract_amount": 100.0,
+        },
+    )
+
+    assert response.status_code == 400
+    data = response.json()
+    assert data["success"] is False
+    assert data["error_code"] == "1001"
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+async def test_update_project_admin_success(
+    client: AsyncClient, admin_token: str, db_session, admin_user
+):
+    """测试管理员更新项目成功"""
+    from src.models.project import Project
+    from datetime import date
+
+    project = Project(
+        project_no="PRJ001",
+        name="原项目名称",
+        engineering_type="基建",
+        contract_amount=100.0,
+        signing_date=date(2026, 1, 1),
+        created_by=admin_user.id,
+    )
+    db_session.add(project)
+    await db_session.commit()
+    await db_session.refresh(project)
+
+    response = await client.put(
+        f"/api/v1/projects/{project.id}",
+        headers={"Authorization": f"Bearer {admin_token}"},
+        json={"name": "更新后的项目名称", "contract_amount": 200.0},
+    )
+
+    assert response.status_code == 200
+    data = response.json()
+    assert data["success"] is True
+    assert data["data"]["name"] == "更新后的项目名称"
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+async def test_update_project_market_own_project(
+    client: AsyncClient, market_token: str, db_session, market_user
+):
+    """测试市场部用户更新自己的项目"""
+    from src.models.project import Project
+    from datetime import date
+
+    project = Project(
+        project_no="PRJ001",
+        name="市场部项目",
+        engineering_type="业扩",
+        contract_amount=100.0,
+        signing_date=date(2026, 1, 1),
+        created_by=market_user.id,
+    )
+    db_session.add(project)
+    await db_session.commit()
+    await db_session.refresh(project)
+
+    response = await client.put(
+        f"/api/v1/projects/{project.id}",
+        headers={"Authorization": f"Bearer {market_token}"},
+        json={"name": "更新后的项目名称"},
+    )
+
+    assert response.status_code == 200
+    data = response.json()
+    assert data["success"] is True
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+async def test_update_project_market_other_project(
+    client: AsyncClient, market_token: str, db_session, admin_user
+):
+    """测试市场部用户更新其他人的项目被禁止"""
+    from src.models.project import Project
+    from datetime import date
+
+    project = Project(
+        project_no="PRJ001",
+        name="管理员项目",
+        engineering_type="基建",
+        contract_amount=100.0,
+        signing_date=date(2026, 1, 1),
+        created_by=admin_user.id,
+    )
+    db_session.add(project)
+    await db_session.commit()
+    await db_session.refresh(project)
+
+    response = await client.put(
+        f"/api/v1/projects/{project.id}",
+        headers={"Authorization": f"Bearer {market_token}"},
+        json={"name": "更新后的项目名称"},
+    )
+
+    assert response.status_code == 403
+    data = response.json()
+    assert data["success"] is False
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+async def test_update_project_not_found(
+    client: AsyncClient, admin_token: str
+):
+    """测试更新不存在的项目"""
+    response = await client.put(
+        "/api/v1/projects/99999",
+        headers={"Authorization": f"Bearer {admin_token}"},
+        json={"name": "更新后的项目名称"},
+    )
+
+    assert response.status_code == 404
+    data = response.json()
+    assert data["success"] is False
+    assert data["error_code"] == "2001"
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+async def test_delete_project_admin_success(
+    client: AsyncClient, admin_token: str, db_session, admin_user
+):
+    """测试管理员删除项目成功"""
+    from src.models.project import Project
+    from datetime import date
+
+    project = Project(
+        project_no="PRJ001",
+        name="待删除项目",
+        engineering_type="基建",
+        contract_amount=100.0,
+        signing_date=date(2026, 1, 1),
+        created_by=admin_user.id,
+    )
+    db_session.add(project)
+    await db_session.commit()
+    await db_session.refresh(project)
+
+    response = await client.delete(
+        f"/api/v1/projects/{project.id}",
+        headers={"Authorization": f"Bearer {admin_token}"},
+    )
+
+    assert response.status_code == 200
+    data = response.json()
+    assert data["success"] is True
+    assert data["message"] == "项目删除成功"
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+async def test_delete_project_market_own_project(
+    client: AsyncClient, market_token: str, db_session, market_user
+):
+    """测试市场部用户删除自己的项目成功"""
+    from src.models.project import Project
+    from datetime import date
+
+    project = Project(
+        project_no="PRJ001",
+        name="市场部项目",
+        engineering_type="业扩",
+        contract_amount=100.0,
+        signing_date=date(2026, 1, 1),
+        created_by=market_user.id,
+    )
+    db_session.add(project)
+    await db_session.commit()
+    await db_session.refresh(project)
+
+    response = await client.delete(
+        f"/api/v1/projects/{project.id}",
+        headers={"Authorization": f"Bearer {market_token}"},
+    )
+
+    assert response.status_code == 200
+    data = response.json()
+    assert data["success"] is True
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+async def test_delete_project_market_other_project_forbidden(
+    client: AsyncClient, market_token: str, db_session, admin_user
+):
+    """测试市场部用户删除其他人的项目被禁止"""
+    from src.models.project import Project
+    from datetime import date
+
+    project = Project(
+        project_no="PRJ001",
+        name="管理员项目",
+        engineering_type="基建",
+        contract_amount=100.0,
+        signing_date=date(2026, 1, 1),
+        created_by=admin_user.id,
+    )
+    db_session.add(project)
+    await db_session.commit()
+    await db_session.refresh(project)
+
+    response = await client.delete(
+        f"/api/v1/projects/{project.id}",
+        headers={"Authorization": f"Bearer {market_token}"},
+    )
+
+    assert response.status_code == 403
+    data = response.json()
+    assert data["success"] is False
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+async def test_delete_project_not_found(
+    client: AsyncClient, admin_token: str
+):
+    """测试删除不存在的项目"""
+    response = await client.delete(
+        "/api/v1/projects/99999",
+        headers={"Authorization": f"Bearer {admin_token}"},
+    )
+
+    assert response.status_code == 404
+    data = response.json()
+    assert data["success"] is False
+    assert data["error_code"] == "2001"
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+async def test_get_project_detail_success(
+    client: AsyncClient, admin_token: str, db_session, admin_user
+):
+    """测试获取项目详情成功"""
+    from src.models.project import Project
+    from datetime import date
+
+    project = Project(
+        project_no="PRJ001",
+        name="测试项目",
+        engineering_type="基建",
+        contract_amount=100.0,
+        signing_date=date(2026, 1, 1),
+        created_by=admin_user.id,
+    )
+    db_session.add(project)
+    await db_session.commit()
+    await db_session.refresh(project)
+
+    response = await client.get(
+        f"/api/v1/projects/{project.id}",
+        headers={"Authorization": f"Bearer {admin_token}"},
+    )
+
+    assert response.status_code == 200
+    data = response.json()
+    assert data["success"] is True
+    assert data["data"]["project_no"] == "PRJ001"

backend/tests/test_statistics.py

@@ -0,0 +1,397 @@
+import pytest
+from httpx import AsyncClient
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+async def test_get_statistics_all_users_success(
+    client: AsyncClient, admin_token: str, db_session, admin_user
+):
+    """测试所有用户获取基础统计成功"""
+    from src.models.project import Project
+    from datetime import date
+
+    # 创建测试项目
+    project1 = Project(
+        project_no="PRJ001",
+        name="项目1",
+        engineering_type="基建",
+        contract_amount=100.0,
+        actual_receipt_amount=50.0,
+        actual_payment_amount=40.0,
+        cumulative_progress=50.0,
+        signing_date=date(2026, 1, 1),
+        created_by=admin_user.id,
+    )
+    project2 = Project(
+        project_no="PRJ002",
+        name="项目2",
+        engineering_type="业扩",
+        contract_amount=200.0,
+        actual_receipt_amount=100.0,
+        actual_payment_amount=80.0,
+        cumulative_progress=50.0,
+        signing_date=date(2026, 1, 1),
+        created_by=admin_user.id,
+    )
+    db_session.add(project1)
+    db_session.add(project2)
+    await db_session.commit()
+
+    response = await client.get(
+        "/api/v1/projects/statistics",
+        headers={"Authorization": f"Bearer {admin_token}"},
+    )
+
+    assert response.status_code == 200
+    data = response.json()
+    assert data["success"] is True
+    assert data["data"]["total_count"] == 2
+    assert data["data"]["total_contract_amount"] == 300.0
+    assert data["data"]["total_receipt_amount"] == 150.0
+    assert data["data"]["total_payment_amount"] == 120.0
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+async def test_get_statistics_with_filters(
+    client: AsyncClient, admin_token: str, db_session, admin_user
+):
+    """测试带筛选条件的统计"""
+    from src.models.project import Project
+    from datetime import date
+
+    # 创建不同类型的项目
+    project1 = Project(
+        project_no="PRJ001",
+        name="基建项目",
+        engineering_type="基建",
+        contract_amount=100.0,
+        signing_date=date(2026, 1, 1),
+        created_by=admin_user.id,
+    )
+    project2 = Project(
+        project_no="PRJ002",
+        name="业扩项目",
+        engineering_type="业扩",
+        contract_amount=200.0,
+        signing_date=date(2026, 2, 1),
+        created_by=admin_user.id,
+    )
+    db_session.add(project1)
+    db_session.add(project2)
+    await db_session.commit()
+
+    # 筛选：工程类别=基建
+    response = await client.get(
+        "/api/v1/projects/statistics?engineering_type=基建",
+        headers={"Authorization": f"Bearer {admin_token}"},
+    )
+
+    assert response.status_code == 200
+    data = response.json()
+    assert data["success"] is True
+    assert data["data"]["total_count"] == 1
+    assert data["data"]["total_contract_amount"] == 100.0
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+async def test_get_statistics_empty_data(
+    client: AsyncClient, admin_token: str
+):
+    """测试无数据时的统计"""
+    response = await client.get(
+        "/api/v1/projects/statistics",
+        headers={"Authorization": f"Bearer {admin_token}"},
+    )
+
+    assert response.status_code == 200
+    data = response.json()
+    assert data["success"] is True
+    assert data["data"]["total_count"] == 0
+    assert data["data"]["total_contract_amount"] == 0
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+async def test_get_group_statistics_by_engineering_type(
+    client: AsyncClient, admin_token: str, db_session, admin_user
+):
+    """测试按工程类别分组统计"""
+    from src.models.project import Project
+    from datetime import date
+
+    # 创建不同类型的项目
+    for i in range(3):
+        project = Project(
+            project_no=f"PRJ{i:03d}",
+            name=f"基建项目{i}",
+            engineering_type="基建",
+            contract_amount=100.0,
+            actual_receipt_amount=80.0,
+            actual_payment_amount=70.0,
+            signing_date=date(2026, 1, 1),
+            created_by=admin_user.id,
+        )
+        db_session.add(project)
+
+    for i in range(2):
+        project = Project(
+            project_no=f"PRJ{i+3:03d}",
+            name=f"业扩项目{i}",
+            engineering_type="业扩",
+            contract_amount=200.0,
+            actual_receipt_amount=150.0,
+            actual_payment_amount=130.0,
+            signing_date=date(2026, 1, 1),
+            created_by=admin_user.id,
+        )
+        db_session.add(project)
+
+    await db_session.commit()
+
+    response = await client.get(
+        "/api/v1/projects/statistics/group?group_by=engineering_type",
+        headers={"Authorization": f"Bearer {admin_token}"},
+    )
+
+    assert response.status_code == 200
+    data = response.json()
+    assert data["success"] is True
+    assert len(data["data"]) == 2
+
+    # 验证基建工程统计
+    infra = next(item for item in data["data"] if item["engineering_type"] == "基建")
+    assert infra["count"] == 3
+    assert infra["total_contract_amount"] == 300.0
+
+    # 验证业扩工程统计
+    market = next(item for item in data["data"] if item["engineering_type"] == "业扩")
+    assert market["count"] == 2
+    assert market["total_contract_amount"] == 400.0
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+async def test_get_group_statistics_by_department(
+    client: AsyncClient, admin_token: str, db_session, admin_user
+):
+    """测试按项目部分组统计"""
+    from src.models.project import Project
+    from datetime import date
+
+    # 创建不同项目部的项目
+    project1 = Project(
+        project_no="PRJ001",
+        name="项目部一项目",
+        engineering_type="基建",
+        contract_amount=100.0,
+        project_department="项目部一",
+        signing_date=date(2026, 1, 1),
+        created_by=admin_user.id,
+    )
+    project2 = Project(
+        project_no="PRJ002",
+        name="项目部二项目",
+        engineering_type="业扩",
+        contract_amount=200.0,
+        project_department="项目部二",
+        signing_date=date(2026, 1, 1),
+        created_by=admin_user.id,
+    )
+    db_session.add(project1)
+    db_session.add(project2)
+    await db_session.commit()
+
+    response = await client.get(
+        "/api/v1/projects/statistics/group?group_by=project_department",
+        headers={"Authorization": f"Bearer {admin_token}"},
+    )
+
+    assert response.status_code == 200
+    data = response.json()
+    assert data["success"] is True
+    assert len(data["data"]) == 2
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+async def test_get_group_statistics_with_date_filter(
+    client: AsyncClient, admin_token: str, db_session, admin_user
+):
+    """测试带日期筛选的分组统计"""
+    from src.models.project import Project
+    from datetime import date
+
+    # 创建不同日期的项目
+    project1 = Project(
+        project_no="PRJ001",
+        name="一月项目",
+        engineering_type="基建",
+        contract_amount=100.0,
+        signing_date=date(2026, 1, 15),
+        created_by=admin_user.id,
+    )
+    project2 = Project(
+        project_no="PRJ002",
+        name="二月项目",
+        engineering_type="业扩",
+        contract_amount=200.0,
+        signing_date=date(2026, 2, 15),
+        created_by=admin_user.id,
+    )
+    db_session.add(project1)
+    db_session.add(project2)
+    await db_session.commit()
+
+    # 只统计1月的项目
+    response = await client.get(
+        "/api/v1/projects/statistics/group?group_by=engineering_type&signing_date_start=2026-01-01&signing_date_end=2026-01-31",
+        headers={"Authorization": f"Bearer {admin_token}"},
+    )
+
+    assert response.status_code == 200
+    data = response.json()
+    assert data["success"] is True
+    assert len(data["data"]) == 1
+    assert data["data"][0]["count"] == 1
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+async def test_get_timeline_statistics_by_month(
+    client: AsyncClient, admin_token: str, db_session, admin_user
+):
+    """测试按月统计"""
+    from src.models.project import Project
+    from datetime import date
+
+    # 创建不同月份的项目
+    for month in [1, 1, 2, 2, 2]:
+        project = Project(
+            project_no=f"PRJ{month:02d}",
+            name=f"{month}月项目",
+            engineering_type="基建",
+            contract_amount=100.0,
+            signing_date=date(2026, month, 15),
+            created_by=admin_user.id,
+        )
+        db_session.add(project)
+
+    await db_session.commit()
+
+    response = await client.get(
+        "/api/v1/projects/statistics/timeline?time_field=signing_date&group_by=month",
+        headers={"Authorization": f"Bearer {admin_token}"},
+    )
+
+    assert response.status_code == 200
+    data = response.json()
+    assert data["success"] is True
+    assert len(data["data"]) == 2
+
+    # 验证1月统计
+    jan = next(item for item in data["data"] if item["month"] == "2026-01")
+    assert jan["count"] == 2
+
+    # 验证2月统计
+    feb = next(item for item in data["data"] if item["month"] == "2026-02")
+    assert feb["count"] == 3
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+async def test_get_timeline_statistics_by_year(
+    client: AsyncClient, admin_token: str, db_session, admin_user
+):
+    """测试按年统计"""
+    from src.models.project import Project
+    from datetime import date
+
+    # 创建不同年份的项目
+    for year in [2025, 2026, 2026]:
+        project = Project(
+            project_no=f"PRJ{year}",
+            name=f"{year}年项目",
+            engineering_type="基建",
+            contract_amount=100.0,
+            signing_date=date(year, 1, 1),
+            created_by=admin_user.id,
+        )
+        db_session.add(project)
+
+    await db_session.commit()
+
+    response = await client.get(
+        "/api/v1/projects/statistics/timeline?time_field=signing_date&group_by=year",
+        headers={"Authorization": f"Bearer {admin_token}"},
+    )
+
+    assert response.status_code == 200
+    data = response.json()
+    assert data["success"] is True
+    assert len(data["data"]) == 2
+
+    # 验证2025年统计
+    y2025 = next(item for item in data["data"] if item["year"] == "2025")
+    assert y2025["count"] == 1
+
+    # 验证2026年统计
+    y2026 = next(item for item in data["data"] if item["year"] == "2026")
+    assert y2026["count"] == 2
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+async def test_get_timeline_statistics_with_filter(
+    client: AsyncClient, admin_token: str, db_session, admin_user
+):
+    """测试带筛选条件的时间统计"""
+    from src.models.project import Project
+    from datetime import date
+
+    # 创建不同类型和日期的项目
+    project1 = Project(
+        project_no="PRJ001",
+        name="基建一月",
+        engineering_type="基建",
+        contract_amount=100.0,
+        signing_date=date(2026, 1, 1),
+        created_by=admin_user.id,
+    )
+    project2 = Project(
+        project_no="PRJ002",
+        name="业扩一月",
+        engineering_type="业扩",
+        contract_amount=200.0,
+        signing_date=date(2026, 1, 1),
+        created_by=admin_user.id,
+    )
+    project3 = Project(
+        project_no="PRJ003",
+        name="业扩二月",
+        engineering_type="业扩",
+        contract_amount=300.0,
+        signing_date=date(2026, 2, 1),
+        created_by=admin_user.id,
+    )
+    db_session.add(project1)
+    db_session.add(project2)
+    db_session.add(project3)
+    await db_session.commit()
+
+    # 只统计业扩项目
+    response = await client.get(
+        "/api/v1/projects/statistics/timeline?time_field=signing_date&group_by=month&engineering_type=业扩",
+        headers={"Authorization": f"Bearer {admin_token}"},
+    )
+
+    assert response.status_code == 200
+    data = response.json()
+    assert data["success"] is True
+    assert len(data["data"]) == 2
+
+    # 验证1月业扩项目统计
+    jan = next(item for item in data["data"] if item["month"] == "2026-01")
+    assert jan["count"] == 1

backend/tests/test_users.py

@@ -0,0 +1,332 @@
+import pytest
+from httpx import AsyncClient
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+async def test_create_user_admin_success(
+    client: AsyncClient, admin_token: str
+):
+    """测试管理员创建用户成功"""
+    response = await client.post(
+        "/api/v1/users",
+        headers={"Authorization": f"Bearer {admin_token}"},
+        json={
+            "username": "newuser",
+            "password": "newpass123",
+            "real_name": "新用户",
+            "department": "技术部",
+            "role": "other",
+            "email": "newuser@test.com",
+            "phone": "13800000005",
+        },
+    )
+
+    assert response.status_code == 200
+    data = response.json()
+    assert data["success"] is True
+    assert data["message"] == "用户创建成功"
+    assert data["data"]["username"] == "newuser"
+    assert data["data"]["role"] == "other"
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+async def test_create_user_market_forbidden(
+    client: AsyncClient, market_token: str
+):
+    """测试市场部用户创建用户被禁止"""
+    response = await client.post(
+        "/api/v1/users",
+        headers={"Authorization": f"Bearer {market_token}"},
+        json={
+            "username": "newuser",
+            "password": "newpass123",
+            "real_name": "新用户",
+            "department": "技术部",
+            "role": "other",
+        },
+    )
+
+    assert response.status_code == 403
+    data = response.json()
+    assert data["success"] is False
+    assert data["error_code"] == "3001"
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+async def test_create_user_duplicate_username(
+    client: AsyncClient, admin_token: str, admin_user
+):
+    """测试创建重复用户名"""
+    response = await client.post(
+        "/api/v1/users",
+        headers={"Authorization": f"Bearer {admin_token}"},
+        json={
+            "username": "admin",
+            "password": "newpass123",
+            "real_name": "重复用户",
+            "department": "技术部",
+            "role": "other",
+        },
+    )
+
+    assert response.status_code == 409
+    data = response.json()
+    assert data["success"] is False
+    assert data["error_code"] == "2002"
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+async def test_create_user_invalid_role(
+    client: AsyncClient, admin_token: str
+):
+    """测试创建用户时使用无效角色"""
+    response = await client.post(
+        "/api/v1/users",
+        headers={"Authorization": f"Bearer {admin_token}"},
+        json={
+            "username": "newuser",
+            "password": "newpass123",
+            "real_name": "新用户",
+            "department": "技术部",
+            "role": "invalid",
+        },
+    )
+
+    assert response.status_code == 400
+    data = response.json()
+    assert data["success"] is False
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+async def test_get_users_admin_success(
+    client: AsyncClient, admin_token: str
+):
+    """测试管理员获取用户列表成功"""
+    response = await client.get(
+        "/api/v1/users",
+        headers={"Authorization": f"Bearer {admin_token}"},
+    )
+
+    assert response.status_code == 200
+    data = response.json()
+    assert data["success"] is True
+    assert "items" in data["data"]
+    assert "total" in data["data"]
+    assert len(data["data"]["items"]) >= 1
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+async def test_get_users_market_forbidden(
+    client: AsyncClient, market_token: str
+):
+    """测试市场部用户获取用户列表被禁止"""
+    response = await client.get(
+        "/api/v1/users",
+        headers={"Authorization": f"Bearer {market_token}"},
+    )
+
+    assert response.status_code == 403
+    data = response.json()
+    assert data["success"] is False
+    assert data["error_code"] == "3001"
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+async def test_get_users_with_pagination(
+    client: AsyncClient, admin_token: str, admin_user, market_user, other_user
+):
+    """测试用户列表分页"""
+    response = await client.get(
+        "/api/v1/users?page=1&page_size=2",
+        headers={"Authorization": f"Bearer {admin_token}"},
+    )
+
+    assert response.status_code == 200
+    data = response.json()
+    assert data["success"] is True
+    assert data["data"]["page"] == 1
+    assert data["data"]["page_size"] == 2
+    assert len(data["data"]["items"]) == 2
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+async def test_get_users_with_filter(
+    client: AsyncClient, admin_token: str, admin_user, market_user, other_user
+):
+    """测试用户列表筛选"""
+    response = await client.get(
+        "/api/v1/users?department=市场部",
+        headers={"Authorization": f"Bearer {admin_token}"},
+    )
+
+    assert response.status_code == 200
+    data = response.json()
+    assert data["success"] is True
+    for user in data["data"]["items"]:
+        assert user["department"] == "市场部"
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+async def test_update_user_admin_success(
+    client: AsyncClient, admin_token: str, market_user
+):
+    """测试管理员更新用户成功"""
+    response = await client.put(
+        f"/api/v1/users/{market_user.id}",
+        headers={"Authorization": f"Bearer {admin_token}"},
+        json={
+            "real_name": "更新后的姓名",
+            "email": "updated@test.com",
+        },
+    )
+
+    assert response.status_code == 200
+    data = response.json()
+    assert data["success"] is True
+    assert data["data"]["real_name"] == "更新后的姓名"
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+async def test_update_user_market_forbidden(
+    client: AsyncClient, market_token: str, other_user
+):
+    """测试市场部用户更新用户被禁止"""
+    response = await client.put(
+        f"/api/v1/users/{other_user.id}",
+        headers={"Authorization": f"Bearer {market_token}"},
+        json={"real_name": "更新后的姓名"},
+    )
+
+    assert response.status_code == 403
+    data = response.json()
+    assert data["success"] is False
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+async def test_update_user_not_found(
+    client: AsyncClient, admin_token: str
+):
+    """测试更新不存在的用户"""
+    response = await client.put(
+        "/api/v1/users/99999",
+        headers={"Authorization": f"Bearer {admin_token}"},
+        json={"real_name": "更新后的姓名"},
+    )
+
+    assert response.status_code == 404
+    data = response.json()
+    assert data["success"] is False
+    assert data["error_code"] == "2001"
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+async def test_delete_user_admin_success(
+    client: AsyncClient, admin_token: str, other_user
+):
+    """测试管理员删除用户成功"""
+    response = await client.delete(
+        f"/api/v1/users/{other_user.id}",
+        headers={"Authorization": f"Bearer {admin_token}"},
+    )
+
+    assert response.status_code == 200
+    data = response.json()
+    assert data["success"] is True
+    assert data["message"] == "用户删除成功"
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+async def test_delete_user_market_forbidden(
+    client: AsyncClient, market_token: str, other_user
+):
+    """测试市场部用户删除用户被禁止"""
+    response = await client.delete(
+        f"/api/v1/users/{other_user.id}",
+        headers={"Authorization": f"Bearer {market_token}"},
+    )
+
+    assert response.status_code == 403
+    data = response.json()
+    assert data["success"] is False
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+async def test_delete_user_not_found(
+    client: AsyncClient, admin_token: str
+):
+    """测试删除不存在的用户"""
+    response = await client.delete(
+        "/api/v1/users/99999",
+        headers={"Authorization": f"Bearer {admin_token}"},
+    )
+
+    assert response.status_code == 404
+    data = response.json()
+    assert data["success"] is False
+    assert data["error_code"] == "2001"
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+async def test_reset_password_admin_success(
+    client: AsyncClient, admin_token: str, other_user
+):
+    """测试管理员重置密码成功"""
+    response = await client.post(
+        f"/api/v1/users/{other_user.id}/reset-password",
+        headers={"Authorization": f"Bearer {admin_token}"},
+        json={"new_password": "newpassword123"},
+    )
+
+    assert response.status_code == 200
+    data = response.json()
+    assert data["success"] is True
+    assert data["message"] == "密码重置成功"
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+async def test_reset_password_market_forbidden(
+    client: AsyncClient, market_token: str, other_user
+):
+    """测试市场部用户重置密码被禁止"""
+    response = await client.post(
+        f"/api/v1/users/{other_user.id}/reset-password",
+        headers={"Authorization": f"Bearer {market_token}"},
+        json={"new_password": "newpassword123"},
+    )
+
+    assert response.status_code == 403
+    data = response.json()
+    assert data["success"] is False
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+async def test_reset_password_invalid_password(
+    client: AsyncClient, admin_token: str, other_user
+):
+    """测试重置密码时使用无效密码格式"""
+    response = await client.post(
+        f"/api/v1/users/{other_user.id}/reset-password",
+        headers={"Authorization": f"Bearer {admin_token}"},
+        json={"new_password": "123"},
+    )
+
+    assert response.status_code == 400
+    data = response.json()
+    assert data["success"] is False