ocean_project_manager/backend/tests/test_users.py

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 == 201
    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