logan
2984f11181
- 删除16个旧Python测试文件(970行→400行) - 重写conftest.py适配真实种子用户(admin/test123/test234/testtest/logan) - 5个精简测试文件覆盖核心API: - test_auth.py: 登录/鉴权/种子用户验证 (6 tests) - test_instances.py: 实例列表/详情/连接测试 (6 tests) - test_tickets.py: 工单CRUD/只读SQL拦截/auditor权限 (7 tests) - test_workflow.py: E2E完整审批流/拒绝/撤销 (3 tests) - test_permissions.py: 角色权限/执行权限/workflow权限 (6 tests) - 新增 cleanup_test_data.py 清理测试脏数据(支持--dry-run) - 全量29测试3.27s通过
93 lines
3.1 KiB
Python
93 lines
3.1 KiB
Python
"""Integration test conftest — Go/Gin backend at localhost:9000"""
|
|
import pytest
|
|
import pytest_asyncio
|
|
from httpx import AsyncClient
|
|
|
|
BASE_URL = "http://localhost:9000"
|
|
|
|
# ═══════════════════════════════════════════════
|
|
# Real test users (seeded by main.go seed_data)
|
|
# ═══════════════════════════════════════════════
|
|
USERS = {
|
|
"admin": {"password": "admin123", "role": "admin"},
|
|
"test123": {"password": "test123", "role": "developer"},
|
|
"test234": {"password": "test234", "role": "developer"},
|
|
"testtest": {"password": "testtest", "role": "developer"},
|
|
"logan": {"password": "zxc987", "role": "dba"},
|
|
}
|
|
|
|
|
|
def auth_header(token: str) -> dict:
|
|
return {"Authorization": f"Bearer {token}"}
|
|
|
|
|
|
async def _login(client: AsyncClient, username: str, password: str) -> str:
|
|
resp = await client.post("/api/v1/auth/login", json={
|
|
"username": username, "password": password,
|
|
})
|
|
assert resp.status_code == 200, f"Login failed for {username}: {resp.status_code} {resp.text}"
|
|
return resp.json()["access_token"]
|
|
|
|
|
|
# ═══════════════════════════════════════════════
|
|
# Fixtures
|
|
# ═══════════════════════════════════════════════
|
|
|
|
@pytest_asyncio.fixture
|
|
async def client():
|
|
async with AsyncClient(base_url=BASE_URL, timeout=30.0) as c:
|
|
yield c
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def admin_token(client):
|
|
return await _login(client, "admin", "admin123")
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def dba_token(client):
|
|
return await _login(client, "logan", "zxc987")
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def developer_token(client):
|
|
return await _login(client, "testtest", "testtest")
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def approver1_token(client):
|
|
"""test123 — 初审人"""
|
|
return await _login(client, "test123", "test123")
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def approver2_token(client):
|
|
"""test234 — 复审人"""
|
|
return await _login(client, "test234", "test234")
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def mysql_instance(client, admin_token):
|
|
"""获取第一个 MySQL 实例"""
|
|
resp = await client.get("/api/v1/instances", headers=auth_header(admin_token))
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
items = data.get("items", data) if isinstance(data, dict) else data
|
|
for inst in items:
|
|
if inst.get("db_type") == "mysql":
|
|
return inst
|
|
pytest.skip("No MySQL instance found")
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def redis_instance(client, admin_token):
|
|
"""获取第一个 Redis 实例"""
|
|
resp = await client.get("/api/v1/instances", headers=auth_header(admin_token))
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
items = data.get("items", data) if isinstance(data, dict) else data
|
|
for inst in items:
|
|
if inst.get("db_type") == "redis":
|
|
return inst
|
|
pytest.skip("No Redis instance found")
|