后端开发完成

2026-01-26 18:21:34 +08:00
parent 40d2f3f6ac
commit b3876bba89
64 changed files with 1736 additions and 0 deletions
@@ -0,0 +1,115 @@
+# 后端测试文档
+
+## 测试结构
+
+本测试目录包含项目管理系统后端的单元测试和集成测试，主要测试以下模块：
+
+- **认证模块**：测试用户登录、token生成等功能
+- **用户管理模块**：测试用户的创建、查询、更新、删除等功能
+- **项目管理模块**：测试项目的创建、查询、更新、删除等功能
+
+## 测试环境
+
+- **测试框架**：Pytest
+- **测试客户端**：FastAPI TestClient
+- **测试数据库**：SQLite（内存数据库）
+- **测试用户**：
+  - 管理员：test_admin / test_password
+  - 市场部：test_marketing / test_password
+  - 其他部门：test_other / test_password
+
+## 测试文件结构
+
+```
+tests/
+├── conftest.py          # 测试配置和 fixtures
+├── test_auth.py         # 认证模块测试
+├── test_user.py         # 用户管理模块测试
+├── test_project.py      # 项目管理模块测试
+└── README.md            # 测试文档
+```
+
+## 运行测试
+
+### 安装依赖
+
+```bash
+# 安装测试依赖
+pip install pytest pytest-cov
+```
+
+### 运行所有测试
+
+```bash
+# 在backend目录下运行
+pytest
+
+# 运行测试并生成覆盖率报告
+pytest --cov=app
+
+# 运行测试并生成HTML格式的覆盖率报告
+pytest --cov=app --cov-report=html
+```
+
+### 运行特定模块的测试
+
+```bash
+# 运行认证模块测试
+pytest tests/test_auth.py
+
+# 运行用户管理模块测试
+pytest tests/test_user.py
+
+# 运行项目管理模块测试
+pytest tests/test_project.py
+```
+
+## 测试用例说明
+
+### 认证模块测试
+- 测试用户登录功能
+- 测试token生成和验证
+- 测试错误登录场景
+
+### 用户管理模块测试
+- 测试管理员获取用户列表
+- 测试管理员创建用户
+- 测试管理员更新用户
+- 测试管理员删除用户
+- 测试权限控制
+
+### 项目管理模块测试
+- 测试市场部创建项目
+- 测试获取项目列表
+- 测试更新项目
+- 测试删除项目
+- 测试项目历史记录
+- 测试权限控制
+
+## 测试覆盖范围
+
+- API路由测试
+- 业务逻辑测试
+- 数据验证测试
+- 权限控制测试
+- 错误处理测试
+
+## 注意事项
+
+1. 测试使用SQLite内存数据库，不会影响生产数据库
+2. 测试会自动创建和清理测试数据
+3. 测试运行时会覆盖数据库依赖，使用测试数据库
+4. 测试完成后会自动删除测试数据库
+
+## 测试结果分析
+
+测试运行完成后，会显示测试结果和覆盖率报告。通过分析测试结果，可以了解：
+
+- 哪些功能通过了测试
+- 哪些功能需要改进
+- 代码覆盖率情况
+- 潜在的问题和bug
+
+## 持续集成
+
+测试可以集成到CI/CD流程中，每次代码提交时自动运行测试，确保代码质量。
@@ -0,0 +1,109 @@
+import pytest
+import os
+from fastapi.testclient import TestClient
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
+from app.main import app as fastapi_app
+from app.database.database import Base, get_db
+from app.config import settings
+from app.models.user import User
+
+# 设置测试环境变量
+os.environ["TESTING"] = "True"
+
+# 创建测试数据库引擎
+SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"
+engine = create_engine(
+    SQLALCHEMY_DATABASE_URL,
+    connect_args={"check_same_thread": False}
+)
+
+# 创建测试会话工厂
+TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
+
+
+from app.common.utils import get_password_hash
+
+
+@pytest.fixture(scope="module")
+def test_db():
+    """创建测试数据库"""
+    # 创建表
+    Base.metadata.create_all(bind=engine)
+    # 创建会话
+    db = TestingSessionLocal()
+    try:
+        # 添加测试用户
+        test_users = [
+            User(
+                username="test_admin",
+                password="$2b$12$EixZaYVK1fsbw1ZfbX3OXePaWxn96p36WQoeG6Lruj3vjPGga31lW",  # test_password
+                department="IT",
+                role="admin"
+            ),
+            User(
+                username="test_marketing",
+                password="$2b$12$EixZaYVK1fsbw1ZfbX3OXePaWxn96p36WQoeG6Lruj3vjPGga31lW",  # test_password
+                department="Marketing",
+                role="marketing"
+            ),
+            User(
+                username="test_other",
+                password="$2b$12$EixZaYVK1fsbw1ZfbX3OXePaWxn96p36WQoeG6Lruj3vjPGga31lW",  # test_password
+                department="Other",
+                role="other"
+            )
+        ]
+        db.add_all(test_users)
+        db.commit()
+        yield db
+    finally:
+        db.close()
+        # 删除表
+        Base.metadata.drop_all(bind=engine)
+
+
+@pytest.fixture(scope="module")
+def client(test_db):
+    """创建测试客户端"""
+    def override_get_db():
+        try:
+            yield test_db
+        finally:
+            pass
+    
+    # 覆盖依赖
+    fastapi_app.dependency_overrides[get_db] = override_get_db
+    
+    with TestClient(fastapi_app) as c:
+        yield c
+
+
+@pytest.fixture(scope="module")
+def admin_token(client):
+    """获取管理员token"""
+    response = client.post(
+        "/api/auth/login",
+        data={"username": "test_admin", "password": "test_password"}
+    )
+    return response.json()["access_token"]
+
+
+@pytest.fixture(scope="module")
+def marketing_token(client):
+    """获取市场部token"""
+    response = client.post(
+        "/api/auth/login",
+        data={"username": "test_marketing", "password": "test_password"}
+    )
+    return response.json()["access_token"]
+
+
+@pytest.fixture(scope="module")
+def other_token(client):
+    """获取其他部门token"""
+    response = client.post(
+        "/api/auth/login",
+        data={"username": "test_other", "password": "test_password"}
+    )
+    return response.json()["access_token"]
@@ -0,0 +1,47 @@
+def test_login_success(client):
+    """测试登录成功"""
+    response = client.post(
+        "/api/auth/login",
+        data={"username": "test_admin", "password": "test_password"}
+    )
+    assert response.status_code == 200
+    assert "access_token" in response.json()
+    assert response.json()["token_type"] == "bearer"
+
+
+def test_login_invalid_username(client):
+    """测试登录失败 - 用户名无效"""
+    response = client.post(
+        "/api/auth/login",
+        data={"username": "invalid_user", "password": "test_password"}
+    )
+    assert response.status_code == 401
+    assert "Incorrect username or password" in response.json()["detail"]
+
+
+def test_login_invalid_password(client):
+    """测试登录失败 - 密码无效"""
+    response = client.post(
+        "/api/auth/login",
+        data={"username": "test_admin", "password": "invalid_password"}
+    )
+    assert response.status_code == 401
+    assert "Incorrect username or password" in response.json()["detail"]
+
+
+def test_login_empty_username(client):
+    """测试登录失败 - 用户名为空"""
+    response = client.post(
+        "/api/auth/login",
+        data={"username": "", "password": "test_password"}
+    )
+    assert response.status_code == 422
+
+
+def test_login_empty_password(client):
+    """测试登录失败 - 密码为空"""
+    response = client.post(
+        "/api/auth/login",
+        data={"username": "test_admin", "password": ""}
+    )
+    assert response.status_code == 422
@@ -0,0 +1,186 @@
+def test_create_project_marketing(client, marketing_token):
+    """测试市场部创建项目"""
+    new_project = {
+        "name": "Test Project",
+        "description": "Test Project Description",
+        "department": "Marketing",
+        "budget": 10000,
+        "status": "pending"
+    }
+    response = client.post(
+        "/api/projects",
+        json=new_project,
+        headers={"Authorization": f"Bearer {marketing_token}"}
+    )
+    assert response.status_code == 200
+    assert response.json()["name"] == "Test Project"
+    assert response.json()["description"] == "Test Project Description"
+    assert response.json()["department"] == "Marketing"
+
+
+def test_create_project_non_marketing(client, other_token):
+    """测试非市场部创建项目（权限不足）"""
+    new_project = {
+        "name": "Test Project",
+        "description": "Test Project Description",
+        "department": "Other",
+        "budget": 10000,
+        "status": "pending"
+    }
+    response = client.post(
+        "/api/projects",
+        json=new_project,
+        headers={"Authorization": f"Bearer {other_token}"}
+    )
+    assert response.status_code == 403
+    assert "Only marketing department can create projects" in response.json()["detail"]
+
+
+def test_get_projects(client, admin_token):
+    """测试获取项目列表"""
+    # 先创建一个项目
+    new_project = {
+        "name": "Test Project",
+        "description": "Test Project Description",
+        "department": "Marketing",
+        "budget": 10000,
+        "status": "pending"
+    }
+    client.post(
+        "/api/projects",
+        json=new_project,
+        headers={"Authorization": f"Bearer {admin_token}"}
+    )
+    
+    # 获取项目列表
+    response = client.get(
+        "/api/projects",
+        headers={"Authorization": f"Bearer {admin_token}"}
+    )
+    assert response.status_code == 200
+    assert len(response.json()) >= 1
+
+
+def test_update_project(client, marketing_token, admin_token):
+    """测试更新项目"""
+    # 先创建一个项目
+    new_project = {
+        "name": "Test Project",
+        "description": "Test Project Description",
+        "department": "Marketing",
+        "budget": 10000,
+        "status": "pending"
+    }
+    create_response = client.post(
+        "/api/projects",
+        json=new_project,
+        headers={"Authorization": f"Bearer {marketing_token}"}
+    )
+    project_id = create_response.json()["id"]
+    
+    # 更新项目
+    update_data = {
+        "name": "Updated Test Project",
+        "description": "Updated Test Project Description",
+        "budget": 15000,
+        "status": "in_progress"
+    }
+    response = client.put(
+        f"/api/projects/{project_id}",
+        json=update_data,
+        headers={"Authorization": f"Bearer {admin_token}"}
+    )
+    assert response.status_code == 200
+    assert response.json()["name"] == "Updated Test Project"
+    assert response.json()["description"] == "Updated Test Project Description"
+    assert response.json()["budget"] == 15000
+    assert response.json()["status"] == "in_progress"
+
+
+def test_delete_project_admin(client, marketing_token, admin_token):
+    """测试管理员删除项目"""
+    # 先创建一个项目
+    new_project = {
+        "name": "Delete Test Project",
+        "description": "Delete Test Project Description",
+        "department": "Marketing",
+        "budget": 10000,
+        "status": "pending"
+    }
+    create_response = client.post(
+        "/api/projects",
+        json=new_project,
+        headers={"Authorization": f"Bearer {marketing_token}"}
+    )
+    project_id = create_response.json()["id"]
+    
+    # 删除项目
+    response = client.delete(
+        f"/api/projects/{project_id}",
+        headers={"Authorization": f"Bearer {admin_token}"}
+    )
+    assert response.status_code == 200
+    assert "Project deleted successfully" in response.json()["message"]
+
+
+def test_delete_project_non_admin(client, marketing_token):
+    """测试非管理员删除项目（权限不足）"""
+    # 先创建一个项目
+    new_project = {
+        "name": "Delete Test Project",
+        "description": "Delete Test Project Description",
+        "department": "Marketing",
+        "budget": 10000,
+        "status": "pending"
+    }
+    create_response = client.post(
+        "/api/projects",
+        json=new_project,
+        headers={"Authorization": f"Bearer {marketing_token}"}
+    )
+    project_id = create_response.json()["id"]
+    
+    # 尝试删除项目
+    response = client.delete(
+        f"/api/projects/{project_id}",
+        headers={"Authorization": f"Bearer {marketing_token}"}
+    )
+    assert response.status_code == 403
+    assert "Not enough permissions" in response.json()["detail"]
+
+
+def test_project_history(client, marketing_token, admin_token):
+    """测试项目历史记录"""
+    # 先创建一个项目
+    new_project = {
+        "name": "History Test Project",
+        "description": "History Test Project Description",
+        "department": "Marketing",
+        "budget": 10000,
+        "status": "pending"
+    }
+    create_response = client.post(
+        "/api/projects",
+        json=new_project,
+        headers={"Authorization": f"Bearer {marketing_token}"}
+    )
+    project_id = create_response.json()["id"]
+    
+    # 更新项目
+    update_data = {
+        "name": "Updated History Test Project",
+        "budget": 15000
+    }
+    client.put(
+        f"/api/projects/{project_id}",
+        json=update_data,
+        headers={"Authorization": f"Bearer {admin_token}"}
+    )
+    
+    # 获取项目历史记录
+    response = client.get(
+        f"/api/projects/{project_id}/history",
+        headers={"Authorization": f"Bearer {admin_token}"}
+    )
+    assert response.status_code == 200
+    assert len(response.json()) >= 2  # 至少有两条历史记录（创建和更新）
@@ -0,0 +1,156 @@
+def test_get_users_admin(client, admin_token):
+    """测试管理员获取用户列表"""
+    response = client.get(
+        "/api/users",
+        headers={"Authorization": f"Bearer {admin_token}"}
+    )
+    assert response.status_code == 200
+    assert len(response.json()) >= 3
+
+
+def test_get_users_non_admin(client, marketing_token):
+    """测试非管理员获取用户列表（权限不足）"""
+    response = client.get(
+        "/api/users",
+        headers={"Authorization": f"Bearer {marketing_token}"}
+    )
+    assert response.status_code == 403
+    assert "Not enough permissions" in response.json()["detail"]
+
+
+def test_create_user_admin(client, admin_token):
+    """测试管理员创建用户"""
+    new_user = {
+        "username": "new_test_user",
+        "password": "new_password",
+        "department": "Test Department",
+        "role": "other"
+    }
+    response = client.post(
+        "/api/users",
+        json=new_user,
+        headers={"Authorization": f"Bearer {admin_token}"}
+    )
+    assert response.status_code == 200
+    assert response.json()["username"] == "new_test_user"
+    assert response.json()["department"] == "Test Department"
+    assert response.json()["role"] == "other"
+
+
+def test_create_user_non_admin(client, marketing_token):
+    """测试非管理员创建用户（权限不足）"""
+    new_user = {
+        "username": "new_test_user",
+        "password": "new_password",
+        "department": "Test Department",
+        "role": "other"
+    }
+    response = client.post(
+        "/api/users",
+        json=new_user,
+        headers={"Authorization": f"Bearer {marketing_token}"}
+    )
+    assert response.status_code == 403
+    assert "Not enough permissions" in response.json()["detail"]
+
+
+def test_create_user_existing_username(client, admin_token):
+    """测试创建用户 - 用户名已存在"""
+    new_user = {
+        "username": "test_admin",
+        "password": "new_password",
+        "department": "Test Department",
+        "role": "other"
+    }
+    response = client.post(
+        "/api/users",
+        json=new_user,
+        headers={"Authorization": f"Bearer {admin_token}"}
+    )
+    assert response.status_code == 400
+    assert "Username already registered" in response.json()["detail"]
+
+
+def test_update_user_admin(client, admin_token):
+    """测试管理员更新用户"""
+    # 先获取用户ID，选择一个非管理员用户
+    response = client.get(
+        "/api/users",
+        headers={"Authorization": f"Bearer {admin_token}"}
+    )
+    users = response.json()
+    
+    # 找到一个非管理员用户
+    user_id = None
+    for user in users:
+        if user["role"] != "admin":
+            user_id = user["id"]
+            break
+    
+    assert user_id is not None, "No non-admin user found for update"
+    
+    # 更新用户
+    update_data = {
+        "password": "updated_password",
+        "department": "Updated Department",
+        "role": "other"
+    }
+    response = client.put(
+        f"/api/users/{user_id}",
+        json=update_data,
+        headers={"Authorization": f"Bearer {admin_token}"}
+    )
+    assert response.status_code == 200
+    assert response.json()["department"] == "Updated Department"
+    assert response.json()["role"] == "other"
+
+
+def test_update_user_non_admin(client, marketing_token):
+    """测试非管理员更新用户（权限不足）"""
+    # 先获取用户ID
+    response = client.get(
+        "/api/users",
+        headers={"Authorization": f"Bearer {marketing_token}"}
+    )
+    assert response.status_code == 403
+
+
+def test_delete_user_admin(client, admin_token):
+    """测试管理员删除用户"""
+    # 先创建一个新的测试用户
+    new_user = {
+        "username": "delete_test_user",
+        "password": "delete_password",
+        "department": "Test Department",
+        "role": "other"
+    }
+    create_response = client.post(
+        "/api/users",
+        json=new_user,
+        headers={"Authorization": f"Bearer {admin_token}"}
+    )
+    
+    # 确保创建用户成功
+    assert create_response.status_code == 200
+    user_id = create_response.json()["id"]
+    
+    # 删除用户
+    response = client.delete(
+        f"/api/users/{user_id}",
+        headers={"Authorization": f"Bearer {admin_token}"}
+    )
+    
+    # 确保删除用户成功
+    assert response.status_code == 200
+    assert "User deleted successfully" in response.json()["message"]
+
+
+def test_delete_user_non_admin(client, marketing_token):
+    """测试非管理员删除用户（权限不足）"""
+    # 假设用户ID为1
+    response = client.delete(
+        "/api/users/1",
+        headers={"Authorization": f"Bearer {marketing_token}"}
+    )
+    assert response.status_code == 403
+    assert "Not enough permissions" in response.json()["detail"]