ForcePilot/backend/test/integration/api/external_systems/conftest.py

"""Shared pytest fixtures for HTTP integration tests of the external_systems routers.

Provides function-scoped fixtures that create (and clean up) minimal external
systems, tools, and environments via the live API under
``/api/system/external-systems``. Parent conftest supplies ``test_client``,
``admin_headers`` and ``standard_user``; do not redefine them here.
"""

from __future__ import annotations

import os
import uuid
from collections.abc import AsyncGenerator

import httpx
import pytest_asyncio
from sqlalchemy import text
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
from sqlalchemy.pool import NullPool
from yuxi.storage.postgres.models_external import ExternalSystemMetric
from yuxi.utils.datetime_utils import utc_now_naive

BASE_URL = "/api/system/external-systems"


@pytest_asyncio.fixture(scope="function")
async def external_system(
    test_client: httpx.AsyncClient,
    admin_headers: dict[str, str],
) -> AsyncGenerator[dict, None]:
    suffix = uuid.uuid4().hex[:8]
    slug = f"pytest_sys_{suffix}"
    name = f"Pytest System {suffix}"

    response = await test_client.post(
        f"{BASE_URL}/systems",
        json={
            "slug": slug,
            "name": name,
            "adapter_type": "http",
            "auth_type": "none",
            "enabled": True,
        },
        headers=admin_headers,
    )
    assert response.status_code == 200, f"Failed to create external system: {response.text}"
    system = response.json()["data"]

    try:
        yield system
    finally:
        try:
            delete_response = await test_client.delete(
                f"{BASE_URL}/systems/{system['id']}",
                headers=admin_headers,
            )
            if delete_response.status_code not in (200, 404):
                print(f"Warning: Failed to cleanup external system {system['id']}: {delete_response.text}")
        except Exception as exc:
            print(f"Warning: Exception during cleanup of external system {system['id']}: {exc}")


@pytest_asyncio.fixture(scope="function")
async def external_tool(
    test_client: httpx.AsyncClient,
    admin_headers: dict[str, str],
    external_system: dict,
) -> AsyncGenerator[dict, None]:
    suffix = uuid.uuid4().hex[:8]
    slug = f"pytest_tool_{suffix}"
    name = f"Pytest Tool {suffix}"

    response = await test_client.post(
        f"{BASE_URL}/tools",
        json={
            "slug": slug,
            "name": name,
            "adapter_type": "http",
            "auth_type": "none",
            "enabled": True,
            "system_id": external_system["id"],
        },
        headers=admin_headers,
    )
    assert response.status_code == 200, f"Failed to create external tool: {response.text}"
    tool = response.json()["data"]

    try:
        yield tool
    finally:
        try:
            delete_response = await test_client.delete(
                f"{BASE_URL}/tools/{tool['id']}",
                headers=admin_headers,
            )
            if delete_response.status_code not in (200, 404):
                print(f"Warning: Failed to cleanup external tool {tool['id']}: {delete_response.text}")
        except Exception as exc:
            print(f"Warning: Exception during cleanup of external tool {tool['id']}: {exc}")


@pytest_asyncio.fixture(scope="function")
async def external_environment(
    test_client: httpx.AsyncClient,
    admin_headers: dict[str, str],
    external_system: dict,
) -> AsyncGenerator[dict, None]:
    suffix = uuid.uuid4().hex[:8]
    env_key = f"env_{suffix}"
    name = f"Pytest Env {suffix}"

    response = await test_client.post(
        f"{BASE_URL}/environments",
        json={
            "system_id": external_system["id"],
            "env_key": env_key,
            "name": name,
            "enabled": True,
        },
        headers=admin_headers,
    )
    assert response.status_code == 200, f"Failed to create external environment: {response.text}"
    environment = response.json()["data"]

    try:
        yield environment
    finally:
        try:
            delete_response = await test_client.delete(
                f"{BASE_URL}/environments/{environment['id']}",
                params={"system_id": external_system["id"]},
                headers=admin_headers,
            )
            if delete_response.status_code not in (200, 404):
                print(f"Warning: Failed to cleanup external environment {environment['id']}: {delete_response.text}")
        except Exception as exc:
            print(f"Warning: Exception during cleanup of external environment {environment['id']}: {exc}")


@pytest_asyncio.fixture(scope="function")
async def db_engine():
    """提供独立于 pg_manager 的 async engine，避免跨 event loop 连接池问题。

    pg_manager 的连接池在 ``ensure_live_api_schema`` fixture 中通过
    ``anyio.run()`` 创建，绑定到临时 event loop。function-scope 的测试用例
    运行在 pytest-asyncio 的 event loop 中，复用 pg_manager 会触发
    "got Future attached to a different loop"。使用 ``NullPool`` 每次新建
    连接，彻底规避连接复用跨 loop 问题。
    """
    db_url = os.getenv("POSTGRES_URL", "postgresql+asyncpg://postgres:postgres@postgres:5432/yuxi")
    engine = create_async_engine(db_url, poolclass=NullPool)
    try:
        yield engine
    finally:
        await engine.dispose()


@pytest_asyncio.fixture(scope="function")
async def metric_bucket(
    external_system: dict,
    external_environment: dict,
    db_engine,
) -> AsyncGenerator[dict, None]:
    """创建一条指标桶数据，用于 by-system/by-env 聚合查询测试。

    依赖 ``external_environment`` 以使 ``env_key`` 与环境记录一致，
    从而验证 ``by-env`` 端点的 LEFT JOIN 能返回 ``env_name``。
    """
    async_session = async_sessionmaker(db_engine, expire_on_commit=False)

    metric = ExternalSystemMetric(
        system_id=external_system["id"],
        env_key=external_environment["env_key"],
        bucket_at=utc_now_naive(),
        total_calls=100,
        success_calls=90,
        failed_calls=10,
        latency_sum_ms=5000,
        latency_count=100,
    )
    async with async_session() as db:
        db.add(metric)
        await db.commit()
        await db.refresh(metric)

    try:
        yield {
            "id": metric.id,
            "system_id": external_system["id"],
            "env_key": external_environment["env_key"],
            "system_name": external_system["name"],
            "env_name": external_environment["name"],
        }
    finally:
        try:
            async with async_session() as db:
                await db.execute(
                    text("DELETE FROM ext_system_metrics WHERE id = :id"),
                    {"id": metric.id},
                )
                await db.commit()
        except Exception as exc:
            print(f"Warning: Exception during cleanup of metric bucket {metric.id}: {exc}")