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通过
84 lines
3.5 KiB
Python
84 lines
3.5 KiB
Python
"""Ticket API tests — CRUD + status filter"""
|
|
import pytest
|
|
from httpx import AsyncClient
|
|
def auth_header(token: str) -> dict:
|
|
return {"Authorization": f"Bearer {token}"}
|
|
|
|
|
|
class TestListTickets:
|
|
async def test_admin_sees_all(self, client: AsyncClient, admin_token):
|
|
resp = await client.get("/api/v1/tickets", headers=auth_header(admin_token))
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert "items" in data
|
|
assert "total" in data
|
|
|
|
async def test_developer_sees_own_and_approver(self, client: AsyncClient, developer_token):
|
|
"""developer 能看到自己提交的 + 自己作为审批人的工单"""
|
|
resp = await client.get("/api/v1/tickets", headers=auth_header(developer_token))
|
|
assert resp.status_code == 200
|
|
|
|
async def test_filter_by_status(self, client: AsyncClient, admin_token):
|
|
resp = await client.get("/api/v1/tickets?status=pending", headers=auth_header(admin_token))
|
|
assert resp.status_code == 200
|
|
items = resp.json().get("items", [])
|
|
for t in items:
|
|
assert t["status"] == "pending"
|
|
|
|
async def test_pagination(self, client: AsyncClient, admin_token):
|
|
resp = await client.get("/api/v1/tickets?page=1&page_size=5", headers=auth_header(admin_token))
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["page"] == 1
|
|
assert data["page_size"] == 5
|
|
|
|
|
|
class TestCreateTicket:
|
|
async def test_create_mysql_ticket(self, client: AsyncClient, developer_token, mysql_instance):
|
|
resp = await client.post("/api/v1/tickets", headers=auth_header(developer_token), json={
|
|
"title": "测试工单-CREATE TABLE",
|
|
"db_type": "mysql",
|
|
"instance_id": mysql_instance["id"],
|
|
"target_database": "opsxsql_test",
|
|
"sql_content": "CREATE TABLE IF NOT EXISTS _test_tbl (id INT)",
|
|
"workflow_id": 4,
|
|
})
|
|
assert resp.status_code in (200, 201)
|
|
data = resp.json()
|
|
assert data["status"] == "pending"
|
|
assert "ticket_no" in data
|
|
return data
|
|
|
|
async def test_reject_readonly_sql(self, client: AsyncClient, developer_token, mysql_instance):
|
|
"""SELECT/SHOW 等只读 SQL 应被拦截"""
|
|
resp = await client.post("/api/v1/tickets", headers=auth_header(developer_token), json={
|
|
"title": "只读SQL",
|
|
"db_type": "mysql",
|
|
"instance_id": mysql_instance["id"],
|
|
"target_database": "opsxsql_test",
|
|
"sql_content": "SELECT * FROM orders",
|
|
"workflow_id": 4,
|
|
})
|
|
assert resp.status_code == 400
|
|
|
|
async def test_auditor_cannot_create(self, client: AsyncClient):
|
|
"""auditor 角色不能提交工单"""
|
|
|
|
resp = await client.post("/api/v1/auth/login", json={
|
|
"username": "wangwu", "password": "admin123",
|
|
})
|
|
if resp.status_code != 200:
|
|
pytest.skip("auditor user not available")
|
|
token = resp.json()["access_token"]
|
|
resp2 = await client.post("/api/v1/tickets", headers=auth_header(token), json={
|
|
"title": "auditor提交",
|
|
"db_type": "mysql", "instance_id": 5,
|
|
"target_database": "opsxsql_test",
|
|
"sql_content": "CREATE TABLE _t(x INT)",
|
|
})
|
|
assert resp2.status_code in (403, 401)
|
|
|
|
async def test_no_auth(self, client: AsyncClient):
|
|
resp = await client.post("/api/v1/tickets", json={"title": "x"})
|
|
assert resp.status_code == 401
|