feat(auth): add admin user management dashboard
- Add UserProfile model (app/models.py) with per-user settings: display_name, daily_upload_limit, notes, is_blocked - Add Alembic migration 013_add_user_profiles for the new table - Add REST API at /api/admin/users/ with list, get, upsert (PUT), delete endpoints (admin-only) - Add HTML template admin_users.html with Alpine.js: filterable user list, paginated table, edit/create modal, delete confirmation modal - Add view handler at /admin/users (admin-only redirect guard) - Register routers in app/api/__init__.py and app/views/__init__.py - Add 'Users' link to admin nav dropdown in base.html (desktop + mobile) - Add 27 tests covering auth, list, get, upsert, delete, and model constraints - Register UserProfile in conftest.py model imports - Document new endpoints in docs/API.md Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com>
This commit is contained in:
copilot-swe-agent[bot]
@@ -0,0 +1,479 @@
|
||||
"""
|
||||
Tests for the admin user management API (/api/admin/users).
|
||||
|
||||
Covers:
|
||||
- Authentication enforcement (403 for non-admins)
|
||||
- List users (empty, with doc-only users, with profile-only users, with both)
|
||||
- Get single user detail
|
||||
- Create / update user profile via PUT (upsert)
|
||||
- Delete user profile
|
||||
- Pagination and search filtering
|
||||
"""
|
||||
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
import pytest
|
||||
from fastapi import HTTPException
|
||||
from fastapi.testclient import TestClient
|
||||
from sqlalchemy import create_engine
|
||||
from sqlalchemy.orm import sessionmaker
|
||||
from sqlalchemy.pool import StaticPool
|
||||
|
||||
from app.database import Base, get_db
|
||||
from app.models import FileRecord, UserProfile
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Fixtures
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def au_engine():
|
||||
"""In-memory SQLite engine for admin-user tests."""
|
||||
engine = create_engine(
|
||||
"sqlite:///:memory:",
|
||||
connect_args={"check_same_thread": False},
|
||||
poolclass=StaticPool,
|
||||
)
|
||||
Base.metadata.create_all(bind=engine)
|
||||
yield engine
|
||||
Base.metadata.drop_all(bind=engine)
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def au_session(au_engine):
|
||||
"""DB session scoped to one test."""
|
||||
Session = sessionmaker(bind=au_engine)
|
||||
session = Session()
|
||||
yield session
|
||||
session.close()
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def au_client(au_engine):
|
||||
"""TestClient that uses an in-memory DB and overrides _require_admin to allow access."""
|
||||
from app.api.admin_users import _require_admin
|
||||
from app.main import app
|
||||
|
||||
def override_db():
|
||||
Session = sessionmaker(bind=au_engine)
|
||||
session = Session()
|
||||
try:
|
||||
yield session
|
||||
finally:
|
||||
session.close()
|
||||
|
||||
def override_require_admin():
|
||||
return {"email": "admin@test.com", "is_admin": True, "name": "Admin"}
|
||||
|
||||
app.dependency_overrides[get_db] = override_db
|
||||
app.dependency_overrides[_require_admin] = override_require_admin
|
||||
with TestClient(app, base_url="http://localhost", raise_server_exceptions=False) as client:
|
||||
yield client
|
||||
app.dependency_overrides.clear()
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def au_client_nonadmin(au_engine):
|
||||
"""TestClient without admin override — _require_admin returns 403."""
|
||||
from app.main import app
|
||||
|
||||
def override_db():
|
||||
Session = sessionmaker(bind=au_engine)
|
||||
session = Session()
|
||||
try:
|
||||
yield session
|
||||
finally:
|
||||
session.close()
|
||||
|
||||
app.dependency_overrides[get_db] = override_db
|
||||
with TestClient(app, base_url="http://localhost", raise_server_exceptions=False) as client:
|
||||
yield client
|
||||
app.dependency_overrides.clear()
|
||||
|
||||
|
||||
def _make_file(session, owner_id: str, n: int = 1) -> list[FileRecord]:
|
||||
"""Insert *n* FileRecord rows for the given owner."""
|
||||
records = []
|
||||
for i in range(n):
|
||||
rec = FileRecord(
|
||||
filehash=f"hash-{owner_id}-{i}",
|
||||
original_filename=f"doc{i}.pdf",
|
||||
local_filename=f"/tmp/{owner_id}_{i}.pdf",
|
||||
file_size=1024,
|
||||
mime_type="application/pdf",
|
||||
is_duplicate=False,
|
||||
owner_id=owner_id,
|
||||
)
|
||||
session.add(rec)
|
||||
records.append(rec)
|
||||
session.commit()
|
||||
return records
|
||||
|
||||
|
||||
def _make_profile(session, user_id: str, **kwargs) -> UserProfile:
|
||||
"""Insert a UserProfile row."""
|
||||
kwargs.setdefault("is_blocked", False)
|
||||
profile = UserProfile(user_id=user_id, **kwargs)
|
||||
session.add(profile)
|
||||
session.commit()
|
||||
session.refresh(profile)
|
||||
return profile
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Authentication tests
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestAdminUsersAuth:
|
||||
"""Endpoints must be restricted to admin users."""
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_require_admin_raises_403_when_no_user(self):
|
||||
"""_require_admin raises 403 when no user in session."""
|
||||
from app.api.admin_users import _require_admin
|
||||
|
||||
mock_request = MagicMock()
|
||||
mock_request.session = {}
|
||||
|
||||
with pytest.raises(HTTPException) as exc_info:
|
||||
_require_admin(mock_request)
|
||||
assert exc_info.value.status_code == 403
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_require_admin_raises_403_for_non_admin(self):
|
||||
"""_require_admin raises 403 for a non-admin user."""
|
||||
from app.api.admin_users import _require_admin
|
||||
|
||||
mock_request = MagicMock()
|
||||
mock_request.session = {"user": {"email": "user@test.com", "is_admin": False}}
|
||||
|
||||
with pytest.raises(HTTPException) as exc_info:
|
||||
_require_admin(mock_request)
|
||||
assert exc_info.value.status_code == 403
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_require_admin_returns_user_for_admin(self):
|
||||
"""_require_admin returns the user dict for an admin."""
|
||||
from app.api.admin_users import _require_admin
|
||||
|
||||
mock_request = MagicMock()
|
||||
user = {"email": "admin@test.com", "is_admin": True}
|
||||
mock_request.session = {"user": user}
|
||||
|
||||
result = _require_admin(mock_request)
|
||||
assert result == user
|
||||
|
||||
@pytest.mark.integration
|
||||
def test_list_users_requires_admin(self, au_client_nonadmin):
|
||||
"""GET /api/admin/users/ returns 403 for non-admins."""
|
||||
resp = au_client_nonadmin.get("/api/admin/users/")
|
||||
assert resp.status_code == 403
|
||||
|
||||
@pytest.mark.integration
|
||||
def test_put_user_requires_admin(self, au_client_nonadmin):
|
||||
"""PUT /api/admin/users/<id> returns 403 for non-admins."""
|
||||
resp = au_client_nonadmin.put(
|
||||
"/api/admin/users/user@example.com",
|
||||
json={"is_blocked": False},
|
||||
)
|
||||
assert resp.status_code == 403
|
||||
|
||||
@pytest.mark.integration
|
||||
def test_delete_user_requires_admin(self, au_client_nonadmin):
|
||||
"""DELETE /api/admin/users/<id> returns 403 for non-admins."""
|
||||
resp = au_client_nonadmin.delete("/api/admin/users/user@example.com")
|
||||
assert resp.status_code == 403
|
||||
|
||||
@pytest.mark.integration
|
||||
def test_get_user_requires_admin(self, au_client_nonadmin):
|
||||
"""GET /api/admin/users/<id> returns 403 for non-admins."""
|
||||
resp = au_client_nonadmin.get("/api/admin/users/user@example.com")
|
||||
assert resp.status_code == 403
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# List endpoint
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestListUsers:
|
||||
"""Tests for GET /api/admin/users/."""
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_empty_returns_empty_list(self, au_client):
|
||||
"""No users → empty list with total=0."""
|
||||
resp = au_client.get("/api/admin/users/")
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["users"] == []
|
||||
assert data["total"] == 0
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_lists_users_with_docs(self, au_client, au_session):
|
||||
"""Users who have documents appear in the list."""
|
||||
_make_file(au_session, "alice@example.com", 3)
|
||||
_make_file(au_session, "bob@example.com", 1)
|
||||
|
||||
resp = au_client.get("/api/admin/users/")
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["total"] == 2
|
||||
|
||||
ids = {u["user_id"] for u in data["users"]}
|
||||
assert "alice@example.com" in ids
|
||||
assert "bob@example.com" in ids
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_document_count_correct(self, au_client, au_session):
|
||||
"""document_count reflects the number of files owned by each user."""
|
||||
_make_file(au_session, "carol@example.com", 5)
|
||||
|
||||
resp = au_client.get("/api/admin/users/")
|
||||
assert resp.status_code == 200
|
||||
carol = next(u for u in resp.json()["users"] if u["user_id"] == "carol@example.com")
|
||||
assert carol["document_count"] == 5
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_lists_profile_only_users(self, au_client, au_session):
|
||||
"""Users with a profile but no documents still appear."""
|
||||
_make_profile(au_session, "profileonly@example.com", display_name="Profile Only")
|
||||
|
||||
resp = au_client.get("/api/admin/users/")
|
||||
assert resp.status_code == 200
|
||||
ids = {u["user_id"] for u in resp.json()["users"]}
|
||||
assert "profileonly@example.com" in ids
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_search_filter(self, au_client, au_session):
|
||||
"""q= parameter filters by user_id substring."""
|
||||
_make_file(au_session, "alice@example.com")
|
||||
_make_file(au_session, "bob@example.com")
|
||||
|
||||
resp = au_client.get("/api/admin/users/?q=alice")
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["total"] == 1
|
||||
assert data["users"][0]["user_id"] == "alice@example.com"
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_pagination(self, au_client, au_session):
|
||||
"""per_page and page parameters paginate results."""
|
||||
for i in range(6):
|
||||
_make_file(au_session, f"user{i:02d}@example.com")
|
||||
|
||||
resp = au_client.get("/api/admin/users/?page=1&per_page=3")
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert len(data["users"]) == 3
|
||||
assert data["total"] == 6
|
||||
assert data["pages"] == 2
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_profile_data_merged(self, au_client, au_session):
|
||||
"""Profile fields (is_blocked, daily_upload_limit, …) are merged into list items."""
|
||||
_make_file(au_session, "managed@example.com")
|
||||
_make_profile(
|
||||
au_session,
|
||||
"managed@example.com",
|
||||
display_name="Managed User",
|
||||
daily_upload_limit=10,
|
||||
is_blocked=True,
|
||||
)
|
||||
|
||||
resp = au_client.get("/api/admin/users/")
|
||||
assert resp.status_code == 200
|
||||
user = next(u for u in resp.json()["users"] if u["user_id"] == "managed@example.com")
|
||||
assert user["display_name"] == "Managed User"
|
||||
assert user["daily_upload_limit"] == 10
|
||||
assert user["is_blocked"] is True
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Get single user
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestGetUser:
|
||||
"""Tests for GET /api/admin/users/{user_id}."""
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_get_user_with_docs(self, au_client, au_session):
|
||||
"""Returns correct document_count and last_upload."""
|
||||
_make_file(au_session, "dana@example.com", 2)
|
||||
|
||||
resp = au_client.get("/api/admin/users/dana@example.com")
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["user_id"] == "dana@example.com"
|
||||
assert data["document_count"] == 2
|
||||
assert data["is_blocked"] is False
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_get_user_with_profile(self, au_client, au_session):
|
||||
"""Returns profile data when a profile exists."""
|
||||
_make_file(au_session, "evan@example.com")
|
||||
_make_profile(au_session, "evan@example.com", notes="VIP user", daily_upload_limit=50)
|
||||
|
||||
resp = au_client.get("/api/admin/users/evan@example.com")
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["notes"] == "VIP user"
|
||||
assert data["daily_upload_limit"] == 50
|
||||
assert data["profile"] is not None
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_get_user_no_docs_no_profile_returns_defaults(self, au_client):
|
||||
"""User with no docs and no profile returns zero counts and defaults."""
|
||||
resp = au_client.get("/api/admin/users/unknown@example.com")
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["document_count"] == 0
|
||||
assert data["profile"] is None
|
||||
assert data["is_blocked"] is False
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Upsert (PUT) endpoint
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestUpsertUserProfile:
|
||||
"""Tests for PUT /api/admin/users/{user_id}."""
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_create_profile(self, au_client, au_session):
|
||||
"""PUT on a user without a profile creates it."""
|
||||
resp = au_client.put(
|
||||
"/api/admin/users/newuser@example.com",
|
||||
json={"display_name": "New User", "daily_upload_limit": 20, "is_blocked": False},
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["user_id"] == "newuser@example.com"
|
||||
assert data["display_name"] == "New User"
|
||||
assert data["daily_upload_limit"] == 20
|
||||
|
||||
# Persisted in DB
|
||||
profile = au_session.query(UserProfile).filter_by(user_id="newuser@example.com").first()
|
||||
assert profile is not None
|
||||
assert profile.display_name == "New User"
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_update_existing_profile(self, au_client, au_session):
|
||||
"""PUT on an existing profile updates it."""
|
||||
_make_profile(au_session, "existing@example.com", display_name="Old Name")
|
||||
|
||||
resp = au_client.put(
|
||||
"/api/admin/users/existing@example.com",
|
||||
json={"display_name": "New Name", "is_blocked": True},
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["display_name"] == "New Name"
|
||||
assert data["is_blocked"] is True
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_block_user(self, au_client, au_session):
|
||||
"""Setting is_blocked=True stores correctly."""
|
||||
resp = au_client.put(
|
||||
"/api/admin/users/blocked@example.com",
|
||||
json={"is_blocked": True},
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
assert resp.json()["is_blocked"] is True
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_null_upload_limit(self, au_client, au_session):
|
||||
"""daily_upload_limit can be null (use global default)."""
|
||||
resp = au_client.put(
|
||||
"/api/admin/users/nulllimit@example.com",
|
||||
json={"daily_upload_limit": None, "is_blocked": False},
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
assert resp.json()["daily_upload_limit"] is None
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_zero_upload_limit_means_unlimited(self, au_client):
|
||||
"""daily_upload_limit=0 is a valid value meaning 'unlimited'."""
|
||||
resp = au_client.put(
|
||||
"/api/admin/users/zerolimit@example.com",
|
||||
json={"daily_upload_limit": 0, "is_blocked": False},
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
assert resp.json()["daily_upload_limit"] == 0
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Delete endpoint
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestDeleteUserProfile:
|
||||
"""Tests for DELETE /api/admin/users/{user_id}."""
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_delete_existing_profile(self, au_client, au_session):
|
||||
"""DELETE removes an existing profile; returns 204."""
|
||||
_make_profile(au_session, "todelete@example.com")
|
||||
|
||||
resp = au_client.delete("/api/admin/users/todelete@example.com")
|
||||
assert resp.status_code == 204
|
||||
|
||||
remaining = au_session.query(UserProfile).filter_by(user_id="todelete@example.com").first()
|
||||
assert remaining is None
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_delete_nonexistent_profile_returns_404(self, au_client):
|
||||
"""DELETE on unknown user_id returns 404."""
|
||||
resp = au_client.delete("/api/admin/users/doesnotexist@example.com")
|
||||
assert resp.status_code == 404
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_delete_profile_does_not_remove_documents(self, au_client, au_session):
|
||||
"""Deleting a profile must not remove documents owned by that user."""
|
||||
_make_file(au_session, "hasfiles@example.com", 3)
|
||||
_make_profile(au_session, "hasfiles@example.com")
|
||||
|
||||
resp = au_client.delete("/api/admin/users/hasfiles@example.com")
|
||||
assert resp.status_code == 204
|
||||
|
||||
doc_count = au_session.query(FileRecord).filter_by(owner_id="hasfiles@example.com").count()
|
||||
assert doc_count == 3
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Model tests
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestUserProfileModel:
|
||||
"""Unit tests for the UserProfile SQLAlchemy model."""
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_user_profile_has_required_columns(self, au_session):
|
||||
"""UserProfile can be created with minimal required fields."""
|
||||
profile = UserProfile(user_id="test@example.com", is_blocked=False)
|
||||
au_session.add(profile)
|
||||
au_session.commit()
|
||||
au_session.refresh(profile)
|
||||
|
||||
assert profile.id is not None
|
||||
assert profile.user_id == "test@example.com"
|
||||
assert profile.is_blocked is False
|
||||
assert profile.display_name is None
|
||||
assert profile.daily_upload_limit is None
|
||||
assert profile.notes is None
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_user_profile_unique_user_id(self, au_session):
|
||||
"""Two profiles with the same user_id should raise an integrity error."""
|
||||
from sqlalchemy.exc import IntegrityError
|
||||
|
||||
au_session.add(UserProfile(user_id="dup@example.com", is_blocked=False))
|
||||
au_session.commit()
|
||||
au_session.add(UserProfile(user_id="dup@example.com", is_blocked=False))
|
||||
with pytest.raises(IntegrityError):
|
||||
au_session.commit()
|
||||
au_session.rollback()
|
||||
Reference in New Issue
Block a user