"""Schema 漂移检测 + 连接池管理 API 测试。""" import pytest pytestmark = pytest.mark.skip(reason="Go backend endpoint not implemented") from unittest.mock import patch, AsyncMock, MagicMock from httpx import AsyncClient from tests.conftest import auth_header # ════════════════════════════════════════════════════════════════════ # Schema Drift Tests # ════════════════════════════════════════════════════════════════════ class TestSchemaDriftCapture: """捕获 Schema 快照。""" @pytest.mark.asyncio async def test_capture_success(self, client, admin_token, test_instance): with patch("app.api.v1.endpoints.schema_drift.SchemaDriftDetector") as MockDet: instance_mock = MagicMock() instance_mock.capture_snapshot = AsyncMock(return_value=[{"col": "id"}]) instance_mock.save_baseline = AsyncMock() MockDet.return_value = instance_mock resp = await client.post( f"/api/v1/schema-drift/instances/{test_instance.id}/capture", headers=auth_header(admin_token), ) assert resp.status_code == 200 body = resp.json() assert body["message"] == "Schema 快照已捕获" assert body["total_columns"] == 1 @pytest.mark.asyncio async def test_capture_instance_not_found(self, client, admin_token): resp = await client.post( "/api/v1/schema-drift/instances/9999/capture", headers=auth_header(admin_token), ) assert resp.status_code == 404 @pytest.mark.asyncio async def test_capture_developer_forbidden(self, client, developer_token, test_instance): resp = await client.post( f"/api/v1/schema-drift/instances/{test_instance.id}/capture", headers=auth_header(developer_token), ) assert resp.status_code == 403 @pytest.mark.asyncio async def test_capture_dba_allowed(self, client, dba_token, test_instance): with patch("app.api.v1.endpoints.schema_drift.SchemaDriftDetector") as MockDet: instance_mock = MagicMock() instance_mock.capture_snapshot = AsyncMock(return_value=[]) instance_mock.save_baseline = AsyncMock() MockDet.return_value = instance_mock resp = await client.post( f"/api/v1/schema-drift/instances/{test_instance.id}/capture", headers=auth_header(dba_token), ) assert resp.status_code == 200 class TestSchemaDriftCheck: """检测 Schema 漂移。""" @pytest.mark.asyncio async def test_check_success(self, client, admin_token, test_instance): with patch("app.api.v1.endpoints.schema_drift.SchemaDriftDetector") as MockDet: report = MagicMock() report.to_dict.return_value = {"drifted": False, "changes": []} instance_mock = MagicMock() instance_mock.detect_drift = AsyncMock(return_value=report) MockDet.return_value = instance_mock resp = await client.post( f"/api/v1/schema-drift/instances/{test_instance.id}/check", headers=auth_header(admin_token), ) assert resp.status_code == 200 body = resp.json() assert body["drifted"] is False @pytest.mark.asyncio async def test_check_instance_not_found(self, client, admin_token): resp = await client.post( "/api/v1/schema-drift/instances/9999/check", headers=auth_header(admin_token), ) assert resp.status_code == 404 class TestSchemaHistory: @pytest.mark.asyncio async def test_history_returns_data(self, client, admin_token, test_instance): resp = await client.get( f"/api/v1/schema-drift/instances/{test_instance.id}/history", headers=auth_header(admin_token), ) assert resp.status_code == 200 body = resp.json() assert body["instance_id"] == test_instance.id assert "history" in body # ════════════════════════════════════════════════════════════════════ # Connection Pool Tests # ════════════════════════════════════════════════════════════════════ class TestPoolStats: @pytest.mark.asyncio async def test_stats_not_initialized(self, client, admin_token, test_instance): with patch("app.api.v1.endpoints.pool.get_pool_manager") as mock_pm: manager = MagicMock() manager.get_pool_stats.return_value = None mock_pm.return_value = manager resp = await client.get( f"/api/v1/pools/instances/{test_instance.id}/stats", headers=auth_header(admin_token), ) assert resp.status_code == 200 body = resp.json() assert body["status"] == "not_initialized" @pytest.mark.asyncio async def test_stats_with_data(self, client, admin_token, test_instance): with patch("app.api.v1.endpoints.pool.get_pool_manager") as mock_pm: stats = MagicMock() stats.to_dict.return_value = {"active": 2, "idle": 3} manager = MagicMock() manager.get_pool_stats.return_value = stats mock_pm.return_value = manager resp = await client.get( f"/api/v1/pools/instances/{test_instance.id}/stats", headers=auth_header(admin_token), ) assert resp.status_code == 200 assert resp.json()["active"] == 2 @pytest.mark.asyncio async def test_stats_instance_not_found(self, client, admin_token): resp = await client.get( "/api/v1/pools/instances/9999/stats", headers=auth_header(admin_token), ) assert resp.status_code == 404 @pytest.mark.asyncio async def test_stats_developer_forbidden(self, client, developer_token, 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 class TestPoolReset: @pytest.mark.asyncio async def test_reset_success(self, client, admin_token, test_instance): with patch("app.api.v1.endpoints.pool.get_pool_manager") as mock_pm: manager = MagicMock() manager.reset_pool = AsyncMock() mock_pm.return_value = manager resp = await client.post( f"/api/v1/pools/instances/{test_instance.id}/reset", headers=auth_header(admin_token), ) assert resp.status_code == 200 assert "已重置" in resp.json()["message"] @pytest.mark.asyncio async def test_reset_dba_allowed(self, client, dba_token, test_instance): with patch("app.api.v1.endpoints.pool.get_pool_manager") as mock_pm: manager = MagicMock() manager.reset_pool = AsyncMock() mock_pm.return_value = manager resp = await client.post( f"/api/v1/pools/instances/{test_instance.id}/reset", headers=auth_header(dba_token), ) assert resp.status_code == 200 @pytest.mark.asyncio async def test_reset_developer_forbidden(self, client, developer_token, 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 class TestPoolsOverview: @pytest.mark.asyncio async def test_overview_empty(self, client, admin_token): with patch("app.api.v1.endpoints.pool.get_pool_manager") as mock_pm: manager = MagicMock() manager.get_all_stats.return_value = {} mock_pm.return_value = manager resp = await client.get( "/api/v1/pools/overview", headers=auth_header(admin_token), ) assert resp.status_code == 200 assert resp.json()["total_pools"] == 0 @pytest.mark.asyncio async def test_overview_with_pools(self, client, admin_token): with patch("app.api.v1.endpoints.pool.get_pool_manager") as mock_pm: stats = MagicMock() stats.to_dict.return_value = {"instance_id": 1, "active": 1} manager = MagicMock() manager.get_all_stats.return_value = {1: stats} mock_pm.return_value = manager resp = await client.get( "/api/v1/pools/overview", headers=auth_header(admin_token), ) assert resp.status_code == 200 body = resp.json() assert body["total_pools"] == 1 assert len(body["pools"]) == 1 @pytest.mark.asyncio async def test_overview_guest_forbidden(self, client, guest_token): resp = await client.get( "/api/v1/pools/overview", headers=auth_header(guest_token), ) assert resp.status_code == 403