opsxsql/tests/test_pool_endpoint.py

"""连接池管理 API 端点测试。"""
import pytest
from unittest.mock import patch, MagicMock, AsyncMock
from httpx import AsyncClient

from app.models.user import User
from app.models.instance import Instance
from app.core.security import hash_password, create_access_token

pytestmark = pytest.mark.skip(reason="Skip: awaiting Go backend alignment")


def auth_header(token: str) -> dict:
    return {"Authorization": f"Bearer {token}"}


@pytest.mark.asyncio
class TestPoolStats:
    """GET /api/v1/pools/instances/{instance_id}/stats"""

    async def test_pool_stats_initialized(
        self, client: AsyncClient, admin_token: str, test_instance
    ):
        mock_stats = MagicMock()
        mock_stats.to_dict.return_value = {
            "instance_id": test_instance.id,
            "status": "active",
            "total_connections": 5,
            "active_connections": 2,
            "idle_connections": 3,
        }
        with patch(
            "app.api.v1.endpoints.pool.get_pool_manager"
        ) as mock_mgr:
            mock_mgr.return_value.get_pool_stats.return_value = mock_stats
            resp = await client.get(
                f"/api/v1/pools/instances/{test_instance.id}/stats",
                headers=auth_header(admin_token),
            )
        assert resp.status_code == 200
        data = resp.json()
        assert data["status"] == "active"
        assert data["total_connections"] == 5

    async def test_pool_stats_not_initialized(
        self, client: AsyncClient, admin_token: str, test_instance
    ):
        with patch(
            "app.api.v1.endpoints.pool.get_pool_manager"
        ) as mock_mgr:
            mock_mgr.return_value.get_pool_stats.return_value = None
            resp = await client.get(
                f"/api/v1/pools/instances/{test_instance.id}/stats",
                headers=auth_header(admin_token),
            )
        assert resp.status_code == 200
        data = resp.json()
        assert data["status"] == "not_initialized"

    async def test_pool_stats_instance_not_found(
        self, client: AsyncClient, admin_token: str
    ):
        resp = await client.get(
            "/api/v1/pools/instances/9999/stats",
            headers=auth_header(admin_token),
        )
        assert resp.status_code == 404

    async def test_pool_stats_forbidden_developer(
        self, client: AsyncClient, developer_token: str, test_instance
    ):
        resp = await client.get(
            f"/api/v1/pools/instances/{test_instance.id}/stats",
            headers=auth_header(developer_token),
        )
        assert resp.status_code == 403

    async def test_pool_stats_dba_allowed(
        self, client: AsyncClient, dba_token: str, test_instance
    ):
        with patch(
            "app.api.v1.endpoints.pool.get_pool_manager"
        ) as mock_mgr:
            mock_mgr.return_value.get_pool_stats.return_value = None
            resp = await client.get(
                f"/api/v1/pools/instances/{test_instance.id}/stats",
                headers=auth_header(dba_token),
            )
        assert resp.status_code == 200


@pytest.mark.asyncio
class TestResetPool:
    """POST /api/v1/pools/instances/{instance_id}/reset"""

    async def test_reset_pool_success(
        self, client: AsyncClient, admin_token: str, test_instance
    ):
        with patch(
            "app.api.v1.endpoints.pool.get_pool_manager"
        ) as mock_mgr:
            mock_mgr.return_value.reset_pool = AsyncMock()
            resp = await client.post(
                f"/api/v1/pools/instances/{test_instance.id}/reset",
                headers=auth_header(admin_token),
            )
        assert resp.status_code == 200
        data = resp.json()
        assert "已重置" in data["message"]
        assert data["instance_id"] == test_instance.id

    async def test_reset_pool_dba_allowed(
        self, client: AsyncClient, dba_token: str, test_instance
    ):
        with patch(
            "app.api.v1.endpoints.pool.get_pool_manager"
        ) as mock_mgr:
            mock_mgr.return_value.reset_pool = AsyncMock()
            resp = await client.post(
                f"/api/v1/pools/instances/{test_instance.id}/reset",
                headers=auth_header(dba_token),
            )
        assert resp.status_code == 200

    async def test_reset_pool_forbidden_developer(
        self, client: AsyncClient, developer_token: str, test_instance
    ):
        resp = await client.post(
            f"/api/v1/pools/instances/{test_instance.id}/reset",
            headers=auth_header(developer_token),
        )
        assert resp.status_code == 403


@pytest.mark.asyncio
class TestPoolsOverview:
    """GET /api/v1/pools/overview"""

    async def test_pools_overview_empty(
        self, client: AsyncClient, admin_token: str
    ):
        with patch(
            "app.api.v1.endpoints.pool.get_pool_manager"
        ) as mock_mgr:
            mock_mgr.return_value.get_all_stats.return_value = {}
            resp = await client.get(
                "/api/v1/pools/overview",
                headers=auth_header(admin_token),
            )
        assert resp.status_code == 200
        data = resp.json()
        assert data["total_pools"] == 0
        assert data["pools"] == []

    async def test_pools_overview_with_pools(
        self, client: AsyncClient, admin_token: str
    ):
        mock_stats = MagicMock()
        mock_stats.to_dict.return_value = {
            "instance_id": 1,
            "status": "active",
            "total_connections": 10,
        }
        with patch(
            "app.api.v1.endpoints.pool.get_pool_manager"
        ) as mock_mgr:
            mock_mgr.return_value.get_all_stats.return_value = {1: mock_stats}
            resp = await client.get(
                "/api/v1/pools/overview",
                headers=auth_header(admin_token),
            )
        assert resp.status_code == 200
        data = resp.json()
        assert data["total_pools"] == 1
        assert len(data["pools"]) == 1

    async def test_pools_overview_unauthorized(self, client: AsyncClient):
        resp = await client.get("/api/v1/pools/overview")
        assert resp.status_code == 401

    async def test_pools_overview_forbidden_guest(
        self, client: AsyncClient, guest_token: str
    ):
        resp = await client.get(
            "/api/v1/pools/overview",
            headers=auth_header(guest_token),
        )
        assert resp.status_code == 403