Files
opsxsql/backend/tests/test_users.py
T
loganandClaude Opus 4.7 c0c731acfc feat: TDD Phase 1 — 44 backend API tests passing
Test infrastructure:
- conftest.py with SQLite in-memory DB, isolated per-test cleanup
- pytest.ini with asyncio auto mode
- Fixtures: admin/dba/developer/guest/disabled users, tokens, test instance

Test coverage (44 tests):
- test_auth: login/logout/me/token expiry (8 tests)
- test_users: CRUD + role permissions (8 tests)
- test_instances: CRUD + connection test mock (8 tests)
- test_tickets: CRUD + review/execute workflow + pagination (11 tests)
- test_query: select/drop/delete rejection + guest deny (4 tests)
- test_dashboard: stats + auth check (2 tests)

Backend fix:
- query endpoint now catches ValueError from QueryService → 400 instead of 500

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
2026-05-22 11:34:47 +08:00

78 lines
2.9 KiB
Python

import pytest
from httpx import AsyncClient
from tests.conftest import auth_header
class TestListUsers:
async def test_admin_list_users(self, client: AsyncClient, admin_token):
resp = await client.get("/api/v1/users", headers=auth_header(admin_token))
assert resp.status_code == 200
assert isinstance(resp.json(), list)
async def test_non_admin_list_users(self, client: AsyncClient, developer_token):
resp = await client.get("/api/v1/users", headers=auth_header(developer_token))
assert resp.status_code == 403
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": "123456",
"role": "developer",
})
assert resp.status_code == 201
data = resp.json()
assert data["username"] == "newuser"
assert data["role"] == "developer"
assert "id" in data
async def test_create_duplicate_username(self, client: AsyncClient, admin_user, admin_token):
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
class TestGetUser:
async def test_get_single_user(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_nonexistent_user(self, client: AsyncClient, admin_token):
resp = await client.get("/api/v1/users/9999", headers=auth_header(admin_token))
assert resp.status_code == 404
class TestUpdateUser:
async def test_update_user(self, client: AsyncClient, admin_user, admin_token):
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"] == "管理员改名"
class TestDeleteUser:
async def test_delete_user(self, client: AsyncClient, admin_user, 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
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