LabelingSystem/backend/test/test_user_api.py

"""
User API tests.
Tests for user management endpoints.
"""
import pytest
import uuid
from fastapi.testclient import TestClient
from main import app
from database import get_db_connection, init_database
from services.jwt_service import JWTService
import bcrypt


# 测试客户端
client = TestClient(app)


@pytest.fixture(scope="module")
def setup_database():
    """初始化测试数据库"""
    init_database()
    yield


@pytest.fixture
def admin_token(setup_database):
    """创建管理员用户并返回 token"""
    admin_id = f"admin_{uuid.uuid4().hex[:8]}"
    password_hash = bcrypt.hashpw("admin123".encode(), bcrypt.gensalt()).decode()
    
    with get_db_connection() as conn:
        cursor = conn.cursor()
        # 检查是否已存在
        cursor.execute("SELECT id FROM users WHERE username = ?", (f"test_admin_{admin_id}",))
        if not cursor.fetchone():
            cursor.execute("""
                INSERT INTO users (id, username, email, password_hash, role)
                VALUES (?, ?, ?, ?, 'admin')
            """, (admin_id, f"test_admin_{admin_id}", f"admin_{admin_id}@test.com", password_hash))
    
    user_data = {
        "id": admin_id,
        "username": f"test_admin_{admin_id}",
        "email": f"admin_{admin_id}@test.com",
        "role": "admin"
    }
    token = JWTService.create_access_token(user_data)
    
    yield {"token": token, "user_id": admin_id}
    
    # 清理
    with get_db_connection() as conn:
        cursor = conn.cursor()
        cursor.execute("DELETE FROM users WHERE id = ?", (admin_id,))


@pytest.fixture
def annotator_token(setup_database):
    """创建标注人员用户并返回 token"""
    annotator_id = f"annotator_{uuid.uuid4().hex[:8]}"
    password_hash = bcrypt.hashpw("annotator123".encode(), bcrypt.gensalt()).decode()
    
    with get_db_connection() as conn:
        cursor = conn.cursor()
        cursor.execute("""
            INSERT INTO users (id, username, email, password_hash, role)
            VALUES (?, ?, ?, ?, 'annotator')
        """, (annotator_id, f"test_annotator_{annotator_id}", f"annotator_{annotator_id}@test.com", password_hash))
    
    user_data = {
        "id": annotator_id,
        "username": f"test_annotator_{annotator_id}",
        "email": f"annotator_{annotator_id}@test.com",
        "role": "annotator"
    }
    token = JWTService.create_access_token(user_data)
    
    yield {"token": token, "user_id": annotator_id}
    
    # 清理
    with get_db_connection() as conn:
        cursor = conn.cursor()
        cursor.execute("DELETE FROM users WHERE id = ?", (annotator_id,))


class TestUserListAPI:
    """用户列表 API 测试"""
    
    def test_list_users_without_auth(self, setup_database):
        """未认证时应返回 401"""
        response = client.get("/api/users")
        assert response.status_code == 401
    
    def test_list_users_as_annotator(self, annotator_token):
        """标注人员访问应返回 403"""
        headers = {"Authorization": f"Bearer {annotator_token['token']}"}
        response = client.get("/api/users", headers=headers)
        assert response.status_code == 403
    
    def test_list_users_as_admin(self, admin_token):
        """管理员可以获取用户列表"""
        headers = {"Authorization": f"Bearer {admin_token['token']}"}
        response = client.get("/api/users", headers=headers)
        assert response.status_code == 200
        data = response.json()
        assert "users" in data
        assert "total" in data
        assert isinstance(data["users"], list)
    
    def test_list_users_with_role_filter(self, admin_token):
        """管理员可以按角色筛选用户"""
        headers = {"Authorization": f"Bearer {admin_token['token']}"}
        response = client.get("/api/users?role=admin", headers=headers)
        assert response.status_code == 200
        data = response.json()
        # 所有返回的用户都应该是 admin 角色
        for user in data["users"]:
            assert user["role"] == "admin"
    
    def test_list_users_with_search(self, admin_token):
        """管理员可以搜索用户"""
        headers = {"Authorization": f"Bearer {admin_token['token']}"}
        response = client.get("/api/users?search=test_admin", headers=headers)
        assert response.status_code == 200
        data = response.json()
        # 搜索结果应该包含匹配的用户
        assert isinstance(data["users"], list)


class TestAnnotatorsAPI:
    """标注人员列表 API 测试"""
    
    def test_list_annotators_without_auth(self, setup_database):
        """未认证时应返回 401"""
        response = client.get("/api/users/annotators")
        assert response.status_code == 401
    
    def test_list_annotators_as_admin(self, admin_token, annotator_token):
        """管理员可以获取标注人员列表"""
        headers = {"Authorization": f"Bearer {admin_token['token']}"}
        response = client.get("/api/users/annotators", headers=headers)
        assert response.status_code == 200
        data = response.json()
        assert isinstance(data, list)
        # 检查返回的用户包含必要字段
        if len(data) > 0:
            assert "id" in data[0]
            assert "username" in data[0]
            assert "email" in data[0]
            assert "current_task_count" in data[0]
            assert "completed_task_count" in data[0]
    
    def test_list_annotators_as_annotator(self, annotator_token):
        """标注人员也可以获取标注人员列表（用于查看同事）"""
        headers = {"Authorization": f"Bearer {annotator_token['token']}"}
        response = client.get("/api/users/annotators", headers=headers)
        assert response.status_code == 200


class TestUserStatsAPI:
    """用户统计 API 测试"""
    
    def test_get_user_stats_without_auth(self, setup_database):
        """未认证时应返回 401"""
        response = client.get("/api/users/some_user_id/stats")
        assert response.status_code == 401
    
    def test_get_user_stats_as_annotator(self, annotator_token):
        """标注人员访问应返回 403"""
        headers = {"Authorization": f"Bearer {annotator_token['token']}"}
        response = client.get(f"/api/users/{annotator_token['user_id']}/stats", headers=headers)
        assert response.status_code == 403
    
    def test_get_user_stats_as_admin(self, admin_token, annotator_token):
        """管理员可以获取用户统计"""
        headers = {"Authorization": f"Bearer {admin_token['token']}"}
        response = client.get(f"/api/users/{annotator_token['user_id']}/stats", headers=headers)
        assert response.status_code == 200
        data = response.json()
        assert "user" in data
        assert "task_stats" in data
        assert "recent_tasks" in data
        # 检查统计字段
        stats = data["task_stats"]
        assert "assigned_count" in stats
        assert "completed_count" in stats
        assert "completion_rate" in stats
    
    def test_get_nonexistent_user_stats(self, admin_token):
        """获取不存在用户的统计应返回 404"""
        headers = {"Authorization": f"Bearer {admin_token['token']}"}
        response = client.get("/api/users/nonexistent_user_id/stats", headers=headers)
        assert response.status_code == 404


class TestUserDetailAPI:
    """用户详情 API 测试"""
    
    def test_get_user_without_auth(self, setup_database):
        """未认证时应返回 401"""
        response = client.get("/api/users/some_user_id")
        assert response.status_code == 401
    
    def test_get_user_as_admin(self, admin_token, annotator_token):
        """管理员可以获取用户详情"""
        headers = {"Authorization": f"Bearer {admin_token['token']}"}
        response = client.get(f"/api/users/{annotator_token['user_id']}", headers=headers)
        assert response.status_code == 200
        data = response.json()
        assert data["id"] == annotator_token["user_id"]
        assert "task_stats" in data
    
    def test_get_nonexistent_user(self, admin_token):
        """获取不存在的用户应返回 404"""
        headers = {"Authorization": f"Bearer {admin_token['token']}"}
        response = client.get("/api/users/nonexistent_user_id", headers=headers)
        assert response.status_code == 404