ocean_project_manager/backend/tests/test_projects.py

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 == 201
    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 == 201
    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"] is None

@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 == 422
    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"] is None


@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"] is None

@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"