Merge pull request #488 from christianlouis/copilot/add-user-admin-dashboard

feat(auth): add admin user management dashboard
This commit is contained in:
Christian Krakau-Louis
2026-03-06 16:28:33 +01:00
committed by GitHub
11 changed files with 1515 additions and 1 deletions
+2
View File
@@ -6,6 +6,7 @@ import logging
from fastapi import APIRouter
from app.api.admin_users import router as admin_users_router
from app.api.azure import router as azure_router
from app.api.database import router as database_router
from app.api.diagnostic import router as diagnostic_router
@@ -35,6 +36,7 @@ logger = logging.getLogger(__name__)
router = APIRouter()
# Include all the routers
router.include_router(admin_users_router)
router.include_router(user_router)
router.include_router(files_router)
router.include_router(process_router)
+259
View File
@@ -0,0 +1,259 @@
"""API endpoints for admin user management.
Provides CRUD operations for user profiles and aggregate statistics so that
administrators can inspect, configure, and manage users in multi-user mode.
"""
import logging
from typing import Annotated, Any
from fastapi import APIRouter, Depends, HTTPException, Query, Request, status
from pydantic import BaseModel, Field
from sqlalchemy import func
from sqlalchemy.orm import Session
from app.database import get_db
from app.models import FileRecord, UserProfile
logger = logging.getLogger(__name__)
router = APIRouter(prefix="/admin/users", tags=["admin-users"])
DbSession = Annotated[Session, Depends(get_db)]
# ---------------------------------------------------------------------------
# Auth helper
# ---------------------------------------------------------------------------
def _require_admin(request: Request) -> dict:
"""Ensure the caller is an admin. Raises 403 otherwise."""
user = request.session.get("user")
if not user or not user.get("is_admin"):
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Admin access required")
return user
AdminUser = Annotated[dict, Depends(_require_admin)]
# ---------------------------------------------------------------------------
# Pydantic schemas
# ---------------------------------------------------------------------------
class UserProfileUpsert(BaseModel):
"""Body for creating or updating a user profile."""
display_name: str | None = Field(default=None, max_length=255, description="Human-readable display name")
daily_upload_limit: int | None = Field(
default=None, ge=0, description="Per-user daily upload cap; null = use global default"
)
notes: str | None = Field(default=None, max_length=4096, description="Admin notes about this user")
is_blocked: bool = Field(default=False, description="Block this user from uploading")
class UserProfileResponse(BaseModel):
"""Response schema for a user profile record."""
id: int
user_id: str
display_name: str | None
daily_upload_limit: int | None
notes: str | None
is_blocked: bool
created_at: str | None
updated_at: str | None
model_config = {"from_attributes": True}
class UserSummary(BaseModel):
"""Per-user summary combining profile data with document statistics."""
user_id: str
display_name: str | None
daily_upload_limit: int | None
notes: str | None
is_blocked: bool
profile_id: int | None
document_count: int
last_upload: str | None
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _get_or_none(db: Session, user_id: str) -> UserProfile | None:
"""Return the UserProfile row for *user_id*, or None if it doesn't exist."""
return db.query(UserProfile).filter(UserProfile.user_id == user_id).first()
def _profile_to_dict(profile: UserProfile) -> dict[str, Any]:
return {
"id": profile.id,
"user_id": profile.user_id,
"display_name": profile.display_name,
"daily_upload_limit": profile.daily_upload_limit,
"notes": profile.notes,
"is_blocked": profile.is_blocked,
"created_at": profile.created_at.isoformat() if profile.created_at else None,
"updated_at": profile.updated_at.isoformat() if profile.updated_at else None,
}
# ---------------------------------------------------------------------------
# Endpoints
# ---------------------------------------------------------------------------
@router.get("/", summary="List all known users with statistics")
def list_users(
db: DbSession,
_admin: AdminUser,
q: str = Query("", description="Filter by user_id substring (case-insensitive)"),
page: int = Query(1, ge=1, description="Page number"),
per_page: int = Query(25, ge=1, le=100, description="Items per page"),
) -> dict[str, Any]:
"""Return every distinct user_id that has at least one document or an explicit profile,
enriched with aggregate document statistics and the admin-managed profile.
Supports substring filtering (``q``) and pagination.
"""
# 1. Collect every distinct owner_id from documents
doc_stats_query = (
db.query(
FileRecord.owner_id.label("user_id"),
func.count(FileRecord.id).label("doc_count"),
func.max(FileRecord.created_at).label("last_upload"),
)
.filter(FileRecord.owner_id.isnot(None))
.group_by(FileRecord.owner_id)
)
# 2. Collect all user_ids that have explicit profiles (may not have docs yet)
profile_query = db.query(UserProfile)
# Build a unified set of user_ids
doc_rows = {row.user_id: row for row in doc_stats_query.all()}
profile_rows = {p.user_id: p for p in profile_query.all()}
all_user_ids = set(doc_rows.keys()) | set(profile_rows.keys())
# Apply optional substring filter
if q.strip():
q_lower = q.strip().lower()
all_user_ids = {uid for uid in all_user_ids if q_lower in uid.lower()}
# Sort and paginate
sorted_ids = sorted(all_user_ids)
total = len(sorted_ids)
start = (page - 1) * per_page
page_ids = sorted_ids[start : start + per_page]
users: list[dict[str, Any]] = []
for uid in page_ids:
doc_row = doc_rows.get(uid)
profile = profile_rows.get(uid)
users.append(
{
"user_id": uid,
"display_name": profile.display_name if profile else None,
"daily_upload_limit": profile.daily_upload_limit if profile else None,
"notes": profile.notes if profile else None,
"is_blocked": profile.is_blocked if profile else False,
"profile_id": profile.id if profile else None,
"document_count": doc_row.doc_count if doc_row else 0,
"last_upload": doc_row.last_upload.isoformat() if (doc_row and doc_row.last_upload) else None,
}
)
return {
"users": users,
"total": total,
"page": page,
"per_page": per_page,
"pages": max(1, (total + per_page - 1) // per_page),
}
@router.get("/{user_id:path}", summary="Get details for a single user")
def get_user(user_id: str, db: DbSession, _admin: AdminUser) -> dict[str, Any]:
"""Return profile and document statistics for a specific user."""
doc_count = db.query(func.count(FileRecord.id)).filter(FileRecord.owner_id == user_id).scalar() or 0
last_row = (
db.query(FileRecord.created_at)
.filter(FileRecord.owner_id == user_id)
.order_by(FileRecord.created_at.desc())
.first()
)
last_upload = last_row[0].isoformat() if last_row and last_row[0] else None
profile = _get_or_none(db, user_id)
return {
"user_id": user_id,
"display_name": profile.display_name if profile else None,
"daily_upload_limit": profile.daily_upload_limit if profile else None,
"notes": profile.notes if profile else None,
"is_blocked": profile.is_blocked if profile else False,
"profile_id": profile.id if profile else None,
"document_count": doc_count,
"last_upload": last_upload,
"profile": _profile_to_dict(profile) if profile else None,
}
@router.put("/{user_id:path}", summary="Create or update a user profile")
def upsert_user_profile(
user_id: str,
body: UserProfileUpsert,
db: DbSession,
_admin: AdminUser,
) -> dict[str, Any]:
"""Create a new profile or update an existing one for *user_id*.
Returns the persisted profile.
"""
profile = _get_or_none(db, user_id)
if profile is None:
profile = UserProfile(user_id=user_id)
db.add(profile)
profile.display_name = body.display_name
profile.daily_upload_limit = body.daily_upload_limit
profile.notes = body.notes
profile.is_blocked = body.is_blocked
try:
db.commit()
db.refresh(profile)
except Exception:
db.rollback()
raise
logger.info("Admin upserted profile for user %s", user_id)
return _profile_to_dict(profile)
@router.delete("/{user_id:path}", status_code=status.HTTP_204_NO_CONTENT, summary="Delete a user profile")
def delete_user_profile(user_id: str, db: DbSession, _admin: AdminUser) -> None:
"""Delete the admin-managed profile for *user_id*.
Documents owned by this user are **not** removed; only the profile record
is deleted. To reassign or purge documents use the files API.
"""
profile = _get_or_none(db, user_id)
if not profile:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User profile not found")
try:
db.delete(profile)
db.commit()
except Exception:
db.rollback()
raise
logger.info("Admin deleted profile for user %s", user_id)
+32
View File
@@ -170,3 +170,35 @@ class WebhookConfig(Base):
description = Column(String, nullable=True) # Optional human-readable description
created_at = Column(DateTime(timezone=True), server_default=func.now())
updated_at = Column(DateTime(timezone=True), server_default=func.now(), onupdate=func.now())
class UserProfile(Base):
"""Per-user profile for admin-managed settings in multi-user mode.
Each row corresponds to one authenticated user (identified by their
``user_id``, which matches ``FileRecord.owner_id``). The admin can
create or update profiles to override global defaults such as the
daily upload limit and to attach notes or block a user.
"""
__tablename__ = "user_profiles"
id = Column(Integer, primary_key=True, index=True)
# Stable user identifier — matches FileRecord.owner_id (OAuth sub / email / username)
user_id = Column(String, unique=True, nullable=False, index=True)
# Optional human-readable display name set by the admin
display_name = Column(String, nullable=True)
# Per-user daily upload limit; NULL means "use global default"
daily_upload_limit = Column(Integer, nullable=True)
# Admin-only free-text notes about this user
notes = Column(Text, nullable=True)
# When True the user is prevented from uploading new documents
is_blocked = Column(Boolean, default=False, nullable=False)
created_at = Column(DateTime(timezone=True), server_default=func.now())
updated_at = Column(DateTime(timezone=True), server_default=func.now(), onupdate=func.now())
+2
View File
@@ -4,6 +4,7 @@ Aggregated view routers for the application.
from fastapi import APIRouter
from app.views.admin_users import router as admin_users_router
from app.views.db_wizard import router as db_wizard_router
from app.views.dropbox import router as dropbox_router
from app.views.filemanager import router as filemanager_router
@@ -23,6 +24,7 @@ from app.views.wizard import router as wizard_router
router = APIRouter()
router.include_router(wizard_router) # Wizard first (for /setup)
router.include_router(db_wizard_router) # Database wizard
router.include_router(admin_users_router) # Admin user management
router.include_router(general_router)
router.include_router(status_router)
router.include_router(onedrive_router)
+44
View File
@@ -0,0 +1,44 @@
"""Admin view: user management dashboard."""
import logging
from fastapi import HTTPException, Request, status
from fastapi.responses import RedirectResponse
from app.views.base import APIRouter, get_db, require_login, settings, templates # noqa: F401
logger = logging.getLogger(__name__)
router = APIRouter()
def _require_admin(request: Request):
"""Return the session user if they are an admin, else redirect."""
user = request.session.get("user")
if not user or not user.get("is_admin"):
logger.warning("Non-admin user attempted to access /admin/users")
return None
return user
@router.get("/admin/users")
@require_login
async def admin_users_page(request: Request):
"""Admin user management dashboard — lists all known users."""
user = _require_admin(request)
if user is None:
return RedirectResponse(url="/", status_code=status.HTTP_302_FOUND)
try:
return templates.TemplateResponse(
"admin_users.html",
{
"request": request,
"app_version": settings.version,
},
)
except Exception as e:
logger.error(f"Error loading admin users page: {e}")
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail="Failed to load admin users page",
)
+84
View File
@@ -728,6 +728,90 @@ curl "http://<your-instance>/api/users/search?q=risti&limit=5"
---
### Admin User Management
**Admin only.** These endpoints let administrators list all known users, view per-user statistics,
and manage per-user settings such as custom upload limits, display names, and blocked status.
---
**GET** `/api/admin/users/`
List all known users — anyone who has uploaded a document or has an explicit profile.
Returns aggregate document statistics merged with profile data.
**Query Parameters**:
- `q` (optional): Substring filter on user ID (case-insensitive)
- `page` (optional): Page number (default: 1)
- `per_page` (optional): Items per page (default: 25, max: 100)
```bash
curl "http://<your-instance>/api/admin/users/" \
-H "Cookie: session=<admin-session>"
```
**Response**:
```json
{
"users": [
{
"user_id": "alice@example.com",
"display_name": "Alice Smith",
"daily_upload_limit": 50,
"notes": null,
"is_blocked": false,
"profile_id": 1,
"document_count": 42,
"last_upload": "2026-02-15T10:23:00"
}
],
"total": 1,
"page": 1,
"per_page": 25,
"pages": 1
}
```
---
**GET** `/api/admin/users/{user_id}`
Return profile and document statistics for a specific user.
```bash
curl "http://<your-instance>/api/admin/users/alice%40example.com"
```
---
**PUT** `/api/admin/users/{user_id}`
Create or update the admin-managed profile for a user. If no profile exists one is created.
**Request body**:
```json
{
"display_name": "Alice Smith",
"daily_upload_limit": 50,
"notes": "VIP customer",
"is_blocked": false
}
```
- `display_name` (optional): Human-readable name shown in the admin UI
- `daily_upload_limit` (optional): Per-user daily cap; `null` = use global default; `0` = unlimited
- `notes` (optional): Admin-only text notes
- `is_blocked`: When `true`, blocks new uploads from this user
---
**DELETE** `/api/admin/users/{user_id}`
Delete the admin-managed profile for a user. Documents owned by the user are **not** removed.
Returns `204 No Content` on success, `404` if no profile exists.
---
### Settings Suggestions (Autocomplete)
**GET** `/api/settings/{key}/suggestions`
+557
View File
@@ -0,0 +1,557 @@
{% extends "base.html" %}
{% block title %}User Management Admin DocuElevate{% endblock %}
{% block content %}
<div class="container mx-auto px-4 py-8" x-data="adminUsersApp()">
<!-- ── Header ─────────────────────────────────────────────────────────────── -->
<div class="mb-6 flex flex-col sm:flex-row sm:items-center sm:justify-between gap-3">
<div>
<h1 class="text-3xl font-bold text-gray-900 flex items-center gap-2">
<i class="fas fa-users text-blue-500" aria-hidden="true"></i>
User Management
<span class="text-xs font-semibold text-red-600 bg-red-50 border border-red-200 rounded px-2 py-0.5 ml-1">Admin Only</span>
</h1>
<p class="text-gray-500 text-sm mt-1">
Manage user profiles, per-user upload limits, and document ownership.
</p>
</div>
<button
type="button"
@click="openCreateModal()"
class="inline-flex items-center px-4 py-2 bg-blue-600 hover:bg-blue-700 text-white text-sm font-medium rounded-md shadow-sm focus:outline-none focus:ring-2 focus:ring-offset-2 focus:ring-blue-500"
>
<i class="fas fa-user-plus mr-2" aria-hidden="true"></i> Add User Profile
</button>
</div>
<!-- ── Alert banner ───────────────────────────────────────────────────────── -->
<div x-show="alert.show" x-transition class="mb-4" role="alert" :aria-live="alert.type === 'error' ? 'assertive' : 'polite'">
<div
:class="alert.type === 'success'
? 'bg-green-50 border-green-400 text-green-800'
: 'bg-red-50 border-red-400 text-red-800'"
class="border-l-4 p-4 rounded"
>
<p class="font-semibold" x-text="alert.title"></p>
<p class="text-sm" x-text="alert.message"></p>
</div>
</div>
<!-- ── Search & pagination controls ──────────────────────────────────────── -->
<div class="mb-4 flex flex-col sm:flex-row gap-3 items-start sm:items-center justify-between">
<div class="flex-1 max-w-sm">
<label for="userSearch" class="sr-only">Search users</label>
<div class="relative">
<span class="absolute inset-y-0 left-0 pl-3 flex items-center pointer-events-none">
<i class="fas fa-search text-gray-400 text-sm" aria-hidden="true"></i>
</span>
<input
id="userSearch"
type="search"
x-model="search"
@input.debounce.300ms="fetchUsers(1)"
placeholder="Filter by user ID…"
class="w-full pl-9 pr-3 py-2 border border-gray-300 rounded-md text-sm focus:outline-none focus:ring-2 focus:ring-blue-400"
/>
</div>
</div>
<div class="text-sm text-gray-500" x-text="totalLabel"></div>
</div>
<!-- ── Table ──────────────────────────────────────────────────────────────── -->
<div class="bg-white shadow rounded-lg overflow-x-auto">
<table class="min-w-full divide-y divide-gray-200" aria-label="User list">
<thead class="bg-gray-50">
<tr>
<th scope="col" class="px-4 py-3 text-left text-xs font-medium text-gray-500 uppercase tracking-wider">User ID</th>
<th scope="col" class="px-4 py-3 text-left text-xs font-medium text-gray-500 uppercase tracking-wider">Display Name</th>
<th scope="col" class="px-4 py-3 text-center text-xs font-medium text-gray-500 uppercase tracking-wider">Documents</th>
<th scope="col" class="px-4 py-3 text-left text-xs font-medium text-gray-500 uppercase tracking-wider">Last Upload</th>
<th scope="col" class="px-4 py-3 text-center text-xs font-medium text-gray-500 uppercase tracking-wider">Upload Limit</th>
<th scope="col" class="px-4 py-3 text-center text-xs font-medium text-gray-500 uppercase tracking-wider">Status</th>
<th scope="col" class="px-4 py-3 text-right text-xs font-medium text-gray-500 uppercase tracking-wider">Actions</th>
</tr>
</thead>
<tbody class="bg-white divide-y divide-gray-200">
<template x-if="loading">
<tr>
<td colspan="7" class="px-4 py-8 text-center text-gray-400">
<i class="fas fa-spinner fa-spin mr-2" aria-hidden="true"></i> Loading users…
</td>
</tr>
</template>
<template x-if="!loading && users.length === 0">
<tr>
<td colspan="7" class="px-4 py-8 text-center text-gray-400">
<i class="fas fa-users-slash text-3xl block mb-2" aria-hidden="true"></i>
No users found.
<span x-show="search"> Try a different search term.</span>
<span x-show="!search"> Upload some documents or add a profile above.</span>
</td>
</tr>
</template>
<template x-for="user in users" :key="user.user_id">
<tr class="hover:bg-gray-50">
<!-- User ID -->
<td class="px-4 py-3 text-sm">
<div class="flex items-center gap-2">
<span class="inline-flex items-center justify-center w-8 h-8 rounded-full bg-blue-100 text-blue-700 font-semibold text-xs flex-shrink-0"
x-text="userInitials(user.user_id)" aria-hidden="true"></span>
<span class="font-mono text-gray-800 text-xs truncate max-w-xs" x-text="user.user_id" :title="user.user_id"></span>
</div>
</td>
<!-- Display name -->
<td class="px-4 py-3 text-sm text-gray-700" x-text="user.display_name || '—'"></td>
<!-- Document count -->
<td class="px-4 py-3 text-sm text-center">
<a
:href="`/files?owner_id=${encodeURIComponent(user.user_id)}`"
class="inline-flex items-center gap-1 text-blue-600 hover:underline font-medium"
:aria-label="`View ${user.document_count} document${user.document_count !== 1 ? 's' : ''} for ${user.user_id}`"
>
<i class="fas fa-file text-xs" aria-hidden="true"></i>
<span x-text="user.document_count"></span>
</a>
</td>
<!-- Last upload -->
<td class="px-4 py-3 text-sm text-gray-500 whitespace-nowrap"
x-text="user.last_upload ? formatDate(user.last_upload) : '—'"></td>
<!-- Upload limit -->
<td class="px-4 py-3 text-sm text-center text-gray-700">
<span x-show="user.daily_upload_limit !== null && user.daily_upload_limit !== undefined"
class="inline-flex items-center px-2 py-0.5 rounded-full text-xs font-medium bg-indigo-100 text-indigo-700"
x-text="user.daily_upload_limit + ' / day'"></span>
<span x-show="user.daily_upload_limit === null || user.daily_upload_limit === undefined"
class="text-gray-400 text-xs">global default</span>
</td>
<!-- Status -->
<td class="px-4 py-3 text-sm text-center">
<span
:class="user.is_blocked
? 'bg-red-100 text-red-700'
: 'bg-green-100 text-green-700'"
class="inline-flex items-center gap-1 px-2 py-0.5 rounded-full text-xs font-medium"
>
<i :class="user.is_blocked ? 'fas fa-ban' : 'fas fa-check-circle'" aria-hidden="true"></i>
<span x-text="user.is_blocked ? 'Blocked' : 'Active'"></span>
</span>
</td>
<!-- Actions -->
<td class="px-4 py-3 text-sm text-right whitespace-nowrap">
<button
type="button"
@click="openEditModal(user)"
class="inline-flex items-center px-2.5 py-1.5 text-xs font-medium rounded border border-gray-300 text-gray-700 bg-white hover:bg-gray-50 focus:outline-none focus:ring-2 focus:ring-offset-1 focus:ring-blue-500 mr-1"
:aria-label="`Edit profile for ${user.user_id}`"
>
<i class="fas fa-edit mr-1" aria-hidden="true"></i> Edit
</button>
<button
type="button"
@click="confirmDelete(user)"
x-show="user.profile_id !== null"
class="inline-flex items-center px-2.5 py-1.5 text-xs font-medium rounded border border-red-300 text-red-600 bg-white hover:bg-red-50 focus:outline-none focus:ring-2 focus:ring-offset-1 focus:ring-red-400"
:aria-label="`Delete profile for ${user.user_id}`"
>
<i class="fas fa-trash mr-1" aria-hidden="true"></i> Delete
</button>
</td>
</tr>
</template>
</tbody>
</table>
</div>
<!-- ── Pagination ─────────────────────────────────────────────────────────── -->
<div class="mt-4 flex items-center justify-between" x-show="pages > 1">
<button
type="button"
@click="fetchUsers(currentPage - 1)"
:disabled="currentPage <= 1"
class="px-3 py-1.5 text-sm border border-gray-300 rounded hover:bg-gray-50 disabled:opacity-40 disabled:cursor-not-allowed"
>
<i class="fas fa-chevron-left mr-1" aria-hidden="true"></i> Previous
</button>
<span class="text-sm text-gray-500">
Page <span x-text="currentPage"></span> of <span x-text="pages"></span>
</span>
<button
type="button"
@click="fetchUsers(currentPage + 1)"
:disabled="currentPage >= pages"
class="px-3 py-1.5 text-sm border border-gray-300 rounded hover:bg-gray-50 disabled:opacity-40 disabled:cursor-not-allowed"
>
Next <i class="fas fa-chevron-right ml-1" aria-hidden="true"></i>
</button>
</div>
<!-- ══════════════════════════════════════════════════════════════════════════
EDIT / CREATE MODAL
═══════════════════════════════════════════════════════════════════════ -->
<div
x-show="modalOpen"
x-transition:enter="transition ease-out duration-150"
x-transition:enter-start="opacity-0"
x-transition:enter-end="opacity-100"
x-transition:leave="transition ease-in duration-100"
x-transition:leave-start="opacity-100"
x-transition:leave-end="opacity-0"
class="fixed inset-0 z-50 flex items-center justify-center bg-black bg-opacity-50 px-4"
role="dialog"
aria-modal="true"
:aria-labelledby="'modal-title'"
>
<div
class="bg-white rounded-lg shadow-xl w-full max-w-lg"
@click.outside="modalOpen = false"
>
<div class="flex items-center justify-between px-6 py-4 border-b">
<h2 id="modal-title" class="text-lg font-semibold text-gray-900" x-text="modalTitle"></h2>
<button
type="button"
@click="modalOpen = false"
class="text-gray-400 hover:text-gray-600 focus:outline-none"
aria-label="Close dialog"
>
<i class="fas fa-times" aria-hidden="true"></i>
</button>
</div>
<form @submit.prevent="saveProfile()">
<div class="px-6 py-5 space-y-4">
<!-- User ID (read-only when editing) -->
<div>
<label for="modal-user-id" class="block text-sm font-medium text-gray-700 mb-1">User ID</label>
<input
id="modal-user-id"
type="text"
x-model="form.user_id"
:readonly="!isCreate"
:class="!isCreate ? 'bg-gray-100 cursor-not-allowed' : ''"
class="w-full px-3 py-2 border border-gray-300 rounded-md text-sm focus:outline-none focus:ring-2 focus:ring-blue-400"
placeholder="user@example.com or OAuth sub"
required
aria-required="true"
autocomplete="off"
/>
<p class="text-xs text-gray-400 mt-1">The stable identifier that matches <code>owner_id</code> in documents.</p>
</div>
<!-- Display name -->
<div>
<label for="modal-display-name" class="block text-sm font-medium text-gray-700 mb-1">Display Name</label>
<input
id="modal-display-name"
type="text"
x-model="form.display_name"
maxlength="255"
class="w-full px-3 py-2 border border-gray-300 rounded-md text-sm focus:outline-none focus:ring-2 focus:ring-blue-400"
placeholder="Alice Smith (optional)"
/>
</div>
<!-- Daily upload limit -->
<div>
<label for="modal-limit" class="block text-sm font-medium text-gray-700 mb-1">
Daily Upload Limit
<span class="ml-1 text-xs font-normal text-gray-400">(leave empty to use global default)</span>
</label>
<input
id="modal-limit"
type="number"
x-model.number="form.daily_upload_limit"
min="0"
class="w-full px-3 py-2 border border-gray-300 rounded-md text-sm focus:outline-none focus:ring-2 focus:ring-blue-400"
placeholder="e.g. 50 (0 = unlimited)"
/>
</div>
<!-- Notes -->
<div>
<label for="modal-notes" class="block text-sm font-medium text-gray-700 mb-1">Admin Notes</label>
<textarea
id="modal-notes"
x-model="form.notes"
rows="3"
maxlength="4096"
class="w-full px-3 py-2 border border-gray-300 rounded-md text-sm focus:outline-none focus:ring-2 focus:ring-blue-400 resize-y"
placeholder="Internal notes visible only to admins…"
></textarea>
</div>
<!-- Blocked toggle -->
<div class="flex items-center gap-3">
<label class="relative inline-flex items-center cursor-pointer">
<input
type="checkbox"
x-model="form.is_blocked"
class="sr-only peer"
id="modal-blocked"
role="switch"
:aria-checked="form.is_blocked"
/>
<div class="w-10 h-6 bg-gray-200 peer-focus:ring-2 peer-focus:ring-blue-400 rounded-full peer peer-checked:bg-red-500 after:content-[''] after:absolute after:top-[2px] after:left-[2px] after:bg-white after:rounded-full after:h-5 after:w-5 after:transition-all peer-checked:after:translate-x-4"></div>
</label>
<label for="modal-blocked" class="text-sm font-medium text-gray-700">
Block this user
<span class="text-xs text-gray-400 font-normal">(prevents new document uploads)</span>
</label>
</div>
</div>
<!-- Footer -->
<div class="px-6 py-4 border-t flex justify-end gap-3">
<button
type="button"
@click="modalOpen = false"
class="px-4 py-2 text-sm font-medium border border-gray-300 rounded-md text-gray-700 bg-white hover:bg-gray-50 focus:outline-none focus:ring-2 focus:ring-offset-1 focus:ring-gray-400"
>
Cancel
</button>
<button
type="submit"
:disabled="saving"
class="px-4 py-2 text-sm font-medium rounded-md text-white bg-blue-600 hover:bg-blue-700 focus:outline-none focus:ring-2 focus:ring-offset-1 focus:ring-blue-500 disabled:opacity-50 disabled:cursor-not-allowed"
>
<span x-show="!saving" x-text="isCreate ? 'Create' : 'Save Changes'"></span>
<span x-show="saving"><i class="fas fa-spinner fa-spin mr-1" aria-hidden="true"></i> Saving…</span>
</button>
</div>
</form>
</div>
</div>
<!-- ── Delete confirmation modal ──────────────────────────────────────────── -->
<div
x-show="deleteModal.open"
x-transition:enter="transition ease-out duration-100"
x-transition:enter-start="opacity-0"
x-transition:enter-end="opacity-100"
class="fixed inset-0 z-50 flex items-center justify-center bg-black bg-opacity-50 px-4"
role="dialog"
aria-modal="true"
aria-labelledby="delete-modal-title"
>
<div class="bg-white rounded-lg shadow-xl w-full max-w-md" @click.outside="deleteModal.open = false">
<div class="px-6 py-4 border-b">
<h2 id="delete-modal-title" class="text-lg font-semibold text-gray-900">Delete User Profile</h2>
</div>
<div class="px-6 py-5">
<p class="text-sm text-gray-700">
Are you sure you want to delete the profile for
<strong class="font-mono" x-text="deleteModal.user_id"></strong>?
</p>
<p class="text-sm text-gray-500 mt-2">
This only removes the admin-managed profile record. Documents owned by this user are <strong>not</strong> deleted.
</p>
</div>
<div class="px-6 py-4 border-t flex justify-end gap-3">
<button
type="button"
@click="deleteModal.open = false"
class="px-4 py-2 text-sm font-medium border border-gray-300 rounded-md text-gray-700 bg-white hover:bg-gray-50"
>
Cancel
</button>
<button
type="button"
@click="executeDelete()"
:disabled="deleteModal.deleting"
class="px-4 py-2 text-sm font-medium rounded-md text-white bg-red-600 hover:bg-red-700 disabled:opacity-50"
>
<span x-show="!deleteModal.deleting">Delete Profile</span>
<span x-show="deleteModal.deleting"><i class="fas fa-spinner fa-spin mr-1" aria-hidden="true"></i> Deleting…</span>
</button>
</div>
</div>
</div>
</div>
<script>
function adminUsersApp() {
return {
// State
users: [],
loading: true,
search: '',
currentPage: 1,
perPage: 25,
total: 0,
pages: 1,
// Edit / create modal
modalOpen: false,
modalTitle: '',
isCreate: false,
saving: false,
form: {
user_id: '',
display_name: '',
daily_upload_limit: null,
notes: '',
is_blocked: false,
},
// Delete modal
deleteModal: {
open: false,
user_id: '',
deleting: false,
},
// Alert
alert: { show: false, type: 'success', title: '', message: '' },
get totalLabel() {
if (this.loading) return '';
if (this.total === 0) return 'No users';
return `${this.total} user${this.total !== 1 ? 's' : ''}`;
},
async init() {
await this.fetchUsers(1);
},
async fetchUsers(page) {
this.loading = true;
this.currentPage = page;
try {
const params = new URLSearchParams({
page: page,
per_page: this.perPage,
});
if (this.search.trim()) params.set('q', this.search.trim());
const resp = await fetch(`/api/admin/users/?${params}`);
if (!resp.ok) {
const err = await resp.json().catch(() => ({}));
this.showAlert('error', 'Failed to load users', err.detail || resp.statusText);
return;
}
const data = await resp.json();
this.users = data.users;
this.total = data.total;
this.pages = data.pages;
} catch (e) {
this.showAlert('error', 'Network error', e.message);
} finally {
this.loading = false;
}
},
openCreateModal() {
this.isCreate = true;
this.modalTitle = 'Add User Profile';
this.form = { user_id: '', display_name: '', daily_upload_limit: null, notes: '', is_blocked: false };
this.modalOpen = true;
},
openEditModal(user) {
this.isCreate = false;
this.modalTitle = 'Edit User Profile';
this.form = {
user_id: user.user_id,
display_name: user.display_name || '',
daily_upload_limit: user.daily_upload_limit !== null && user.daily_upload_limit !== undefined
? user.daily_upload_limit : null,
notes: user.notes || '',
is_blocked: !!user.is_blocked,
};
this.modalOpen = true;
},
async saveProfile() {
this.saving = true;
try {
const body = {
display_name: this.form.display_name || null,
daily_upload_limit: this.form.daily_upload_limit !== '' && this.form.daily_upload_limit !== null
? Number(this.form.daily_upload_limit) : null,
notes: this.form.notes || null,
is_blocked: !!this.form.is_blocked,
};
const uid = encodeURIComponent(this.form.user_id);
const resp = await fetch(`/api/admin/users/${uid}`, {
method: 'PUT',
headers: {
'Content-Type': 'application/json',
'X-CSRF-Token': document.querySelector('meta[name="csrf-token"]')?.content || '',
},
body: JSON.stringify(body),
});
if (!resp.ok) {
const err = await resp.json().catch(() => ({}));
this.showAlert('error', 'Save failed', err.detail || resp.statusText);
return;
}
this.modalOpen = false;
this.showAlert('success', 'Saved', `Profile for "${this.form.user_id}" has been saved.`);
await this.fetchUsers(this.currentPage);
} catch (e) {
this.showAlert('error', 'Network error', e.message);
} finally {
this.saving = false;
}
},
confirmDelete(user) {
this.deleteModal.user_id = user.user_id;
this.deleteModal.deleting = false;
this.deleteModal.open = true;
},
async executeDelete() {
this.deleteModal.deleting = true;
try {
const uid = encodeURIComponent(this.deleteModal.user_id);
const resp = await fetch(`/api/admin/users/${uid}`, {
method: 'DELETE',
headers: {
'X-CSRF-Token': document.querySelector('meta[name="csrf-token"]')?.content || '',
},
});
if (resp.status === 204) {
this.deleteModal.open = false;
this.showAlert('success', 'Deleted', `Profile for "${this.deleteModal.user_id}" has been removed.`);
await this.fetchUsers(this.currentPage);
} else {
const err = await resp.json().catch(() => ({}));
this.showAlert('error', 'Delete failed', err.detail || resp.statusText);
this.deleteModal.open = false;
}
} catch (e) {
this.showAlert('error', 'Network error', e.message);
this.deleteModal.open = false;
} finally {
this.deleteModal.deleting = false;
}
},
userInitials(userId) {
if (!userId) return '?';
// Try to get initials from email or username
const parts = userId.split(/[@._\s-]+/);
return parts.slice(0, 2).map(p => p[0]?.toUpperCase() || '').join('') || userId[0].toUpperCase();
},
formatDate(iso) {
if (!iso) return '—';
try {
return new Date(iso).toLocaleDateString(undefined, { year: 'numeric', month: 'short', day: 'numeric' });
} catch {
return iso;
}
},
showAlert(type, title, message) {
this.alert = { show: true, type, title, message };
setTimeout(() => { this.alert.show = false; }, type === 'success' ? 5000 : 10000);
},
};
}
</script>
{% endblock %}
+6
View File
@@ -95,6 +95,9 @@
<a href="/settings" role="menuitem" class="flex items-center px-4 py-2 text-sm text-gray-700 hover:bg-gray-100">
<i class="fas fa-cog w-4 mr-2 text-gray-500" aria-hidden="true"></i> Settings
</a>
<a href="/admin/users" role="menuitem" class="flex items-center px-4 py-2 text-sm text-gray-700 hover:bg-gray-100">
<i class="fas fa-users w-4 mr-2 text-blue-500" aria-hidden="true"></i> Users
</a>
<a href="/admin/credentials" role="menuitem" class="flex items-center px-4 py-2 text-sm text-gray-700 hover:bg-gray-100">
<i class="fas fa-key w-4 mr-2 text-yellow-500" aria-hidden="true"></i> Credentials
</a>
@@ -178,6 +181,9 @@
<a href="/settings" class="block px-3 py-3 rounded-md text-base font-medium text-gray-700 hover:text-gray-900 hover:bg-gray-50">
<i class="fas fa-cog mr-2 text-gray-400" aria-hidden="true"></i> Settings
</a>
<a href="/admin/users" class="block px-3 py-3 rounded-md text-base font-medium text-gray-700 hover:text-gray-900 hover:bg-gray-50">
<i class="fas fa-users mr-2 text-blue-400" aria-hidden="true"></i> Users
</a>
<a href="/admin/credentials" class="block px-3 py-3 rounded-md text-base font-medium text-gray-700 hover:text-gray-900 hover:bg-gray-50">
<i class="fas fa-key mr-2 text-yellow-400" aria-hidden="true"></i> Credentials
</a>
@@ -0,0 +1,42 @@
"""Add user_profiles table for per-user admin settings
Revision ID: 013_add_user_profiles
Revises: 012_add_multi_user_support
Create Date: 2026-03-06
"""
from typing import Union
import sqlalchemy as sa
from alembic import op
# revision identifiers, used by Alembic.
revision: str = "013_add_user_profiles"
down_revision: Union[str, None] = "012_add_multi_user_support"
depends_on: Union[str, None] = None
def upgrade() -> None:
"""Create the user_profiles table."""
op.create_table(
"user_profiles",
sa.Column("id", sa.Integer(), nullable=False),
sa.Column("user_id", sa.String(255), nullable=False),
sa.Column("display_name", sa.String(255), nullable=True),
sa.Column("daily_upload_limit", sa.Integer(), nullable=True),
sa.Column("notes", sa.Text(), nullable=True),
sa.Column("is_blocked", sa.Boolean(), nullable=False, server_default=sa.text("0")),
sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.text("CURRENT_TIMESTAMP")),
sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.text("CURRENT_TIMESTAMP")),
sa.PrimaryKeyConstraint("id"),
)
op.create_index("ix_user_profiles_id", "user_profiles", ["id"])
op.create_index("ix_user_profiles_user_id", "user_profiles", ["user_id"], unique=True)
def downgrade() -> None:
"""Drop the user_profiles table."""
op.drop_index("ix_user_profiles_user_id", table_name="user_profiles")
op.drop_index("ix_user_profiles_id", table_name="user_profiles")
op.drop_table("user_profiles")
+8 -1
View File
@@ -59,7 +59,14 @@ from app.database import Base # noqa: E402
from app.main import app as fastapi_app # noqa: E402
# Import models to register them with SQLAlchemy Base
from app.models import DocumentMetadata, FileRecord, ProcessingLog, SavedSearch, WebhookConfig # noqa: F401, E402
from app.models import ( # noqa: F401, E402
DocumentMetadata,
FileRecord,
ProcessingLog,
SavedSearch,
UserProfile,
WebhookConfig,
)
@pytest.fixture(scope="session")
+479
View File
@@ -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()