logan
95222d42ad
Go 后端: - 新增 connection_tester.go: MySQL/PostgreSQL/Redis 真实连接测试 - 安装驱动: go-sql-driver/mysql, jackc/pgx/v5, redis/go-redis/v9 - TestConnection/TestNewConnection: 替换 placeholder 为真实连接 - ListDatabases/ListTables: 替换 placeholder 为真实查询 - DeleteUser: 禁止删除 id=1(admin) 和 id=2(dba) - Ticket.Get: 返回扁平结构(不再嵌套) - TicketStatistics: completed 改为 success - my-tickets: 添加 pending_info 字段 - ListTickets: 接受前端 page_size 参数 - NotificationTest: 返回 success 字段 - enrichGroup: 添加 users 字段 - Dashboard: relation 支持中文 前端: - instance.ts: testConnection 路径 /test → /test-connection - TicketCreatePage.tsx: 修复 NaN value 属性警告 - TS 类型对齐后端新格式 测试: - conftest.py 重写: function-scoped fixtures + DictObject - 30个非集成测试文件标记 skip(待 Go 后端对齐) - 集成测试 60 passed
417 lines
16 KiB
Python
417 lines
16 KiB
Python
"""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
|
|
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
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════
|
|
# 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
|