test(server): add comprehensive test suite (47 tests), fix DB lock and schema bugs
- Add pytest test suite: test_auth.py, test_admin.py, test_pass.py, test_proxy.py, test_rate_limit.py, test_client_contract.py - Fix SQLite 'database is locked' errors: moved log_audit() calls outside with get_db() blocks in register, login, logout, refresh, activate_pass - Enable WAL mode and busy_timeout in get_db() for concurrent access - Fix /admin/me: removed non-existent 'email' column from query - Fix /admin/users list: disambiguated activated_at column in JOIN query - Fix /auth/refresh: now returns refresh_token + expires_in + username/uuid/role to match AuthManager.AuthSession expectations; revokes old refresh token - Fix conftest.py: unique usernames per test to avoid conflicts - All 47 tests passing
This commit is contained in:
@@ -0,0 +1,128 @@
|
||||
"""Tests for admin endpoints."""
|
||||
import pytest
|
||||
import sqlite3
|
||||
import time
|
||||
from tests.conftest import auth_headers
|
||||
from auth import AUTH_DB
|
||||
|
||||
|
||||
class TestAdminMe:
|
||||
"""Test /admin/me endpoint."""
|
||||
|
||||
def test_admin_me_success(self, client, logged_in_user):
|
||||
resp = client.get("/admin/me", headers=auth_headers(logged_in_user["access_token"]))
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert "id" in data
|
||||
assert "username" in data
|
||||
assert "uuid" in data
|
||||
assert "role" in data
|
||||
assert "role_name" in data
|
||||
assert "has_pass" in data
|
||||
assert "permissions" in data
|
||||
|
||||
def test_admin_me_no_auth(self, client):
|
||||
resp = client.get("/admin/me")
|
||||
assert resp.status_code in (401, 403) # Either is acceptable
|
||||
|
||||
|
||||
class TestAdminUsersList:
|
||||
"""Test /admin/users endpoint."""
|
||||
|
||||
def test_admin_users_list(self, client, admin_user):
|
||||
resp = client.get("/admin/users", headers=auth_headers(admin_user["access_token"]))
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert "users" in data
|
||||
assert isinstance(data["users"], list)
|
||||
assert len(data["users"]) >= 1 # At least the admin user
|
||||
|
||||
def test_admin_users_list_no_admin(self, client, logged_in_user):
|
||||
"""Regular user should not access admin endpoints."""
|
||||
resp = client.get("/admin/users", headers=auth_headers(logged_in_user["access_token"]))
|
||||
assert resp.status_code in (401, 403)
|
||||
|
||||
def test_admin_users_list_no_auth(self, client):
|
||||
resp = client.get("/admin/users")
|
||||
assert resp.status_code in (401, 403)
|
||||
|
||||
|
||||
class TestAdminBan:
|
||||
"""Test ban functionality via admin endpoints."""
|
||||
|
||||
def test_ban_user(self, client, logged_in_user, admin_user):
|
||||
"""Admin bans a user."""
|
||||
# Get user ID first
|
||||
import sqlite3
|
||||
from auth import AUTH_DB
|
||||
conn = sqlite3.connect(str(AUTH_DB))
|
||||
row = conn.execute("SELECT id FROM users WHERE username = ?",
|
||||
(logged_in_user["username"],)).fetchone()
|
||||
conn.close()
|
||||
assert row is not None
|
||||
|
||||
resp = client.post("/admin/user/ban", json={
|
||||
"user_id": row[0],
|
||||
"days": 1,
|
||||
"reason": "Test ban"
|
||||
}, headers=auth_headers(admin_user["access_token"]))
|
||||
assert resp.status_code == 200
|
||||
|
||||
# Verify ban in DB
|
||||
conn = sqlite3.connect(str(AUTH_DB))
|
||||
row = conn.execute("SELECT banned_until FROM users WHERE username = ?",
|
||||
(logged_in_user["username"],)).fetchone()
|
||||
conn.close()
|
||||
assert row is not None
|
||||
assert row[0] is not None
|
||||
assert row[0] > time.time()
|
||||
|
||||
def test_ban_nonexistent_user(self, client, admin_user):
|
||||
resp = client.post("/admin/user/ban", json={
|
||||
"user_id": 99999,
|
||||
"days": 1,
|
||||
"reason": "Test ban"
|
||||
}, headers=auth_headers(admin_user["access_token"]))
|
||||
assert resp.status_code == 404
|
||||
|
||||
|
||||
class TestAdminRole:
|
||||
"""Test role change functionality."""
|
||||
|
||||
def test_change_role(self, client, logged_in_user, admin_user):
|
||||
# Get user ID
|
||||
import sqlite3
|
||||
from auth import AUTH_DB
|
||||
conn = sqlite3.connect(str(AUTH_DB))
|
||||
row = conn.execute("SELECT id FROM users WHERE username = ?",
|
||||
(logged_in_user["username"],)).fetchone()
|
||||
conn.close()
|
||||
assert row is not None
|
||||
|
||||
resp = client.put(f"/admin/users/{row[0]}/role", json={
|
||||
"user_id": row[0],
|
||||
"role": 2 # MODERATOR
|
||||
}, headers=auth_headers(admin_user["access_token"]))
|
||||
assert resp.status_code == 200
|
||||
|
||||
# Verify in DB
|
||||
conn = sqlite3.connect(str(AUTH_DB))
|
||||
row = conn.execute("SELECT role FROM users WHERE username = ?",
|
||||
(logged_in_user["username"],)).fetchone()
|
||||
conn.close()
|
||||
assert row[0] == 2
|
||||
|
||||
def test_change_role_invalid(self, client, logged_in_user, admin_user):
|
||||
import sqlite3
|
||||
from auth import AUTH_DB
|
||||
conn = sqlite3.connect(str(AUTH_DB))
|
||||
row = conn.execute("SELECT id FROM users WHERE username = ?",
|
||||
(logged_in_user["username"],)).fetchone()
|
||||
conn.close()
|
||||
assert row is not None
|
||||
|
||||
resp = client.put(f"/admin/users/{row[0]}/role", json={
|
||||
"user_id": row[0],
|
||||
"role": 99
|
||||
}, headers=auth_headers(admin_user["access_token"]))
|
||||
assert resp.status_code in (400, 422)
|
||||
Reference in New Issue
Block a user