fix(merge): resolve conflicts with main v0.92.0 keeping path-param regression tests

Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com>
This commit is contained in:
copilot-swe-agent[bot]
2026-03-08 11:19:22 +00:00
parent c31b72810e
commit 2d754d52ef
22 changed files with 2256 additions and 122 deletions
+1 -1
View File
@@ -24,7 +24,7 @@ jobs:
include:
- language: actions
build-mode: none
- language: javascript-typescript
- language: javascript
build-mode: none
- language: python
build-mode: none
+1 -1
View File
@@ -1 +1 @@
2026-03-08T10:58:56Z
2026-03-08T11:14:23Z
+36
View File
@@ -10,6 +10,42 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
<!-- version list -->
## v0.92.0 (2026-03-08)
### Bug Fixes
- **admin-users**: Restore missing @router.get decorator on get_user endpoint
([`9b8eb91`](https://github.com/christianlouis/DocuElevate/commit/9b8eb911c3bf2b0be84a73030e82cff12151d4ba))
### Chores
- Update plan to include forgot-username and login label clarification
([`44ea43f`](https://github.com/christianlouis/DocuElevate/commit/44ea43f9cf948aabe8f8ae98744cd41061ab8770))
### Continuous Integration
- Fix CodeQL javascript language identifier mismatch
([`3f8a95d`](https://github.com/christianlouis/DocuElevate/commit/3f8a95d8081c4ce3ad380e6391afa3bebb57bbb7))
### Documentation
- **changelog**: Update changelog [skip ci]
([`f2d2aaa`](https://github.com/christianlouis/DocuElevate/commit/f2d2aaa3545081de1ff7317672199085daa3c3c0))
### Features
- **auth**: Password reset, forgot username, and admin user management for local accounts
([`d36ba88`](https://github.com/christianlouis/DocuElevate/commit/d36ba88de765b688888c6e661256f8508da86d89))
## v0.91.0 (2026-03-08)
### Features
- **backup**: Extend backup and restore to PostgreSQL and MySQL/MariaDB
([`a0f5ba1`](https://github.com/christianlouis/DocuElevate/commit/a0f5ba179978d7c63bbdb8c564a32eaf3f78e5ad))
## v0.90.3 (2026-03-08)
### Bug Fixes
+1 -1
View File
@@ -1 +1 @@
da47283
4109bf6
+6 -6
View File
@@ -1,10 +1,10 @@
DocuElevate Build Information
==============================
Version: 0.90.3
Build Date: 2026-03-08T10:58:56Z
Git Commit: da47283e0afb48e2992c7698b5ff74028c83d55e
Git Short SHA: da47283
Version: 0.92.0
Build Date: 2026-03-08T11:14:23Z
Git Commit: 4109bf65d2e127b9fa0186a6f22b5bcb685580ea
Git Short SHA: 4109bf6
Git Branch: main
Commit Date: 2026-03-08T11:58:40+01:00
Build Timestamp: 2026-03-08T10:58:56Z
Commit Date: 2026-03-08T12:14:04+01:00
Build Timestamp: 2026-03-08T11:14:23Z
==============================
+1 -1
View File
@@ -1 +1 @@
0.90.3
0.92.0
+149 -2
View File
@@ -7,7 +7,7 @@ user accounts directly, without requiring email verification.
"""
import logging
from datetime import datetime
from datetime import datetime, timezone
from typing import Annotated, Any
from fastapi import APIRouter, Depends, HTTPException, Query, Request, status
@@ -15,9 +15,10 @@ from pydantic import BaseModel, Field
from sqlalchemy import func
from sqlalchemy.orm import Session
from app.config import settings
from app.database import get_db
from app.models import FileRecord, LocalUser, UserProfile
from app.utils.local_auth import hash_password
from app.utils.local_auth import generate_token, hash_password, send_password_reset_email
logger = logging.getLogger(__name__)
router = APIRouter(prefix="/admin/users", tags=["admin-users"])
@@ -123,6 +124,21 @@ class LocalUserCreate(BaseModel):
is_admin: bool = Field(default=False, description="Grant admin privileges")
class LocalUserUpdate(BaseModel):
"""Body for admin-updating a local (email/password) user account."""
email: str | None = Field(default=None, max_length=255, description="New email address")
display_name: str | None = Field(default=None, max_length=255, description="New display name")
is_admin: bool | None = Field(default=None, description="Grant or revoke admin privileges")
is_active: bool | None = Field(default=None, description="Activate or deactivate the account")
class LocalUserSetPassword(BaseModel):
"""Body for admin setting a temporary password for a local user."""
password: str = Field(..., min_length=8, max_length=128, description="New temporary password")
class LocalUserResponse(BaseModel):
"""Summary of a local user account."""
@@ -355,6 +371,137 @@ def delete_local_user(local_user_id: int, db: DbSession, _admin: AdminUser) -> N
logger.info("Admin deleted local user account: %s", user.email)
@router.patch("/local/{local_user_id}", summary="Update a local user account")
def update_local_user(local_user_id: int, body: LocalUserUpdate, db: DbSession, _admin: AdminUser) -> dict[str, Any]:
"""Update the email address, display name, admin flag, or active status of a local user account.
Only fields explicitly provided (non-None) are modified. If the email is changed
the associated UserProfile row is also updated to keep ``user_id`` in sync.
Raises:
404: Local user not found.
409: The new email is already taken by another account.
"""
user = db.query(LocalUser).filter(LocalUser.id == local_user_id).first()
if not user:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Local user not found.")
old_email = user.email
if body.email is not None and body.email != user.email:
if db.query(LocalUser).filter(LocalUser.email == body.email, LocalUser.id != local_user_id).first():
raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="Email already registered.")
user.email = body.email
if body.display_name is not None:
# Normalise empty string to None so that clearing the field removes the display name
user.display_name = body.display_name or None
if body.is_admin is not None:
user.is_admin = body.is_admin
if body.is_active is not None:
user.is_active = body.is_active
try:
db.flush()
# Keep UserProfile.user_id in sync when email changes
if body.email is not None and body.email != old_email:
profile = db.query(UserProfile).filter(UserProfile.user_id == old_email).first()
if profile:
profile.user_id = body.email
db.commit()
db.refresh(user)
except Exception:
db.rollback()
raise
logger.info("Admin updated local user %s (id=%d)", user.email, user.id)
return {
"id": user.id,
"email": user.email,
"username": user.username,
"display_name": user.display_name,
"is_active": user.is_active,
"is_admin": user.is_admin,
"created_at": user.created_at.isoformat() if user.created_at else None,
}
@router.post(
"/local/{local_user_id}/send-password-reset",
status_code=status.HTTP_200_OK,
summary="Send a password reset email to a local user",
)
def admin_send_password_reset(local_user_id: int, request: Request, db: DbSession, _admin: AdminUser) -> dict[str, Any]:
"""Generate a password reset token and email the reset link to the local user.
This is a last-resort tool for admins to help users who are locked out.
Returns ``{"sent": true}`` on success and ``{"sent": false, "reason": "..."}`` when
SMTP is not configured or sending fails.
Raises:
404: Local user not found.
"""
user = db.query(LocalUser).filter(LocalUser.id == local_user_id).first()
if not user:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Local user not found.")
if not settings.email_host:
logger.warning("Admin requested password reset for %s but SMTP is not configured", user.email)
return {"sent": False, "reason": "SMTP is not configured on this server."}
token = generate_token()
user.password_reset_token = token
user.password_reset_sent_at = datetime.now(tz=timezone.utc)
db.commit()
base_url = str(request.base_url).rstrip("/")
try:
send_password_reset_email(user.email, user.username, token, base_url)
except Exception as exc:
logger.warning("Admin-triggered password reset email failed for %s: %s", user.email, exc)
return {"sent": False, "reason": str(exc)}
logger.info("[SECURITY] ADMIN_PASSWORD_RESET_EMAIL user=%s admin=%s", user.email, _admin.get("email", "unknown"))
return {"sent": True, "email": user.email}
@router.post(
"/local/{local_user_id}/set-password",
status_code=status.HTTP_200_OK,
summary="Set a temporary password for a local user account",
)
def admin_set_password(
local_user_id: int, body: LocalUserSetPassword, db: DbSession, _admin: AdminUser
) -> dict[str, Any]:
"""Directly set a new password for a local user without requiring an email token.
Use this as a last resort when email delivery is unavailable. The user
should be advised to change their password after logging in.
Raises:
404: Local user not found.
"""
user = db.query(LocalUser).filter(LocalUser.id == local_user_id).first()
if not user:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Local user not found.")
user.hashed_password = hash_password(body.password)
# Clear any outstanding reset tokens
user.password_reset_token = None
user.password_reset_sent_at = None
try:
db.commit()
except Exception:
db.rollback()
raise
logger.info("[SECURITY] ADMIN_SET_PASSWORD user=%s admin=%s", user.email, _admin.get("email", "unknown"))
return {"updated": True, "email": user.email}
@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."""
+84 -68
View File
@@ -117,88 +117,104 @@ async def restore_backup(
"""Restore the database from an uploaded gzip-compressed SQL dump.
**Warning**: This overwrites the current database contents.
Only SQLite databases are supported.
The uploaded file must be a ``.db.gz`` file produced by the DocuElevate
backup task (a gzip-compressed SQLite ``.dump()`` SQL script).
Supported formats (must match the currently configured database backend):
- ``*.db.gz`` gzip-compressed SQLite ``.dump()`` SQL script (SQLite backend)
- ``*.pgsql.gz`` gzip-compressed ``pg_dump --format=plain`` output (PostgreSQL backend)
- ``*.mysql.gz`` gzip-compressed ``mysqldump`` output (MySQL / MariaDB backend)
"""
from app.tasks.backup_tasks import _db_path
db_path = _db_path()
if db_path is None:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Restore is only supported for SQLite databases.",
)
if not file.filename or not file.filename.endswith(".db.gz"):
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Uploaded file must be a .db.gz backup archive.",
)
import gzip
import sqlite3
import tempfile
from pathlib import Path
# Write the upload to a temp file first so we can validate it
with tempfile.NamedTemporaryFile(suffix=".db.gz", delete=False) as tmp:
from sqlalchemy.engine.url import make_url
from app.config import settings as app_settings
from app.tasks.backup_tasks import (
_archive_ext_for_backend,
_db_path,
_restore_mysql,
_restore_postgresql,
_restore_sqlite,
)
url = make_url(app_settings.database_url)
backend = url.get_backend_name()
expected_ext = _archive_ext_for_backend(backend)
if not file.filename or not file.filename.endswith(expected_ext):
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=(
f"Uploaded file must be a '{expected_ext}' backup archive for the current database backend ({backend})."
),
)
# Write upload to a temp file
with tempfile.NamedTemporaryFile(suffix=expected_ext, delete=False) as tmp:
tmp_path = Path(tmp.name)
content = await file.read()
tmp.write(content)
try:
# Decompress and read SQL statements
with gzip.open(str(tmp_path), "rt", encoding="utf-8") as gz:
sql_script = gz.read()
except Exception as exc:
tmp_path.unlink(missing_ok=True)
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"Failed to decompress backup file: {exc}",
) from exc
if backend == "sqlite":
db_path = _db_path()
if db_path is None:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Restore is only supported for file-based SQLite databases.",
)
# Close the application DB session before replacing the file
db.close()
try:
_restore_sqlite(db_path, tmp_path)
except ValueError as exc:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=str(exc),
) from exc
except RuntimeError as exc:
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=str(exc),
) from exc
# Create a fresh in-memory DB from the script to validate it
try:
mem_conn = sqlite3.connect(":memory:")
mem_conn.executescript(sql_script)
mem_conn.close()
except sqlite3.Error as exc:
tmp_path.unlink(missing_ok=True)
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"Backup file contains invalid SQL: {exc}",
) from exc
elif backend == "postgresql":
db.close()
try:
_restore_postgresql(app_settings.database_url, tmp_path)
except FileNotFoundError as exc:
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=f"psql binary not found is PostgreSQL client installed? ({exc})",
) from exc
except RuntimeError as exc:
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=f"PostgreSQL restore failed: {exc}",
) from exc
# Close the application DB session before replacing the file
db.close()
elif backend == "mysql":
db.close()
try:
_restore_mysql(app_settings.database_url, tmp_path)
except FileNotFoundError as exc:
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=f"mysql binary not found is MySQL client installed? ({exc})",
) from exc
except RuntimeError as exc:
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=f"MySQL restore failed: {exc}",
) from exc
# Preserve the current DB before overwriting
import shutil
else:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"Database backend '{backend}' does not support restore.",
)
bak = str(db_path) + ".pre_restore"
try:
shutil.copy2(str(db_path), bak)
except OSError as exc:
logger.warning(f"Could not create pre-restore backup at {bak}: {exc}")
try:
# Write the restored database
restore_conn = sqlite3.connect(str(db_path))
restore_conn.executescript(sql_script)
restore_conn.close()
except sqlite3.Error as exc:
# Attempt rollback
try:
if os.path.exists(bak):
shutil.copy2(bak, str(db_path))
except OSError as rollback_exc:
logger.error(f"Rollback failed; database may be corrupted: {rollback_exc}")
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=f"Restore failed: {exc}",
) from exc
finally:
tmp_path.unlink(missing_ok=True)
+49
View File
@@ -31,6 +31,7 @@ from app.utils.local_auth import (
generate_token,
hash_password,
is_token_expired,
send_forgot_username_email,
send_password_reset_email,
send_verification_email,
)
@@ -79,6 +80,12 @@ class PasswordResetBody(BaseModel):
new_password_confirm: str
class ForgotUsernameBody(BaseModel):
"""Body for the forgot-username endpoint."""
email: str
# ---------------------------------------------------------------------------
# Page routes (return HTML)
# ---------------------------------------------------------------------------
@@ -107,6 +114,32 @@ async def verify_email_sent_page(request: Request) -> Any:
return templates.TemplateResponse("verify_email_sent.html", {"request": request})
@router.get("/forgot-username", include_in_schema=False)
async def forgot_username_page(request: Request) -> Any:
"""Render the forgot-username page where users can request a username reminder email."""
return templates.TemplateResponse(
"forgot_username.html",
{
"request": request,
"csrf_token": getattr(request.state, "csrf_token", ""),
"app_version": settings.version,
},
)
@router.get("/forgot-password", include_in_schema=False)
async def forgot_password_page(request: Request) -> Any:
"""Render the forgot-password page where users can request a reset email."""
return templates.TemplateResponse(
"forgot_password.html",
{
"request": request,
"csrf_token": getattr(request.state, "csrf_token", ""),
"app_version": settings.version,
},
)
@router.get("/reset-password", include_in_schema=False)
async def reset_password_page(request: Request) -> Any:
"""Render the password reset form page."""
@@ -337,3 +370,19 @@ async def reset_password(body: PasswordResetBody, db: DbSession) -> dict[str, st
logger.info("[SECURITY] PASSWORD_RESET_SUCCESS user=%s", user.email)
return {"message": "Password updated successfully."}
@router.post("/api/auth/forgot-username")
async def forgot_username(body: ForgotUsernameBody, db: DbSession) -> dict[str, str]:
"""Send a username reminder email.
Always returns 200 to avoid leaking whether an email is registered.
"""
user = db.query(LocalUser).filter(LocalUser.email == body.email).first()
if user:
try:
send_forgot_username_email(user.email, user.username)
except Exception as exc:
logger.warning("Failed to send forgot-username email to %s: %s", user.email, exc)
return {"message": "Username reminder sent if account exists."}
+337 -9
View File
@@ -16,12 +16,19 @@ Three separate Celery-beat entries call ``create_backup`` with the appropriate
After each backup is created ``_apply_retention`` prunes old local backups for
that tier. Remote copies are pruned by ``_prune_remote_backups`` which mirrors
the same retention limits.
Supported database backends
----------------------------
- **SQLite** dumped via Python's built-in ``sqlite3.iterdump()``; archive extension ``.db.gz``
- **PostgreSQL** dumped via ``pg_dump --format=plain``; archive extension ``.pgsql.gz``
- **MySQL / MariaDB** dumped via ``mysqldump --single-transaction``; archive extension ``.mysql.gz``
"""
import gzip
import hashlib
import logging
import os
import subprocess
from datetime import datetime, timezone
from pathlib import Path
@@ -42,6 +49,13 @@ _BACKUP_TYPE_RETAIN: dict[str, str] = {
"weekly": "backup_retain_weekly",
}
#: Map of backend name → archive file extension.
_BACKEND_EXTENSIONS: dict[str, str] = {
"sqlite": ".db.gz",
"postgresql": ".pgsql.gz",
"mysql": ".mysql.gz",
}
def _backup_dir() -> Path:
"""Return (and create) the local backup directory."""
@@ -51,6 +65,14 @@ def _backup_dir() -> Path:
return path
def _db_backend() -> str:
"""Return the database backend name (e.g. ``'sqlite'``, ``'postgresql'``, ``'mysql'``)."""
from sqlalchemy.engine.url import make_url
url = make_url(settings.database_url)
return url.get_backend_name()
def _db_path() -> Path | None:
"""Return the SQLite database file path, or None for non-SQLite databases."""
from sqlalchemy.engine.url import make_url
@@ -64,6 +86,20 @@ def _db_path() -> Path | None:
return Path(db)
def _archive_ext_for_backend(backend: str) -> str:
"""Return the archive file extension for the given database backend.
Args:
backend: Backend name as returned by
``sqlalchemy.engine.url.URL.get_backend_name()`` (e.g. ``'sqlite'``).
Returns:
File extension string including the leading dot, e.g. ``'.db.gz'``.
Falls back to ``'.sql.gz'`` for unknown backends.
"""
return _BACKEND_EXTENSIONS.get(backend, ".sql.gz")
def _sha256(path: Path) -> str:
"""Return the SHA-256 hex digest of *path*."""
h = hashlib.sha256()
@@ -86,6 +122,277 @@ def _dump_sqlite(db_path: Path, dest: Path) -> None:
conn.close()
def _dump_postgresql(db_url: str, dest: Path) -> None:
"""Write a gzip-compressed ``pg_dump`` of the PostgreSQL database to *dest*.
Uses ``PGPASSWORD`` environment variable so the password is never exposed on
the process command line.
Args:
db_url: Full SQLAlchemy database URL (e.g. ``postgresql://user:pass@host/db``).
dest: Destination path for the ``.pgsql.gz`` archive.
Raises:
RuntimeError: If ``pg_dump`` exits with a non-zero return code.
FileNotFoundError: If the ``pg_dump`` binary is not found.
"""
from sqlalchemy.engine.url import make_url
url = make_url(db_url)
env = os.environ.copy()
if url.password:
env["PGPASSWORD"] = str(url.password)
# Command arguments are built from the SQLAlchemy URL (admin-configured DATABASE_URL),
# not from user-controlled input. shell=False (the default when passing a list) is used
# so there is no shell interpretation of the argument values.
cmd: list[str] = ["pg_dump", "--format=plain", "--no-password"]
if url.host:
cmd.extend(["-h", url.host])
if url.port:
cmd.extend(["-p", str(url.port)])
if url.username:
cmd.extend(["-U", url.username])
if url.database:
cmd.append(url.database)
with gzip.open(str(dest), "wb") as gz:
proc = subprocess.Popen( # noqa: S603
cmd,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
env=env,
)
stdout = proc.stdout
if stdout is None: # pragma: no cover guaranteed by stdout=PIPE
raise RuntimeError("pg_dump produced no stdout pipe")
try:
while True:
chunk = stdout.read(65536)
if not chunk:
break
gz.write(chunk)
finally:
stdout.close()
stderr_bytes = proc.stderr.read() if proc.stderr else b""
proc.wait()
if proc.returncode != 0:
dest.unlink(missing_ok=True)
raise RuntimeError(
f"pg_dump exited with code {proc.returncode}: {stderr_bytes.decode(errors='replace').strip()}"
)
def _dump_mysql(db_url: str, dest: Path) -> None:
"""Write a gzip-compressed ``mysqldump`` of the MySQL database to *dest*.
Uses the ``MYSQL_PWD`` environment variable so the password is never exposed
on the process command line.
Args:
db_url: Full SQLAlchemy database URL
(e.g. ``mysql+pymysql://user:pass@host/db``).
dest: Destination path for the ``.mysql.gz`` archive.
Raises:
RuntimeError: If ``mysqldump`` exits with a non-zero return code.
FileNotFoundError: If the ``mysqldump`` binary is not found.
"""
from sqlalchemy.engine.url import make_url
url = make_url(db_url)
env = os.environ.copy()
if url.password:
env["MYSQL_PWD"] = str(url.password)
# Command arguments are built from the SQLAlchemy URL (admin-configured DATABASE_URL).
# shell=False (list form) prevents shell interpretation of argument values.
cmd: list[str] = ["mysqldump", "--single-transaction", "--routines", "--triggers"]
if url.host:
cmd.extend(["-h", url.host])
if url.port:
cmd.extend(["-P", str(url.port)])
if url.username:
cmd.extend(["-u", url.username])
if url.database:
cmd.append(url.database)
with gzip.open(str(dest), "wb") as gz:
proc = subprocess.Popen( # noqa: S603
cmd,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
env=env,
)
stdout = proc.stdout
if stdout is None: # pragma: no cover guaranteed by stdout=PIPE
raise RuntimeError("mysqldump produced no stdout pipe")
try:
while True:
chunk = stdout.read(65536)
if not chunk:
break
gz.write(chunk)
finally:
stdout.close()
stderr_bytes = proc.stderr.read() if proc.stderr else b""
proc.wait()
if proc.returncode != 0:
dest.unlink(missing_ok=True)
raise RuntimeError(
f"mysqldump exited with code {proc.returncode}: {stderr_bytes.decode(errors='replace').strip()}"
)
def _restore_sqlite(db_path: Path, archive_path: Path) -> None:
"""Restore a SQLite database from a gzip-compressed SQL dump archive.
Validates the SQL by replaying it on an in-memory database before touching
the live file. Saves a ``<db_path>.pre_restore`` rollback copy first.
Args:
db_path: Path to the live SQLite database file to overwrite.
archive_path: Path to the ``.db.gz`` gzip-compressed SQL dump.
Raises:
ValueError: If the archive cannot be decompressed or contains invalid SQL.
RuntimeError: If writing the restored database fails.
"""
import shutil
import sqlite3
# Decompress and read SQL statements
try:
with gzip.open(str(archive_path), "rt", encoding="utf-8") as gz:
sql_script = gz.read()
except Exception as exc:
raise ValueError(f"Failed to decompress backup file: {exc}") from exc
# Validate by replaying on an in-memory database
try:
mem_conn = sqlite3.connect(":memory:")
mem_conn.executescript(sql_script)
mem_conn.close()
except sqlite3.Error as exc:
raise ValueError(f"Backup file contains invalid SQL: {exc}") from exc
# Preserve the current DB before overwriting
bak = str(db_path) + ".pre_restore"
try:
shutil.copy2(str(db_path), bak)
except OSError as exc:
logger.warning(f"Could not create pre-restore backup at {bak}: {exc}")
try:
restore_conn = sqlite3.connect(str(db_path))
restore_conn.executescript(sql_script)
restore_conn.close()
except sqlite3.Error as exc:
# Attempt rollback to the pre-restore copy
try:
if os.path.exists(bak):
shutil.copy2(bak, str(db_path))
except OSError as rollback_exc:
logger.error(f"Rollback failed; database may be corrupted: {rollback_exc}")
raise RuntimeError(f"SQLite restore failed: {exc}") from exc
def _restore_postgresql(db_url: str, archive_path: Path) -> None:
"""Restore a PostgreSQL database from a gzip-compressed SQL dump archive.
Pipes the decompressed dump to ``psql``. Uses ``PGPASSWORD`` so the
password is never exposed on the process command line.
Args:
db_url: Full SQLAlchemy database URL.
archive_path: Path to the ``.pgsql.gz`` gzip-compressed ``pg_dump`` archive.
Raises:
RuntimeError: If ``psql`` exits with a non-zero return code.
FileNotFoundError: If the ``psql`` binary is not found.
"""
from sqlalchemy.engine.url import make_url
url = make_url(db_url)
env = os.environ.copy()
if url.password:
env["PGPASSWORD"] = str(url.password)
# Command arguments are built from the SQLAlchemy URL (admin-configured DATABASE_URL).
# shell=False (list form) prevents shell interpretation of argument values.
cmd: list[str] = ["psql", "--no-password"]
if url.host:
cmd.extend(["-h", url.host])
if url.port:
cmd.extend(["-p", str(url.port)])
if url.username:
cmd.extend(["-U", url.username])
if url.database:
cmd.append(url.database)
with gzip.open(str(archive_path), "rb") as gz:
proc = subprocess.Popen( # noqa: S603
cmd,
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
env=env,
)
_, stderr_bytes = proc.communicate(input=gz.read())
if proc.returncode != 0:
raise RuntimeError(f"psql exited with code {proc.returncode}: {stderr_bytes.decode(errors='replace').strip()}")
def _restore_mysql(db_url: str, archive_path: Path) -> None:
"""Restore a MySQL database from a gzip-compressed SQL dump archive.
Pipes the decompressed dump to ``mysql``. Uses the ``MYSQL_PWD``
environment variable so the password is never exposed on the command line.
Args:
db_url: Full SQLAlchemy database URL.
archive_path: Path to the ``.mysql.gz`` gzip-compressed ``mysqldump`` archive.
Raises:
RuntimeError: If ``mysql`` exits with a non-zero return code.
FileNotFoundError: If the ``mysql`` binary is not found.
"""
from sqlalchemy.engine.url import make_url
url = make_url(db_url)
env = os.environ.copy()
if url.password:
env["MYSQL_PWD"] = str(url.password)
# Command arguments are built from the SQLAlchemy URL (admin-configured DATABASE_URL).
# shell=False (list form) prevents shell interpretation of argument values.
cmd: list[str] = ["mysql"]
if url.host:
cmd.extend(["-h", url.host])
if url.port:
cmd.extend(["-P", str(url.port)])
if url.username:
cmd.extend(["-u", url.username])
if url.database:
cmd.append(url.database)
with gzip.open(str(archive_path), "rb") as gz:
proc = subprocess.Popen( # noqa: S603
cmd,
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
env=env,
)
_, stderr_bytes = proc.communicate(input=gz.read())
if proc.returncode != 0:
raise RuntimeError(f"mysql exited with code {proc.returncode}: {stderr_bytes.decode(errors='replace').strip()}")
def _apply_retention(backup_type: str, db: object) -> None:
"""Delete local backups beyond the retention limit for *backup_type*.
@@ -314,6 +621,11 @@ def _email_backup(archive_path: Path, filename: str) -> None:
def create_backup(self, backup_type: str = "hourly") -> dict:
"""Create a database backup archive and apply retention.
Supports SQLite (``.db.gz``), PostgreSQL (``.pgsql.gz``), and
MySQL / MariaDB (``.mysql.gz``) databases. The native dump tool for the
configured backend (``sqlite3``, ``pg_dump``, or ``mysqldump``) must be
available on the worker's ``PATH``.
Args:
backup_type: ``"hourly"``, ``"daily"``, or ``"weekly"``.
@@ -327,19 +639,27 @@ def create_backup(self, backup_type: str = "hourly") -> dict:
logger.debug("Backup is disabled; skipping create_backup task.")
return {"status": "disabled"}
backend = _db_backend()
ext = _archive_ext_for_backend(backend)
ts = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H-%M-%S")
filename = f"backup_{backup_type}_{ts}.db.gz"
filename = f"backup_{backup_type}_{ts}{ext}"
archive_path = _backup_dir() / filename
db_path = _db_path()
if db_path is None:
logger.warning("Backup task skipped: non-SQLite databases are not supported for file-based backups.")
# SQLite: verify the database file exists before attempting to dump it
db_path: Path | None = None
if backend == "sqlite":
db_path = _db_path()
if db_path is None:
logger.warning("Backup task skipped: in-memory SQLite databases are not supported.")
return {"status": "unsupported_db"}
if not db_path.exists():
logger.error(f"Database file not found: {db_path}")
return {"status": "error", "detail": f"DB file missing: {db_path}"}
elif backend not in ("postgresql", "mysql"):
logger.warning(f"Backup task skipped: unsupported database backend '{backend}'.")
return {"status": "unsupported_db"}
if not db_path.exists():
logger.error(f"Database file not found: {db_path}")
return {"status": "error", "detail": f"DB file missing: {db_path}"}
status = "ok"
checksum: str | None = None
size_bytes = 0
@@ -347,7 +667,15 @@ def create_backup(self, backup_type: str = "hourly") -> dict:
remote_path: str | None = None
try:
_dump_sqlite(db_path, archive_path)
if backend == "sqlite":
# db_path is guaranteed non-None: we returned early if it were None
if db_path is None: # pragma: no cover
return {"status": "error", "detail": "db_path unexpectedly None"}
_dump_sqlite(db_path, archive_path)
elif backend == "postgresql":
_dump_postgresql(settings.database_url, archive_path)
elif backend == "mysql":
_dump_mysql(settings.database_url, archive_path)
size_bytes = archive_path.stat().st_size
checksum = _sha256(archive_path)
logger.info(f"Created {backup_type} backup: {archive_path} ({size_bytes:,} bytes)")
+35
View File
@@ -164,6 +164,41 @@ def send_password_reset_email(email: str, username: str, token: str, base_url: s
_smtp_send(subject, html_body, plain_body, email)
def send_forgot_username_email(email: str, username: str) -> None:
"""Send an email reminding the user of their username.
Args:
email: Recipient email address.
username: The user's username to include in the message.
"""
subject = "Your DocuElevate username"
html_body = f"""<!DOCTYPE html>
<html lang="en">
<head><meta charset="UTF-8"></head>
<body style="font-family:Arial,sans-serif;background:#f4f4f5;margin:0;padding:32px;">
<div style="max-width:480px;margin:0 auto;background:#fff;border-radius:12px;padding:32px;box-shadow:0 2px 8px rgba(0,0,0,.08);">
<h1 style="color:#4f46e5;font-size:24px;margin-bottom:8px;">Your Username</h1>
<p style="color:#374151;">You requested a reminder of your DocuElevate username.</p>
<div style="text-align:center;margin:32px 0;background:#f3f4f6;border-radius:8px;padding:20px;">
<p style="color:#6b7280;font-size:13px;margin-bottom:4px;">Your username is:</p>
<p style="color:#111827;font-size:22px;font-weight:700;font-family:monospace;">{username}</p>
</div>
<p style="color:#374151;font-size:14px;">You can sign in using your username <strong>or</strong> your email address.</p>
<p style="color:#6b7280;font-size:13px;margin-top:16px;">If you did not request this reminder, you can safely ignore this email.</p>
<hr style="border:none;border-top:1px solid #e5e7eb;margin:24px 0;">
<p style="color:#9ca3af;font-size:12px;text-align:center;">DocuElevate &middot; Intelligent Document Processing</p>
</div>
</body>
</html>"""
plain_body = (
f"You requested a reminder of your DocuElevate username.\n\n"
f"Your username is: {username}\n\n"
"You can sign in using your username or your email address.\n\n"
"If you did not request this, please ignore this email."
)
_smtp_send(subject, html_body, plain_body, email)
def build_session_user(user: object) -> dict:
"""Build the session user dict for a LocalUser, matching the OAuth session format.
+125
View File
@@ -844,6 +844,131 @@ problem.
---
**GET** `/api/admin/users/local`
List all local (email/password) user accounts with basic metadata.
---
**POST** `/api/admin/users/local`
Create a new local user account (admin-only, immediately active — no email verification required).
**Request body**:
```json
{
"email": "user@example.com",
"username": "alice",
"display_name": "Alice Smith",
"password": "securepassword",
"is_admin": false
}
```
---
**PATCH** `/api/admin/users/local/{local_user_id}`
Update an existing local user account. Only the provided (non-null) fields are modified.
If the email is changed, the associated `UserProfile.user_id` is also updated automatically.
**Request body** (all fields optional):
```json
{
"email": "newemail@example.com",
"display_name": "Alice Wonderland",
"is_admin": true,
"is_active": false
}
```
**Error Responses**:
- `404`: Local user not found
- `409`: New email already taken by another account
---
**POST** `/api/admin/users/local/{local_user_id}/send-password-reset`
Send a password reset email to a local user on their behalf. Useful when a user is locked out.
Returns `{"sent": true}` on success or `{"sent": false, "reason": "..."}` when SMTP is not
configured or sending fails (never returns an error status so the admin always gets feedback).
**Error Responses**:
- `404`: Local user not found
---
**POST** `/api/admin/users/local/{local_user_id}/set-password`
Directly set a new password for a local user without requiring an email token (last resort when
email delivery is unavailable). The user should be advised to change their password after logging in.
**Request body**:
```json
{
"password": "temporarypassword"
}
```
**Error Responses**:
- `404`: Local user not found
- `422`: Password shorter than 8 characters
---
**DELETE** `/api/admin/users/local/{local_user_id}`
Delete a local user account by numeric ID. The associated `UserProfile` is also removed. Documents
owned by this user are **not** deleted. Returns `204 No Content` on success.
---
### Local Authentication (self-service)
These endpoints are for local (email/password) users and do not require authentication.
**POST** `/api/auth/request-password-reset`
Send a password reset email. Always returns 200 to avoid leaking whether an email is registered.
**Request body**:
```json
{ "email": "user@example.com" }
```
---
**POST** `/api/auth/reset-password`
Set a new password using a valid reset token (received via email).
**Request body**:
```json
{
"token": "the-token-from-email",
"new_password": "newpassword",
"new_password_confirm": "newpassword"
}
```
**Error Responses**:
- `400`: Token is invalid or expired
- `422`: Passwords do not match
---
**POST** `/api/auth/forgot-username`
Send a username reminder email. Always returns 200 to avoid leaking whether an email is registered.
**Request body**:
```json
{ "email": "user@example.com" }
```
---
### Settings Suggestions (Autocomplete)
**GET** `/api/settings/{key}/suggestions`
+5 -1
View File
@@ -1056,9 +1056,13 @@ Webhook URLs, secrets, and subscribed events are configured per-webhook via the
### Backup & Restore
DocuElevate can automatically back up the SQLite database on a scheduled basis.
DocuElevate automatically backs up the database on a scheduled basis.
Backups are managed from the **Admin → Backup & Restore** dashboard.
Supported database backends: **SQLite** (`.db.gz`), **PostgreSQL** (`.pgsql.gz`), **MySQL / MariaDB** (`.mysql.gz`).
For PostgreSQL and MySQL backups the respective CLI client (`pg_dump` / `psql` or `mysqldump` / `mysql`) must be installed on the Celery worker host.
See the [Database Configuration Guide](DatabaseConfiguration.md#backup-procedures) for setup details.
| **Variable** | **Description** | **Default** |
|--------------------------------|-----------------------------------------------------------------------------------------------|---------------------|
| `BACKUP_ENABLED` | Enable or disable automatic scheduled backups (`True`/`False`). | `True` |
+71 -9
View File
@@ -341,29 +341,91 @@ Disable `prepared_statements` when using PgBouncer in transaction mode.
## Backup Procedures
### PostgreSQL
DocuElevate's built-in **Backup & Restore** feature (Admin → Backup & Restore) supports all three
database backends natively, using the native dump tools of each database.
**Manual backup:**
| Backend | Backup tool | Archive extension | Restore tool |
|----------------|--------------|-------------------|--------------|
| SQLite | `sqlite3` (built-in Python) | `.db.gz` | `sqlite3` (built-in Python) |
| PostgreSQL | `pg_dump` | `.pgsql.gz` | `psql` |
| MySQL/MariaDB | `mysqldump` | `.mysql.gz` | `mysql` |
Passwords are passed via the `PGPASSWORD` (PostgreSQL) and `MYSQL_PWD` (MySQL) environment
variables so they are never exposed on the process command line.
### Prerequisites
For PostgreSQL and MySQL backups the corresponding CLI client must be installed on the
worker host (the container / server that runs Celery workers):
```bash
# PostgreSQL clients (Debian/Ubuntu)
apt-get install -y postgresql-client
# MySQL clients (Debian/Ubuntu)
apt-get install -y default-mysql-client
```
The binaries required are:
- **PostgreSQL**: `pg_dump` (backup) and `psql` (restore)
- **MySQL / MariaDB**: `mysqldump` (backup) and `mysql` (restore)
### Using the Admin Dashboard
Navigate to **Admin → Backup & Restore** to:
- Trigger manual backups (hourly / daily / weekly)
- Download backup archives
- Upload and restore a backup archive
- Configure retention and remote storage destinations
### PostgreSQL manual backup/restore
**Manual backup using DocuElevate's archive format (for use with the UI restore):**
```bash
pg_dump --format=plain --no-password \
-h localhost -U docuelevate docuelevate \
| gzip > docuelevate_$(date +%Y%m%d_%H%M).pgsql.gz
```
**Restore via the DocuElevate UI:** upload the `.pgsql.gz` file on the Backup & Restore page.
**Manual restore using native tools (custom format):**
```bash
pg_dump -h localhost -U docuelevate -F c docuelevate > docuelevate_$(date +%Y%m%d_%H%M).dump
```
**Restore:**
```bash
pg_restore -h localhost -U docuelevate -d docuelevate docuelevate_20240101_1200.dump
```
**Automated daily backup (cron example):**
```cron
0 2 * * * pg_dump -h localhost -U docuelevate -F c docuelevate | gzip > /backups/docuelevate_$(date +\%Y\%m\%d).dump.gz
0 2 * * * pg_dump --format=plain -h localhost -U docuelevate docuelevate | gzip > /backups/docuelevate_$(date +\%Y\%m\%d).pgsql.gz
```
Use your cloud provider's automated backup feature when available (e.g., RDS automated snapshots, Cloud SQL backups).
### SQLite
### MySQL / MariaDB manual backup/restore
**Manual backup using DocuElevate's archive format (for use with the UI restore):**
```bash
MYSQL_PWD=yourpassword mysqldump --single-transaction --routines --triggers \
-h localhost -u docuelevate docuelevate \
| gzip > docuelevate_$(date +%Y%m%d_%H%M).mysql.gz
```
**Restore via the DocuElevate UI:** upload the `.mysql.gz` file on the Backup & Restore page.
**Manual restore using native tools:**
```bash
gunzip -c docuelevate_20240101_1200.mysql.gz | mysql -h localhost -u docuelevate -p docuelevate
```
### SQLite manual backup/restore
```bash
# Stop the application first, or use SQLite's online backup API
+23 -2
View File
@@ -28,8 +28,29 @@ If OpenID Connect authentication is configured:
3. Log in with your existing credentials on that platform
4. You'll be redirected back to DocuElevate after successful authentication
#### User Sessions
- Once authenticated, your session will remain active until you log out or it expires
#### Local User Accounts
If your administrator has created a local (email/password) account for you:
1. You'll see a "Sign in with username" form on the login page
2. Enter your **username or email address** — both are accepted
3. Enter your password and click **Sign in**
##### Forgot your password?
If you can't remember your password:
1. Click **Forgot password?** below the sign-in form
2. Enter your email address and click **Send reset link**
3. Check your inbox for a password reset email (valid for 24 hours)
4. Click the link in the email and enter your new password
##### Forgot your username?
If you can't remember your username:
1. Click **Forgot username?** below the sign-in form
2. Enter your email address and click **Send username reminder**
3. You'll receive an email with your username
> **Tip:** You can always sign in with your email address directly — you don't need to look up your username.
- Click the "Logout" button in the top navigation bar to end your session
- For security, sessions automatically expire after a period of inactivity
+253 -2
View File
@@ -508,13 +508,38 @@
</td>
<td class="px-4 py-3 text-sm text-gray-500 whitespace-nowrap" x-text="lu.created_at ? formatDate(lu.created_at) : '—'"></td>
<td class="px-4 py-3 text-sm text-right">
<button
type="button"
@click="openEditLocalUserModal(lu)"
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 account for ${lu.username}`"
>
<i class="fas fa-edit mr-1" aria-hidden="true"></i> Edit
</button>
<button
type="button"
@click="openSetPasswordModal(lu)"
class="inline-flex items-center px-2.5 py-1.5 text-xs font-medium rounded border border-yellow-300 text-yellow-700 bg-white hover:bg-yellow-50 focus:outline-none focus:ring-2 focus:ring-offset-1 focus:ring-yellow-400 mr-1"
:aria-label="`Set password for ${lu.username}`"
>
<i class="fas fa-key mr-1" aria-hidden="true"></i> Password
</button>
<button
type="button"
@click="sendPasswordReset(lu)"
class="inline-flex items-center px-2.5 py-1.5 text-xs font-medium rounded border border-indigo-300 text-indigo-600 bg-white hover:bg-indigo-50 focus:outline-none focus:ring-2 focus:ring-offset-1 focus:ring-indigo-400 mr-1"
:aria-label="`Send password reset email to ${lu.username}`"
title="Send password reset email"
>
<i class="fas fa-envelope mr-1" aria-hidden="true"></i> Reset
</button>
<button
type="button"
@click="confirmDeleteLocalUser(lu)"
class="text-red-600 hover:text-red-800 focus:outline-none"
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 account for ${lu.username}`"
>
<i class="fas fa-trash" aria-hidden="true"></i>
<i class="fas fa-trash mr-1" aria-hidden="true"></i> Delete
</button>
</td>
</tr>
@@ -595,6 +620,116 @@
</div>
</div>
<!-- ── Edit local user modal ────────────────────────────────────────────── -->
<div
x-show="editLocalUserModal.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="edit-local-user-title"
>
<div class="bg-white rounded-lg shadow-xl w-full max-w-lg" @click.outside="editLocalUserModal.open = false">
<div class="px-6 py-4 border-b flex items-center justify-between">
<h2 id="edit-local-user-title" class="text-lg font-semibold text-gray-900">Edit Local Account</h2>
<button type="button" @click="editLocalUserModal.open = false" aria-label="Close" class="text-gray-400 hover:text-gray-600">
<i class="fas fa-times" aria-hidden="true"></i>
</button>
</div>
<form @submit.prevent="submitEditLocalUser" class="px-6 py-5 space-y-4">
<div>
<label for="elu-email" class="block text-sm font-medium text-gray-700">Email <span aria-hidden="true" class="text-red-500">*</span></label>
<input type="email" id="elu-email" x-model="editLocalUserModal.form.email" required autocomplete="off"
class="mt-1 block w-full rounded-md border-gray-300 shadow-sm focus:border-blue-500 focus:ring focus:ring-blue-500 focus:ring-opacity-50 text-sm"
style="min-height:40px;" aria-required="true">
</div>
<div>
<label for="elu-display-name" class="block text-sm font-medium text-gray-700">Display Name <span class="text-gray-400">(optional)</span></label>
<input type="text" id="elu-display-name" x-model="editLocalUserModal.form.display_name" autocomplete="off" maxlength="255"
class="mt-1 block w-full rounded-md border-gray-300 shadow-sm focus:border-blue-500 focus:ring focus:ring-blue-500 focus:ring-opacity-50 text-sm"
style="min-height:40px;">
</div>
<div class="flex items-center gap-2">
<input type="checkbox" id="elu-is-admin" x-model="editLocalUserModal.form.is_admin"
class="h-4 w-4 rounded border-gray-300 text-blue-600 focus:ring-blue-500">
<label for="elu-is-admin" class="text-sm text-gray-700">Admin privileges</label>
</div>
<div class="flex items-center gap-2">
<input type="checkbox" id="elu-is-active" x-model="editLocalUserModal.form.is_active"
class="h-4 w-4 rounded border-gray-300 text-blue-600 focus:ring-blue-500">
<label for="elu-is-active" class="text-sm text-gray-700">Account active</label>
</div>
<div x-show="editLocalUserModal.error" x-cloak
class="bg-red-50 border-l-4 border-red-500 text-red-700 p-3 rounded text-sm"
role="alert" aria-live="assertive" x-text="editLocalUserModal.error">
</div>
<div class="flex justify-end gap-3 pt-2">
<button type="button" @click="editLocalUserModal.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="submit" :disabled="editLocalUserModal.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">
<span x-show="!editLocalUserModal.saving">Save Changes</span>
<span x-show="editLocalUserModal.saving" x-cloak><i class="fas fa-spinner fa-spin mr-1" aria-hidden="true"></i> Saving…</span>
</button>
</div>
</form>
</div>
</div>
<!-- ── Set password modal ────────────────────────────────────────────────── -->
<div
x-show="setPasswordModal.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="set-password-title"
>
<div class="bg-white rounded-lg shadow-xl w-full max-w-md" @click.outside="setPasswordModal.open = false">
<div class="px-6 py-4 border-b flex items-center justify-between">
<h2 id="set-password-title" class="text-lg font-semibold text-gray-900">Set Temporary Password</h2>
<button type="button" @click="setPasswordModal.open = false" aria-label="Close" class="text-gray-400 hover:text-gray-600">
<i class="fas fa-times" aria-hidden="true"></i>
</button>
</div>
<form @submit.prevent="submitSetPassword" class="px-6 py-5 space-y-4">
<p class="text-sm text-gray-600">
Set a new password directly for <strong class="font-mono" x-text="setPasswordModal.username"></strong>.
The user should change this password after logging in.
</p>
<div>
<label for="sp-password" class="block text-sm font-medium text-gray-700">New Password <span aria-hidden="true" class="text-red-500">*</span></label>
<input type="password" id="sp-password" x-model="setPasswordModal.password" required autocomplete="new-password"
minlength="8" maxlength="128"
class="mt-1 block w-full rounded-md border-gray-300 shadow-sm focus:border-yellow-500 focus:ring focus:ring-yellow-500 focus:ring-opacity-50 text-sm"
style="min-height:40px;" aria-required="true" aria-describedby="sp-password-hint">
<p id="sp-password-hint" class="mt-1 text-xs text-gray-500">Minimum 8 characters.</p>
</div>
<div x-show="setPasswordModal.error" x-cloak
class="bg-red-50 border-l-4 border-red-500 text-red-700 p-3 rounded text-sm"
role="alert" aria-live="assertive" x-text="setPasswordModal.error">
</div>
<div class="flex justify-end gap-3 pt-2">
<button type="button" @click="setPasswordModal.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="submit" :disabled="setPasswordModal.saving"
class="px-4 py-2 text-sm font-medium rounded-md text-white bg-yellow-600 hover:bg-yellow-700 focus:outline-none focus:ring-2 focus:ring-offset-1 focus:ring-yellow-500 disabled:opacity-50">
<span x-show="!setPasswordModal.saving">Set Password</span>
<span x-show="setPasswordModal.saving" x-cloak><i class="fas fa-spinner fa-spin mr-1" aria-hidden="true"></i> Setting…</span>
</button>
</div>
</form>
</div>
</div>
<!-- ── Delete local user confirmation modal ───────────────────────────────── -->
<div
x-show="deleteLocalUserModal.open"
@@ -724,6 +859,21 @@ function adminUsersApp() {
error: '',
form: { email: '', username: '', display_name: '', password: '', is_admin: false },
},
editLocalUserModal: {
open: false,
id: null,
saving: false,
error: '',
form: { email: '', display_name: '', is_admin: false, is_active: true },
},
setPasswordModal: {
open: false,
id: null,
username: '',
saving: false,
error: '',
password: '',
},
deleteLocalUserModal: {
open: false,
id: null,
@@ -947,6 +1097,107 @@ function adminUsersApp() {
this.deleteLocalUserModal.open = true;
},
openEditLocalUserModal(lu) {
this.editLocalUserModal.id = lu.id;
this.editLocalUserModal.form = {
email: lu.email,
display_name: lu.display_name || '',
is_admin: !!lu.is_admin,
is_active: !!lu.is_active,
};
this.editLocalUserModal.error = '';
this.editLocalUserModal.saving = false;
this.editLocalUserModal.open = true;
},
async submitEditLocalUser() {
this.editLocalUserModal.error = '';
this.editLocalUserModal.saving = true;
try {
const resp = await fetch(`/api/admin/users/local/${this.editLocalUserModal.id}`, {
method: 'PATCH',
headers: {
'Content-Type': 'application/json',
'X-CSRF-Token': document.querySelector('meta[name="csrf-token"]')?.content || '',
},
body: JSON.stringify({
email: this.editLocalUserModal.form.email || null,
display_name: this.editLocalUserModal.form.display_name,
is_admin: this.editLocalUserModal.form.is_admin,
is_active: this.editLocalUserModal.form.is_active,
}),
});
if (resp.ok) {
this.editLocalUserModal.open = false;
this.showAlert('success', 'Updated', `Account has been updated.`);
await this.fetchLocalUsers();
} else {
const err = await resp.json().catch(() => ({}));
this.editLocalUserModal.error = err.detail || 'Failed to update account.';
}
} catch (e) {
this.editLocalUserModal.error = 'Network error: ' + e.message;
} finally {
this.editLocalUserModal.saving = false;
}
},
openSetPasswordModal(lu) {
this.setPasswordModal.id = lu.id;
this.setPasswordModal.username = lu.username;
this.setPasswordModal.password = '';
this.setPasswordModal.error = '';
this.setPasswordModal.saving = false;
this.setPasswordModal.open = true;
},
async submitSetPassword() {
this.setPasswordModal.error = '';
this.setPasswordModal.saving = true;
try {
const resp = await fetch(`/api/admin/users/local/${this.setPasswordModal.id}/set-password`, {
method: 'POST',
headers: {
'Content-Type': 'application/json',
'X-CSRF-Token': document.querySelector('meta[name="csrf-token"]')?.content || '',
},
body: JSON.stringify({ password: this.setPasswordModal.password }),
});
if (resp.ok) {
this.setPasswordModal.open = false;
this.showAlert('success', 'Password set', `Password for "${this.setPasswordModal.username}" has been updated.`);
} else {
const err = await resp.json().catch(() => ({}));
this.setPasswordModal.error = err.detail || 'Failed to set password.';
}
} catch (e) {
this.setPasswordModal.error = 'Network error: ' + e.message;
} finally {
this.setPasswordModal.saving = false;
}
},
async sendPasswordReset(lu) {
try {
const resp = await fetch(`/api/admin/users/local/${lu.id}/send-password-reset`, {
method: 'POST',
headers: {
'X-CSRF-Token': document.querySelector('meta[name="csrf-token"]')?.content || '',
},
});
const data = await resp.json().catch(() => ({}));
if (resp.ok && data.sent) {
this.showAlert('success', 'Email sent', `Password reset email sent to "${lu.email}".`);
} else if (resp.ok && !data.sent) {
this.showAlert('error', 'Email not sent', data.reason || 'SMTP is not configured.');
} else {
this.showAlert('error', 'Failed', data.detail || resp.statusText);
}
} catch (e) {
this.showAlert('error', 'Network error', e.message);
}
},
async executeDeleteLocalUser() {
this.deleteLocalUserModal.deleting = true;
try {
+118
View File
@@ -0,0 +1,118 @@
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>DocuElevate - Forgot Password</title>
<link href="https://cdn.jsdelivr.net/npm/tailwindcss@2.2.19/dist/tailwind.min.css" rel="stylesheet">
<link rel="stylesheet" href="https://cdnjs.cloudflare.com/ajax/libs/font-awesome/6.5.1/css/all.min.css"
integrity="sha512-DTOQO9RWCH3ppGqcWaEA1BIZOC6xxalwEsw9c2QQeAIftl+Vegovlnee1c9QX4TctnWMn13TZye+giMm8e2LwA=="
crossorigin="anonymous" referrerpolicy="no-referrer" />
<script defer src="https://cdn.jsdelivr.net/npm/alpinejs@3.x.x/dist/cdn.min.js"></script>
</head>
<body class="bg-gray-100 min-h-screen flex items-center justify-center py-8">
<main class="bg-white rounded-lg shadow-lg p-8 max-w-md w-full" role="main">
<div class="flex justify-center mb-6">
<img src="/static/images/logo_writing.svg" alt="DocuElevate Logo" class="h-16">
</div>
<h1 class="text-2xl font-bold text-center text-gray-800 mb-2">Forgot your password?</h1>
<p class="text-center text-gray-500 text-sm mb-6">
Enter your email address and we'll send you a link to reset your password.
</p>
<div
x-data="{
email: '',
loading: false,
error: '',
success: false,
async submit() {
this.error = '';
this.loading = true;
try {
const resp = await fetch('/api/auth/request-password-reset', {
method: 'POST',
headers: {
'Content-Type': 'application/json',
'X-CSRF-Token': '{{ csrf_token }}'
},
body: JSON.stringify({ email: this.email })
});
if (resp.ok) {
this.success = true;
} else {
const data = await resp.json().catch(() => ({}));
this.error = data.detail || 'Something went wrong. Please try again.';
}
} catch(e) {
this.error = 'Network error. Please try again.';
} finally {
this.loading = false;
}
}
}"
>
<div x-show="success" x-cloak class="text-center py-4">
<div class="flex justify-center mb-4">
<div class="bg-green-100 rounded-full p-4">
<i class="fas fa-envelope-open-text text-green-600 text-4xl" aria-hidden="true"></i>
</div>
</div>
<p class="text-green-700 font-semibold mb-2">Check your inbox</p>
<p class="text-gray-500 text-sm mb-4">
If an account exists for that email address, a password reset link has been sent. The link expires in 24 hours.
</p>
<a href="/login"
class="inline-block py-2 px-6 rounded-md bg-indigo-600 text-white text-sm font-medium hover:bg-indigo-700 focus:outline-none focus:ring-2 focus:ring-offset-2 focus:ring-indigo-500"
style="min-height:44px;display:flex;align-items:center;justify-content:center;"
>Back to sign in</a>
</div>
<form x-show="!success" @submit.prevent="submit" class="space-y-4" novalidate>
<div x-show="error" x-cloak
class="bg-red-100 border-l-4 border-red-500 text-red-700 p-4 rounded"
role="alert" aria-live="polite">
<p x-text="error"></p>
</div>
<div>
<label for="email" class="block text-sm font-medium text-gray-700">
Email address <span aria-hidden="true" class="text-red-500">*</span>
</label>
<input
type="email" id="email" name="email" required
x-model="email"
autocomplete="email"
class="mt-1 block w-full rounded-md border-gray-300 shadow-sm focus:border-indigo-500 focus:ring focus:ring-indigo-500 focus:ring-opacity-50"
style="min-height:44px;"
aria-required="true"
placeholder="you@example.com"
>
</div>
<button
type="submit"
:disabled="loading"
class="w-full flex justify-center py-2 px-4 border border-transparent rounded-md shadow-sm text-sm font-medium text-white bg-indigo-600 hover:bg-indigo-700 focus:outline-none focus:ring-2 focus:ring-offset-2 focus:ring-indigo-500 disabled:opacity-50"
style="min-height:44px;"
>
<span x-show="!loading">Send reset link</span>
<span x-show="loading" x-cloak>
<i class="fas fa-spinner fa-spin mr-2" aria-hidden="true"></i>Sending&hellip;
</span>
</button>
</form>
</div>
<div class="mt-6 text-center">
<a href="/login" class="text-sm font-medium text-blue-600 hover:text-blue-500">
<i class="fas fa-arrow-left mr-1" aria-hidden="true"></i> Back to sign in
</a>
</div>
</main>
<div class="fixed bottom-4 text-center w-full text-xs text-gray-500">
DocuElevate {{ app_version|default('', true) }}
</div>
</body>
</html>
+125
View File
@@ -0,0 +1,125 @@
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>DocuElevate - Forgot Username</title>
<link href="https://cdn.jsdelivr.net/npm/tailwindcss@2.2.19/dist/tailwind.min.css" rel="stylesheet">
<link rel="stylesheet" href="https://cdnjs.cloudflare.com/ajax/libs/font-awesome/6.5.1/css/all.min.css"
integrity="sha512-DTOQO9RWCH3ppGqcWaEA1BIZOC6xxalwEsw9c2QQeAIftl+Vegovlnee1c9QX4TctnWMn13TZye+giMm8e2LwA=="
crossorigin="anonymous" referrerpolicy="no-referrer" />
<script defer src="https://cdn.jsdelivr.net/npm/alpinejs@3.x.x/dist/cdn.min.js"></script>
</head>
<body class="bg-gray-100 min-h-screen flex items-center justify-center py-8">
<main class="bg-white rounded-lg shadow-lg p-8 max-w-md w-full" role="main">
<div class="flex justify-center mb-6">
<img src="/static/images/logo_writing.svg" alt="DocuElevate Logo" class="h-16">
</div>
<h1 class="text-2xl font-bold text-center text-gray-800 mb-2">Forgot your username?</h1>
<p class="text-center text-gray-500 text-sm mb-6">
Enter the email address associated with your account and we'll send you your username.
You can also sign in directly with your email address.
</p>
<div
x-data="{
email: '',
loading: false,
error: '',
success: false,
async submit() {
this.error = '';
this.loading = true;
try {
const resp = await fetch('/api/auth/forgot-username', {
method: 'POST',
headers: {
'Content-Type': 'application/json',
'X-CSRF-Token': '{{ csrf_token }}'
},
body: JSON.stringify({ email: this.email })
});
if (resp.ok) {
this.success = true;
} else {
const data = await resp.json().catch(() => ({}));
this.error = data.detail || 'Something went wrong. Please try again.';
}
} catch(e) {
this.error = 'Network error. Please try again.';
} finally {
this.loading = false;
}
}
}"
>
<div x-show="success" x-cloak class="text-center py-4">
<div class="flex justify-center mb-4">
<div class="bg-green-100 rounded-full p-4">
<i class="fas fa-envelope-open-text text-green-600 text-4xl" aria-hidden="true"></i>
</div>
</div>
<p class="text-green-700 font-semibold mb-2">Check your inbox</p>
<p class="text-gray-500 text-sm mb-4">
If an account exists for that email address, your username has been sent.
Remember: you can also sign in using your email address directly.
</p>
<a href="/login"
class="inline-block py-2 px-6 rounded-md bg-indigo-600 text-white text-sm font-medium hover:bg-indigo-700 focus:outline-none focus:ring-2 focus:ring-offset-2 focus:ring-indigo-500"
style="min-height:44px;display:flex;align-items:center;justify-content:center;"
>Back to sign in</a>
</div>
<form x-show="!success" @submit.prevent="submit" class="space-y-4" novalidate>
<div x-show="error" x-cloak
class="bg-red-100 border-l-4 border-red-500 text-red-700 p-4 rounded"
role="alert" aria-live="polite">
<p x-text="error"></p>
</div>
<div class="bg-blue-50 border border-blue-200 rounded-md p-3 text-sm text-blue-700">
<i class="fas fa-info-circle mr-1" aria-hidden="true"></i>
<strong>Tip:</strong> You can sign in with either your username <em>or</em> your email address — no lookup needed.
</div>
<div>
<label for="email" class="block text-sm font-medium text-gray-700">
Email address <span aria-hidden="true" class="text-red-500">*</span>
</label>
<input
type="email" id="email" name="email" required
x-model="email"
autocomplete="email"
class="mt-1 block w-full rounded-md border-gray-300 shadow-sm focus:border-indigo-500 focus:ring focus:ring-indigo-500 focus:ring-opacity-50"
style="min-height:44px;"
aria-required="true"
placeholder="you@example.com"
>
</div>
<button
type="submit"
:disabled="loading"
class="w-full flex justify-center py-2 px-4 border border-transparent rounded-md shadow-sm text-sm font-medium text-white bg-indigo-600 hover:bg-indigo-700 focus:outline-none focus:ring-2 focus:ring-offset-2 focus:ring-indigo-500 disabled:opacity-50"
style="min-height:44px;"
>
<span x-show="!loading">Send username reminder</span>
<span x-show="loading" x-cloak>
<i class="fas fa-spinner fa-spin mr-2" aria-hidden="true"></i>Sending&hellip;
</span>
</button>
</form>
</div>
<div class="mt-6 text-center">
<a href="/login" class="text-sm font-medium text-blue-600 hover:text-blue-500">
<i class="fas fa-arrow-left mr-1" aria-hidden="true"></i> Back to sign in
</a>
</div>
</main>
<div class="fixed bottom-4 text-center w-full text-xs text-gray-500">
DocuElevate {{ app_version|default('', true) }}
</div>
</body>
</html>
+11 -1
View File
@@ -35,8 +35,9 @@
<form method="POST" action="/auth" class="space-y-4">
<input type="hidden" name="csrf_token" value="{{ csrf_token | default('', true) }}">
<div>
<label for="username" class="block text-sm font-medium text-gray-700">Username</label>
<label for="username" class="block text-sm font-medium text-gray-700">Username or Email</label>
<input type="text" id="username" name="username" required
autocomplete="username"
class="mt-1 block w-full rounded-md border-gray-300 shadow-sm focus:border-blue-500 focus:ring focus:ring-blue-500 focus:ring-opacity-50">
</div>
@@ -50,6 +51,15 @@
Sign in
</button>
</form>
<div class="mt-3 text-center space-x-3">
<a href="/forgot-password" class="text-sm text-blue-600 hover:text-blue-500">
Forgot password?
</a>
<span class="text-gray-300" aria-hidden="true">|</span>
<a href="/forgot-username" class="text-sm text-blue-600 hover:text-blue-500">
Forgot username?
</a>
</div>
</div>
{% if show_oauth %}
+265 -1
View File
@@ -10,6 +10,7 @@ Covers:
- Pagination and search filtering
"""
from datetime import datetime, timezone
from unittest.mock import MagicMock
import pytest
@@ -20,7 +21,7 @@ from sqlalchemy.orm import sessionmaker
from sqlalchemy.pool import StaticPool
from app.database import Base, get_db
from app.models import FileRecord, UserProfile
from app.models import FileRecord, LocalUser, UserProfile
# ---------------------------------------------------------------------------
# Fixtures
@@ -663,3 +664,266 @@ class TestEnsureUserProfileAdmin:
# No profile should have been created
count = au_session.query(UserProfile).count()
assert count == 0
# ---------------------------------------------------------------------------
# Local user admin management: update, send-password-reset, set-password
# ---------------------------------------------------------------------------
def _make_local_user(session, email: str = "lu@example.com", username: str = "luuser", **kwargs) -> LocalUser:
"""Insert a LocalUser row and return it."""
from app.utils.local_auth import hash_password
defaults = {
"hashed_password": hash_password("password123"),
"is_active": True,
"is_admin": False,
}
defaults.update(kwargs)
user = LocalUser(email=email, username=username, **defaults)
session.add(user)
session.commit()
session.refresh(user)
return user
class TestAdminUpdateLocalUser:
"""Tests for PATCH /api/admin/users/local/{id}."""
@pytest.mark.unit
def test_update_email(self, au_client, au_session):
"""PATCH can change the email address of a local user."""
user = _make_local_user(au_session, email="old@example.com", username="updateemail")
resp = au_client.patch(
f"/api/admin/users/local/{user.id}",
json={"email": "new@example.com"},
)
assert resp.status_code == 200
assert resp.json()["email"] == "new@example.com"
au_session.refresh(user)
assert user.email == "new@example.com"
@pytest.mark.unit
def test_update_email_syncs_user_profile(self, au_client, au_session):
"""PATCH email also updates UserProfile.user_id for the matching profile."""
user = _make_local_user(au_session, email="synced@example.com", username="synceduser")
_make_profile(au_session, "synced@example.com")
au_client.patch(
f"/api/admin/users/local/{user.id}",
json={"email": "synced_new@example.com"},
)
from app.models import UserProfile
old_profile = au_session.query(UserProfile).filter_by(user_id="synced@example.com").first()
new_profile = au_session.query(UserProfile).filter_by(user_id="synced_new@example.com").first()
assert old_profile is None
assert new_profile is not None
@pytest.mark.unit
def test_update_email_conflict_returns_409(self, au_client, au_session):
"""PATCH returns 409 when the new email is already taken."""
_make_local_user(au_session, email="taken@example.com", username="takenuser")
user = _make_local_user(au_session, email="mine@example.com", username="myuser")
resp = au_client.patch(
f"/api/admin/users/local/{user.id}",
json={"email": "taken@example.com"},
)
assert resp.status_code == 409
@pytest.mark.unit
def test_update_is_admin(self, au_client, au_session):
"""PATCH can grant or revoke admin privileges."""
user = _make_local_user(au_session, email="grantadmin@example.com", username="grantadmin")
assert user.is_admin is False
resp = au_client.patch(
f"/api/admin/users/local/{user.id}",
json={"is_admin": True},
)
assert resp.status_code == 200
assert resp.json()["is_admin"] is True
au_session.refresh(user)
assert user.is_admin is True
@pytest.mark.unit
def test_update_is_active(self, au_client, au_session):
"""PATCH can deactivate a user account."""
user = _make_local_user(au_session, email="deactivate@example.com", username="deactivateuser")
resp = au_client.patch(
f"/api/admin/users/local/{user.id}",
json={"is_active": False},
)
assert resp.status_code == 200
assert resp.json()["is_active"] is False
au_session.refresh(user)
assert user.is_active is False
@pytest.mark.unit
def test_update_display_name(self, au_client, au_session):
"""PATCH can update the display name."""
user = _make_local_user(au_session, email="displayname@example.com", username="displaynameuser")
resp = au_client.patch(
f"/api/admin/users/local/{user.id}",
json={"display_name": "Alice Wonderland"},
)
assert resp.status_code == 200
assert resp.json()["display_name"] == "Alice Wonderland"
@pytest.mark.unit
def test_update_nonexistent_user_returns_404(self, au_client):
"""PATCH on unknown ID returns 404."""
resp = au_client.patch("/api/admin/users/local/99999", json={"email": "x@example.com"})
assert resp.status_code == 404
class TestAdminSendPasswordReset:
"""Tests for POST /api/admin/users/local/{id}/send-password-reset."""
@pytest.mark.unit
def test_send_reset_email_success(self, au_client, au_session):
"""Returns sent=True when SMTP is configured and sending succeeds."""
from unittest.mock import patch
user = _make_local_user(au_session, email="resetme@example.com", username="resetmeuser")
with (
patch("app.api.admin_users.settings") as mock_settings,
patch("app.api.admin_users.send_password_reset_email") as mock_send,
):
mock_settings.email_host = "smtp.example.com"
mock_settings.version = "test"
resp = au_client.post(f"/api/admin/users/local/{user.id}/send-password-reset")
assert resp.status_code == 200
data = resp.json()
assert data["sent"] is True
assert data["email"] == "resetme@example.com"
mock_send.assert_called_once()
@pytest.mark.unit
def test_send_reset_email_no_smtp_returns_not_sent(self, au_client, au_session):
"""Returns sent=False with reason when SMTP is not configured."""
from unittest.mock import patch
user = _make_local_user(au_session, email="nosmtp@example.com", username="nosmtpuser")
with patch("app.api.admin_users.settings") as mock_settings:
mock_settings.email_host = ""
resp = au_client.post(f"/api/admin/users/local/{user.id}/send-password-reset")
assert resp.status_code == 200
data = resp.json()
assert data["sent"] is False
assert "smtp" in data["reason"].lower()
@pytest.mark.unit
def test_send_reset_email_smtp_failure_returns_not_sent(self, au_client, au_session):
"""Returns sent=False with reason when SMTP sending fails."""
from unittest.mock import patch
user = _make_local_user(au_session, email="smtperr@example.com", username="smtperruser")
with (
patch("app.api.admin_users.settings") as mock_settings,
patch("app.api.admin_users.send_password_reset_email", side_effect=RuntimeError("connection refused")),
):
mock_settings.email_host = "smtp.example.com"
resp = au_client.post(f"/api/admin/users/local/{user.id}/send-password-reset")
assert resp.status_code == 200
assert resp.json()["sent"] is False
@pytest.mark.unit
def test_send_reset_email_unknown_user_returns_404(self, au_client):
"""Returns 404 for unknown local_user_id."""
resp = au_client.post("/api/admin/users/local/99999/send-password-reset")
assert resp.status_code == 404
@pytest.mark.unit
def test_send_reset_stores_token(self, au_client, au_session):
"""Password reset token is persisted to the DB."""
from unittest.mock import patch
user = _make_local_user(au_session, email="tokenstore@example.com", username="tokenstoreuser")
assert user.password_reset_token is None
with (
patch("app.api.admin_users.settings") as mock_settings,
patch("app.api.admin_users.send_password_reset_email"),
):
mock_settings.email_host = "smtp.example.com"
au_client.post(f"/api/admin/users/local/{user.id}/send-password-reset")
au_session.refresh(user)
assert user.password_reset_token is not None
assert user.password_reset_sent_at is not None
class TestAdminSetPassword:
"""Tests for POST /api/admin/users/local/{id}/set-password."""
@pytest.mark.unit
def test_set_password_success(self, au_client, au_session):
"""Returns updated=True and changes the hashed password."""
from app.utils.local_auth import verify_password
user = _make_local_user(au_session, email="setpw@example.com", username="setpwuser")
resp = au_client.post(
f"/api/admin/users/local/{user.id}/set-password",
json={"password": "brandnewpassword"},
)
assert resp.status_code == 200
assert resp.json()["updated"] is True
au_session.refresh(user)
assert verify_password("brandnewpassword", user.hashed_password)
@pytest.mark.unit
def test_set_password_too_short_returns_422(self, au_client, au_session):
"""Returns 422 when password is shorter than 8 characters."""
user = _make_local_user(au_session, email="shortpw@example.com", username="shortpwuser")
resp = au_client.post(
f"/api/admin/users/local/{user.id}/set-password",
json={"password": "short"},
)
assert resp.status_code == 422
@pytest.mark.unit
def test_set_password_clears_reset_token(self, au_client, au_session):
"""Setting a password clears any outstanding password_reset_token."""
from app.utils.local_auth import generate_token
user = _make_local_user(au_session, email="cleartok@example.com", username="cleartokuser")
user.password_reset_token = generate_token()
user.password_reset_sent_at = datetime.now(tz=timezone.utc)
au_session.commit()
au_client.post(
f"/api/admin/users/local/{user.id}/set-password",
json={"password": "clearedpassword"},
)
au_session.refresh(user)
assert user.password_reset_token is None
assert user.password_reset_sent_at is None
@pytest.mark.unit
def test_set_password_unknown_user_returns_404(self, au_client):
"""Returns 404 for unknown local_user_id."""
resp = au_client.post(
"/api/admin/users/local/99999/set-password",
json={"password": "doesnotmatter"},
)
assert resp.status_code == 404
+444 -17
View File
@@ -236,6 +236,254 @@ class TestBackupTaskHelpers:
result = _db_path()
assert result is None
def test_db_backend_sqlite(self):
"""_db_backend() returns 'sqlite' for SQLite URLs."""
from app.tasks.backup_tasks import _db_backend
with patch("app.tasks.backup_tasks.settings") as mock_settings:
mock_settings.database_url = "sqlite:////tmp/test.db"
assert _db_backend() == "sqlite"
def test_db_backend_postgresql(self):
"""_db_backend() returns 'postgresql' for PostgreSQL URLs."""
from app.tasks.backup_tasks import _db_backend
with patch("app.tasks.backup_tasks.settings") as mock_settings:
mock_settings.database_url = "postgresql://user:pass@localhost/db"
assert _db_backend() == "postgresql"
def test_db_backend_mysql(self):
"""_db_backend() returns 'mysql' for MySQL URLs."""
from app.tasks.backup_tasks import _db_backend
with patch("app.tasks.backup_tasks.settings") as mock_settings:
mock_settings.database_url = "mysql+pymysql://user:pass@localhost/db"
assert _db_backend() == "mysql"
def test_archive_ext_sqlite(self):
"""_archive_ext_for_backend() returns '.db.gz' for sqlite."""
from app.tasks.backup_tasks import _archive_ext_for_backend
assert _archive_ext_for_backend("sqlite") == ".db.gz"
def test_archive_ext_postgresql(self):
"""_archive_ext_for_backend() returns '.pgsql.gz' for postgresql."""
from app.tasks.backup_tasks import _archive_ext_for_backend
assert _archive_ext_for_backend("postgresql") == ".pgsql.gz"
def test_archive_ext_mysql(self):
"""_archive_ext_for_backend() returns '.mysql.gz' for mysql."""
from app.tasks.backup_tasks import _archive_ext_for_backend
assert _archive_ext_for_backend("mysql") == ".mysql.gz"
def test_archive_ext_unknown(self):
"""_archive_ext_for_backend() falls back to '.sql.gz' for unknown backends."""
from app.tasks.backup_tasks import _archive_ext_for_backend
assert _archive_ext_for_backend("mssql") == ".sql.gz"
def test_dump_postgresql_success(self, tmp_path):
"""_dump_postgresql() streams pg_dump output into a gzip archive."""
from unittest.mock import MagicMock
from app.tasks.backup_tasks import _dump_postgresql
dest = tmp_path / "dump.pgsql.gz"
fake_sql = b"-- PostgreSQL database dump\nSELECT 1;\n"
mock_proc = MagicMock()
mock_proc.stdout.read.side_effect = [fake_sql, b""]
mock_proc.stderr.read.return_value = b""
mock_proc.returncode = 0
with patch("app.tasks.backup_tasks.subprocess.Popen", return_value=mock_proc):
_dump_postgresql("postgresql://user:pass@localhost/testdb", dest)
assert dest.exists()
with gzip.open(str(dest), "rb") as gz:
assert gz.read() == fake_sql
def test_dump_postgresql_failure(self, tmp_path):
"""_dump_postgresql() raises RuntimeError when pg_dump fails."""
from unittest.mock import MagicMock
from app.tasks.backup_tasks import _dump_postgresql
dest = tmp_path / "dump.pgsql.gz"
mock_proc = MagicMock()
mock_proc.stdout.read.side_effect = [b""]
mock_proc.stderr.read.return_value = b"FATAL: connection refused"
mock_proc.returncode = 1
with patch("app.tasks.backup_tasks.subprocess.Popen", return_value=mock_proc):
with pytest.raises(RuntimeError, match="pg_dump exited with code 1"):
_dump_postgresql("postgresql://user:pass@localhost/testdb", dest)
def test_dump_mysql_success(self, tmp_path):
"""_dump_mysql() streams mysqldump output into a gzip archive."""
from unittest.mock import MagicMock
from app.tasks.backup_tasks import _dump_mysql
dest = tmp_path / "dump.mysql.gz"
fake_sql = b"-- MySQL dump\nCREATE TABLE t (id INT);\n"
mock_proc = MagicMock()
mock_proc.stdout.read.side_effect = [fake_sql, b""]
mock_proc.stderr.read.return_value = b""
mock_proc.returncode = 0
with patch("app.tasks.backup_tasks.subprocess.Popen", return_value=mock_proc):
_dump_mysql("mysql+pymysql://user:pass@localhost/testdb", dest)
assert dest.exists()
with gzip.open(str(dest), "rb") as gz:
assert gz.read() == fake_sql
def test_dump_mysql_failure(self, tmp_path):
"""_dump_mysql() raises RuntimeError when mysqldump fails."""
from unittest.mock import MagicMock
from app.tasks.backup_tasks import _dump_mysql
dest = tmp_path / "dump.mysql.gz"
mock_proc = MagicMock()
mock_proc.stdout.read.side_effect = [b""]
mock_proc.stderr.read.return_value = b"ERROR: Access denied"
mock_proc.returncode = 1
with patch("app.tasks.backup_tasks.subprocess.Popen", return_value=mock_proc):
with pytest.raises(RuntimeError, match="mysqldump exited with code 1"):
_dump_mysql("mysql+pymysql://user:pass@localhost/testdb", dest)
def test_restore_sqlite_success(self, tmp_path):
"""_restore_sqlite() applies a valid SQL dump to a SQLite file."""
from app.tasks.backup_tasks import _restore_sqlite
db_file = tmp_path / "test.db"
conn = sqlite3.connect(str(db_file))
conn.execute("CREATE TABLE old (id INTEGER)")
conn.commit()
conn.close()
# Create a valid dump archive
sql = "BEGIN TRANSACTION;\nCREATE TABLE new_tbl (x TEXT);\nCOMMIT;\n"
archive = tmp_path / "dump.db.gz"
with gzip.open(str(archive), "wt") as gz:
gz.write(sql)
_restore_sqlite(db_file, archive)
conn2 = sqlite3.connect(str(db_file))
tables = [r[0] for r in conn2.execute("SELECT name FROM sqlite_master WHERE type='table'")]
conn2.close()
assert "new_tbl" in tables
def test_restore_sqlite_invalid_gz(self, tmp_path):
"""_restore_sqlite() raises ValueError for corrupt gzip content."""
from app.tasks.backup_tasks import _restore_sqlite
db_file = tmp_path / "test.db"
db_file.write_bytes(b"")
archive = tmp_path / "bad.db.gz"
archive.write_bytes(b"not gzip data")
with pytest.raises(ValueError, match="Failed to decompress"):
_restore_sqlite(db_file, archive)
def test_restore_sqlite_invalid_sql(self, tmp_path):
"""_restore_sqlite() raises ValueError for invalid SQL content."""
from app.tasks.backup_tasks import _restore_sqlite
db_file = tmp_path / "test.db"
db_file.write_bytes(b"")
archive = tmp_path / "bad.db.gz"
with gzip.open(str(archive), "wt") as gz:
gz.write("THIS IS NOT VALID SQL!!!;\n")
with pytest.raises(ValueError, match="invalid SQL"):
_restore_sqlite(db_file, archive)
def test_restore_postgresql_success(self, tmp_path):
"""_restore_postgresql() pipes the archive to psql."""
from unittest.mock import MagicMock
from app.tasks.backup_tasks import _restore_postgresql
fake_sql = b"-- PostgreSQL dump\nSELECT 1;\n"
archive = tmp_path / "dump.pgsql.gz"
with gzip.open(str(archive), "wb") as gz:
gz.write(fake_sql)
mock_proc = MagicMock()
mock_proc.communicate.return_value = (b"", b"")
mock_proc.returncode = 0
with patch("app.tasks.backup_tasks.subprocess.Popen", return_value=mock_proc):
_restore_postgresql("postgresql://user:pass@localhost/testdb", archive)
mock_proc.communicate.assert_called_once_with(input=fake_sql)
def test_restore_postgresql_failure(self, tmp_path):
"""_restore_postgresql() raises RuntimeError when psql fails."""
from unittest.mock import MagicMock
from app.tasks.backup_tasks import _restore_postgresql
archive = tmp_path / "dump.pgsql.gz"
with gzip.open(str(archive), "wb") as gz:
gz.write(b"SELECT 1;")
mock_proc = MagicMock()
mock_proc.communicate.return_value = (b"", b"ERROR: invalid input")
mock_proc.returncode = 1
with patch("app.tasks.backup_tasks.subprocess.Popen", return_value=mock_proc):
with pytest.raises(RuntimeError, match="psql exited with code 1"):
_restore_postgresql("postgresql://user:pass@localhost/testdb", archive)
def test_restore_mysql_success(self, tmp_path):
"""_restore_mysql() pipes the archive to mysql."""
from unittest.mock import MagicMock
from app.tasks.backup_tasks import _restore_mysql
fake_sql = b"-- MySQL dump\nSELECT 1;\n"
archive = tmp_path / "dump.mysql.gz"
with gzip.open(str(archive), "wb") as gz:
gz.write(fake_sql)
mock_proc = MagicMock()
mock_proc.communicate.return_value = (b"", b"")
mock_proc.returncode = 0
with patch("app.tasks.backup_tasks.subprocess.Popen", return_value=mock_proc):
_restore_mysql("mysql+pymysql://user:pass@localhost/testdb", archive)
mock_proc.communicate.assert_called_once_with(input=fake_sql)
def test_restore_mysql_failure(self, tmp_path):
"""_restore_mysql() raises RuntimeError when mysql fails."""
from unittest.mock import MagicMock
from app.tasks.backup_tasks import _restore_mysql
archive = tmp_path / "dump.mysql.gz"
with gzip.open(str(archive), "wb") as gz:
gz.write(b"SELECT 1;")
mock_proc = MagicMock()
mock_proc.communicate.return_value = (b"", b"ERROR: Access denied")
mock_proc.returncode = 1
with patch("app.tasks.backup_tasks.subprocess.Popen", return_value=mock_proc):
with pytest.raises(RuntimeError, match="mysql exited with code 1"):
_restore_mysql("mysql+pymysql://user:pass@localhost/testdb", archive)
def test_apply_retention_prunes_old(self, tmp_path, db_session):
"""_apply_retention() deletes backups beyond the retention limit."""
from app.tasks.backup_tasks import _apply_retention
@@ -352,20 +600,28 @@ class TestCreateBackupTask:
result = create_backup("hourly")
assert result["status"] == "disabled"
def test_non_sqlite_db(self):
"""create_backup returns unsupported_db for non-SQLite databases."""
def test_unsupported_db_backend(self):
"""create_backup returns unsupported_db for backends other than sqlite/postgresql/mysql."""
from app.tasks.backup_tasks import create_backup
with (
patch("app.tasks.backup_tasks.settings") as mock_settings,
patch("app.tasks.backup_tasks._db_path", return_value=None),
):
with patch("app.tasks.backup_tasks.settings") as mock_settings:
mock_settings.backup_enabled = True
mock_settings.database_url = "mssql+pyodbc://user:pass@server/db"
result = create_backup("hourly")
assert result["status"] == "unsupported_db"
def test_in_memory_sqlite_unsupported(self):
"""create_backup returns unsupported_db for in-memory SQLite."""
from app.tasks.backup_tasks import create_backup
with patch("app.tasks.backup_tasks.settings") as mock_settings:
mock_settings.backup_enabled = True
mock_settings.database_url = "sqlite:///:memory:"
result = create_backup("hourly")
assert result["status"] == "unsupported_db"
def test_missing_db_file(self, tmp_path):
"""create_backup returns error when the DB file does not exist."""
"""create_backup returns error when the SQLite DB file does not exist."""
from app.tasks.backup_tasks import create_backup
missing = tmp_path / "does_not_exist.db"
@@ -375,6 +631,7 @@ class TestCreateBackupTask:
patch("app.tasks.backup_tasks._db_path", return_value=missing),
):
mock_settings.backup_enabled = True
mock_settings.database_url = f"sqlite:///{missing}"
result = create_backup("hourly")
assert result["status"] == "error"
@@ -400,6 +657,7 @@ class TestCreateBackupTask:
):
backup_dir.mkdir(parents=True, exist_ok=True)
mock_settings.backup_enabled = True
mock_settings.database_url = f"sqlite:///{db_file}"
mock_db = MagicMock()
mock_sl.return_value.__enter__ = MagicMock(return_value=mock_db)
mock_sl.return_value.__exit__ = MagicMock(return_value=False)
@@ -408,6 +666,69 @@ class TestCreateBackupTask:
assert result["status"] == "ok"
assert "filename" in result
assert result["filename"].startswith("backup_hourly_")
assert result["filename"].endswith(".db.gz")
def test_successful_backup_postgresql(self, tmp_path):
"""create_backup creates a .pgsql.gz archive for PostgreSQL databases."""
from app.tasks.backup_tasks import create_backup
backup_dir = tmp_path / "backups"
backup_dir.mkdir()
def fake_pg_dump(db_url: str, dest: Path) -> None:
with gzip.open(str(dest), "wb") as gz:
gz.write(b"-- PostgreSQL dump\n")
with (
patch("app.tasks.backup_tasks.settings") as mock_settings,
patch("app.tasks.backup_tasks._backup_dir", return_value=backup_dir),
patch("app.tasks.backup_tasks._dump_postgresql", side_effect=fake_pg_dump),
patch("app.tasks.backup_tasks._upload_remote", return_value=None),
patch("app.tasks.backup_tasks._apply_retention"),
patch("app.tasks.backup_tasks._prune_remote_backups"),
patch("app.tasks.backup_tasks.SessionLocal") as mock_sl,
):
mock_settings.backup_enabled = True
mock_settings.database_url = "postgresql://user:pass@localhost/testdb"
mock_db = MagicMock()
mock_sl.return_value.__enter__ = MagicMock(return_value=mock_db)
mock_sl.return_value.__exit__ = MagicMock(return_value=False)
result = create_backup("daily")
assert result["status"] == "ok"
assert result["filename"].endswith(".pgsql.gz")
assert "daily" in result["filename"]
def test_successful_backup_mysql(self, tmp_path):
"""create_backup creates a .mysql.gz archive for MySQL databases."""
from app.tasks.backup_tasks import create_backup
backup_dir = tmp_path / "backups"
backup_dir.mkdir()
def fake_mysql_dump(db_url: str, dest: Path) -> None:
with gzip.open(str(dest), "wb") as gz:
gz.write(b"-- MySQL dump\n")
with (
patch("app.tasks.backup_tasks.settings") as mock_settings,
patch("app.tasks.backup_tasks._backup_dir", return_value=backup_dir),
patch("app.tasks.backup_tasks._dump_mysql", side_effect=fake_mysql_dump),
patch("app.tasks.backup_tasks._upload_remote", return_value=None),
patch("app.tasks.backup_tasks._apply_retention"),
patch("app.tasks.backup_tasks._prune_remote_backups"),
patch("app.tasks.backup_tasks.SessionLocal") as mock_sl,
):
mock_settings.backup_enabled = True
mock_settings.database_url = "mysql+pymysql://user:pass@localhost/testdb"
mock_db = MagicMock()
mock_sl.return_value.__enter__ = MagicMock(return_value=mock_db)
mock_sl.return_value.__exit__ = MagicMock(return_value=False)
result = create_backup("weekly")
assert result["status"] == "ok"
assert result["filename"].endswith(".mysql.gz")
assert "weekly" in result["filename"]
def test_invalid_backup_type_defaults_to_hourly(self, tmp_path):
"""create_backup normalises unknown backup_type to 'hourly'."""
@@ -431,6 +752,7 @@ class TestCreateBackupTask:
patch("app.tasks.backup_tasks.SessionLocal") as mock_sl,
):
mock_settings.backup_enabled = True
mock_settings.database_url = f"sqlite:///{db_file}"
mock_db = MagicMock()
mock_sl.return_value.__enter__ = MagicMock(return_value=mock_db)
mock_sl.return_value.__exit__ = MagicMock(return_value=False)
@@ -549,19 +871,24 @@ class TestBackupAPIEndpoints:
assert resp.status_code == 403
def test_restore_wrong_extension(self, admin_client):
"""POST /api/admin/backup/restore rejects non-.db.gz files."""
"""POST /api/admin/backup/restore rejects files with wrong extension for current backend."""
# Default test env uses sqlite:///:memory: → expects .db.gz
resp = admin_client.post(
"/api/admin/backup/restore",
files={"file": ("backup.zip", b"data", "application/zip")},
)
assert resp.status_code == 400
def test_restore_invalid_gz_content(self, admin_client):
def test_restore_invalid_gz_content(self, admin_client, tmp_path):
"""POST /api/admin/backup/restore rejects corrupt gzip data."""
resp = admin_client.post(
"/api/admin/backup/restore",
files={"file": ("backup.db.gz", b"not gzip data at all", "application/gzip")},
)
db_file = tmp_path / "test.db"
db_file.write_bytes(b"")
with patch("app.tasks.backup_tasks._db_path", return_value=db_file):
resp = admin_client.post(
"/api/admin/backup/restore",
files={"file": ("backup.db.gz", b"not gzip data at all", "application/gzip")},
)
assert resp.status_code == 400
def test_restore_valid_archive(self, admin_client, tmp_path):
@@ -581,11 +908,11 @@ class TestBackupAPIEndpoints:
assert resp.status_code == 200
assert resp.json()["status"] == "restored"
def test_restore_non_sqlite_db(self, admin_client):
"""POST /api/admin/backup/restore returns 400 for non-SQLite database."""
sql = "BEGIN TRANSACTION;\nCOMMIT;\n"
gz_data = gzip.compress(sql.encode())
def test_restore_in_memory_sqlite(self, admin_client):
"""POST /api/admin/backup/restore returns 400 for in-memory SQLite (no file to restore to)."""
gz_data = gzip.compress(b"BEGIN TRANSACTION;\nCOMMIT;\n")
# _db_path() returns None for :memory: URLs → 400
with patch("app.tasks.backup_tasks._db_path", return_value=None):
resp = admin_client.post(
"/api/admin/backup/restore",
@@ -593,6 +920,106 @@ class TestBackupAPIEndpoints:
)
assert resp.status_code == 400
def test_restore_wrong_extension_for_postgresql(self, admin_client):
"""POST /api/admin/backup/restore returns 400 when uploading .db.gz for PostgreSQL backend."""
gz_data = gzip.compress(b"-- PostgreSQL dump")
with patch("app.config.settings") as mock_settings:
mock_settings.database_url = "postgresql://user:pass@localhost/testdb"
resp = admin_client.post(
"/api/admin/backup/restore",
files={"file": ("backup.db.gz", gz_data, "application/gzip")},
)
assert resp.status_code == 400
def test_restore_wrong_extension_for_mysql(self, admin_client):
"""POST /api/admin/backup/restore returns 400 when uploading .db.gz for MySQL backend."""
gz_data = gzip.compress(b"-- MySQL dump")
with patch("app.config.settings") as mock_settings:
mock_settings.database_url = "mysql+pymysql://user:pass@localhost/testdb"
resp = admin_client.post(
"/api/admin/backup/restore",
files={"file": ("backup.db.gz", gz_data, "application/gzip")},
)
assert resp.status_code == 400
def test_restore_postgresql_success(self, admin_client):
"""POST /api/admin/backup/restore succeeds for PostgreSQL database."""
gz_data = gzip.compress(b"-- PostgreSQL dump\n")
with (
patch("app.config.settings") as mock_settings,
patch("app.tasks.backup_tasks._restore_postgresql") as mock_restore,
):
mock_settings.database_url = "postgresql://user:pass@localhost/testdb"
resp = admin_client.post(
"/api/admin/backup/restore",
files={"file": ("backup.pgsql.gz", gz_data, "application/gzip")},
)
assert resp.status_code == 200
assert resp.json()["status"] == "restored"
mock_restore.assert_called_once()
def test_restore_mysql_success(self, admin_client):
"""POST /api/admin/backup/restore succeeds for MySQL database."""
gz_data = gzip.compress(b"-- MySQL dump\n")
with (
patch("app.config.settings") as mock_settings,
patch("app.tasks.backup_tasks._restore_mysql") as mock_restore,
):
mock_settings.database_url = "mysql+pymysql://user:pass@localhost/testdb"
resp = admin_client.post(
"/api/admin/backup/restore",
files={"file": ("backup.mysql.gz", gz_data, "application/gzip")},
)
assert resp.status_code == 200
assert resp.json()["status"] == "restored"
mock_restore.assert_called_once()
def test_restore_postgresql_runtime_error(self, admin_client):
"""POST /api/admin/backup/restore returns 500 when psql command fails."""
gz_data = gzip.compress(b"-- PostgreSQL dump\n")
with (
patch("app.config.settings") as mock_settings,
patch("app.tasks.backup_tasks._restore_postgresql", side_effect=RuntimeError("psql failed")),
):
mock_settings.database_url = "postgresql://user:pass@localhost/testdb"
resp = admin_client.post(
"/api/admin/backup/restore",
files={"file": ("backup.pgsql.gz", gz_data, "application/gzip")},
)
assert resp.status_code == 500
def test_restore_postgresql_missing_binary(self, admin_client):
"""POST /api/admin/backup/restore returns 500 when psql binary is missing."""
gz_data = gzip.compress(b"-- PostgreSQL dump\n")
with (
patch("app.config.settings") as mock_settings,
patch("app.tasks.backup_tasks._restore_postgresql", side_effect=FileNotFoundError("psql not found")),
):
mock_settings.database_url = "postgresql://user:pass@localhost/testdb"
resp = admin_client.post(
"/api/admin/backup/restore",
files={"file": ("backup.pgsql.gz", gz_data, "application/gzip")},
)
assert resp.status_code == 500
def test_restore_unsupported_backend(self, admin_client):
"""POST /api/admin/backup/restore returns 400 for an unsupported database backend."""
gz_data = gzip.compress(b"-- some dump\n")
with patch("app.config.settings") as mock_settings:
mock_settings.database_url = "mssql+pyodbc://user:pass@server/db"
resp = admin_client.post(
"/api/admin/backup/restore",
files={"file": ("backup.sql.gz", gz_data, "application/gzip")},
)
assert resp.status_code == 400
# ---------------------------------------------------------------------------
# View tests
+116
View File
@@ -6,9 +6,12 @@ Covers:
- POST /api/auth/resend-verification
- POST /api/auth/request-password-reset
- POST /api/auth/reset-password
- POST /api/auth/forgot-username
- GET /signup (page route)
- GET /verify-email-sent (page route)
- GET /reset-password (page route)
- GET /forgot-password (page route)
- GET /forgot-username (page route)
- app/utils/local_auth utility functions
- auth() login flow with LocalUser
"""
@@ -801,3 +804,116 @@ def test_admin_local_user_list_after_create(admin_session_client):
assert resp.status_code == 200
users = resp.json()
assert any(u["email"] == "listed@example.com" for u in users)
# ---------------------------------------------------------------------------
# Integration tests: forgot-username endpoint
# ---------------------------------------------------------------------------
@pytest.mark.integration
def test_forgot_username_returns_200_for_existing_email(la_client, la_session):
"""POST /api/auth/forgot-username returns 200 and sends email when account exists."""
la_session.add(
LocalUser(
email="remindme@example.com",
username="remindmeuser",
hashed_password=hash_password("pw123456"),
is_active=True,
)
)
la_session.commit()
with patch("app.api.local_auth.send_forgot_username_email") as mock_send:
resp = la_client.post("/api/auth/forgot-username", json={"email": "remindme@example.com"})
assert resp.status_code == 200
assert "reminder" in resp.json()["message"].lower()
mock_send.assert_called_once_with("remindme@example.com", "remindmeuser")
@pytest.mark.integration
def test_forgot_username_returns_200_for_unknown_email(la_client):
"""POST /api/auth/forgot-username always returns 200 (no info leak)."""
with patch("app.api.local_auth.send_forgot_username_email") as mock_send:
resp = la_client.post("/api/auth/forgot-username", json={"email": "nobody@example.com"})
assert resp.status_code == 200
mock_send.assert_not_called()
@pytest.mark.integration
def test_forgot_username_smtp_failure_does_not_raise(la_client, la_session):
"""POST /api/auth/forgot-username returns 200 even when SMTP fails."""
la_session.add(
LocalUser(
email="smtpfail@example.com",
username="smtpfailuser",
hashed_password=hash_password("pw123456"),
is_active=True,
)
)
la_session.commit()
with patch("app.api.local_auth.send_forgot_username_email", side_effect=RuntimeError("SMTP down")):
resp = la_client.post("/api/auth/forgot-username", json={"email": "smtpfail@example.com"})
assert resp.status_code == 200
# ---------------------------------------------------------------------------
# Integration tests: new page routes
# ---------------------------------------------------------------------------
@pytest.mark.integration
def test_forgot_password_page(la_client):
"""GET /forgot-password returns 200."""
with patch("app.api.local_auth.settings") as mock_settings:
mock_settings.version = "test"
resp = la_client.get("/forgot-password")
assert resp.status_code == 200
assert b"password" in resp.content.lower()
@pytest.mark.integration
def test_forgot_username_page(la_client):
"""GET /forgot-username returns 200."""
with patch("app.api.local_auth.settings") as mock_settings:
mock_settings.version = "test"
resp = la_client.get("/forgot-username")
assert resp.status_code == 200
assert b"username" in resp.content.lower()
# ---------------------------------------------------------------------------
# Unit tests: send_forgot_username_email utility
# ---------------------------------------------------------------------------
@pytest.mark.unit
def test_send_forgot_username_email_calls_smtp():
"""send_forgot_username_email calls _smtp_send with the username."""
from app.utils.local_auth import send_forgot_username_email
with patch("app.utils.local_auth._smtp_send") as mock_smtp:
send_forgot_username_email("u@example.com", "myusername")
mock_smtp.assert_called_once()
args = mock_smtp.call_args[0]
# subject, html_body, plain_body, recipient
assert "myusername" in args[1] # HTML body
assert "myusername" in args[2] # plain body
assert args[3] == "u@example.com"
@pytest.mark.unit
def test_send_forgot_username_email_no_smtp_raises():
"""send_forgot_username_email raises RuntimeError when EMAIL_HOST is not set."""
from app.utils.local_auth import send_forgot_username_email
with patch("app.utils.local_auth.settings") as mock_settings:
mock_settings.email_host = ""
with pytest.raises(RuntimeError, match="SMTP"):
send_forgot_username_email("u@example.com", "myusername")