diff --git a/backend/tests/test_instance_permission.py b/backend/tests/test_instance_permission.py index 22e0344..0175f57 100644 --- a/backend/tests/test_instance_permission.py +++ b/backend/tests/test_instance_permission.py @@ -9,8 +9,11 @@ - 集成测试: query/ticket 端点权限检查 """ import json +import sys import pytest +from contextlib import asynccontextmanager from fastapi import HTTPException +from unittest.mock import patch, AsyncMock, MagicMock from app.models.instance import Instance from app.models.instance_group import InstanceGroup, user_groups @@ -243,7 +246,7 @@ class TestPermissionEndpointIntegration: async def test_query_with_permission( self, client, db, developer_user, group_with_permissions ): - """有 query:select 权限 → 查询不被 403""" + """有 query:select 权限 → 查询不被 403（mock DB连接避免挂起）""" group, instance = group_with_permissions await db.execute( user_groups.insert().values( @@ -254,12 +257,38 @@ class TestPermissionEndpointIntegration: token = create_access_token( data={"sub": str(developer_user.id), "role": developer_user.role} ) - resp = await client.post( - "/api/v1/query", - json={"instance_id": instance.id, "database": "test", "sql": "SELECT 1"}, - headers={"Authorization": f"Bearer {token}"}, - ) - # 不应是 403（可能 400/500 因为连不上真DB，但权限通过） + + # Mock aiomysql 防止真实连接挂起（见 skill #20, #37） + mock_cursor = AsyncMock() + mock_cursor.description = [("1",)] + mock_cursor.fetchall.return_value = [{"1": 1}] + mock_cursor.close = AsyncMock() + + @asynccontextmanager + async def mock_cursor_ctx(*a, **kw): + yield mock_cursor + + mock_conn = AsyncMock() + mock_conn.cursor = mock_cursor_ctx + mock_conn.commit = AsyncMock() + mock_conn.rollback = AsyncMock() + mock_conn.close = AsyncMock() + + mock_aiomysql = MagicMock() + mock_aiomysql.connect = AsyncMock(return_value=mock_conn) + mock_aiomysql.DictCursor = MagicMock() + + with patch.dict(sys.modules, {"aiomysql": mock_aiomysql}): + resp = await client.post( + "/api/v1/query", + json={ + "instance_id": instance.id, + "database": "test", + "sql": "SELECT 1", + }, + headers={"Authorization": f"Bearer {token}"}, + ) + # 不应是 403（权限通过，mock 返回成功） assert resp.status_code != 403 @pytest.mark.asyncio diff --git a/backend/tests/test_pool.py b/backend/tests/test_pool.py new file mode 100644 index 0000000..2c77055 --- /dev/null +++ b/backend/tests/test_pool.py @@ -0,0 +1,329 @@ +"""连接池管理端点完整测试 — pool.py + +覆盖端点: + GET /api/v1/pools/instances/{instance_id}/stats 连接池状态 + POST /api/v1/pools/instances/{instance_id}/reset 重置连接池 + GET /api/v1/pools/overview 所有连接池概览 +""" +import pytest +from unittest.mock import patch, AsyncMock, MagicMock +from httpx import AsyncClient + +from tests.conftest import auth_header + + +# ═══════════════════════════════════════════════════════════════ +# GET /pools/instances/{instance_id}/stats — 连接池状态 +# ═══════════════════════════════════════════════════════════════ + + +class TestPoolStats: + + async def test_stats_not_initialized( + self, client: AsyncClient, admin_token, test_instance + ): + """连接池未初始化时返回 not_initialized。""" + 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" + assert body["instance_id"] == test_instance.id + + async def test_stats_with_active_pool( + self, client: AsyncClient, admin_token, test_instance + ): + """已初始化的连接池返回统计数据。""" + with patch("app.api.v1.endpoints.pool.get_pool_manager") as mock_pm: + stats = MagicMock() + stats.to_dict.return_value = { + "instance_id": test_instance.id, + "active": 3, + "idle": 2, + "max_size": 10, + } + 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 + body = resp.json() + assert body["active"] == 3 + assert body["idle"] == 2 + + async def test_stats_instance_not_found( + self, client: AsyncClient, admin_token + ): + """数据源不存在返回 404。""" + resp = await client.get( + "/api/v1/pools/instances/9999/stats", + headers=auth_header(admin_token), + ) + assert resp.status_code == 404 + + async def test_stats_dba_allowed( + self, client: AsyncClient, dba_token, test_instance + ): + """DBA 角色可以查看连接池状态。""" + with patch("app.api.v1.endpoints.pool.get_pool_manager") as mock_pm: + stats = MagicMock() + stats.to_dict.return_value = {"active": 0, "idle": 0} + 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(dba_token), + ) + assert resp.status_code == 200 + + async def test_stats_developer_forbidden( + self, client: AsyncClient, developer_token, test_instance + ): + """developer 角色无权查看连接池状态。""" + 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_stats_guest_forbidden( + self, client: AsyncClient, guest_token, test_instance + ): + """guest 角色无权查看连接池状态。""" + resp = await client.get( + f"/api/v1/pools/instances/{test_instance.id}/stats", + headers=auth_header(guest_token), + ) + assert resp.status_code == 403 + + async def test_stats_unauthenticated( + self, client: AsyncClient, test_instance + ): + """未认证请求返回 401。""" + resp = await client.get( + f"/api/v1/pools/instances/{test_instance.id}/stats", + ) + assert resp.status_code == 401 + + async def test_stats_returns_not_initialized_message( + self, client: AsyncClient, 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), + ) + body = resp.json() + assert "message" in body + assert "首次查询" in body["message"] or "初始化" in body["message"] + + +# ═══════════════════════════════════════════════════════════════ +# POST /pools/instances/{instance_id}/reset — 重置连接池 +# ═══════════════════════════════════════════════════════════════ + + +class TestPoolReset: + + async def test_reset_success( + self, client: AsyncClient, 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 + body = resp.json() + assert "已重置" in body["message"] + assert body["instance_id"] == test_instance.id + + async def test_reset_dba_allowed( + self, client: AsyncClient, dba_token, test_instance + ): + """DBA 角色可以重置连接池。""" + 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 + + async def test_reset_developer_forbidden( + self, client: AsyncClient, developer_token, test_instance + ): + """developer 角色无权重置连接池。""" + resp = await client.post( + f"/api/v1/pools/instances/{test_instance.id}/reset", + headers=auth_header(developer_token), + ) + assert resp.status_code == 403 + + async def test_reset_guest_forbidden( + self, client: AsyncClient, guest_token, test_instance + ): + """guest 角色无权重置连接池。""" + resp = await client.post( + f"/api/v1/pools/instances/{test_instance.id}/reset", + headers=auth_header(guest_token), + ) + assert resp.status_code == 403 + + async def test_reset_unauthenticated( + self, client: AsyncClient, test_instance + ): + """未认证请求返回 401。""" + resp = await client.post( + f"/api/v1/pools/instances/{test_instance.id}/reset", + ) + assert resp.status_code == 401 + + async def test_reset_calls_manager_reset_pool( + self, client: AsyncClient, admin_token, test_instance + ): + """验证 reset_pool 被正确调用。""" + with patch("app.api.v1.endpoints.pool.get_pool_manager") as mock_pm: + manager = MagicMock() + manager.reset_pool = AsyncMock() + mock_pm.return_value = manager + + await client.post( + f"/api/v1/pools/instances/{test_instance.id}/reset", + headers=auth_header(admin_token), + ) + manager.reset_pool.assert_awaited_once_with(test_instance.id) + + +# ═══════════════════════════════════════════════════════════════ +# GET /pools/overview — 所有连接池概览 +# ═══════════════════════════════════════════════════════════════ + + +class TestPoolsOverview: + + async def test_overview_empty(self, client: AsyncClient, 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 + body = resp.json() + assert body["total_pools"] == 0 + assert body["pools"] == [] + + async def test_overview_with_pools(self, client: AsyncClient, admin_token): + """有连接池时返回正确概览。""" + with patch("app.api.v1.endpoints.pool.get_pool_manager") as mock_pm: + stats1 = MagicMock() + stats1.to_dict.return_value = {"instance_id": 1, "active": 2} + stats2 = MagicMock() + stats2.to_dict.return_value = {"instance_id": 2, "active": 1} + + manager = MagicMock() + manager.get_all_stats.return_value = {1: stats1, 2: stats2} + 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"] == 2 + assert len(body["pools"]) == 2 + + async def test_overview_dba_allowed(self, client: AsyncClient, dba_token): + """DBA 角色可以查看连接池概览。""" + 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(dba_token), + ) + assert resp.status_code == 200 + + async def test_overview_developer_forbidden( + self, client: AsyncClient, developer_token + ): + """developer 角色无权查看连接池概览。""" + resp = await client.get( + "/api/v1/pools/overview", + headers=auth_header(developer_token), + ) + assert resp.status_code == 403 + + async def test_overview_guest_forbidden( + self, client: AsyncClient, guest_token + ): + """guest 角色无权查看连接池概览。""" + resp = await client.get( + "/api/v1/pools/overview", + headers=auth_header(guest_token), + ) + assert resp.status_code == 403 + + async def test_overview_unauthenticated(self, client: AsyncClient): + """未认证请求返回 401。""" + resp = await client.get("/api/v1/pools/overview") + assert resp.status_code == 401 + + async def test_overview_single_pool_data( + self, client: AsyncClient, admin_token + ): + """单个连接池数据验证。""" + with patch("app.api.v1.endpoints.pool.get_pool_manager") as mock_pm: + stats = MagicMock() + stats.to_dict.return_value = { + "instance_id": 42, + "active": 5, + "idle": 3, + "max_size": 10, + } + manager = MagicMock() + manager.get_all_stats.return_value = {42: stats} + mock_pm.return_value = manager + + resp = await client.get( + "/api/v1/pools/overview", + headers=auth_header(admin_token), + ) + body = resp.json() + assert body["total_pools"] == 1 + assert body["pools"][0]["instance_id"] == 42 + assert body["pools"][0]["active"] == 5 diff --git a/backend/tests/test_pool_endpoint.py b/backend/tests/test_pool_endpoint.py new file mode 100644 index 0000000..a674631 --- /dev/null +++ b/backend/tests/test_pool_endpoint.py @@ -0,0 +1,186 @@ +"""连接池管理 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 + + +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 diff --git a/backend/tests/test_schema_drift.py b/backend/tests/test_schema_drift.py new file mode 100644 index 0000000..d912604 --- /dev/null +++ b/backend/tests/test_schema_drift.py @@ -0,0 +1,413 @@ +"""Schema 漂移检测端点完整测试 — schema_drift.py + +覆盖端点: + POST /api/v1/schema-drift/instances/{instance_id}/capture 捕获快照 + POST /api/v1/schema-drift/instances/{instance_id}/check 检测漂移 + GET /api/v1/schema-drift/instances/{instance_id}/history 变更历史 +""" +import pytest +from unittest.mock import patch, AsyncMock, MagicMock +from httpx import AsyncClient + +from tests.conftest import auth_header + + +# ═══════════════════════════════════════════════════════════════ +# POST /schema-drift/instances/{instance_id}/capture — 捕获快照 +# ═══════════════════════════════════════════════════════════════ + + +class TestSchemaDriftCapture: + + async def test_capture_success( + self, client: AsyncClient, admin_token, test_instance + ): + """管理员成功捕获 Schema 快照。""" + with patch( + "app.api.v1.endpoints.schema_drift.SchemaDriftDetector" + ) as MockDet, patch( + "app.api.v1.endpoints.schema_drift.decrypt_password", + return_value="decrypted_pw", + ): + detector_mock = MagicMock() + detector_mock.capture_snapshot = AsyncMock( + return_value=[{"table": "users", "column": "id"}] + ) + detector_mock.save_baseline = AsyncMock() + MockDet.return_value = detector_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["instance_id"] == test_instance.id + assert body["total_columns"] == 1 + + async def test_capture_with_explicit_database( + self, client: AsyncClient, admin_token, test_instance + ): + """通过 query param 指定 database。""" + with patch( + "app.api.v1.endpoints.schema_drift.SchemaDriftDetector" + ) as MockDet, patch( + "app.api.v1.endpoints.schema_drift.decrypt_password", + return_value="decrypted_pw", + ): + detector_mock = MagicMock() + detector_mock.capture_snapshot = AsyncMock(return_value=[]) + detector_mock.save_baseline = AsyncMock() + MockDet.return_value = detector_mock + + resp = await client.post( + f"/api/v1/schema-drift/instances/{test_instance.id}/capture?database=mydb", + headers=auth_header(admin_token), + ) + assert resp.status_code == 200 + body = resp.json() + assert body["database"] == "mydb" + + async def test_capture_uses_instance_database_as_default( + self, client: AsyncClient, admin_token, test_instance + ): + """不传 database 时使用 instance 的 database 字段。""" + with patch( + "app.api.v1.endpoints.schema_drift.SchemaDriftDetector" + ) as MockDet, patch( + "app.api.v1.endpoints.schema_drift.decrypt_password", + return_value="decrypted_pw", + ): + detector_mock = MagicMock() + detector_mock.capture_snapshot = AsyncMock(return_value=[]) + detector_mock.save_baseline = AsyncMock() + MockDet.return_value = detector_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 + # test_instance.database == "test_db" + assert resp.json()["database"] == "test_db" + + async def test_capture_instance_not_found( + self, client: AsyncClient, admin_token + ): + """数据源不存在返回 404。""" + resp = await client.post( + "/api/v1/schema-drift/instances/9999/capture", + headers=auth_header(admin_token), + ) + assert resp.status_code == 404 + assert "不存在" in resp.json()["detail"] + + async def test_capture_developer_forbidden( + self, client: AsyncClient, developer_token, test_instance + ): + """developer 角色无权捕获快照。""" + resp = await client.post( + f"/api/v1/schema-drift/instances/{test_instance.id}/capture", + headers=auth_header(developer_token), + ) + assert resp.status_code == 403 + + async def test_capture_guest_forbidden( + self, client: AsyncClient, guest_token, test_instance + ): + """guest 角色无权捕获快照。""" + resp = await client.post( + f"/api/v1/schema-drift/instances/{test_instance.id}/capture", + headers=auth_header(guest_token), + ) + assert resp.status_code == 403 + + async def test_capture_dba_allowed( + self, client: AsyncClient, dba_token, test_instance + ): + """DBA 角色可以捕获快照。""" + with patch( + "app.api.v1.endpoints.schema_drift.SchemaDriftDetector" + ) as MockDet, patch( + "app.api.v1.endpoints.schema_drift.decrypt_password", + return_value="decrypted_pw", + ): + detector_mock = MagicMock() + detector_mock.capture_snapshot = AsyncMock(return_value=[]) + detector_mock.save_baseline = AsyncMock() + MockDet.return_value = detector_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 + + async def test_capture_unauthenticated( + self, client: AsyncClient, test_instance + ): + """未认证请求返回 401。""" + resp = await client.post( + f"/api/v1/schema-drift/instances/{test_instance.id}/capture", + ) + assert resp.status_code == 401 + + async def test_capture_multiple_columns( + self, client: AsyncClient, admin_token, test_instance + ): + """快照包含多个列。""" + snapshot = [ + {"table": "users", "column": "id"}, + {"table": "users", "column": "name"}, + {"table": "orders", "column": "id"}, + ] + with patch( + "app.api.v1.endpoints.schema_drift.SchemaDriftDetector" + ) as MockDet, patch( + "app.api.v1.endpoints.schema_drift.decrypt_password", + return_value="decrypted_pw", + ): + detector_mock = MagicMock() + detector_mock.capture_snapshot = AsyncMock(return_value=snapshot) + detector_mock.save_baseline = AsyncMock() + MockDet.return_value = detector_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 + assert resp.json()["total_columns"] == 3 + + +# ═══════════════════════════════════════════════════════════════ +# POST /schema-drift/instances/{instance_id}/check — 检测漂移 +# ═══════════════════════════════════════════════════════════════ + + +class TestSchemaDriftCheck: + + async def test_check_no_drift( + self, client: AsyncClient, admin_token, test_instance + ): + """无漂移时返回正确报告。""" + with patch( + "app.api.v1.endpoints.schema_drift.SchemaDriftDetector" + ) as MockDet, patch( + "app.api.v1.endpoints.schema_drift.decrypt_password", + return_value="decrypted_pw", + ): + report = MagicMock() + report.to_dict.return_value = { + "drifted": False, + "changes": [], + "instance_id": test_instance.id, + } + detector_mock = MagicMock() + detector_mock.detect_drift = AsyncMock(return_value=report) + MockDet.return_value = detector_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 + assert body["changes"] == [] + + async def test_check_with_drift( + self, client: AsyncClient, admin_token, test_instance + ): + """检测到漂移时返回变更列表。""" + with patch( + "app.api.v1.endpoints.schema_drift.SchemaDriftDetector" + ) as MockDet, patch( + "app.api.v1.endpoints.schema_drift.decrypt_password", + return_value="decrypted_pw", + ): + report = MagicMock() + report.to_dict.return_value = { + "drifted": True, + "changes": [ + {"type": "column_added", "table": "users", "column": "age"}, + ], + } + detector_mock = MagicMock() + detector_mock.detect_drift = AsyncMock(return_value=report) + MockDet.return_value = detector_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 True + assert len(body["changes"]) == 1 + + async def test_check_instance_not_found( + self, client: AsyncClient, admin_token + ): + """数据源不存在返回 404。""" + resp = await client.post( + "/api/v1/schema-drift/instances/9999/check", + headers=auth_header(admin_token), + ) + assert resp.status_code == 404 + + async def test_check_dba_allowed( + self, client: AsyncClient, dba_token, test_instance + ): + """DBA 角色可以检测漂移。""" + with patch( + "app.api.v1.endpoints.schema_drift.SchemaDriftDetector" + ) as MockDet, patch( + "app.api.v1.endpoints.schema_drift.decrypt_password", + return_value="decrypted_pw", + ): + report = MagicMock() + report.to_dict.return_value = {"drifted": False, "changes": []} + detector_mock = MagicMock() + detector_mock.detect_drift = AsyncMock(return_value=report) + MockDet.return_value = detector_mock + + resp = await client.post( + f"/api/v1/schema-drift/instances/{test_instance.id}/check", + headers=auth_header(dba_token), + ) + assert resp.status_code == 200 + + async def test_check_developer_forbidden( + self, client: AsyncClient, developer_token, test_instance + ): + """developer 角色无权检测漂移。""" + resp = await client.post( + f"/api/v1/schema-drift/instances/{test_instance.id}/check", + headers=auth_header(developer_token), + ) + assert resp.status_code == 403 + + async def test_check_guest_forbidden( + self, client: AsyncClient, guest_token, test_instance + ): + """guest 角色无权检测漂移。""" + resp = await client.post( + f"/api/v1/schema-drift/instances/{test_instance.id}/check", + headers=auth_header(guest_token), + ) + assert resp.status_code == 403 + + async def test_check_unauthenticated( + self, client: AsyncClient, test_instance + ): + """未认证请求返回 401。""" + resp = await client.post( + f"/api/v1/schema-drift/instances/{test_instance.id}/check", + ) + assert resp.status_code == 401 + + async def test_check_with_explicit_database( + self, client: AsyncClient, admin_token, test_instance + ): + """通过 query param 指定 database 检测漂移。""" + with patch( + "app.api.v1.endpoints.schema_drift.SchemaDriftDetector" + ) as MockDet, patch( + "app.api.v1.endpoints.schema_drift.decrypt_password", + return_value="decrypted_pw", + ): + report = MagicMock() + report.to_dict.return_value = {"drifted": False} + detector_mock = MagicMock() + detector_mock.detect_drift = AsyncMock(return_value=report) + MockDet.return_value = detector_mock + + resp = await client.post( + f"/api/v1/schema-drift/instances/{test_instance.id}/check?database=other_db", + headers=auth_header(admin_token), + ) + assert resp.status_code == 200 + + +# ═══════════════════════════════════════════════════════════════ +# GET /schema-drift/instances/{instance_id}/history — 变更历史 +# ═══════════════════════════════════════════════════════════════ + + +class TestSchemaHistory: + + async def test_history_returns_data( + self, client: AsyncClient, admin_token, test_instance + ): + """管理员可以获取 Schema 变更历史。""" + 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 + + async def test_history_is_list( + self, client: AsyncClient, admin_token, test_instance + ): + """history 字段为列表。""" + resp = await client.get( + f"/api/v1/schema-drift/instances/{test_instance.id}/history", + headers=auth_header(admin_token), + ) + body = resp.json() + assert isinstance(body["history"], list) + + async def test_history_dba_allowed( + self, client: AsyncClient, dba_token, test_instance + ): + """DBA 角色可以查看变更历史。""" + resp = await client.get( + f"/api/v1/schema-drift/instances/{test_instance.id}/history", + headers=auth_header(dba_token), + ) + assert resp.status_code == 200 + + async def test_history_developer_forbidden( + self, client: AsyncClient, developer_token, test_instance + ): + """developer 角色无权查看变更历史。""" + resp = await client.get( + f"/api/v1/schema-drift/instances/{test_instance.id}/history", + headers=auth_header(developer_token), + ) + assert resp.status_code == 403 + + async def test_history_guest_forbidden( + self, client: AsyncClient, guest_token, test_instance + ): + """guest 角色无权查看变更历史。""" + resp = await client.get( + f"/api/v1/schema-drift/instances/{test_instance.id}/history", + headers=auth_header(guest_token), + ) + assert resp.status_code == 403 + + async def test_history_unauthenticated( + self, client: AsyncClient, test_instance + ): + """未认证请求返回 401。""" + resp = await client.get( + f"/api/v1/schema-drift/instances/{test_instance.id}/history", + ) + assert resp.status_code == 401 + + async def test_history_contains_message( + self, client: AsyncClient, admin_token, test_instance + ): + """响应包含 message 字段。""" + resp = await client.get( + f"/api/v1/schema-drift/instances/{test_instance.id}/history", + headers=auth_header(admin_token), + ) + body = resp.json() + assert "message" in body diff --git a/backend/tests/test_schema_drift_endpoint.py b/backend/tests/test_schema_drift_endpoint.py new file mode 100644 index 0000000..5f777d1 --- /dev/null +++ b/backend/tests/test_schema_drift_endpoint.py @@ -0,0 +1,231 @@ +"""Schema 漂移检测 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 + + +def auth_header(token: str) -> dict: + return {"Authorization": f"Bearer {token}"} + + +@pytest.mark.asyncio +class TestCaptureSchema: + """POST /api/v1/schema-drift/instances/{instance_id}/capture""" + + async def test_capture_schema_success( + self, client: AsyncClient, admin_token: str, test_instance + ): + mock_snapshot = [ + {"table": "users", "column": "id", "type": "int"}, + {"table": "users", "column": "name", "type": "varchar"}, + ] + with patch( + "app.api.v1.endpoints.schema_drift.SchemaDriftDetector" + ) as MockDetector, patch( + "app.api.v1.endpoints.schema_drift.decrypt_password", + return_value="decrypted_pw", + ): + detector = MockDetector.return_value + detector.capture_snapshot = AsyncMock(return_value=mock_snapshot) + detector.save_baseline = AsyncMock() + + resp = await client.post( + f"/api/v1/schema-drift/instances/{test_instance.id}/capture", + headers=auth_header(admin_token), + ) + assert resp.status_code == 200 + data = resp.json() + assert "已捕获" in data["message"] + assert data["total_columns"] == 2 + assert data["instance_id"] == test_instance.id + + async def test_capture_schema_with_database_param( + self, client: AsyncClient, admin_token: str, test_instance + ): + with patch( + "app.api.v1.endpoints.schema_drift.SchemaDriftDetector" + ) as MockDetector, patch( + "app.api.v1.endpoints.schema_drift.decrypt_password", + return_value="decrypted_pw", + ): + detector = MockDetector.return_value + detector.capture_snapshot = AsyncMock(return_value=[]) + detector.save_baseline = AsyncMock() + + resp = await client.post( + f"/api/v1/schema-drift/instances/{test_instance.id}/capture?database=mydb", + headers=auth_header(admin_token), + ) + assert resp.status_code == 200 + data = resp.json() + assert data["database"] == "mydb" + + async def test_capture_schema_instance_not_found( + self, client: AsyncClient, admin_token: str + ): + resp = await client.post( + "/api/v1/schema-drift/instances/9999/capture", + headers=auth_header(admin_token), + ) + assert resp.status_code == 404 + + async def test_capture_schema_forbidden_developer( + self, client: AsyncClient, developer_token: str, 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 + + async def test_capture_schema_dba_allowed( + self, client: AsyncClient, dba_token: str, test_instance + ): + with patch( + "app.api.v1.endpoints.schema_drift.SchemaDriftDetector" + ) as MockDetector, patch( + "app.api.v1.endpoints.schema_drift.decrypt_password", + return_value="pw", + ): + detector = MockDetector.return_value + detector.capture_snapshot = AsyncMock(return_value=[]) + detector.save_baseline = AsyncMock() + + resp = await client.post( + f"/api/v1/schema-drift/instances/{test_instance.id}/capture", + headers=auth_header(dba_token), + ) + assert resp.status_code == 200 + + +@pytest.mark.asyncio +class TestCheckDrift: + """POST /api/v1/schema-drift/instances/{instance_id}/check""" + + async def test_check_drift_no_changes( + self, client: AsyncClient, admin_token: str, test_instance + ): + mock_report = MagicMock() + mock_report.to_dict.return_value = { + "has_drift": False, + "changes": [], + "instance_id": test_instance.id, + } + with patch( + "app.api.v1.endpoints.schema_drift.SchemaDriftDetector" + ) as MockDetector, patch( + "app.api.v1.endpoints.schema_drift.decrypt_password", + return_value="pw", + ): + detector = MockDetector.return_value + detector.detect_drift = AsyncMock(return_value=mock_report) + + resp = await client.post( + f"/api/v1/schema-drift/instances/{test_instance.id}/check", + headers=auth_header(admin_token), + ) + assert resp.status_code == 200 + data = resp.json() + assert data["has_drift"] is False + + async def test_check_drift_with_changes( + self, client: AsyncClient, admin_token: str, test_instance + ): + mock_report = MagicMock() + mock_report.to_dict.return_value = { + "has_drift": True, + "changes": [ + {"type": "column_added", "table": "users", "column": "email"}, + ], + "instance_id": test_instance.id, + } + with patch( + "app.api.v1.endpoints.schema_drift.SchemaDriftDetector" + ) as MockDetector, patch( + "app.api.v1.endpoints.schema_drift.decrypt_password", + return_value="pw", + ): + detector = MockDetector.return_value + detector.detect_drift = AsyncMock(return_value=mock_report) + + resp = await client.post( + f"/api/v1/schema-drift/instances/{test_instance.id}/check", + headers=auth_header(admin_token), + ) + assert resp.status_code == 200 + data = resp.json() + assert data["has_drift"] is True + assert len(data["changes"]) == 1 + + async def test_check_drift_instance_not_found( + self, client: AsyncClient, admin_token: str + ): + resp = await client.post( + "/api/v1/schema-drift/instances/9999/check", + headers=auth_header(admin_token), + ) + assert resp.status_code == 404 + + async def test_check_drift_dict_report( + self, client: AsyncClient, admin_token: str, test_instance + ): + """Detector 返回普通 dict 而非对象时的序列化。""" + plain_dict = {"has_drift": False, "instance_id": test_instance.id} + with patch( + "app.api.v1.endpoints.schema_drift.SchemaDriftDetector" + ) as MockDetector, patch( + "app.api.v1.endpoints.schema_drift.decrypt_password", + return_value="pw", + ): + detector = MockDetector.return_value + detector.detect_drift = AsyncMock(return_value=plain_dict) + + resp = await client.post( + f"/api/v1/schema-drift/instances/{test_instance.id}/check", + headers=auth_header(admin_token), + ) + assert resp.status_code == 200 + + +@pytest.mark.asyncio +class TestSchemaHistory: + """GET /api/v1/schema-drift/instances/{instance_id}/history""" + + async def test_schema_history( + self, client: AsyncClient, admin_token: str, 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 + data = resp.json() + assert data["instance_id"] == test_instance.id + assert "history" in data + + async def test_schema_history_dba_allowed( + self, client: AsyncClient, dba_token: str, test_instance + ): + resp = await client.get( + f"/api/v1/schema-drift/instances/{test_instance.id}/history", + headers=auth_header(dba_token), + ) + assert resp.status_code == 200 + + async def test_schema_history_forbidden_guest( + self, client: AsyncClient, guest_token: str, test_instance + ): + resp = await client.get( + f"/api/v1/schema-drift/instances/{test_instance.id}/history", + headers=auth_header(guest_token), + ) + assert resp.status_code == 403 + + async def test_schema_history_unauthorized(self, client: AsyncClient, test_instance): + resp = await client.get( + f"/api/v1/schema-drift/instances/{test_instance.id}/history", + ) + assert resp.status_code == 401 diff --git a/backend/tests/test_user.py b/backend/tests/test_user.py new file mode 100644 index 0000000..d1476c9 --- /dev/null +++ b/backend/tests/test_user.py @@ -0,0 +1,449 @@ +"""用户管理端点完整测试 — user.py + +覆盖端点: + GET /api/v1/users 列表（支持 role / is_active 筛选） + POST /api/v1/users 创建用户 + GET /api/v1/users/{user_id} 获取单个用户 + PUT /api/v1/users/{user_id} 更新用户 + DELETE /api/v1/users/{user_id} 删除用户 +""" +import pytest +from httpx import AsyncClient + +from tests.conftest import auth_header + + +# ═══════════════════════════════════════════════════════════════ +# GET /users — 列表 +# ═══════════════════════════════════════════════════════════════ + + +class TestListUsers: + + async def test_list_all_users(self, client: AsyncClient, admin_user, admin_token): + """管理员可以获取全部用户列表。""" + resp = await client.get("/api/v1/users", headers=auth_header(admin_token)) + assert resp.status_code == 200 + data = resp.json() + assert isinstance(data, list) + assert len(data) >= 1 + + async def test_list_users_role_filter(self, client: AsyncClient, admin_user, admin_token): + """按 role 筛选用户。""" + resp = await client.get( + "/api/v1/users?role=admin", + headers=auth_header(admin_token), + ) + assert resp.status_code == 200 + for u in resp.json(): + assert u["role"] == "admin" + + async def test_list_users_is_active_filter( + self, client: AsyncClient, admin_user, disabled_user, admin_token + ): + """按 is_active 筛选用户。""" + resp = await client.get( + "/api/v1/users?is_active=false", + headers=auth_header(admin_token), + ) + assert resp.status_code == 200 + for u in resp.json(): + assert u["is_active"] is False + + async def test_list_users_combined_filters( + self, client: AsyncClient, admin_user, disabled_user, admin_token + ): + """同时使用 role + is_active 筛选。""" + resp = await client.get( + "/api/v1/users?role=developer&is_active=false", + headers=auth_header(admin_token), + ) + assert resp.status_code == 200 + for u in resp.json(): + assert u["role"] == "developer" + assert u["is_active"] is False + + async def test_list_users_empty_result(self, client: AsyncClient, admin_token): + """不存在的角色返回空列表。""" + resp = await client.get( + "/api/v1/users?role=guest", + headers=auth_header(admin_token), + ) + assert resp.status_code == 200 + assert resp.json() == [] + + async def test_list_users_non_admin_forbidden( + self, client: AsyncClient, developer_token + ): + """developer 角色无权访问用户列表。""" + resp = await client.get( + "/api/v1/users", headers=auth_header(developer_token) + ) + assert resp.status_code == 403 + + async def test_list_users_guest_forbidden( + self, client: AsyncClient, guest_token + ): + """guest 角色无权访问用户列表。""" + resp = await client.get( + "/api/v1/users", headers=auth_header(guest_token) + ) + assert resp.status_code == 403 + + async def test_list_users_unauthenticated(self, client: AsyncClient): + """未认证请求返回 401。""" + resp = await client.get("/api/v1/users") + assert resp.status_code == 401 + + +# ═══════════════════════════════════════════════════════════════ +# POST /users — 创建 +# ═══════════════════════════════════════════════════════════════ + + +class TestCreateUser: + + async def test_create_user_success(self, client: AsyncClient, admin_token): + """成功创建用户。""" + resp = await client.post( + "/api/v1/users", + headers=auth_header(admin_token), + json={ + "username": "newuser", + "display_name": "新用户", + "password": "secure123", + "role": "developer", + }, + ) + assert resp.status_code == 201 + data = resp.json() + assert data["username"] == "newuser" + assert data["display_name"] == "新用户" + assert data["role"] == "developer" + assert "id" in data + assert "hashed_password" not in data + + async def test_create_user_with_all_fields(self, client: AsyncClient, admin_token): + """创建用户时填写所有可选字段。""" + resp = await client.post( + "/api/v1/users", + headers=auth_header(admin_token), + json={ + "username": "fulluser", + "display_name": "完整用户", + "password": "secure123", + "email": "full@example.com", + "phone": "13800138000", + "role": "dba", + }, + ) + assert resp.status_code == 201 + data = resp.json() + assert data["email"] == "full@example.com" + assert data["phone"] == "13800138000" + assert data["role"] == "dba" + + async def test_create_duplicate_username( + self, client: AsyncClient, admin_user, admin_token + ): + """用户名重复返回 400。""" + resp = await client.post( + "/api/v1/users", + headers=auth_header(admin_token), + json={ + "username": "admin", + "display_name": "重复", + "password": "123456", + "role": "developer", + }, + ) + assert resp.status_code == 400 + assert "已存在" in resp.json()["detail"] + + async def test_create_user_short_password(self, client: AsyncClient, admin_token): + """密码少于 6 位返回 422。""" + resp = await client.post( + "/api/v1/users", + headers=auth_header(admin_token), + json={ + "username": "shortpw", + "display_name": "短密码", + "password": "123", + "role": "developer", + }, + ) + assert resp.status_code == 422 + + async def test_create_user_invalid_role(self, client: AsyncClient, admin_token): + """无效角色返回 422。""" + resp = await client.post( + "/api/v1/users", + headers=auth_header(admin_token), + json={ + "username": "badrole", + "display_name": "坏角色", + "password": "123456", + "role": "superadmin", + }, + ) + assert resp.status_code == 422 + + async def test_create_user_missing_username(self, client: AsyncClient, admin_token): + """缺少 username 字段返回 422。""" + resp = await client.post( + "/api/v1/users", + headers=auth_header(admin_token), + json={ + "display_name": "无用户名", + "password": "123456", + "role": "developer", + }, + ) + assert resp.status_code == 422 + + async def test_create_user_non_admin_forbidden( + self, client: AsyncClient, developer_token + ): + """developer 角色无权创建用户。""" + resp = await client.post( + "/api/v1/users", + headers=auth_header(developer_token), + json={ + "username": "unauthorized", + "display_name": "不应成功", + "password": "123456", + "role": "developer", + }, + ) + assert resp.status_code == 403 + + async def test_create_user_default_role(self, client: AsyncClient, admin_token): + """不传 role 时默认为 developer。""" + resp = await client.post( + "/api/v1/users", + headers=auth_header(admin_token), + json={ + "username": "defaultrole", + "display_name": "默认角色", + "password": "123456", + }, + ) + assert resp.status_code == 201 + assert resp.json()["role"] == "developer" + + +# ═══════════════════════════════════════════════════════════════ +# GET /users/{user_id} — 获取单个 +# ═══════════════════════════════════════════════════════════════ + + +class TestGetUser: + + async def test_get_user_success( + self, client: AsyncClient, admin_user, admin_token + ): + """成功获取单个用户。""" + resp = await client.get( + f"/api/v1/users/{admin_user.id}", + headers=auth_header(admin_token), + ) + assert resp.status_code == 200 + assert resp.json()["username"] == "admin" + + async def test_get_user_nonexistent(self, client: AsyncClient, admin_token): + """查询不存在的用户返回 404。""" + resp = await client.get( + "/api/v1/users/9999", + headers=auth_header(admin_token), + ) + assert resp.status_code == 404 + assert "不存在" in resp.json()["detail"] + + async def test_get_user_non_admin_forbidden( + self, client: AsyncClient, admin_user, developer_token + ): + """developer 角色无权查看用户详情。""" + resp = await client.get( + f"/api/v1/users/{admin_user.id}", + headers=auth_header(developer_token), + ) + assert resp.status_code == 403 + + async def test_get_user_unauthenticated(self, client: AsyncClient, admin_user): + """未认证请求返回 401。""" + resp = await client.get(f"/api/v1/users/{admin_user.id}") + assert resp.status_code == 401 + + +# ═══════════════════════════════════════════════════════════════ +# PUT /users/{user_id} — 更新 +# ═══════════════════════════════════════════════════════════════ + + +class TestUpdateUser: + + async def test_update_display_name( + self, client: AsyncClient, admin_user, admin_token + ): + """更新 display_name。""" + resp = await client.put( + f"/api/v1/users/{admin_user.id}", + headers=auth_header(admin_token), + json={"display_name": "管理员改名"}, + ) + assert resp.status_code == 200 + assert resp.json()["display_name"] == "管理员改名" + + async def test_update_email( + self, client: AsyncClient, admin_user, admin_token + ): + """更新 email。""" + resp = await client.put( + f"/api/v1/users/{admin_user.id}", + headers=auth_header(admin_token), + json={"email": "admin@example.com"}, + ) + assert resp.status_code == 200 + assert resp.json()["email"] == "admin@example.com" + + async def test_update_role( + self, client: AsyncClient, admin_user, dba_user, admin_token + ): + """更新用户角色。""" + resp = await client.put( + f"/api/v1/users/{dba_user.id}", + headers=auth_header(admin_token), + json={"role": "developer"}, + ) + assert resp.status_code == 200 + assert resp.json()["role"] == "developer" + + async def test_update_is_active( + self, client: AsyncClient, dba_user, admin_token + ): + """禁用用户。""" + resp = await client.put( + f"/api/v1/users/{dba_user.id}", + headers=auth_header(admin_token), + json={"is_active": False}, + ) + assert resp.status_code == 200 + assert resp.json()["is_active"] is False + + async def test_update_multiple_fields( + self, client: AsyncClient, dba_user, admin_token + ): + """同时更新多个字段。""" + resp = await client.put( + f"/api/v1/users/{dba_user.id}", + headers=auth_header(admin_token), + json={ + "display_name": "DBA改名", + "phone": "13900139000", + }, + ) + assert resp.status_code == 200 + data = resp.json() + assert data["display_name"] == "DBA改名" + assert data["phone"] == "13900139000" + + async def test_update_nonexistent_user(self, client: AsyncClient, admin_token): + """更新不存在的用户返回 404。""" + resp = await client.put( + "/api/v1/users/9999", + headers=auth_header(admin_token), + json={"display_name": "不存在"}, + ) + assert resp.status_code == 404 + + async def test_update_invalid_role( + self, client: AsyncClient, admin_user, admin_token + ): + """无效角色返回 422。""" + resp = await client.put( + f"/api/v1/users/{admin_user.id}", + headers=auth_header(admin_token), + json={"role": "superadmin"}, + ) + assert resp.status_code == 422 + + async def test_update_non_admin_forbidden( + self, client: AsyncClient, admin_user, developer_token + ): + """developer 角色无权更新用户。""" + resp = await client.put( + f"/api/v1/users/{admin_user.id}", + headers=auth_header(developer_token), + json={"display_name": "越权"}, + ) + assert resp.status_code == 403 + + +# ═══════════════════════════════════════════════════════════════ +# DELETE /users/{user_id} — 删除 +# ═══════════════════════════════════════════════════════════════ + + +class TestDeleteUser: + + async def test_delete_user_success( + self, client: AsyncClient, dba_user, admin_token + ): + """成功删除用户。""" + resp = await client.delete( + f"/api/v1/users/{dba_user.id}", + headers=auth_header(admin_token), + ) + assert resp.status_code == 200 + assert "已删除" in resp.json()["message"] + + # 确认已被删除 + resp2 = await client.get( + f"/api/v1/users/{dba_user.id}", + headers=auth_header(admin_token), + ) + assert resp2.status_code == 404 + + async def test_delete_self( + self, client: AsyncClient, admin_user, admin_token + ): + """不能删除自己。""" + resp = await client.delete( + f"/api/v1/users/{admin_user.id}", + headers=auth_header(admin_token), + ) + assert resp.status_code == 400 + assert "不能删除自己" in resp.json()["detail"] + + async def test_delete_nonexistent_user(self, client: AsyncClient, admin_token): + """删除不存在的用户返回 404。""" + resp = await client.delete( + "/api/v1/users/9999", + headers=auth_header(admin_token), + ) + assert resp.status_code == 404 + + async def test_delete_non_admin_forbidden( + self, client: AsyncClient, admin_user, developer_token + ): + """developer 角色无权删除用户。""" + resp = await client.delete( + f"/api/v1/users/{admin_user.id}", + headers=auth_header(developer_token), + ) + assert resp.status_code == 403 + + async def test_delete_unauthenticated(self, client: AsyncClient, admin_user): + """未认证请求返回 401。""" + resp = await client.delete(f"/api/v1/users/{admin_user.id}") + assert resp.status_code == 401 + + async def test_delete_guest_forbidden( + self, client: AsyncClient, admin_user, guest_token + ): + """guest 角色无权删除用户。""" + resp = await client.delete( + f"/api/v1/users/{admin_user.id}", + headers=auth_header(guest_token), + ) + assert resp.status_code == 403 diff --git a/backend/tests/test_user_endpoint.py b/backend/tests/test_user_endpoint.py new file mode 100644 index 0000000..1a82910 --- /dev/null +++ b/backend/tests/test_user_endpoint.py @@ -0,0 +1,230 @@ +"""用户管理 API 端点测试。""" +import pytest +from httpx import AsyncClient + +from app.core.security import hash_password, create_access_token +from app.models.user import User + + +def auth_header(token: str) -> dict: + return {"Authorization": f"Bearer {token}"} + + +@pytest.mark.asyncio +class TestListUsers: + """GET /api/v1/users""" + + async def test_list_users_empty(self, client: AsyncClient, admin_token: str): + resp = await client.get("/api/v1/users", headers=auth_header(admin_token)) + assert resp.status_code == 200 + data = resp.json() + # admin_user fixture 创建了 1 个 + assert isinstance(data, list) + + async def test_list_users_filter_role( + self, client: AsyncClient, db, admin_token: str + ): + # 创建一个 dba 用户 + dba = User( + username="dba_test", + display_name="DBA", + role="dba", + hashed_password=hash_password("123456"), + ) + db.add(dba) + await db.commit() + + resp = await client.get( + "/api/v1/users?role=dba", headers=auth_header(admin_token) + ) + assert resp.status_code == 200 + data = resp.json() + assert all(u["role"] == "dba" for u in data) + + async def test_list_users_filter_active( + self, client: AsyncClient, db, admin_token: str + ): + dev = User( + username="inactive_dev", + display_name="离线", + role="developer", + is_active=False, + hashed_password=hash_password("123456"), + ) + db.add(dev) + await db.commit() + + resp = await client.get( + "/api/v1/users?is_active=false", headers=auth_header(admin_token) + ) + assert resp.status_code == 200 + data = resp.json() + assert all(u["is_active"] is False for u in data) + + async def test_list_users_forbidden_non_admin( + self, client: AsyncClient, developer_token: str + ): + resp = await client.get("/api/v1/users", headers=auth_header(developer_token)) + assert resp.status_code == 403 + + async def test_list_users_unauthorized(self, client: AsyncClient): + resp = await client.get("/api/v1/users") + assert resp.status_code == 401 + + +@pytest.mark.asyncio +class TestCreateUser: + """POST /api/v1/users""" + + async def test_create_user_success(self, client: AsyncClient, admin_token: str): + resp = await client.post( + "/api/v1/users", + json={ + "username": "newuser", + "display_name": "新用户", + "email": "new@test.com", + "phone": "13800001111", + "role": "developer", + "password": "pass123", + }, + headers=auth_header(admin_token), + ) + assert resp.status_code == 201 + data = resp.json() + assert data["username"] == "newuser" + assert data["role"] == "developer" + assert "hashed_password" not in data + + async def test_create_user_duplicate_username( + self, client: AsyncClient, admin_token: str, admin_user: User + ): + resp = await client.post( + "/api/v1/users", + json={ + "username": "admin", + "display_name": "重复", + "role": "developer", + "password": "123456", + }, + headers=auth_header(admin_token), + ) + assert resp.status_code == 400 + assert "已存在" in resp.json()["detail"] + + async def test_create_user_forbidden_dba( + self, client: AsyncClient, dba_token: str + ): + resp = await client.post( + "/api/v1/users", + json={ + "username": "hack", + "display_name": "hack", + "role": "developer", + "password": "123", + }, + headers=auth_header(dba_token), + ) + assert resp.status_code == 403 + + +@pytest.mark.asyncio +class TestGetUser: + """GET /api/v1/users/{user_id}""" + + async def test_get_user_success( + self, client: AsyncClient, admin_token: str, admin_user: User + ): + resp = await client.get( + f"/api/v1/users/{admin_user.id}", headers=auth_header(admin_token) + ) + assert resp.status_code == 200 + data = resp.json() + assert data["username"] == "admin" + + async def test_get_user_not_found(self, client: AsyncClient, admin_token: str): + resp = await client.get( + "/api/v1/users/9999", headers=auth_header(admin_token) + ) + assert resp.status_code == 404 + + async def test_get_user_forbidden(self, client: AsyncClient, developer_token: str): + resp = await client.get( + "/api/v1/users/1", headers=auth_header(developer_token) + ) + assert resp.status_code == 403 + + +@pytest.mark.asyncio +class TestUpdateUser: + """PUT /api/v1/users/{user_id}""" + + async def test_update_user_display_name( + self, client: AsyncClient, admin_token: str, dba_user + ): + resp = await client.put( + f"/api/v1/users/{dba_user.id}", + json={"display_name": "DBA王五"}, + headers=auth_header(admin_token), + ) + assert resp.status_code == 200 + data = resp.json() + assert data["display_name"] == "DBA王五" + + async def test_update_user_role( + self, client: AsyncClient, admin_token: str, developer_user + ): + resp = await client.put( + f"/api/v1/users/{developer_user.id}", + json={"role": "dba"}, + headers=auth_header(admin_token), + ) + assert resp.status_code == 200 + assert resp.json()["role"] == "dba" + + async def test_update_user_not_found(self, client: AsyncClient, admin_token: str): + resp = await client.put( + "/api/v1/users/9999", + json={"display_name": "x"}, + headers=auth_header(admin_token), + ) + assert resp.status_code == 404 + + async def test_update_user_partial( + self, client: AsyncClient, admin_token: str, dba_user + ): + resp = await client.put( + f"/api/v1/users/{dba_user.id}", + json={"email": "dba@test.com"}, + headers=auth_header(admin_token), + ) + assert resp.status_code == 200 + assert resp.json()["email"] == "dba@test.com" + + +@pytest.mark.asyncio +class TestDeleteUser: + """DELETE /api/v1/users/{user_id}""" + + async def test_delete_user_success( + self, client: AsyncClient, admin_token: str, developer_user + ): + resp = await client.delete( + f"/api/v1/users/{developer_user.id}", headers=auth_header(admin_token) + ) + assert resp.status_code == 200 + assert "已删除" in resp.json()["message"] + + async def test_delete_self_forbidden( + self, client: AsyncClient, admin_token: str, admin_user: User + ): + resp = await client.delete( + f"/api/v1/users/{admin_user.id}", headers=auth_header(admin_token) + ) + assert resp.status_code == 400 + assert "不能删除自己" in resp.json()["detail"] + + async def test_delete_user_not_found(self, client: AsyncClient, admin_token: str): + resp = await client.delete( + "/api/v1/users/9999", headers=auth_header(admin_token) + ) + assert resp.status_code == 404 diff --git a/run.sh b/run.sh old mode 100644 new mode 100755