gitea-admin/gh-christianlouis-docuelevate
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #260 from christianlouis/copilot/format-code-with-linters

Browse Source 
style: fix black, isort, and flake8 violations across app/ and tests/
This commit is contained in:
Christian Krakau-Louis
2026-02-12 04:24:04 +01:00
committed by GitHub
parent e827cfffe2 ff9a3ff49f
commit 3710aca6ce
87 changed files with 874 additions and 729 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/api/files.py
+4 -8
View File
@@ -397,9 +397,9 @@ def reprocess_single_file(request: Request, file_id: int, db: DbSession):
def reprocess_with_cloud_ocr(request: Request, file_id: int, db: DbSession):
"""
Reprocess a single file with forced Cloud OCR processing.
This endpoint forces Azure Document Intelligence OCR processing regardless
of whether the PDF contains embedded text. Useful for documents with
of whether the PDF contains embedded text. Useful for documents with
low-quality embedded text or when higher quality OCR is needed.
Args:
@@ -425,16 +425,12 @@ def reprocess_with_cloud_ocr(request: Request, file_id: int, db: DbSession):
logger.info(f"Using local file for Cloud OCR reprocessing: {source_file}")
else:
raise HTTPException(
status_code=400,
detail="Neither original nor local file found on disk. Cannot reprocess."
status_code=400, detail="Neither original nor local file found on disk. Cannot reprocess."
)
# Queue the file for processing with force_cloud_ocr=True
task = process_document.delay(
source_file,
original_filename=file_record.original_filename,
file_id=file_record.id,
force_cloud_ocr=True
source_file, original_filename=file_record.original_filename, file_id=file_record.id, force_cloud_ocr=True
)
logger.info(
app/api/url_upload.py
+1 -1
View File
@@ -277,7 +277,7 @@ async def process_url(request: Request, url_request: URLUploadRequest):
return {
"task_id": task.id,
"status": "queued",
"message": f"File downloaded from URL and queued for processing",
"message": "File downloaded from URL and queued for processing",
"filename": safe_filename,
"size": downloaded_size,
}
app/celery_worker.py
+1 -1
View File
@@ -13,6 +13,7 @@ from app.tasks.convert_to_pdf import convert_to_pdf # noqa: F401
from app.tasks.embed_metadata_into_pdf import embed_metadata_into_pdf # noqa: F401
from app.tasks.extract_metadata_with_gpt import extract_metadata_with_gpt # noqa: F401
from app.tasks.imap_tasks import pull_all_inboxes # noqa: F401
from app.tasks.monitor_stalled_steps import monitor_stalled_steps # noqa: F401
# **Ensure all tasks are imported before Celery starts**
from app.tasks.process_document import process_document # noqa: F401
@@ -33,7 +34,6 @@ from app.tasks.upload_to_s3 import upload_to_s3 # noqa: F401
from app.tasks.upload_to_sftp import upload_to_sftp # noqa: F401
from app.tasks.upload_to_webdav import upload_to_webdav # noqa: F401
from app.tasks.uptime_kuma_tasks import ping_uptime_kuma # noqa: F401
from app.tasks.monitor_stalled_steps import monitor_stalled_steps # noqa: F401
celery.conf.task_routes = {
"app.tasks.*": {"queue": "default"},
app/config.py
+22 -6
View File
@@ -33,7 +33,8 @@ class Settings(BaseSettings):
paperless_host: Optional[str] = None
paperless_custom_field_absender: Optional[str] = None # Name of the "absender" custom field in Paperless
# JSON mapping of metadata field names to Paperless custom field names
# Example: {"absender": "Sender", "empfaenger": "Recipient", "language": "Language", "correspondent": "Correspondent"}
# Example: {"absender": "Sender", "empfaenger": "Recipient",
# "language": "Language", "correspondent": "Correspondent"}
paperless_custom_fields_mapping: Optional[str] = None
azure_ai_key: str
@@ -176,23 +177,35 @@ class Settings(BaseSettings):
)
max_single_file_size: Optional[int] = Field(
default=None,
description="Maximum size for a single file chunk in bytes. If set and file exceeds this, it will be split into smaller chunks for processing. Default: None (no splitting).",
description=(
"Maximum size for a single file chunk in bytes. If set and file exceeds this,"
" it will be split into smaller chunks for processing. Default: None (no splitting)."
),
)
# Deduplication settings - prevents processing of duplicate files
enable_deduplication: bool = Field(
default=True,
description="Enable deduplication check before processing. If enabled, files with the same SHA-256 hash as previously processed files will not be processed again. Default: True (enabled).",
description=(
"Enable deduplication check before processing. If enabled, files with the same SHA-256 hash"
" as previously processed files will not be processed again. Default: True (enabled)."
),
)
show_deduplication_step: bool = Field(
default=True,
description="Show the 'Check for Duplicates' step in processing history. If False, the check is still performed but not displayed. Default: True.",
description=(
"Show the 'Check for Duplicates' step in processing history."
" If False, the check is still performed but not displayed. Default: True."
),
)
# Processing step timeout - prevents files from getting stuck in "in_progress" state
step_timeout: int = Field(
default=600,
description="Timeout in seconds for processing steps. If a step is 'in_progress' for longer than this, it will be marked as failed. Default: 600 seconds (10 minutes).",
description=(
"Timeout in seconds for processing steps. If a step is 'in_progress' for longer than this,"
" it will be marked as failed. Default: 600 seconds (10 minutes)."
),
)
# Security Headers Configuration (see SECURITY_AUDIT.md and docs/DeploymentGuide.md)
@@ -215,7 +228,10 @@ class Settings(BaseSettings):
# Content-Security-Policy (CSP) - Controls resource loading
security_header_csp_enabled: bool = Field(default=True, description="Enable CSP header.")
security_header_csp_value: str = Field(
default="default-src 'self'; script-src 'self' 'unsafe-inline'; style-src 'self' 'unsafe-inline'; img-src 'self' data: https:; font-src 'self' data:;",
default=(
"default-src 'self'; script-src 'self' 'unsafe-inline';"
" style-src 'self' 'unsafe-inline'; img-src 'self' data: https:; font-src 'self' data:;"
),
description="CSP header value. Customize based on your application's resource loading needs.",
)
app/main.py
+1 -1
View File
@@ -8,6 +8,7 @@ from fastapi import FastAPI, HTTPException, Request, status
from fastapi.responses import JSONResponse
from fastapi.staticfiles import StaticFiles
from fastapi.templating import Jinja2Templates
from slowapi.errors import RateLimitExceeded
from starlette.config import Config
from starlette.middleware.sessions import SessionMiddleware
from starlette.middleware.trustedhost import TrustedHostMiddleware
@@ -21,7 +22,6 @@ from app.middleware.rate_limit import create_limiter, get_rate_limit_exceeded_ha
from app.middleware.security_headers import SecurityHeadersMiddleware
from app.utils.config_validator import check_all_configs
from app.utils.notification import init_apprise, notify_shutdown, notify_startup
from slowapi.errors import RateLimitExceeded
# Import the routers - now using views directly instead of frontend
from app.views import router as frontend_router
app/middleware/rate_limit.py
+12 -13
View File
@@ -21,7 +21,6 @@ from typing import Callable
from fastapi import Request
from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.errors import RateLimitExceeded
from slowapi.util import get_remote_address
logger = logging.getLogger(__name__)
@@ -30,14 +29,14 @@ logger = logging.getLogger(__name__)
def get_identifier(request: Request) -> str:
"""
Get unique identifier for rate limiting.
Uses authenticated user ID if available, otherwise falls back to IP address.
This provides better rate limiting for authenticated users and prevents
IP-based bypassing for authenticated endpoints.
Args:
request: FastAPI request object
Returns:
Unique identifier string for rate limiting
"""
@@ -50,7 +49,7 @@ def get_identifier(request: Request) -> str:
if identifier:
logger.debug(f"Rate limiting by user: {identifier}")
return f"user:{identifier}"
# Fall back to IP address for unauthenticated requests
ip = get_remote_address(request)
logger.debug(f"Rate limiting by IP: {ip}")
@@ -60,11 +59,11 @@ def get_identifier(request: Request) -> str:
def create_limiter(redis_url: str = None, enabled: bool = True) -> Limiter:
"""
Create and configure the rate limiter.
Args:
redis_url: Redis connection URL for distributed rate limiting
enabled: Whether rate limiting is enabled (default: True)
Returns:
Configured Limiter instance
"""
@@ -76,10 +75,10 @@ def create_limiter(redis_url: str = None, enabled: bool = True) -> Limiter:
default_limits=["10000/minute"], # Effectively unlimited
enabled=False,
)
# Use Redis if available, otherwise fall back to in-memory
storage_uri = redis_url if redis_url else "memory://"
if redis_url:
logger.info(f"Rate limiting enabled with Redis backend: {redis_url}")
else:
@@ -87,7 +86,7 @@ def create_limiter(redis_url: str = None, enabled: bool = True) -> Limiter:
"Rate limiting using in-memory storage (not suitable for production with multiple workers). "
"Configure REDIS_URL for distributed rate limiting."
)
# Create limiter with default limits
# Default: 100 requests per minute per IP/user
limiter = Limiter(
@@ -97,7 +96,7 @@ def create_limiter(redis_url: str = None, enabled: bool = True) -> Limiter:
strategy="fixed-window", # Can be: fixed-window, moving-window, or fixed-window-elastic-expiry
enabled=True,
)
logger.info("Rate limiter initialized successfully")
return limiter
@@ -105,10 +104,10 @@ def create_limiter(redis_url: str = None, enabled: bool = True) -> Limiter:
def get_rate_limit_exceeded_handler() -> Callable:
"""
Get the rate limit exceeded exception handler.
Returns a handler that provides user-friendly 429 responses with
Retry-After header when rate limit is exceeded.
Returns:
Exception handler function
"""
app/middleware/rate_limit_decorators.py
+5 -9
View File
@@ -7,10 +7,6 @@ This module provides convenient decorators to apply rate limits to specific endp
Import the limiter from main.py state and use these decorators to protect endpoints.
"""
from functools import wraps
from fastapi import Request
# Import will happen at runtime to avoid circular dependencies
_limiter = None
@@ -28,13 +24,13 @@ def get_limiter():
def limit(rate_limit: str):
"""
Apply a rate limit to an endpoint.
Args:
rate_limit: Rate limit string (e.g., "10/minute", "100/hour")
Returns:
Decorator function
Example:
@router.post("/login")
@limit("10/minute")
@@ -53,10 +49,10 @@ def limit(rate_limit: str):
def exempt():
"""
Exempt an endpoint from rate limiting.
Returns:
Decorator function
Example:
@router.get("/health")
@exempt()
app/models.py
+3 -3
View File
@@ -1,6 +1,6 @@
# app/models.py
from sqlalchemy import Column, DateTime, ForeignKey, Integer, String, Text, UniqueConstraint, func, Boolean
from sqlalchemy import Boolean, Column, DateTime, ForeignKey, Integer, String, Text, UniqueConstraint, func
from app.database import Base
@@ -44,11 +44,11 @@ class FileRecord(Base):
# MIME type or extension (optional)
mime_type = Column(String)
# Deduplication tracking: True if this file is a duplicate of another file
# When a duplicate is detected, this file record is created but marked as duplicate
is_duplicate = Column(Boolean, default=False, nullable=False, index=True)
# If this is a duplicate, record the ID of the original file for reference
duplicate_of_id = Column(Integer, ForeignKey("files.id"), nullable=True)
app/tasks/embed_metadata_into_pdf.py
+26 -14
View File
@@ -35,28 +35,28 @@ def persist_metadata(metadata, final_pdf_path, original_file_path=None, processe
Saves the metadata dictionary to a JSON file with the same base name as the final PDF.
For example, if final_pdf_path is "<workdir>/processed/MyFile.pdf",
the metadata will be saved as "<workdir>/processed/MyFile.json".
Optionally augments the metadata with file path references for traceability.
Args:
metadata: Dictionary of metadata to save
final_pdf_path: Path to the final PDF file
original_file_path: Optional path to the immutable original file
processed_file_path: Optional path to the processed file
Returns:
str: Path to the created JSON file
"""
base, _ = os.path.splitext(final_pdf_path)
json_path = base + ".json"
# Augment metadata with file path references if provided
metadata_with_paths = metadata.copy()
if original_file_path:
metadata_with_paths["original_file_path"] = original_file_path
if processed_file_path:
metadata_with_paths["processed_file_path"] = processed_file_path
with open(json_path, "w", encoding="utf-8") as f:
json.dump(metadata_with_paths, f, ensure_ascii=False, indent=2)
return json_path
@@ -102,7 +102,11 @@ def embed_metadata_into_pdf(self, local_file_path: str, extracted_text: str, met
else:
logger.error(f"[{task_id}] Local file {local_file_path} not found, cannot embed metadata.")
log_task_progress(
task_id, "embed_metadata_into_pdf", "failure", "File not found", file_id=file_id,
task_id,
"embed_metadata_into_pdf",
"failure",
"File not found",
file_id=file_id,
detail=(
f"Local file not found, cannot embed metadata.\n"
f"Tried path: {local_file_path}\n"
@@ -199,10 +203,7 @@ def embed_metadata_into_pdf(self, local_file_path: str, extracted_text: str, met
logger.info(f"[{task_id}] Persisting metadata to JSON")
log_task_progress(task_id, "save_metadata_json", "in_progress", "Saving metadata JSON", file_id=file_id)
json_path = persist_metadata(
metadata,
final_file_path,
original_file_path=original_file_path,
processed_file_path=final_file_path
metadata, final_file_path, original_file_path=original_file_path, processed_file_path=final_file_path
)
logger.info(f"[{task_id}] Metadata persisted to {json_path}")
log_task_progress(
@@ -212,7 +213,11 @@ def embed_metadata_into_pdf(self, local_file_path: str, extracted_text: str, met
# Trigger the next step: final storage.
logger.info(f"[{task_id}] Queueing final storage task")
log_task_progress(
task_id, "embed_metadata_into_pdf", "success", "Metadata embedded, queuing finalization", file_id=file_id,
task_id,
"embed_metadata_into_pdf",
"success",
"Metadata embedded, queuing finalization",
file_id=file_id,
detail=(
f"Metadata embedded into PDF successfully.\n"
f"Original file: {original_file}\n"
@@ -229,7 +234,7 @@ def embed_metadata_into_pdf(self, local_file_path: str, extracted_text: str, met
try:
original_file_path = Path(original_file).resolve()
workdir_tmp_resolved = workdir_tmp_path.resolve()
# Check if file is within workdir/tmp and exists
if original_file_path.is_relative_to(workdir_tmp_resolved) and original_file_path.exists():
try:
@@ -245,8 +250,15 @@ def embed_metadata_into_pdf(self, local_file_path: str, extracted_text: str, met
except Exception as e:
logger.exception(f"[{task_id}] Failed to embed metadata into {processed_file}: {e}")
log_task_progress(
task_id, "embed_metadata_into_pdf", "failure", f"Exception: {str(e)}", file_id=file_id,
detail=f"Failed to embed metadata into {processed_file}.\nOriginal file: {original_file}\nException: {str(e)}",
task_id,
"embed_metadata_into_pdf",
"failure",
f"Exception: {str(e)}",
file_id=file_id,
detail=(
f"Failed to embed metadata into {processed_file}.\n"
f"Original file: {original_file}\nException: {str(e)}"
),
)
# Clean up temporary file in case of error
if os.path.exists(processed_file):
app/tasks/extract_metadata_with_gpt.py
+24 -10
View File
@@ -112,7 +112,11 @@ def extract_metadata_with_gpt(self, filename: str, cleaned_text: str, file_id: i
content = completion.choices[0].message.content
logger.info(f"[{task_id}] Raw classification response for {filename}: {content[:200]}...")
log_task_progress(
task_id, "call_openai", "success", "Received OpenAI response", file_id=file_id,
task_id,
"call_openai",
"success",
"Received OpenAI response",
file_id=file_id,
detail=f"Raw classification response:\n{content}",
)
@@ -120,13 +124,17 @@ def extract_metadata_with_gpt(self, filename: str, cleaned_text: str, file_id: i
if not json_text:
logger.error(f"[{task_id}] Could not find valid JSON in GPT response for {filename}.")
log_task_progress(
task_id, "extract_metadata_with_gpt", "failure", "Invalid JSON in response", file_id=file_id,
task_id,
"extract_metadata_with_gpt",
"failure",
"Invalid JSON in response",
file_id=file_id,
detail=f"Could not parse valid JSON from GPT response.\nRaw response:\n{content}",
)
return {}
metadata = json.loads(json_text)
# SECURITY: Validate filename format from GPT to prevent path traversal
# The prompt requests filenames with only letters, numbers, periods, and underscores
# Enforce this constraint to prevent malicious filenames
@@ -138,16 +146,18 @@ def extract_metadata_with_gpt(self, filename: str, cleaned_text: str, file_id: i
# 1. Potential locale-specific \w behavior
# 2. Files literally named ".." which are valid but problematic
# 3. Future code changes that might relax the regex
if not re.match(r'^[\w\-\. ]+$', suggested_filename) or ".." in suggested_filename:
logger.warning(
f"[{task_id}] Invalid filename format from GPT: '{suggested_filename}', using fallback"
)
if not re.match(r"^[\w\-\. ]+$", suggested_filename) or ".." in suggested_filename:
logger.warning(f"[{task_id}] Invalid filename format from GPT: '{suggested_filename}', using fallback")
# Reset to empty to trigger fallback to original filename
metadata["filename"] = ""
logger.info(f"[{task_id}] Extracted metadata: {metadata}")
log_task_progress(
task_id, "parse_metadata", "success", f"Parsed metadata: {list(metadata.keys())}", file_id=file_id,
task_id,
"parse_metadata",
"success",
f"Parsed metadata: {list(metadata.keys())}",
file_id=file_id,
detail=f"Extracted metadata:\n{json.dumps(metadata, ensure_ascii=False, indent=2)}",
)
@@ -164,7 +174,11 @@ def extract_metadata_with_gpt(self, filename: str, cleaned_text: str, file_id: i
except Exception as e:
logger.exception(f"[{task_id}] OpenAI classification failed for {filename}: {e}")
log_task_progress(
task_id, "extract_metadata_with_gpt", "failure", f"Exception: {str(e)}", file_id=file_id,
task_id,
"extract_metadata_with_gpt",
"failure",
f"Exception: {str(e)}",
file_id=file_id,
detail=f"OpenAI classification failed for {filename}.\nException: {str(e)}",
)
return {}
app/tasks/monitor_stalled_steps.py
+4 -7
View File
@@ -34,7 +34,7 @@ def monitor_stalled_steps():
try:
with SessionLocal() as db:
stalled_count = mark_stalled_steps_as_failed(db)
if stalled_count > 0:
logger.warning(
f"[{datetime.utcnow().isoformat()}] "
@@ -42,13 +42,10 @@ def monitor_stalled_steps():
f"Marked as failed due to timeout."
)
else:
logger.debug(
f"[{datetime.utcnow().isoformat()}] "
f"No stalled steps found."
)
logger.debug(f"[{datetime.utcnow().isoformat()}] " f"No stalled steps found.")
return {"recovered": stalled_count}
except Exception as e:
logger.error(f"Error in monitor_stalled_steps task: {e}", exc_info=True)
return {"error": str(e), "recovered": 0}
app/tasks/process_document.py
+23 -16
View File
@@ -24,7 +24,9 @@ logger = logging.getLogger(__name__)
@celery.task(base=BaseTaskWithRetry, bind=True)
def process_document(self, original_local_file: str, original_filename: str = None, file_id: int = None, force_cloud_ocr: bool = False):
def process_document(
self, original_local_file: str, original_filename: str = None, file_id: int = None, force_cloud_ocr: bool = False
):
"""
Process a document file and trigger appropriate text extraction.
@@ -58,7 +60,10 @@ def process_document(self, original_local_file: str, original_filename: str = No
if not os.path.exists(original_local_file):
logger.error(f"[{task_id}] File {original_local_file} not found.")
log_task_progress(
task_id, "process_document", "failure", "File not found",
task_id,
"process_document",
"failure",
"File not found",
detail=f"File not found on disk: {original_local_file}",
)
return {"error": "File not found"}
@@ -71,7 +76,7 @@ def process_document(self, original_local_file: str, original_filename: str = No
else:
logger.info(f"[{task_id}] Computing file hash (deduplication disabled)...")
filehash = hash_file(original_local_file)
# Use provided original_filename or fall back to basename of path
if original_filename is None:
original_filename = os.path.basename(original_local_file)
@@ -81,7 +86,7 @@ def process_document(self, original_local_file: str, original_filename: str = No
mime_type = "application/octet-stream"
logger.info(f"[{task_id}] File hash: {filehash[:10]}..., Size: {file_size} bytes, MIME: {mime_type}")
# Log deduplication step result (only if enabled)
if settings.enable_deduplication:
log_task_progress(
@@ -113,13 +118,15 @@ def process_document(self, original_local_file: str, original_filename: str = No
else:
# Check for duplicate only if this is a new file (not reprocessing)
# IMPORTANT: Only consider it a duplicate if it matches a DIFFERENT file
existing = db.query(FileRecord).filter(FileRecord.filehash == filehash, FileRecord.is_duplicate.is_(False)).order_by(FileRecord.created_at.asc()).first()
existing = (
db.query(FileRecord)
.filter(FileRecord.filehash == filehash, FileRecord.is_duplicate.is_(False))
.order_by(FileRecord.created_at.asc())
.first()
)
if existing is None:
existing = (
db.query(FileRecord)
.filter(FileRecord.filehash == filehash)
.order_by(FileRecord.id.asc())
.first()
db.query(FileRecord).filter(FileRecord.filehash == filehash).order_by(FileRecord.id.asc()).first()
)
# A file is only a duplicate if it matches a different file's hash
@@ -215,11 +222,11 @@ def process_document(self, original_local_file: str, original_filename: str = No
# This copy serves as the permanent, untouched reference of the ingested file
original_dir = os.path.join(settings.workdir, "original")
os.makedirs(original_dir, exist_ok=True)
# Use collision-resistant naming with -0001, -0002 suffixes
base_name = os.path.splitext(new_filename)[0]
original_file_path = get_unique_filepath_with_counter(original_dir, base_name, file_ext)
logger.info(f"[{task_id}] Saving immutable original to: {original_file_path}")
log_task_progress(
task_id,
@@ -236,7 +243,7 @@ def process_document(self, original_local_file: str, original_filename: str = No
f"Original saved: {os.path.basename(original_file_path)}",
file_id=new_record.id,
)
# Update the DB with original_file_path
new_record.original_file_path = original_file_path
else:
@@ -308,7 +315,7 @@ def process_document(self, original_local_file: str, original_filename: str = No
)
process_with_azure_document_intelligence.delay(new_filename, file_id)
return {"file": new_local_path, "status": "Queued for forced OCR", "file_id": file_id}
# If the file is not a PDF, skip embedded text check and convert to PDF first
is_pdf = mime_type == "application/pdf" or os.path.splitext(new_local_path)[1].lower() == ".pdf"
if not is_pdf:
@@ -403,7 +410,7 @@ def process_document(self, original_local_file: str, original_filename: str = No
f"Extracted {len(extracted_text)} characters",
file_id=file_id,
)
# Mark Azure OCR as skipped since we extracted text locally
log_task_progress(
task_id,
@@ -438,7 +445,7 @@ def process_document(self, original_local_file: str, original_filename: str = No
"No embedded text, queuing OCR",
file_id=file_id,
)
# Mark local text extraction as skipped since we're using Azure OCR
log_task_progress(
task_id,
@@ -447,7 +454,7 @@ def process_document(self, original_local_file: str, original_filename: str = No
"No embedded text, using Azure OCR instead",
file_id=file_id,
)
log_task_progress(
task_id,
"process_document",
app/tasks/process_with_azure_document_intelligence.py
+43 -16
View File
@@ -101,8 +101,11 @@ def process_with_azure_document_intelligence(self, filename: str, file_id: int =
"""
task_id = self.request.id
log_task_progress(
task_id, "process_with_azure_document_intelligence", "in_progress",
f"Starting OCR for {filename}", file_id=file_id,
task_id,
"process_with_azure_document_intelligence",
"in_progress",
f"Starting OCR for {filename}",
file_id=file_id,
)
try:
tmp_file_path = os.path.join(settings.workdir, "tmp", filename)
@@ -117,8 +120,12 @@ def process_with_azure_document_intelligence(self, filename: str, file_id: int =
)
logger.error(f"[{task_id}] {error_msg}")
log_task_progress(
task_id, "validate_file", "failure",
f"File too large: {filename}", file_id=file_id, detail=error_msg,
task_id,
"validate_file",
"failure",
f"File too large: {filename}",
file_id=file_id,
detail=error_msg,
)
return {"error": error_msg, "file": filename, "status": "Failed - Size limit exceeded"}
@@ -130,8 +137,12 @@ def process_with_azure_document_intelligence(self, filename: str, file_id: int =
error_msg = f"PDF page count ({page_count}) exceeds Azure Document Intelligence limit of 2000 pages"
logger.error(f"[{task_id}] {error_msg}")
log_task_progress(
task_id, "validate_file", "failure",
f"Too many pages: {filename}", file_id=file_id, detail=error_msg,
task_id,
"validate_file",
"failure",
f"Too many pages: {filename}",
file_id=file_id,
detail=error_msg,
)
return {"error": error_msg, "file": filename, "status": "Failed - Page limit exceeded"}
if page_count is None:
@@ -140,14 +151,20 @@ def process_with_azure_document_intelligence(self, filename: str, file_id: int =
)
log_task_progress(
task_id, "validate_file", "success",
f"File validation passed for {filename}", file_id=file_id,
task_id,
"validate_file",
"success",
f"File validation passed for {filename}",
file_id=file_id,
)
logger.info(f"[{task_id}] Processing {filename} with Azure Document Intelligence OCR.")
log_task_progress(
task_id, "call_azure_ocr", "in_progress",
f"Sending {filename} to Azure Document Intelligence", file_id=file_id,
task_id,
"call_azure_ocr",
"in_progress",
f"Sending {filename} to Azure Document Intelligence",
file_id=file_id,
)
# Open and send the document for processing
@@ -173,8 +190,11 @@ def process_with_azure_document_intelligence(self, filename: str, file_id: int =
logger.info(f"[{task_id}] Extracted text for {filename}: {len(extracted_text)} characters")
log_task_progress(
task_id, "call_azure_ocr", "success",
f"Azure OCR completed for {filename}", file_id=file_id,
task_id,
"call_azure_ocr",
"success",
f"Azure OCR completed for {filename}",
file_id=file_id,
detail=f"Extracted {len(extracted_text)} characters, {len(rotation_data)} rotated pages detected",
)
@@ -182,8 +202,11 @@ def process_with_azure_document_intelligence(self, filename: str, file_id: int =
rotate_pdf_pages.delay(filename, extracted_text, rotation_data, file_id)
log_task_progress(
task_id, "process_with_azure_document_intelligence", "success",
f"OCR processing complete for {filename}", file_id=file_id,
task_id,
"process_with_azure_document_intelligence",
"success",
f"OCR processing complete for {filename}",
file_id=file_id,
detail=f"Searchable PDF saved, {len(extracted_text)} characters extracted",
)
@@ -191,7 +214,11 @@ def process_with_azure_document_intelligence(self, filename: str, file_id: int =
except Exception as e:
logger.error(f"[{task_id}] Error processing {filename} with Azure Document Intelligence: {e}")
log_task_progress(
task_id, "process_with_azure_document_intelligence", "failure",
f"OCR failed for {filename}", file_id=file_id, detail=str(e),
task_id,
"process_with_azure_document_intelligence",
"failure",
f"OCR failed for {filename}",
file_id=file_id,
detail=str(e),
)
raise
app/tasks/rotate_pdf_pages.py
+28 -15
View File
@@ -63,8 +63,11 @@ def rotate_pdf_pages(self, filename: str, extracted_text: str, rotation_data=Non
try:
task_id = self.request.id
log_task_progress(
task_id, "rotate_pdf_pages", "in_progress",
f"Checking page rotation for {filename}", file_id=file_id,
task_id,
"rotate_pdf_pages",
"in_progress",
f"Checking page rotation for {filename}",
file_id=file_id,
)
pdf_path = os.path.join(settings.workdir, "tmp", filename)
if not os.path.exists(pdf_path):
@@ -72,12 +75,13 @@ def rotate_pdf_pages(self, filename: str, extracted_text: str, rotation_data=Non
# Skip rotation if no rotation data provided
if not rotation_data:
logger.info(
f"[{task_id}] No rotation data provided for {filename}, proceeding with metadata extraction"
)
logger.info(f"[{task_id}] No rotation data provided for {filename}, proceeding with metadata extraction")
log_task_progress(
task_id, "rotate_pdf_pages", "success",
"No rotation needed, proceeding to metadata extraction", file_id=file_id,
task_id,
"rotate_pdf_pages",
"success",
"No rotation needed, proceeding to metadata extraction",
file_id=file_id,
)
extract_metadata_with_gpt.delay(filename, extracted_text, file_id)
return {"file": filename, "status": "no_rotation_needed"}
@@ -95,16 +99,22 @@ def rotate_pdf_pages(self, filename: str, extracted_text: str, rotation_data=Non
f"[{task_id}] No significant rotations detected in {filename}, proceeding with metadata extraction"
)
log_task_progress(
task_id, "rotate_pdf_pages", "success",
"No rotation needed, proceeding to metadata extraction", file_id=file_id,
task_id,
"rotate_pdf_pages",
"success",
"No rotation needed, proceeding to metadata extraction",
file_id=file_id,
)
extract_metadata_with_gpt.delay(filename, extracted_text, file_id)
return {"file": filename, "status": "no_rotation_needed"}
logger.info(f"[{task_id}] Rotating {len(normalized_rotation_data)} pages in {filename}")
log_task_progress(
task_id, "apply_rotation", "in_progress",
f"Rotating {len(normalized_rotation_data)} pages", file_id=file_id,
task_id,
"apply_rotation",
"in_progress",
f"Rotating {len(normalized_rotation_data)} pages",
file_id=file_id,
)
applied_rotations = {}
@@ -144,8 +154,7 @@ def rotate_pdf_pages(self, filename: str, extracted_text: str, rotation_data=Non
if applied_rotations:
logger.info(
f"[{task_id}] Successfully rotated PDF: {filename} with rotations: "
f"{json.dumps(applied_rotations)}"
f"[{task_id}] Successfully rotated PDF: {filename} with rotations: " f"{json.dumps(applied_rotations)}"
)
else:
logger.info(
@@ -157,7 +166,9 @@ def rotate_pdf_pages(self, filename: str, extracted_text: str, rotation_data=Non
extract_metadata_with_gpt.delay(filename, extracted_text, file_id)
log_task_progress(
task_id, "rotate_pdf_pages", "success",
task_id,
"rotate_pdf_pages",
"success",
f"Rotation complete for {filename}",
file_id=file_id,
detail={"applied_rotations": applied_rotations},
@@ -173,7 +184,9 @@ def rotate_pdf_pages(self, filename: str, extracted_text: str, rotation_data=Non
except Exception as e:
logger.error(f"[{task_id}] Error rotating PDF {filename}: {e}")
log_task_progress(
task_id, "rotate_pdf_pages", "failure",
task_id,
"rotate_pdf_pages",
"failure",
f"Rotation failed: {str(e)}",
file_id=file_id,
detail={"error": str(e), "filename": filename},
app/tasks/upload_to_paperless.py
+11 -4
View File
@@ -274,8 +274,15 @@ def upload_to_paperless(self, file_path: str, file_id: int = None):
response_text,
)
log_task_progress(
task_id, "upload_to_paperless", "failure", error_msg, file_id=file_id,
detail=f"Failed to upload document to Paperless.\nFile: {file_path}\nError: {exc}\nResponse: {response_text}",
task_id,
"upload_to_paperless",
"failure",
error_msg,
file_id=file_id,
detail=(
f"Failed to upload document to Paperless.\n"
f"File: {file_path}\nError: {exc}\nResponse: {response_text}"
),
)
raise
@@ -322,9 +329,9 @@ def upload_to_paperless(self, file_path: str, file_id: int = None):
# Map each metadata field to its corresponding Paperless custom field
for metadata_field, paperless_field in field_mapping.items():
if metadata_field in metadata and metadata[metadata_field]:
# Convert to string to ensure consistent comparison with UNKNOWN_VALUE
# Convert to string to ensure consistent comparison
value = str(metadata[metadata_field]) if metadata[metadata_field] is not None else ""
if value and value != UNKNOWN_VALUE:
if value and value != METADATA_UNKNOWN_PLACEHOLDER:
custom_fields_to_set[paperless_field] = value
logger.debug(f"[{task_id}] Mapping {metadata_field}='{value}' to field '{paperless_field}'")
except json.JSONDecodeError as e:
app/tasks/upload_with_rclone.py
+1 -3
View File
@@ -103,9 +103,7 @@ def upload_with_rclone(self, file_path: str, destination: str):
except (OSError, ValueError) as e:
error_msg = f"Error uploading {filename} to {destination}: {str(e)}"
logger.error(f"[{task_id}] {error_msg}")
log_task_progress(
task_id, "upload_with_rclone", "failure", f"Upload failed for {filename}", detail=error_msg
)
log_task_progress(task_id, "upload_with_rclone", "failure", f"Upload failed for {filename}", detail=error_msg)
raise RuntimeError(error_msg) from e
app/utils/file_queries.py
+10 -12
View File
@@ -10,7 +10,7 @@ from typing import Optional
from sqlalchemy import or_
from sqlalchemy.orm import Query, Session
from app.models import FileRecord, FileProcessingStep
from app.models import FileProcessingStep, FileRecord
def apply_status_filter(query: Query, db: Session, status: Optional[str]) -> Query:
@@ -19,13 +19,13 @@ def apply_status_filter(query: Query, db: Session, status: Optional[str]) -> Que
This function modifies a SQLAlchemy query to filter files based on their
processing status by examining associated FileProcessingStep entries.
Only tracks "real" processing steps that represent user-facing status:
- Main steps: create_file_record, check_text, extract_text, process_with_azure_document_intelligence,
extract_metadata_with_gpt, embed_metadata_into_pdf, finalize_document_storage,
send_to_all_destinations
- Upload steps: queue_*, upload_to_*
Diagnostic/internal steps (poll_task, upload_file, set_custom_fields, etc.) are ignored
as they may not complete properly and don't affect the actual status.
@@ -53,7 +53,7 @@ def apply_status_filter(query: Query, db: Session, status: Optional[str]) -> Que
# Define which steps are "real" status-determining steps
# Only high-level logical steps and actual upload destinations (not queue_* steps)
from app.config import settings
REAL_STEPS = {
"create_file_record",
"check_text",
@@ -75,16 +75,13 @@ def apply_status_filter(query: Query, db: Session, status: Optional[str]) -> Que
"upload_to_email",
"upload_to_s3",
}
# Add check_for_duplicates if deduplication is enabled
if settings.enable_deduplication:
REAL_STEPS.add("check_for_duplicates")
# Filter to only real steps
real_steps_subq = (
db.query(FileProcessingStep)
.filter(FileProcessingStep.step_name.in_(REAL_STEPS))
)
real_steps_subq = db.query(FileProcessingStep).filter(FileProcessingStep.step_name.in_(REAL_STEPS))
if status == "pending":
# Files with no real steps (never started processing)
@@ -102,14 +99,15 @@ def apply_status_filter(query: Query, db: Session, status: Optional[str]) -> Que
# Files where all real steps are either success or skipped (no failures or in_progress)
# Exclude duplicates from completed
query = query.filter(FileRecord.is_duplicate.is_(False))
# Get files that have real steps
files_with_real_steps = real_steps_subq.distinct().subquery()
# Get files with failures or in_progress on real steps
files_with_issues = (
real_steps_subq
.filter(or_(FileProcessingStep.status == "failure", FileProcessingStep.status == "in_progress"))
real_steps_subq.filter(
or_(FileProcessingStep.status == "failure", FileProcessingStep.status == "in_progress")
)
.distinct()
.subquery()
)
app/utils/file_status.py
+4 -4
View File
@@ -7,7 +7,7 @@ from typing import Dict, List
from sqlalchemy.orm import Session
from app.models import FileProcessingStep, FileRecord, ProcessingLog
from app.utils.step_manager import get_file_overall_status, get_step_summary
from app.utils.step_manager import get_file_overall_status
def get_file_processing_status(db: Session, file_id: int) -> Dict:
@@ -60,7 +60,7 @@ def get_files_processing_status(db: Session, file_ids: List[int]) -> Dict[int, D
extract_metadata_with_gpt, embed_metadata_into_pdf, finalize_document_storage,
send_to_all_destinations
- Upload steps: upload_to_*
Diagnostic/internal steps (poll_task, upload_file, set_custom_fields, etc.) are ignored.
Args:
@@ -78,7 +78,7 @@ def get_files_processing_status(db: Session, file_ids: List[int]) -> Dict[int, D
# Define which steps are "real" status-determining steps
from app.config import settings
REAL_STEPS = {
"create_file_record",
"check_text",
@@ -100,7 +100,7 @@ def get_files_processing_status(db: Session, file_ids: List[int]) -> Dict[int, D
"upload_to_email",
"upload_to_s3",
}
# Add check_for_duplicates if deduplication is enabled
if settings.enable_deduplication:
REAL_STEPS.add("check_for_duplicates")
app/utils/filename_utils.py
+7 -7
View File
@@ -73,22 +73,22 @@ def get_unique_filepath_with_counter(directory, base_filename, extension=".pdf")
"""
Returns a unique filepath in the specified directory using a numeric counter suffix.
If 'base_filename.pdf' exists, it will append '-0001', '-0002', etc.
This function implements robust collision handling with zero-padded numeric suffixes
as required for document storage organization.
Args:
directory (str): Directory path where the file will be stored
base_filename (str): Base name for the file (without extension)
extension (str): File extension including the dot (default: ".pdf")
Returns:
str: Full path to a unique filename
Examples:
>>> get_unique_filepath_with_counter("/workdir/original", "2024-01-01_Invoice")
"/workdir/original/2024-01-01_Invoice.pdf" # If doesn't exist
>>> get_unique_filepath_with_counter("/workdir/original", "2024-01-01_Invoice")
"/workdir/original/2024-01-01_Invoice-0001.pdf" # If original exists
"""
@@ -96,7 +96,7 @@ def get_unique_filepath_with_counter(directory, base_filename, extension=".pdf")
candidate = os.path.join(directory, base_filename + extension)
if not os.path.exists(candidate):
return candidate
# If base exists, try with counter suffix
counter = 1
while True:
@@ -106,7 +106,7 @@ def get_unique_filepath_with_counter(directory, base_filename, extension=".pdf")
if not os.path.exists(candidate):
return candidate
counter += 1
# Sanity check to prevent infinite loops (very unlikely to reach)
if counter > 9999:
# Fall back to timestamp + UUID if somehow we have 10000 collisions
app/utils/logging.py
+4 -4
View File
@@ -96,7 +96,7 @@ def log_task_progress(task_id, step_name, status, message=None, file_id=None, de
detail=detail,
)
db.add(log_entry)
# Update FileProcessingStep table (for status tracking) if file_id is provided
if file_id and step_name:
# Find or create the step record
@@ -105,9 +105,9 @@ def log_task_progress(task_id, step_name, status, message=None, file_id=None, de
.filter(FileProcessingStep.file_id == file_id, FileProcessingStep.step_name == step_name)
.first()
)
now = datetime.utcnow()
if not step_record:
# Create new step record
step_record = FileProcessingStep(
@@ -128,5 +128,5 @@ def log_task_progress(task_id, step_name, status, message=None, file_id=None, de
step_record.completed_at = now
if status == "failure":
step_record.error_message = message or detail
db.commit()
app/utils/migrate_logs_to_steps.py
+1 -4
View File
@@ -6,14 +6,11 @@ for files that were processed before the status tracking table was created.
"""
import logging
from datetime import datetime
from typing import Dict, List
from sqlalchemy import func
from sqlalchemy.orm import Session
from app.models import FileProcessingStep, FileRecord, ProcessingLog
from app.utils.step_manager import MAIN_PROCESSING_STEPS
from app.models import FileProcessingStep, ProcessingLog
logger = logging.getLogger(__name__)
app/utils/step_manager.py
+10 -12
View File
@@ -167,7 +167,7 @@ def get_file_step_status(db: Session, file_id: int) -> Dict[str, Dict]:
def get_file_overall_status(db: Session, file_id: int) -> Dict:
"""
Get the overall processing status for a file based on its steps.
Only considers "real" processing steps that represent user-facing status.
Ignores diagnostic/internal steps like poll_task, upload_file, etc.
@@ -211,19 +211,18 @@ def get_file_overall_status(db: Session, file_id: int) -> Dict:
"finalize_document_storage",
"send_to_all_destinations",
}
# Add check_for_duplicates if deduplication is enabled
if settings.enable_deduplication:
REAL_MAIN_STEPS.add("check_for_duplicates")
all_steps = db.query(FileProcessingStep).filter(FileProcessingStep.file_id == file_id).all()
# Filter to only real steps
steps = [
s for s in all_steps
if s.step_name in REAL_MAIN_STEPS
or s.step_name.startswith("queue_")
or s.step_name.startswith("upload_to_")
s
for s in all_steps
if s.step_name in REAL_MAIN_STEPS or s.step_name.startswith("queue_") or s.step_name.startswith("upload_to_")
]
if not steps:
@@ -297,11 +296,11 @@ def get_step_summary(db: Session, file_id: int) -> Dict:
"finalize_document_storage",
"send_to_all_destinations",
}
# Add check_for_duplicates if deduplication is enabled
if settings.enable_deduplication:
REAL_MAIN_STEPS.add("check_for_duplicates")
steps = db.query(FileProcessingStep).filter(FileProcessingStep.file_id == file_id).all()
main_counts = {"queued": 0, "in_progress": 0, "success": 0, "failure": 0, "skipped": 0}
@@ -318,7 +317,7 @@ def get_step_summary(db: Session, file_id: int) -> Dict:
# Check if it's an upload task (only count actual upload_to_* steps, not queue_* steps)
is_upload = step.step_name.startswith("upload_to_")
# Only count "real" steps
is_real_step = step.step_name in REAL_MAIN_STEPS or is_upload or step.step_name.startswith("queue_")
@@ -341,4 +340,3 @@ def get_step_summary(db: Session, file_id: int) -> Dict:
"total_main_steps": main_steps_count,
"total_upload_tasks": upload_steps_count,
}
app/utils/step_timeout.py
+2 -5
View File
@@ -32,9 +32,7 @@ def get_step_timeout() -> int:
def mark_stalled_steps_as_failed(
db: Session,
timeout_seconds: Optional[int] = None,
file_id: Optional[int] = None
db: Session, timeout_seconds: Optional[int] = None, file_id: Optional[int] = None
) -> int:
"""
Find and mark any in-progress steps that have exceeded the timeout as failed.
@@ -75,8 +73,7 @@ def mark_stalled_steps_as_failed(
return 0
logger.warning(
f"Found {len(stalled_steps)} stalled step(s) that exceeded "
f"{timeout_seconds}s timeout. Marking as failed."
f"Found {len(stalled_steps)} stalled step(s) that exceeded " f"{timeout_seconds}s timeout. Marking as failed."
)
count = 0
app/views/files.py
+17 -16
View File
@@ -2,13 +2,11 @@
File management views for displaying and managing files.
"""
import os
from typing import Optional
from fastapi import Depends, Query, Request
from sqlalchemy.orm import Session
from app.config import settings
from app.utils.file_queries import apply_status_filter
from app.utils.file_status import get_files_processing_status
from app.views.base import APIRouter, get_db, logger, require_login, templates
@@ -34,9 +32,9 @@ def files_page(
"""
try:
# Import the model here to avoid circular imports
from sqlalchemy import asc, desc, or_
from sqlalchemy import asc, desc
from app.models import FileRecord, ProcessingLog
from app.models import FileRecord
# Start with base query
query = db.query(FileRecord)
@@ -233,9 +231,10 @@ def _compute_processing_flow(logs):
"finalize_document_storage": {"label": "Finalize & Queue Distribution", "next": ["send_to_all_destinations"]},
"send_to_all_destinations": {"label": "Upload to Destinations", "next": [], "has_branches": True},
}
# Filter out deduplication step if not enabled or if not showing it
from app.config import settings
if not settings.enable_deduplication or not settings.show_deduplication_step:
stages.pop("check_for_duplicates", None)
# Update the next pointer for create_file_record
@@ -354,21 +353,23 @@ def _compute_step_summary(logs):
based on timestamp, regardless of input log ordering.
"""
from app.config import settings
# Count statuses for main processing steps (not uploads)
main_steps = []
if settings.enable_deduplication and settings.show_deduplication_step:
main_steps.append("check_for_duplicates")
main_steps.extend([
"create_file_record",
"check_text",
"extract_text",
"process_with_azure_document_intelligence",
"extract_metadata_with_gpt",
"embed_metadata_into_pdf",
"finalize_document_storage",
"send_to_all_destinations",
])
main_steps.extend(
[
"create_file_record",
"check_text",
"extract_text",
"process_with_azure_document_intelligence",
"extract_metadata_with_gpt",
"embed_metadata_into_pdf",
"finalize_document_storage",
"send_to_all_destinations",
]
)
upload_prefixes = ["upload_to_", "queue_"]
tests/fixtures_integration.py
+5 -3
View File
@@ -14,16 +14,17 @@ These tests exercise the full application stack end-to-end.
import os
import time
import pytest
from typing import Generator
from pathlib import Path
from typing import Generator
import pytest
# Import testcontainers
pytest.importorskip("testcontainers", reason="testcontainers not installed")
from testcontainers.core.container import DockerContainer
from testcontainers.minio import MinioContainer
from testcontainers.postgres import PostgresContainer
from testcontainers.redis import RedisContainer
from testcontainers.minio import MinioContainer
_TEST_CREDENTIAL = "testpass" # noqa: S105
@@ -300,6 +301,7 @@ def db_session_real(postgres_container):
"""
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from app.database import Base
# Create engine using PostgreSQL container
tests/test_api_common.py
+3 -1
View File
@@ -1,8 +1,10 @@
"""Tests for app/api/common.py module."""
import os
import pytest
from unittest.mock import patch
import pytest
from app.api.common import resolve_file_path
tests/test_api_error_handling.py
+20 -18
View File
@@ -2,10 +2,12 @@
Tests for API error handling - ensuring JSON responses for API routes.
"""
from unittest.mock import patch
import pytest
from fastapi.testclient import TestClient
from app.models import FileRecord
from unittest.mock import patch
@pytest.mark.integration
@@ -16,18 +18,18 @@ class TestAPIErrorHandling:
def test_delete_nonexistent_file_returns_json_404(self, client: TestClient):
"""Test that deleting a non-existent file returns JSON 404, not HTML."""
response = client.delete("/api/files/99999")
# Should return 404
assert response.status_code == 404
# Should be JSON, not HTML
content_type = response.headers.get("content-type", "")
assert "application/json" in content_type, f"Expected JSON but got {content_type}"
# Should not contain HTML
assert not response.text.startswith("<!DOCTYPE"), "Response should be JSON, not HTML"
assert not response.text.startswith("<html"), "Response should be JSON, not HTML"
# Should have valid JSON with detail field
data = response.json()
assert "detail" in data
@@ -50,18 +52,18 @@ class TestAPIErrorHandling:
# Mock settings to disable file deletion
with patch("app.api.files.settings.allow_file_delete", False):
response = client.delete(f"/api/files/{file_id}")
# Should return 403
assert response.status_code == 403
# Should be JSON, not HTML
content_type = response.headers.get("content-type", "")
assert "application/json" in content_type, f"Expected JSON but got {content_type}"
# Should not contain HTML
assert not response.text.startswith("<!DOCTYPE"), "Response should be JSON, not HTML"
assert not response.text.startswith("<html"), "Response should be JSON, not HTML"
# Should have valid JSON with detail field
data = response.json()
assert "detail" in data
@@ -84,18 +86,18 @@ class TestAPIErrorHandling:
# Mock db.delete to raise an exception
with patch.object(db_session, "delete", side_effect=Exception("Database error")):
response = client.delete(f"/api/files/{file_id}")
# Should return 500
assert response.status_code == 500
# Should be JSON, not HTML
content_type = response.headers.get("content-type", "")
assert "application/json" in content_type, f"Expected JSON but got {content_type}"
# Should not contain HTML
assert not response.text.startswith("<!DOCTYPE"), "Response should be JSON, not HTML"
assert not response.text.startswith("<html"), "Response should be JSON, not HTML"
# Should have valid JSON with detail field
data = response.json()
assert "detail" in data
@@ -103,14 +105,14 @@ class TestAPIErrorHandling:
def test_list_files_api_returns_json(self, client: TestClient, db_session):
"""Test that the files listing API returns JSON."""
response = client.get("/api/files")
# Should return 200
assert response.status_code == 200
# Should be JSON
content_type = response.headers.get("content-type", "")
assert "application/json" in content_type
# Should return a dict with files and pagination
data = response.json()
assert isinstance(data, dict)
@@ -126,10 +128,10 @@ class TestAPIErrorHandling:
def test_frontend_404_returns_html(self, client: TestClient):
"""Test that 404 on non-API routes returns HTML (not JSON)."""
response = client.get("/nonexistent-page")
# Should return 404
assert response.status_code == 404
# Note: FastAPI's HTTPException handler now returns JSON for all routes
# because HTTPException is a general exception, not a 404-specific one
# Our handler checks if it's an API route, but for non-existent routes,
tests/test_api_logs.py
+3 -1
View File
@@ -1,7 +1,9 @@
"""Tests for app/api/logs.py module."""
import pytest
from datetime import datetime
import pytest
from app.models import FileRecord, ProcessingLog
tests/test_api_openai.py
+1
View File
@@ -1,4 +1,5 @@
"""Tests for app/api/openai.py module."""
import pytest
tests/test_api_process.py
+2
View File
@@ -1,4 +1,5 @@
"""Tests for app/api/process.py module."""
import pytest
@@ -44,6 +45,7 @@ class TestProcessEndpoints:
def test_processall_endpoint(self, client, tmp_path):
"""Test POST /api/processall with no PDF files in workdir."""
from unittest.mock import patch
with patch("app.api.process.settings") as mock_settings:
mock_settings.workdir = str(tmp_path)
response = client.post("/api/processall")
tests/test_api_settings.py
+3 -1
View File
@@ -1,6 +1,8 @@
"""Tests for app/api/settings.py module."""
from unittest.mock import MagicMock, patch
import pytest
from unittest.mock import patch, MagicMock
from fastapi import HTTPException
from app.api.settings import require_admin
tests/test_api_user.py
+3 -1
View File
@@ -1,8 +1,10 @@
"""Tests for app/api/user.py module."""
import pytest
from hashlib import md5
from unittest.mock import MagicMock
import pytest
from app.api.user import whoami_handler
tests/test_auth_integration.py
+11 -7
View File
@@ -1,9 +1,11 @@
"""Integration tests for auth.py with AUTH_ENABLED=True scenarios."""
import pytest
from unittest.mock import patch, MagicMock, AsyncMock
import hashlib
from app.auth import get_gravatar_url, get_current_user, require_login
import hashlib
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from app.auth import get_current_user, get_gravatar_url, require_login
@pytest.mark.unit
@@ -20,8 +22,9 @@ class TestRequireLoginWithAuth:
return {"success": True}
# Manually create the decorator behavior
from functools import wraps
import inspect
from functools import wraps
from fastapi import status
from starlette.responses import RedirectResponse
@@ -45,10 +48,11 @@ class TestRequireLoginWithAuth:
@pytest.mark.asyncio
async def test_require_login_allows_authenticated_user(self):
"""Test that require_login allows through when user exists in session."""
from functools import wraps
import inspect
from starlette.responses import RedirectResponse
from functools import wraps
from fastapi import status
from starlette.responses import RedirectResponse
async def my_route(request):
return {"success": True}
tests/test_check_credentials.py
+6 -1
View File
@@ -1,8 +1,10 @@
"""Tests for app/tasks/check_credentials.py module."""
import json
import os
from unittest.mock import MagicMock, patch
import pytest
from unittest.mock import patch, MagicMock
from app.tasks.check_credentials import (
MockRequest,
@@ -95,6 +97,7 @@ class TestUnwrapDecoratedFunction:
def test_returns_same_function_if_not_decorated(self):
"""Test returns same function if not decorated."""
def my_func():
return "hello"
@@ -103,6 +106,7 @@ class TestUnwrapDecoratedFunction:
def test_unwraps_decorated_function(self):
"""Test unwraps decorated function."""
def inner():
return "hello"
@@ -116,6 +120,7 @@ class TestUnwrapDecoratedFunction:
def test_unwraps_multiple_levels(self):
"""Test unwraps multiple levels of decoration."""
def original():
return "hello"
tests/test_check_credentials_extended.py
+4 -2
View File
@@ -1,13 +1,15 @@
"""Extended tests for app/tasks/check_credentials.py module."""
from unittest.mock import MagicMock, patch
import pytest
from unittest.mock import patch, MagicMock
from app.tasks.check_credentials import (
sync_test_openai_connection,
sync_test_azure_connection,
sync_test_dropbox_token,
sync_test_google_drive_token,
sync_test_onedrive_token,
sync_test_openai_connection,
)
tests/test_config.py
+39 -36
View File
@@ -1,16 +1,19 @@
"""
Unit tests for configuration and security validation.
"""
import pytest
import os
import pytest
from pydantic import ValidationError
from app.config import Settings
@pytest.mark.unit
class TestConfigurationValidation:
"""Tests for configuration validation."""
def test_session_secret_required_with_auth(self):
"""Test that SESSION_SECRET is required when auth is enabled."""
with pytest.raises(ValidationError) as exc_info:
@@ -24,10 +27,10 @@ class TestConfigurationValidation:
gotenberg_url="http://localhost:3000",
workdir="/tmp",
auth_enabled=True,
session_secret=None
session_secret=None,
)
assert "SESSION_SECRET must be set" in str(exc_info.value)
def test_session_secret_minimum_length(self):
"""Test that SESSION_SECRET must be at least 32 characters."""
with pytest.raises(ValidationError) as exc_info:
@@ -41,10 +44,10 @@ class TestConfigurationValidation:
gotenberg_url="http://localhost:3000",
workdir="/tmp",
auth_enabled=True,
session_secret="short"
session_secret="short",
)
assert "at least 32 characters" in str(exc_info.value)
def test_valid_configuration(self):
"""Test that valid configuration is accepted."""
config = Settings(
@@ -57,11 +60,11 @@ class TestConfigurationValidation:
gotenberg_url="http://localhost:3000",
workdir="/tmp",
auth_enabled=True,
session_secret="a" * 32 # 32 character secret
session_secret="a" * 32, # 32 character secret
)
assert config.auth_enabled is True
assert len(config.session_secret) == 32
def test_auth_disabled_no_session_secret_required(self):
"""Test that SESSION_SECRET is not required when auth is disabled."""
config = Settings(
@@ -74,18 +77,16 @@ class TestConfigurationValidation:
gotenberg_url="http://localhost:3000",
workdir="/tmp",
auth_enabled=False,
session_secret=None
session_secret=None,
)
assert config.auth_enabled is False
assert config.session_secret is None
@pytest.mark.unit
class TestBuildMetadataConfiguration:
"""Tests for build metadata configuration."""
def test_version_from_environment(self, monkeypatch):
"""Test that version is read from APP_VERSION environment variable."""
monkeypatch.setenv("APP_VERSION", "1.2.3-test")
@@ -98,10 +99,10 @@ class TestBuildMetadataConfiguration:
azure_endpoint="https://test.example.com",
gotenberg_url="http://localhost:3000",
workdir="/tmp",
auth_enabled=False
auth_enabled=False,
)
assert config.version == "1.2.3-test"
def test_build_date_from_environment(self, monkeypatch):
"""Test that build_date is read from BUILD_DATE environment variable."""
monkeypatch.setenv("BUILD_DATE", "2026-01-15")
@@ -114,7 +115,7 @@ class TestBuildMetadataConfiguration:
azure_endpoint="https://test.example.com",
gotenberg_url="http://localhost:3000",
workdir="/tmp",
auth_enabled=False
auth_enabled=False,
)
assert config.build_date == "2026-01-15"
@@ -130,10 +131,10 @@ class TestBuildMetadataConfiguration:
azure_endpoint="https://test.example.com",
gotenberg_url="http://localhost:3000",
workdir="/tmp",
auth_enabled=False
auth_enabled=False,
)
assert config.build_date == "2026-01-15T10:30:00Z"
def test_git_sha_from_environment(self, monkeypatch):
"""Test that git_sha is read from GIT_COMMIT_SHA environment variable."""
monkeypatch.setenv("GIT_COMMIT_SHA", "abc1234")
@@ -146,16 +147,17 @@ class TestBuildMetadataConfiguration:
azure_endpoint="https://test.example.com",
gotenberg_url="http://localhost:3000",
workdir="/tmp",
auth_enabled=False
auth_enabled=False,
)
assert config.git_sha == "abc1234"
def test_git_sha_default(self, monkeypatch, tmp_path):
"""Test that git_sha defaults to 'unknown' when not set."""
# Mock the file system to ensure no GIT_SHA file exists
import app.config
monkeypatch.setattr(app.config.os.path, 'dirname', lambda x: str(tmp_path))
monkeypatch.setattr(app.config.os.path, "dirname", lambda x: str(tmp_path))
config = Settings(
database_url="sqlite:///test.db",
redis_url="redis://localhost:6379",
@@ -165,7 +167,7 @@ class TestBuildMetadataConfiguration:
azure_endpoint="https://test.example.com",
gotenberg_url="http://localhost:3000",
workdir="/tmp",
auth_enabled=False
auth_enabled=False,
)
# When no file or env var exists, should return "unknown"
assert config.git_sha == "unknown"
@@ -173,7 +175,8 @@ class TestBuildMetadataConfiguration:
def test_version_default_when_no_file_or_env(self, monkeypatch, tmp_path):
"""Test that version defaults to 'unknown' when no VERSION file or env var exists."""
import app.config
monkeypatch.setattr(app.config.os.path, 'dirname', lambda x: str(tmp_path))
monkeypatch.setattr(app.config.os.path, "dirname", lambda x: str(tmp_path))
config = Settings(
database_url="sqlite:///test.db",
@@ -184,11 +187,11 @@ class TestBuildMetadataConfiguration:
azure_endpoint="https://test.example.com",
gotenberg_url="http://localhost:3000",
workdir="/tmp",
auth_enabled=False
auth_enabled=False,
)
# When no file or env var exists, should return "unknown"
assert config.version == "unknown"
def test_runtime_info_property(self):
"""Test that runtime_info returns build information."""
config = Settings(
@@ -200,7 +203,7 @@ class TestBuildMetadataConfiguration:
azure_endpoint="https://test.example.com",
gotenberg_url="http://localhost:3000",
workdir="/tmp",
auth_enabled=False
auth_enabled=False,
)
runtime_info = config.runtime_info
# Should contain version, build_date, and git_sha in some form
@@ -210,7 +213,7 @@ class TestBuildMetadataConfiguration:
@pytest.mark.unit
class TestNotificationConfiguration:
"""Tests for notification configuration parsing."""
def test_notification_urls_from_string(self):
"""Test parsing notification URLs from comma-separated string."""
config = Settings(
@@ -223,12 +226,12 @@ class TestNotificationConfiguration:
gotenberg_url="http://localhost:3000",
workdir="/tmp",
auth_enabled=False,
notification_urls="discord://webhook1,telegram://webhook2"
notification_urls="discord://webhook1,telegram://webhook2",
)
assert len(config.notification_urls) == 2
assert "discord://webhook1" in config.notification_urls
assert "telegram://webhook2" in config.notification_urls
def test_notification_urls_from_list(self):
"""Test that notification URLs can be provided as a list."""
config = Settings(
@@ -241,7 +244,7 @@ class TestNotificationConfiguration:
gotenberg_url="http://localhost:3000",
workdir="/tmp",
auth_enabled=False,
notification_urls=["discord://webhook1", "telegram://webhook2"]
notification_urls=["discord://webhook1", "telegram://webhook2"],
)
assert len(config.notification_urls) == 2
@@ -250,7 +253,7 @@ class TestNotificationConfiguration:
@pytest.mark.security
class TestSecurityConfiguration:
"""Tests for security-related configuration."""
def test_no_default_credentials_in_config(self):
"""Test that no default credentials are present in configuration."""
# This test ensures we don't accidentally have hardcoded credentials
@@ -263,15 +266,15 @@ class TestSecurityConfiguration:
azure_endpoint="https://test.example.com",
gotenberg_url="http://localhost:3000",
workdir="/tmp",
auth_enabled=False
auth_enabled=False,
)
# Ensure optional credentials are actually optional (None)
assert config.dropbox_app_key is None
assert config.dropbox_app_secret is None
assert config.nextcloud_password is None
assert config.paperless_ngx_api_token is None
def test_optional_services_dont_require_credentials(self):
"""Test that application can start without optional service credentials."""
config = Settings(
@@ -283,8 +286,8 @@ class TestSecurityConfiguration:
azure_endpoint="https://test.example.com",
gotenberg_url="http://localhost:3000",
workdir="/tmp",
auth_enabled=False
auth_enabled=False,
)
# Should not raise an error
assert config.database_url == "sqlite:///test.db"
tests/test_config_loader.py
+1
View File
@@ -1,4 +1,5 @@
"""Tests for app/utils/config_loader.py module."""
import pytest
from app.utils.config_loader import convert_setting_value
tests/test_config_validators.py
+5 -3
View File
@@ -1,12 +1,14 @@
"""Tests for app/utils/config_validator/validators.py module."""
import pytest
from unittest.mock import patch
import pytest
from app.utils.config_validator.validators import (
validate_storage_configs,
check_all_configs,
validate_email_config,
validate_notification_config,
check_all_configs,
validate_storage_configs,
)
tests/test_convert_to_pdf.py
+6 -1
View File
@@ -1,7 +1,9 @@
"""Tests for app/tasks/convert_to_pdf.py module."""
import os
from unittest.mock import MagicMock, patch
import pytest
from unittest.mock import patch, MagicMock
@pytest.mark.unit
@@ -11,6 +13,7 @@ class TestConvertToPdfMimeTypes:
def test_office_extensions_set(self):
"""Test that the task module is importable."""
from app.tasks.convert_to_pdf import convert_to_pdf
assert callable(convert_to_pdf)
@patch("app.tasks.convert_to_pdf.requests")
@@ -25,6 +28,7 @@ class TestConvertToPdfMimeTypes:
mock_self.request.id = "test-task-id"
from app.tasks.convert_to_pdf import convert_to_pdf
result = convert_to_pdf.__wrapped__(mock_self, "/tmp/test.docx")
assert result is None
@@ -46,5 +50,6 @@ class TestConvertToPdfMimeTypes:
mock_self.request.id = "test-task-id"
from app.tasks.convert_to_pdf import convert_to_pdf
result = convert_to_pdf.__wrapped__(mock_self, str(test_file))
assert result is None
tests/test_coverage_boost.py
+24 -1
View File
@@ -1,6 +1,8 @@
"""Tests to boost coverage for various small modules."""
from unittest.mock import MagicMock, patch
import pytest
from unittest.mock import patch, MagicMock
@pytest.mark.unit
@@ -10,11 +12,13 @@ class TestUtilsCompat:
def test_imports_hash_file(self):
"""Test that hash_file can be imported from utils."""
from app.utils import hash_file
assert callable(hash_file)
def test_imports_log_task_progress(self):
"""Test that log_task_progress can be imported from utils."""
from app.utils import log_task_progress
assert callable(log_task_progress)
@@ -25,31 +29,37 @@ class TestConfigValidatorCompat:
def test_imports_validate_email_config(self):
"""Test backward compatible import."""
from app.utils.config_validator import validate_email_config
assert callable(validate_email_config)
def test_imports_validate_storage_configs(self):
"""Test backward compatible import."""
from app.utils.config_validator import validate_storage_configs
assert callable(validate_storage_configs)
def test_imports_mask_sensitive_value(self):
"""Test backward compatible import."""
from app.utils.config_validator import mask_sensitive_value
assert callable(mask_sensitive_value)
def test_imports_get_provider_status(self):
"""Test backward compatible import."""
from app.utils.config_validator import get_provider_status
assert callable(get_provider_status)
def test_imports_dump_all_settings(self):
"""Test backward compatible import."""
from app.utils.config_validator import dump_all_settings
assert callable(dump_all_settings)
def test_imports_check_all_configs(self):
"""Test backward compatible import."""
from app.utils.config_validator import check_all_configs
assert callable(check_all_configs)
@@ -60,6 +70,7 @@ class TestCeleryWorkerImport:
def test_celery_worker_module_exists(self):
"""Test that celery_worker module can be found."""
import importlib
spec = importlib.util.find_spec("app.celery_worker")
assert spec is not None
@@ -71,22 +82,26 @@ class TestSettingsDisplayMasking:
def test_dump_all_settings_masks_passwords(self):
"""Test that passwords are masked in settings dump."""
from app.utils.config_validator.settings_display import dump_all_settings
# Should not raise
dump_all_settings()
def test_dump_all_settings_masks_tokens(self):
"""Test that tokens are masked in settings dump."""
from app.utils.config_validator.settings_display import dump_all_settings
dump_all_settings()
def test_dump_all_settings_masks_keys(self):
"""Test that API keys are masked in settings dump."""
from app.utils.config_validator.settings_display import dump_all_settings
dump_all_settings()
def test_get_settings_for_display_categories(self):
"""Test that all expected categories are returned."""
from app.utils.config_validator.settings_display import get_settings_for_display
result = get_settings_for_display(show_values=True)
# Should have multiple categories
assert len(result) > 3
@@ -102,6 +117,7 @@ class TestNotificationInit:
"""Test init_apprise when no URLs configured."""
mock_settings.notification_urls = []
from app.utils.notification import init_apprise
result = init_apprise()
assert result is not None
@@ -111,6 +127,7 @@ class TestNotificationInit:
"""Test init_apprise with URLs configured."""
mock_settings.notification_urls = ["json://localhost"]
from app.utils.notification import init_apprise
result = init_apprise()
assert result is not None
@@ -127,6 +144,7 @@ class TestNotificationFileProcessed:
mock_send.return_value = True
from app.utils.notification import notify_file_processed
result = notify_file_processed(
filename="test.pdf",
file_size=1048576,
@@ -144,6 +162,7 @@ class TestNotificationFileProcessed:
mock_send.return_value = True
from app.utils.notification import notify_file_processed
result = notify_file_processed(
filename="small.pdf",
file_size=512, # Less than 1KB
@@ -165,6 +184,7 @@ class TestNotificationCeleryFailure:
mock_send.return_value = True
from app.utils.notification import notify_celery_failure
result = notify_celery_failure(
task_name="process_document",
task_id="task-123",
@@ -188,6 +208,7 @@ class TestNotificationCredentialFailure:
mock_send.return_value = True
from app.utils.notification import notify_credential_failure
result = notify_credential_failure(
service_name="OpenAI",
error="Invalid API key",
@@ -208,6 +229,7 @@ class TestNotificationStartupShutdown:
mock_send.return_value = True
from app.utils.notification import notify_startup
result = notify_startup()
assert result is True
@@ -220,5 +242,6 @@ class TestNotificationStartupShutdown:
mock_send.return_value = True
from app.utils.notification import notify_shutdown
result = notify_shutdown()
assert result is True
tests/test_coverage_final.py
+20 -1
View File
@@ -1,6 +1,8 @@
"""Final tests to push coverage over 60%."""
from unittest.mock import MagicMock, patch
import pytest
from unittest.mock import patch, MagicMock
@pytest.mark.unit
@@ -10,6 +12,7 @@ class TestCheckCredentialsFunctions:
def test_sync_test_openai_connection(self):
"""Test sync_test_openai_connection."""
from app.tasks.check_credentials import sync_test_openai_connection
result = sync_test_openai_connection()
assert isinstance(result, dict)
assert "status" in result
@@ -17,6 +20,7 @@ class TestCheckCredentialsFunctions:
def test_sync_test_azure_connection(self):
"""Test sync_test_azure_connection."""
from app.tasks.check_credentials import sync_test_azure_connection
result = sync_test_azure_connection()
assert isinstance(result, dict)
assert "status" in result
@@ -24,6 +28,7 @@ class TestCheckCredentialsFunctions:
def test_sync_test_dropbox_token(self):
"""Test sync_test_dropbox_token."""
from app.tasks.check_credentials import sync_test_dropbox_token
result = sync_test_dropbox_token()
assert isinstance(result, dict)
assert "status" in result
@@ -31,6 +36,7 @@ class TestCheckCredentialsFunctions:
def test_sync_test_google_drive_token(self):
"""Test sync_test_google_drive_token."""
from app.tasks.check_credentials import sync_test_google_drive_token
result = sync_test_google_drive_token()
assert isinstance(result, dict)
assert "status" in result
@@ -38,6 +44,7 @@ class TestCheckCredentialsFunctions:
def test_sync_test_onedrive_token(self):
"""Test sync_test_onedrive_token."""
from app.tasks.check_credentials import sync_test_onedrive_token
result = sync_test_onedrive_token()
assert isinstance(result, dict)
assert "status" in result
@@ -45,29 +52,34 @@ class TestCheckCredentialsFunctions:
def test_sync_test_nextcloud_credentials(self):
"""Test that check_credentials module has check_credentials task."""
from app.tasks.check_credentials import check_credentials
assert callable(check_credentials)
def test_sync_test_sftp_credentials(self):
"""Test MockRequest scope attribute."""
from app.tasks.check_credentials import MockRequest
req = MockRequest()
assert hasattr(req, "session")
def test_sync_test_email_credentials(self):
"""Test MockRequest path_params attribute."""
from app.tasks.check_credentials import MockRequest
req = MockRequest()
assert isinstance(req.query_params, dict)
def test_sync_test_ftp_credentials(self):
"""Test MockRequest headers attribute."""
from app.tasks.check_credentials import MockRequest
req = MockRequest()
assert isinstance(req.headers, dict)
def test_sync_test_paperless_credentials(self):
"""Test get_failure_state returns dict."""
from app.tasks.check_credentials import get_failure_state
result = get_failure_state()
assert isinstance(result, dict)
@@ -75,7 +87,9 @@ class TestCheckCredentialsFunctions:
"""Test save_failure_state accepts dict."""
import os
from unittest.mock import patch
from app.tasks.check_credentials import save_failure_state
with patch("app.tasks.check_credentials.FAILURE_STATE_FILE", "/tmp/test_failure_state_final.json"):
save_failure_state({"test": "value"})
if os.path.exists("/tmp/test_failure_state_final.json"):
@@ -89,6 +103,7 @@ class TestImapPullInboxes:
def test_pull_all_inboxes_is_callable(self):
"""Test that pull_all_inboxes is callable."""
from app.tasks.imap_tasks import pull_all_inboxes
assert callable(pull_all_inboxes)
@@ -99,6 +114,7 @@ class TestViewsProviderStatus:
def test_get_provider_status_returns_dict(self):
"""Test get_provider_status."""
from app.utils.config_validator.providers import get_provider_status
result = get_provider_status()
assert isinstance(result, dict)
assert len(result) > 0
@@ -111,6 +127,7 @@ class TestSettingsService:
def test_get_settings_by_category(self):
"""Test get_settings_by_category."""
from app.utils.settings_service import get_settings_by_category
result = get_settings_by_category()
assert isinstance(result, dict)
assert len(result) > 0
@@ -118,12 +135,14 @@ class TestSettingsService:
def test_get_setting_metadata(self):
"""Test get_setting_metadata for a known key."""
from app.utils.settings_service import get_setting_metadata
result = get_setting_metadata("openai_api_key")
assert isinstance(result, dict)
def test_validate_setting_value(self):
"""Test validate_setting_value."""
from app.utils.settings_service import validate_setting_value
# Should return tuple of (is_valid, error_message or None)
result = validate_setting_value("openai_api_key", "test-key")
assert isinstance(result, tuple)
tests/test_database.py
+6 -4
View File
@@ -1,9 +1,11 @@
"""Tests for app/database.py module."""
import os
import pytest
from unittest.mock import patch, MagicMock
from app.database import init_db, get_db
import os
from unittest.mock import MagicMock, patch
import pytest
from app.database import get_db, init_db
@pytest.mark.unit
tests/test_diagnostic.py
+1
View File
@@ -1,4 +1,5 @@
"""Tests for app/api/diagnostic.py module."""
import pytest
tests/test_e2e_full_stack.py
+11 -9
View File
@@ -7,9 +7,10 @@ and test the complete application workflow from API request to file upload.
import os
import time
from unittest.mock import patch
import pytest
import requests
from unittest.mock import patch
# Import testcontainers requirement
pytest.importorskip("testcontainers", reason="testcontainers not installed")
@@ -22,16 +23,16 @@ except ModuleNotFoundError:
_has_psycopg2 = False
from tests.fixtures_integration import (
postgres_container,
redis_container,
gotenberg_container,
webdav_container,
sftp_container,
minio_container,
full_infrastructure,
celery_app,
celery_worker,
db_session_real,
full_infrastructure,
gotenberg_container,
minio_container,
postgres_container,
redis_container,
sftp_container,
webdav_container,
)
_TEST_CREDENTIAL = "pass" # noqa: S105
@@ -117,9 +118,10 @@ class TestEndToEndWithRedis:
This verifies the Redis broker is working correctly.
"""
from app.tasks.upload_to_webdav import upload_to_webdav
import redis
from app.tasks.upload_to_webdav import upload_to_webdav
# Connect to Redis directly
r = redis.from_url(redis_container["url"])
tests/test_embed_metadata.py
+3 -1
View File
@@ -1,7 +1,9 @@
"""Tests for app/tasks/embed_metadata_into_pdf.py module."""
import os
from unittest.mock import MagicMock, patch
import pytest
from unittest.mock import patch, MagicMock
from app.tasks.embed_metadata_into_pdf import persist_metadata
from app.utils.filename_utils import get_unique_filepath_with_counter
tests/test_endpoint_registration.py
+4 -9
View File
@@ -5,9 +5,10 @@ This test module serves as a regression prevention mechanism to ensure
that endpoints remain accessible after code refactoring or reorganization.
"""
import pytest
from unittest.mock import Mock, patch
import pytest
# Test constants
TEST_URL = "https://example.com/test.pdf"
@@ -34,10 +35,7 @@ class TestEndpointRegistration:
mock_process_document.delay.return_value = mock_task
# Make a request to the endpoint - it should not return 404
response = client.post(
"/api/process-url",
json={"url": TEST_URL}
)
response = client.post("/api/process-url", json={"url": TEST_URL})
# The endpoint exists if we don't get a 404
# We may get other errors (401, 400, 500, etc.) due to validation or missing mocks,
@@ -66,10 +64,7 @@ class TestEndpointRegistration:
mock_process_document.delay.return_value = mock_task
# Try POST request
response = client.post(
"/api/process-url",
json={"url": TEST_URL}
)
response = client.post("/api/process-url", json={"url": TEST_URL})
# Should not return 405 (Method Not Allowed)
assert response.status_code != 405, (
tests/test_extract_metadata.py
+1
View File
@@ -1,4 +1,5 @@
"""Tests for app/tasks/extract_metadata_with_gpt.py module."""
import pytest
from app.tasks.extract_metadata_with_gpt import extract_json_from_text
tests/test_file_listing.py
+50 -43
View File
@@ -1,17 +1,20 @@
"""
Tests for file listing, pagination, filtering, and detail endpoints.
"""
from datetime import datetime
import pytest
from fastapi.testclient import TestClient
from app.models import FileProcessingStep, FileRecord, ProcessingLog
from datetime import datetime
@pytest.mark.integration
@pytest.mark.requires_db
class TestFileListingPagination:
"""Tests for file listing with pagination, sorting, and filtering."""
def test_list_files_empty_with_pagination(self, client: TestClient):
"""Test listing files when database is empty returns pagination structure."""
response = client.get("/api/files")
@@ -22,7 +25,7 @@ class TestFileListingPagination:
assert isinstance(data["files"], list)
assert len(data["files"]) == 0
assert data["pagination"]["total_items"] == 0
def test_list_files_with_data(self, client: TestClient, db_session):
"""Test listing files with sample data."""
# Create sample files
@@ -32,18 +35,18 @@ class TestFileListingPagination:
original_filename=f"test{i}.pdf",
local_filename=f"/tmp/test{i}.pdf",
file_size=1024 * (i + 1),
mime_type="application/pdf"
mime_type="application/pdf",
)
db_session.add(file_record)
db_session.commit()
response = client.get("/api/files")
assert response.status_code == 200
data = response.json()
assert len(data["files"]) == 5
assert data["pagination"]["total_items"] == 5
assert data["pagination"]["page"] == 1
def test_pagination_works(self, client: TestClient, db_session):
"""Test that pagination correctly limits results."""
# Create 10 sample files
@@ -53,11 +56,11 @@ class TestFileListingPagination:
original_filename=f"test{i}.pdf",
local_filename=f"/tmp/test{i}.pdf",
file_size=1024,
mime_type="application/pdf"
mime_type="application/pdf",
)
db_session.add(file_record)
db_session.commit()
# Get first page with 5 items per page
response = client.get("/api/files?page=1&per_page=5")
assert response.status_code == 200
@@ -65,14 +68,14 @@ class TestFileListingPagination:
assert len(data["files"]) == 5
assert data["pagination"]["page"] == 1
assert data["pagination"]["total_pages"] == 2
# Get second page
response = client.get("/api/files?page=2&per_page=5")
assert response.status_code == 200
data = response.json()
assert len(data["files"]) == 5
assert data["pagination"]["page"] == 2
def test_sorting_by_filename(self, client: TestClient, db_session):
"""Test sorting files by filename."""
# Create files with different names
@@ -82,25 +85,25 @@ class TestFileListingPagination:
original_filename=name,
local_filename=f"/tmp/{name}",
file_size=1024,
mime_type="application/pdf"
mime_type="application/pdf",
)
db_session.add(file_record)
db_session.commit()
# Sort ascending
response = client.get("/api/files?sort_by=original_filename&sort_order=asc")
assert response.status_code == 200
data = response.json()
filenames = [f["original_filename"] for f in data["files"]]
assert filenames == ["apple.pdf", "middle.pdf", "zebra.pdf"]
# Sort descending
response = client.get("/api/files?sort_by=original_filename&sort_order=desc")
assert response.status_code == 200
data = response.json()
filenames = [f["original_filename"] for f in data["files"]]
assert filenames == ["zebra.pdf", "middle.pdf", "apple.pdf"]
def test_sorting_by_file_size(self, client: TestClient, db_session):
"""Test sorting files by size."""
# Create files with different sizes
@@ -110,18 +113,18 @@ class TestFileListingPagination:
original_filename=f"file{i}.pdf",
local_filename=f"/tmp/file{i}.pdf",
file_size=size,
mime_type="application/pdf"
mime_type="application/pdf",
)
db_session.add(file_record)
db_session.commit()
# Sort by size ascending
response = client.get("/api/files?sort_by=file_size&sort_order=asc")
assert response.status_code == 200
data = response.json()
sizes = [f["file_size"] for f in data["files"]]
assert sizes == [1000, 3000, 5000]
def test_search_filter(self, client: TestClient, db_session):
"""Test searching files by filename."""
# Create files with different names
@@ -131,11 +134,11 @@ class TestFileListingPagination:
original_filename=name,
local_filename=f"/tmp/{name}",
file_size=1024,
mime_type="application/pdf"
mime_type="application/pdf",
)
db_session.add(file_record)
db_session.commit()
# Search for "2024"
response = client.get("/api/files?search=2024")
assert response.status_code == 200
@@ -145,7 +148,7 @@ class TestFileListingPagination:
assert "invoice_2024.pdf" in filenames
assert "receipt_2024.pdf" in filenames
assert "report.pdf" not in filenames
def test_mime_type_filter(self, client: TestClient, db_session):
"""Test filtering files by MIME type."""
# Create files with different MIME types
@@ -160,11 +163,11 @@ class TestFileListingPagination:
original_filename=filename,
local_filename=f"/tmp/{filename}",
file_size=1024,
mime_type=mime_type
mime_type=mime_type,
)
db_session.add(file_record)
db_session.commit()
# Filter by PDF mime type
response = client.get("/api/files?mime_type=application/pdf")
assert response.status_code == 200
@@ -172,7 +175,7 @@ class TestFileListingPagination:
assert len(data["files"]) == 2
for file in data["files"]:
assert file["mime_type"] == "application/pdf"
def test_processing_status_included(self, client: TestClient, db_session):
"""Test that processing status is included in file listing."""
# Create a file
@@ -181,11 +184,11 @@ class TestFileListingPagination:
original_filename="test.pdf",
local_filename="/tmp/test.pdf",
file_size=1024,
mime_type="application/pdf"
mime_type="application/pdf",
)
db_session.add(file_record)
db_session.commit()
# Add a processing step (used for status determination)
step = FileProcessingStep(
file_id=file_record.id,
@@ -194,7 +197,7 @@ class TestFileListingPagination:
)
db_session.add(step)
db_session.commit()
response = client.get("/api/files")
assert response.status_code == 200
data = response.json()
@@ -207,7 +210,7 @@ class TestFileListingPagination:
@pytest.mark.requires_db
class TestFileDetailEndpoint:
"""Tests for file detail endpoint."""
def test_get_file_detail_success(self, client: TestClient, db_session):
"""Test getting details for an existing file."""
# Create a file
@@ -216,13 +219,17 @@ class TestFileDetailEndpoint:
original_filename="test.pdf",
local_filename="/tmp/test.pdf",
file_size=1024,
mime_type="application/pdf"
mime_type="application/pdf",
)
db_session.add(file_record)
db_session.commit()
# Add processing steps (used for status determination)
for step_name, status in [("extract_text", "success"), ("extract_metadata_with_gpt", "in_progress"), ("embed_metadata_into_pdf", "success")]:
for step_name, status in [
("extract_text", "success"),
("extract_metadata_with_gpt", "in_progress"),
("embed_metadata_into_pdf", "success"),
]:
step = FileProcessingStep(
file_id=file_record.id,
step_name=step_name,
@@ -237,33 +244,33 @@ class TestFileDetailEndpoint:
task_id=f"task_{i}",
step_name=f"step_{i}",
status=status,
message=f"Message {i}"
message=f"Message {i}",
)
db_session.add(log)
db_session.commit()
response = client.get(f"/api/files/{file_record.id}")
assert response.status_code == 200
data = response.json()
# Check file details
assert "file" in data
assert data["file"]["id"] == file_record.id
assert data["file"]["original_filename"] == "test.pdf"
# Check processing status
assert "processing_status" in data
assert data["processing_status"]["status"] in ["completed", "processing"]
# Check logs
assert "logs" in data
assert len(data["logs"]) == 3
def test_get_nonexistent_file_detail(self, client: TestClient):
"""Test getting details for a non-existent file returns 404."""
response = client.get("/api/files/99999")
assert response.status_code == 404
def test_file_detail_status_determination(self, client: TestClient, db_session):
"""Test that processing status is correctly determined."""
# Create a file
@@ -272,16 +279,16 @@ class TestFileDetailEndpoint:
original_filename="test.pdf",
local_filename="/tmp/test.pdf",
file_size=1024,
mime_type="application/pdf"
mime_type="application/pdf",
)
db_session.add(file_record)
db_session.commit()
# Test 1: No steps = pending
response = client.get(f"/api/files/{file_record.id}")
assert response.status_code == 200
assert response.json()["processing_status"]["status"] == "pending"
# Test 2: Success step = completed
step = FileProcessingStep(
file_id=file_record.id,
@@ -290,11 +297,11 @@ class TestFileDetailEndpoint:
)
db_session.add(step)
db_session.commit()
response = client.get(f"/api/files/{file_record.id}")
assert response.status_code == 200
assert response.json()["processing_status"]["status"] == "completed"
# Test 3: Failure step = failed
step2 = FileProcessingStep(
file_id=file_record.id,
@@ -304,7 +311,7 @@ class TestFileDetailEndpoint:
)
db_session.add(step2)
db_session.commit()
response = client.get(f"/api/files/{file_record.id}")
assert response.status_code == 200
assert response.json()["processing_status"]["status"] == "failed"
tests/test_file_queries.py
+1 -1
View File
@@ -9,7 +9,7 @@ from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from app.database import Base
from app.models import FileRecord, FileProcessingStep
from app.models import FileProcessingStep, FileRecord
from app.utils.file_queries import apply_status_filter
tests/test_file_status_fix.py
+11 -25
View File
@@ -14,7 +14,7 @@ from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from app.database import Base
from app.models import FileRecord, FileProcessingStep
from app.models import FileProcessingStep, FileRecord
from app.utils.step_manager import get_file_overall_status, get_step_summary, initialize_file_steps, update_step_status
@@ -40,10 +40,7 @@ class TestFileStatusCalculation:
"""
# Create file and initialize steps
file_record = FileRecord(
filehash="test1",
original_filename="test.pdf",
local_filename="/tmp/test.pdf",
file_size=1024
filehash="test1", original_filename="test.pdf", local_filename="/tmp/test.pdf", file_size=1024
)
db_session.add(file_record)
db_session.commit()
@@ -52,6 +49,7 @@ class TestFileStatusCalculation:
# Mark all steps as success
from app.utils.step_manager import MAIN_PROCESSING_STEPS
for step_name in MAIN_PROCESSING_STEPS:
update_step_status(db_session, file_record.id, step_name, "success")
@@ -65,10 +63,7 @@ class TestFileStatusCalculation:
Test that status shows "processing" when there are in_progress steps.
"""
file_record = FileRecord(
filehash="test2",
original_filename="test2.pdf",
local_filename="/tmp/test2.pdf",
file_size=2048
filehash="test2", original_filename="test2.pdf", local_filename="/tmp/test2.pdf", file_size=2048
)
db_session.add(file_record)
db_session.commit()
@@ -89,10 +84,7 @@ class TestFileStatusCalculation:
Test that status shows "failed" when any step has failure status.
"""
file_record = FileRecord(
filehash="test3",
original_filename="test3.pdf",
local_filename="/tmp/test3.pdf",
file_size=3072
filehash="test3", original_filename="test3.pdf", local_filename="/tmp/test3.pdf", file_size=3072
)
db_session.add(file_record)
db_session.commit()
@@ -119,10 +111,7 @@ class TestMetricsCounting:
Test that main processing steps are counted correctly.
"""
file_record = FileRecord(
filehash="test4",
original_filename="test4.pdf",
local_filename="/tmp/test4.pdf",
file_size=4096
filehash="test4", original_filename="test4.pdf", local_filename="/tmp/test4.pdf", file_size=4096
)
db_session.add(file_record)
db_session.commit()
@@ -138,6 +127,7 @@ class TestMetricsCounting:
# Should count each step once
from app.utils.step_manager import MAIN_PROCESSING_STEPS
assert summary["total_main_steps"] == len(MAIN_PROCESSING_STEPS)
assert summary["main"]["success"] == 2
assert summary["main"]["in_progress"] == 1
@@ -147,16 +137,14 @@ class TestMetricsCounting:
Test that upload tasks are counted correctly.
"""
file_record = FileRecord(
filehash="test5",
original_filename="test5.pdf",
local_filename="/tmp/test5.pdf",
file_size=5120
filehash="test5", original_filename="test5.pdf", local_filename="/tmp/test5.pdf", file_size=5120
)
db_session.add(file_record)
db_session.commit()
initialize_file_steps(db_session, file_record.id)
from app.utils.step_manager import add_upload_steps
add_upload_steps(db_session, file_record.id, ["dropbox", "s3", "nextcloud"])
# Mark upload steps with different statuses
@@ -177,16 +165,14 @@ class TestMetricsCounting:
Test metrics for a file with multiple successful uploads.
"""
file_record = FileRecord(
filehash="test6",
original_filename="test6.pdf",
local_filename="/tmp/test6.pdf",
file_size=6144
filehash="test6", original_filename="test6.pdf", local_filename="/tmp/test6.pdf", file_size=6144
)
db_session.add(file_record)
db_session.commit()
initialize_file_steps(db_session, file_record.id)
from app.utils.step_manager import add_upload_steps
services = ["dropbox", "s3", "nextcloud", "google_drive", "onedrive", "webdav"]
add_upload_steps(db_session, file_record.id, services)
tests/test_filename_utils.py
+3 -3
View File
@@ -4,11 +4,12 @@ Tests for app/utils/filename_utils.py
Tests filename sanitization and manipulation functions.
"""
import pytest
import os
from datetime import datetime
from pathlib import Path
from unittest.mock import Mock, patch
from datetime import datetime
import pytest
@pytest.mark.unit
@@ -259,7 +260,6 @@ class TestFilenameUtilsEdgeCases:
assert "?" not in result
@pytest.mark.unit
class TestUniqueFilepathWithCounter:
"""Test unique filepath generation with numeric counter suffix"""
tests/test_files_view.py
+18 -17
View File
@@ -1,8 +1,10 @@
"""
Test the /files view UI endpoint to ensure template rendering works correctly.
"""
import pytest
from fastapi.testclient import TestClient
from app.models import FileRecord
@@ -16,7 +18,7 @@ def _assert_no_template_errors(content: str):
@pytest.mark.requires_db
class TestFilesView:
"""Tests for the /files UI view."""
def test_files_view_renders_without_error(self, client: TestClient, db_session):
"""Test that the /files view renders without 'min' undefined error."""
# Create some test files to ensure pagination works
@@ -26,25 +28,25 @@ class TestFilesView:
original_filename=f"test{i}.pdf",
local_filename=f"/tmp/test{i}.pdf",
file_size=1024 * (i + 1),
mime_type="application/pdf"
mime_type="application/pdf",
)
db_session.add(file_record)
db_session.commit()
# Access the /files view
response = client.get("/files")
assert response.status_code == 200
# Check that the response is HTML
assert "text/html" in response.headers.get("content-type", "")
# Check that the response contains expected content
content = response.text
assert "File Records" in content
# Ensure no 'min' or 'max' is undefined error - this is the key fix we're testing
_assert_no_template_errors(content)
def test_files_view_pagination_with_many_pages(self, client: TestClient, db_session):
"""Test that pagination works correctly with many pages and min/max functions work."""
# Create enough files to span multiple pages (e.g., 150 files with 50 per page = 3 pages)
@@ -54,39 +56,38 @@ class TestFilesView:
original_filename=f"test{i}.pdf",
local_filename=f"/tmp/test{i}.pdf",
file_size=1024,
mime_type="application/pdf"
mime_type="application/pdf",
)
db_session.add(file_record)
db_session.commit()
# Access the /files view with pagination
response = client.get("/files?page=2&per_page=50")
assert response.status_code == 200
content = response.text
# The key test: ensure the min/max functions work in the template
# (they're used for pagination on lines 397 and 406 of files.html)
_assert_no_template_errors(content)
def test_files_view_includes_drag_drop_elements(self, client: TestClient, db_session):
"""Test that the /files view includes drag-and-drop upload elements."""
# Access the /files view
response = client.get("/files")
assert response.status_code == 200
content = response.text
# Check that drag-and-drop elements are present
assert "dropOverlay" in content, "Drop overlay element should be present"
assert "uploadModal" in content, "Upload modal element should be present"
assert "drop-overlay" in content, "Drop overlay CSS class should be present"
assert "upload-modal" in content, "Upload modal CSS class should be present"
# Check that the upload.js script is included
assert "/static/js/upload.js" in content, "upload.js script should be included"
# Check for drag-and-drop event handlers
assert "dragenter" in content or "drag" in content, "Drag event handlers should be present"
assert "Drop files anywhere to upload" in content, "Drop message should be present"
tests/test_finalize_storage.py
+4 -1
View File
@@ -1,7 +1,9 @@
"""Tests for app/tasks/finalize_document_storage.py module."""
import os
from unittest.mock import MagicMock, patch
import pytest
from unittest.mock import patch, MagicMock
@pytest.mark.unit
@@ -18,4 +20,5 @@ class TestFinalizeDocumentStorageHelpers:
def test_module_imports(self):
"""Test that the module can be imported without errors."""
from app.tasks.finalize_document_storage import finalize_document_storage
assert callable(finalize_document_storage)
tests/test_imap_extended.py
+8 -8
View File
@@ -1,16 +1,18 @@
"""Extended tests for app/tasks/imap_tasks.py module."""
import os
import json
import pytest
import os
from datetime import datetime, timezone
from unittest.mock import patch, MagicMock
from email.message import EmailMessage
from unittest.mock import MagicMock, patch
import pytest
from app.tasks.imap_tasks import (
save_processed_emails,
load_processed_emails,
fetch_attachments_and_enqueue,
find_all_mail_xlist,
load_processed_emails,
save_processed_emails,
)
@@ -41,9 +43,7 @@ class TestFetchAttachmentsExtended:
msg = EmailMessage()
msg["Subject"] = "Test"
# Create attachment with wrong MIME type but .pdf extension
msg.add_attachment(
b"%PDF-1.4", maintype="application", subtype="octet-stream", filename="invoice.pdf"
)
msg.add_attachment(b"%PDF-1.4", maintype="application", subtype="octet-stream", filename="invoice.pdf")
with patch("app.tasks.imap_tasks.settings") as mock_settings:
mock_settings.workdir = str(tmp_path)
tests/test_imap_tasks.py
+10 -9
View File
@@ -1,23 +1,24 @@
"""Tests for app/tasks/imap_tasks.py module."""
import os
import json
import pytest
import os
from datetime import datetime, timedelta, timezone
from unittest.mock import patch, MagicMock
from email.message import EmailMessage
from unittest.mock import MagicMock, patch
import pytest
from app.tasks.imap_tasks import (
load_processed_emails,
save_processed_emails,
cleanup_old_entries,
check_and_pull_mailbox,
fetch_attachments_and_enqueue,
cleanup_old_entries,
email_already_has_label,
mark_as_processed_with_star,
mark_as_processed_with_label,
fetch_attachments_and_enqueue,
find_all_mail_folder,
get_capabilities,
load_processed_emails,
mark_as_processed_with_label,
mark_as_processed_with_star,
save_processed_emails,
)
_TEST_CREDENTIAL = "pass" # noqa: S105
tests/test_logging_utils.py
+3 -7
View File
@@ -5,9 +5,9 @@ Tests task progress logging functionality.
"""
import logging
from unittest.mock import MagicMock, Mock, patch
import pytest
from unittest.mock import Mock, patch, MagicMock
@pytest.mark.unit
@@ -64,9 +64,7 @@ class TestTaskLogging:
mock_processing_log.return_value = mock_log_entry
# Call without message
log_task_progress(
task_id="task-456", step_name="upload", status="completed", message=None, file_id=None
)
log_task_progress(task_id="task-456", step_name="upload", status="completed", message=None, file_id=None)
# Verify called with None for optional parameters
mock_processing_log.assert_called_once_with(
@@ -89,9 +87,7 @@ class TestTaskLogging:
mock_processing_log.return_value = mock_log_entry
# Call without file_id
log_task_progress(
task_id="task-789", step_name="metadata", status="running", message="Extracting metadata"
)
log_task_progress(task_id="task-789", step_name="metadata", status="running", message="Extracting metadata")
# file_id should default to None
mock_processing_log.assert_called_once()
tests/test_notification.py
+6 -5
View File
@@ -4,8 +4,9 @@ Tests for app/utils/notification.py
Tests notification utilities and URL masking.
"""
from unittest.mock import MagicMock, Mock, patch
import pytest
from unittest.mock import Mock, patch, MagicMock
_TEST_CREDENTIAL_URL = "https://user:password@example.com/notify" # noqa: S105
_TEST_QUERY_URL = "https://example.com/api?key=secret1&password=secret2&public=visible" # noqa: S105
@@ -133,8 +134,8 @@ class TestAppriseInitialization:
@patch("app.utils.notification.settings")
def test_init_apprise_with_configured_urls(self, mock_settings, mock_apprise_class):
"""Test Apprise initialization with configured URLs"""
from app.utils.notification import init_apprise
import app.utils.notification
from app.utils.notification import init_apprise
# Reset global
app.utils.notification._apprise = None
@@ -162,8 +163,8 @@ class TestAppriseInitialization:
@patch("app.utils.notification.settings")
def test_init_apprise_no_urls_configured(self, mock_settings, mock_apprise_class):
"""Test Apprise initialization without configured URLs"""
from app.utils.notification import init_apprise
import app.utils.notification
from app.utils.notification import init_apprise
# Reset global
app.utils.notification._apprise = None
@@ -188,8 +189,8 @@ class TestAppriseInitialization:
@patch("app.utils.notification.settings")
def test_init_apprise_caches_instance(self, mock_settings, mock_apprise_class):
"""Test that Apprise instance is cached"""
from app.utils.notification import init_apprise
import app.utils.notification
from app.utils.notification import init_apprise
# Reset global
app.utils.notification._apprise = None
@@ -214,8 +215,8 @@ class TestAppriseInitialization:
@patch("app.utils.notification.settings")
def test_init_apprise_handles_add_failure(self, mock_settings, mock_apprise_class):
"""Test handling when adding notification URL fails"""
from app.utils.notification import init_apprise
import app.utils.notification
from app.utils.notification import init_apprise
# Reset global
app.utils.notification._apprise = None
tests/test_notification_utils.py
+7 -5
View File
@@ -1,16 +1,18 @@
"""Tests for app/utils/notification.py module."""
from unittest.mock import MagicMock, patch
import pytest
from unittest.mock import patch, MagicMock
from app.utils.notification import (
_mask_sensitive_url,
send_notification,
init_apprise,
notify_celery_failure,
notify_credential_failure,
notify_startup,
notify_shutdown,
notify_file_processed,
init_apprise,
notify_shutdown,
notify_startup,
send_notification,
)
tests/test_oauth_helper.py
+5 -10
View File
@@ -4,8 +4,9 @@ Tests for app/utils/oauth_helper.py
Tests OAuth token exchange helper functions.
"""
import pytest
from unittest.mock import Mock, patch
import pytest
import requests
from fastapi import HTTPException
@@ -51,9 +52,7 @@ class TestOAuthTokenExchange:
assert result["expires_in"] == 3600
# Verify request was made correctly
mock_post.assert_called_once_with(
"https://oauth.example.com/token", data=payload, timeout=30
)
mock_post.assert_called_once_with("https://oauth.example.com/token", data=payload, timeout=30)
@patch("app.utils.oauth_helper.requests.post")
@patch("app.utils.oauth_helper.settings")
@@ -81,9 +80,7 @@ class TestOAuthTokenExchange:
)
# Verify custom timeout was used
mock_post.assert_called_once_with(
"https://oauth.example.com/token", data=payload, timeout=60
)
mock_post.assert_called_once_with("https://oauth.example.com/token", data=payload, timeout=60)
@patch("app.utils.oauth_helper.requests.post")
@patch("app.utils.oauth_helper.settings")
@@ -200,9 +197,7 @@ class TestOAuthTokenExchange:
# Mock error response with invalid JSON
mock_response = Mock()
mock_response.status_code = 400
mock_response.json.side_effect = requests.exceptions.JSONDecodeError(
"Invalid JSON", "", 0
)
mock_response.json.side_effect = requests.exceptions.JSONDecodeError("Invalid JSON", "", 0)
mock_post.return_value = mock_response
payload = {"grant_type": "authorization_code"}
tests/test_ocr_processing.py
+91 -127
View File
@@ -6,18 +6,19 @@ These tests verify OCR processing logic with mocked external AI/ML services
"""
import os
from unittest.mock import MagicMock, Mock, patch
import pytest
from unittest.mock import patch, MagicMock, Mock
from azure.ai.documentintelligence.models import AnalyzeResult
from app.tasks.process_with_azure_document_intelligence import (
process_with_azure_document_intelligence,
get_pdf_page_count,
check_page_rotation,
AZURE_DOC_INTELLIGENCE_LIMITS,
check_page_rotation,
get_pdf_page_count,
process_with_azure_document_intelligence,
)
from app.tasks.refine_text_with_gpt import refine_text_with_gpt
from app.tasks.rotate_pdf_pages import rotate_pdf_pages, determine_rotation_angle
from app.tasks.rotate_pdf_pages import determine_rotation_angle, rotate_pdf_pages
@pytest.mark.unit
@@ -86,22 +87,20 @@ startxref
mock_client.begin_analyze_document.return_value = mock_poller
mock_client.get_analyze_result_pdf.return_value = iter([b"searchable pdf content"])
with patch(
"app.tasks.process_with_azure_document_intelligence.document_intelligence_client",
mock_client,
), patch(
"app.tasks.process_with_azure_document_intelligence.settings"
) as mock_settings, patch(
"app.tasks.process_with_azure_document_intelligence.rotate_pdf_pages"
) as mock_rotate:
with (
patch(
"app.tasks.process_with_azure_document_intelligence.document_intelligence_client",
mock_client,
),
patch("app.tasks.process_with_azure_document_intelligence.settings") as mock_settings,
patch("app.tasks.process_with_azure_document_intelligence.rotate_pdf_pages") as mock_rotate,
):
mock_settings.workdir = str(tmp_path)
mock_rotate.delay = MagicMock()
# Run the task
result = process_with_azure_document_intelligence.run(
filename=test_pdf.name, file_id=1
)
result = process_with_azure_document_intelligence.run(filename=test_pdf.name, file_id=1)
# Verify results
assert result["file"] == test_pdf.name
@@ -124,15 +123,11 @@ startxref
@patch("app.tasks.process_with_azure_document_intelligence.log_task_progress")
def test_file_not_found_error(self, mock_log, tmp_path):
"""Test that FileNotFoundError is raised when file doesn't exist."""
with patch(
"app.tasks.process_with_azure_document_intelligence.settings"
) as mock_settings:
with patch("app.tasks.process_with_azure_document_intelligence.settings") as mock_settings:
mock_settings.workdir = str(tmp_path)
with pytest.raises(FileNotFoundError) as exc_info:
process_with_azure_document_intelligence.run(
filename="nonexistent.pdf", file_id=1
)
process_with_azure_document_intelligence.run(filename="nonexistent.pdf", file_id=1)
assert "Local file not found" in str(exc_info.value)
@@ -145,21 +140,16 @@ startxref
test_pdf = tmp_dir / "large.pdf"
test_pdf.write_bytes(b"dummy content")
with patch(
"app.tasks.process_with_azure_document_intelligence.settings"
) as mock_settings, patch(
"app.tasks.process_with_azure_document_intelligence.os.path.getsize"
) as mock_getsize:
with (
patch("app.tasks.process_with_azure_document_intelligence.settings") as mock_settings,
patch("app.tasks.process_with_azure_document_intelligence.os.path.getsize") as mock_getsize,
):
mock_settings.workdir = str(tmp_path)
# Mock file size to be larger than 500 MB
mock_getsize.return_value = AZURE_DOC_INTELLIGENCE_LIMITS[
"max_file_size_bytes"
] + 1024
mock_getsize.return_value = AZURE_DOC_INTELLIGENCE_LIMITS["max_file_size_bytes"] + 1024
result = process_with_azure_document_intelligence.run(
filename=test_pdf.name, file_id=1
)
result = process_with_azure_document_intelligence.run(filename=test_pdf.name, file_id=1)
# Verify error response
assert "error" in result
@@ -175,19 +165,16 @@ startxref
test_pdf = tmp_dir / "many_pages.pdf"
test_pdf.write_bytes(b"%PDF-1.4\n%%EOF") # Minimal valid PDF
with patch(
"app.tasks.process_with_azure_document_intelligence.settings"
) as mock_settings, patch(
"app.tasks.process_with_azure_document_intelligence.get_pdf_page_count"
) as mock_page_count:
with (
patch("app.tasks.process_with_azure_document_intelligence.settings") as mock_settings,
patch("app.tasks.process_with_azure_document_intelligence.get_pdf_page_count") as mock_page_count,
):
mock_settings.workdir = str(tmp_path)
# Mock page count to exceed limit
mock_page_count.return_value = AZURE_DOC_INTELLIGENCE_LIMITS["max_pages"] + 1
result = process_with_azure_document_intelligence.run(
filename=test_pdf.name, file_id=1
)
result = process_with_azure_document_intelligence.run(filename=test_pdf.name, file_id=1)
# Verify error response
assert "error" in result
@@ -219,15 +206,14 @@ startxref
mock_client.begin_analyze_document.return_value = mock_poller
mock_client.get_analyze_result_pdf.return_value = iter([b"content"])
with patch(
"app.tasks.process_with_azure_document_intelligence.document_intelligence_client",
mock_client,
), patch(
"app.tasks.process_with_azure_document_intelligence.settings"
) as mock_settings, patch(
"app.tasks.process_with_azure_document_intelligence.get_pdf_page_count"
) as mock_page_count, patch(
"app.tasks.process_with_azure_document_intelligence.rotate_pdf_pages"
with (
patch(
"app.tasks.process_with_azure_document_intelligence.document_intelligence_client",
mock_client,
),
patch("app.tasks.process_with_azure_document_intelligence.settings") as mock_settings,
patch("app.tasks.process_with_azure_document_intelligence.get_pdf_page_count") as mock_page_count,
patch("app.tasks.process_with_azure_document_intelligence.rotate_pdf_pages"),
):
mock_settings.workdir = str(tmp_path)
@@ -235,9 +221,7 @@ startxref
mock_page_count.return_value = None
# Should not raise an error
result = process_with_azure_document_intelligence.run(
filename=test_pdf.name, file_id=1
)
result = process_with_azure_document_intelligence.run(filename=test_pdf.name, file_id=1)
# Verify processing continued
assert "error" not in result
@@ -275,21 +259,19 @@ startxref
mock_client.begin_analyze_document.return_value = mock_poller
mock_client.get_analyze_result_pdf.return_value = iter([b"content"])
with patch(
"app.tasks.process_with_azure_document_intelligence.document_intelligence_client",
mock_client,
), patch(
"app.tasks.process_with_azure_document_intelligence.settings"
) as mock_settings, patch(
"app.tasks.process_with_azure_document_intelligence.rotate_pdf_pages"
) as mock_rotate:
with (
patch(
"app.tasks.process_with_azure_document_intelligence.document_intelligence_client",
mock_client,
),
patch("app.tasks.process_with_azure_document_intelligence.settings") as mock_settings,
patch("app.tasks.process_with_azure_document_intelligence.rotate_pdf_pages") as mock_rotate,
):
mock_settings.workdir = str(tmp_path)
mock_rotate.delay = MagicMock()
result = process_with_azure_document_intelligence.run(
filename=test_pdf.name, file_id=2
)
result = process_with_azure_document_intelligence.run(filename=test_pdf.name, file_id=2)
# Verify rotation data was passed correctly
mock_rotate.delay.assert_called_once()
@@ -311,24 +293,21 @@ startxref
test_pdf.write_bytes(b"%PDF-1.4\n%%EOF")
mock_client = Mock()
mock_client.begin_analyze_document.side_effect = Exception(
"Azure API connection failed"
)
mock_client.begin_analyze_document.side_effect = Exception("Azure API connection failed")
with patch(
"app.tasks.process_with_azure_document_intelligence.document_intelligence_client",
mock_client,
), patch(
"app.tasks.process_with_azure_document_intelligence.settings"
) as mock_settings:
with (
patch(
"app.tasks.process_with_azure_document_intelligence.document_intelligence_client",
mock_client,
),
patch("app.tasks.process_with_azure_document_intelligence.settings") as mock_settings,
):
mock_settings.workdir = str(tmp_path)
# Should raise the exception
with pytest.raises(Exception) as exc_info:
process_with_azure_document_intelligence.run(
filename=test_pdf.name, file_id=1
)
process_with_azure_document_intelligence.run(filename=test_pdf.name, file_id=1)
assert "Azure API connection failed" in str(exc_info.value)
@@ -456,13 +435,11 @@ class TestRefineTextWithGPT:
# Import the module to patch the correct function
from app.tasks import extract_metadata_with_gpt as metadata_module
with patch(
"app.tasks.refine_text_with_gpt.client", mock_client
), patch.object(
metadata_module, "extract_metadata_with_gpt"
) as mock_extract, patch(
"app.tasks.refine_text_with_gpt.settings"
) as mock_settings:
with (
patch("app.tasks.refine_text_with_gpt.client", mock_client),
patch.object(metadata_module, "extract_metadata_with_gpt") as mock_extract,
patch("app.tasks.refine_text_with_gpt.settings") as mock_settings,
):
mock_settings.openai_model = "gpt-4"
mock_extract.delay = MagicMock()
@@ -481,9 +458,7 @@ class TestRefineTextWithGPT:
assert call_kwargs["messages"][1]["content"] == raw_text
# Verify metadata extraction was queued
mock_extract.delay.assert_called_once_with(
filename, "This is some text with OCR errors"
)
mock_extract.delay.assert_called_once_with(filename, "This is some text with OCR errors")
@patch("app.tasks.refine_text_with_gpt.log_task_progress")
def test_openai_api_error(self, mock_log):
@@ -492,15 +467,12 @@ class TestRefineTextWithGPT:
filename = "test.pdf"
mock_client = Mock()
mock_client.chat.completions.create.side_effect = Exception(
"OpenAI API error"
)
mock_client.chat.completions.create.side_effect = Exception("OpenAI API error")
with patch(
"app.tasks.refine_text_with_gpt.client", mock_client
), patch(
"app.tasks.refine_text_with_gpt.settings"
) as mock_settings:
with (
patch("app.tasks.refine_text_with_gpt.client", mock_client),
patch("app.tasks.refine_text_with_gpt.settings") as mock_settings,
):
mock_settings.openai_model = "gpt-4"
@@ -606,11 +578,10 @@ startxref
rotation_data = {0: 90} # Rotate first page by 90 degrees
extracted_text = "Test text"
with patch(
"app.tasks.rotate_pdf_pages.settings"
) as mock_settings, patch(
"app.tasks.rotate_pdf_pages.extract_metadata_with_gpt"
) as mock_extract:
with (
patch("app.tasks.rotate_pdf_pages.settings") as mock_settings,
patch("app.tasks.rotate_pdf_pages.extract_metadata_with_gpt") as mock_extract,
):
mock_settings.workdir = str(tmp_path)
mock_extract.delay = MagicMock()
@@ -629,9 +600,7 @@ startxref
assert "0" in result["applied_rotations"]
# Verify metadata extraction was queued
mock_extract.delay.assert_called_once_with(
test_pdf.name, extracted_text, 1
)
mock_extract.delay.assert_called_once_with(test_pdf.name, extracted_text, 1)
@patch("app.tasks.rotate_pdf_pages.log_task_progress")
def test_rotate_pdf_pages_no_rotation_needed(self, mock_log, tmp_path):
@@ -644,11 +613,10 @@ startxref
extracted_text = "Test text"
with patch(
"app.tasks.rotate_pdf_pages.settings"
) as mock_settings, patch(
"app.tasks.rotate_pdf_pages.extract_metadata_with_gpt"
) as mock_extract:
with (
patch("app.tasks.rotate_pdf_pages.settings") as mock_settings,
patch("app.tasks.rotate_pdf_pages.extract_metadata_with_gpt") as mock_extract,
):
mock_settings.workdir = str(tmp_path)
mock_extract.delay = MagicMock()
@@ -679,11 +647,10 @@ startxref
rotation_data = {0: 0, 1: 0} # All pages have 0 rotation
extracted_text = "Test text"
with patch(
"app.tasks.rotate_pdf_pages.settings"
) as mock_settings, patch(
"app.tasks.rotate_pdf_pages.extract_metadata_with_gpt"
) as mock_extract:
with (
patch("app.tasks.rotate_pdf_pages.settings") as mock_settings,
patch("app.tasks.rotate_pdf_pages.extract_metadata_with_gpt") as mock_extract,
):
mock_settings.workdir = str(tmp_path)
mock_extract.delay = MagicMock()
@@ -705,11 +672,10 @@ startxref
tmp_dir = tmp_path / "tmp"
tmp_dir.mkdir()
with patch(
"app.tasks.rotate_pdf_pages.settings"
) as mock_settings, patch(
"app.tasks.rotate_pdf_pages.extract_metadata_with_gpt"
) as mock_extract:
with (
patch("app.tasks.rotate_pdf_pages.settings") as mock_settings,
patch("app.tasks.rotate_pdf_pages.extract_metadata_with_gpt") as mock_extract,
):
mock_settings.workdir = str(tmp_path)
mock_extract.delay = MagicMock()
@@ -742,11 +708,10 @@ startxref
rotation_data = {0: 90}
extracted_text = "Test text"
with patch(
"app.tasks.rotate_pdf_pages.settings"
) as mock_settings, patch(
"app.tasks.rotate_pdf_pages.extract_metadata_with_gpt"
) as mock_extract:
with (
patch("app.tasks.rotate_pdf_pages.settings") as mock_settings,
patch("app.tasks.rotate_pdf_pages.extract_metadata_with_gpt") as mock_extract,
):
mock_settings.workdir = str(tmp_path)
mock_extract.delay = MagicMock()
@@ -814,11 +779,10 @@ startxref
rotation_data = {"0": "90"}
extracted_text = "Test text"
with patch(
"app.tasks.rotate_pdf_pages.settings"
) as mock_settings, patch(
"app.tasks.rotate_pdf_pages.extract_metadata_with_gpt"
) as mock_extract:
with (
patch("app.tasks.rotate_pdf_pages.settings") as mock_settings,
patch("app.tasks.rotate_pdf_pages.extract_metadata_with_gpt") as mock_extract,
):
mock_settings.workdir = str(tmp_path)
mock_extract.delay = MagicMock()
tests/test_original_filename_preservation.py
+12 -10
View File
@@ -87,11 +87,12 @@ startxref
original_filename = "Apostille Sverige.pdf"
# Mock environment and dependencies
with patch("app.tasks.process_document.SessionLocal") as mock_session_local, patch(
"app.tasks.process_document.settings"
) as mock_settings, patch("app.tasks.process_document.log_task_progress"), patch(
"app.tasks.process_document.extract_metadata_with_gpt"
) as mock_extract:
with (
patch("app.tasks.process_document.SessionLocal") as mock_session_local,
patch("app.tasks.process_document.settings") as mock_settings,
patch("app.tasks.process_document.log_task_progress"),
patch("app.tasks.process_document.extract_metadata_with_gpt") as mock_extract,
):
# Setup mocks
mock_settings.workdir = str(tmp_path)
@@ -187,11 +188,12 @@ startxref
test_pdf.write_bytes(pdf_content)
# Mock environment and dependencies
with patch("app.tasks.process_document.SessionLocal") as mock_session_local, patch(
"app.tasks.process_document.settings"
) as mock_settings, patch("app.tasks.process_document.log_task_progress"), patch(
"app.tasks.process_document.extract_metadata_with_gpt"
) as mock_extract:
with (
patch("app.tasks.process_document.SessionLocal") as mock_session_local,
patch("app.tasks.process_document.settings") as mock_settings,
patch("app.tasks.process_document.log_task_progress"),
patch("app.tasks.process_document.extract_metadata_with_gpt") as mock_extract,
):
# Setup mocks
mock_settings.workdir = str(tmp_path)
tests/test_path_traversal_security.py
+8 -5
View File
@@ -4,11 +4,11 @@ Security tests for path traversal vulnerabilities.
Tests all file path operations to ensure they properly prevent path traversal attacks.
"""
import os
import json
import os
import tempfile
from pathlib import Path
from unittest.mock import Mock, patch, MagicMock
from unittest.mock import MagicMock, Mock, patch
import pytest
@@ -389,9 +389,10 @@ class TestFileUploadSecurity:
def test_sanitize_after_basename(self):
"""Test that sanitization happens after basename extraction."""
from app.utils.filename_utils import sanitize_filename
import os
from app.utils.filename_utils import sanitize_filename
malicious = "../../../passwd.pdf"
# Step 1: Extract basename (as ui_upload does)
@@ -442,10 +443,11 @@ class TestEndToEndPathTraversal:
def test_full_upload_flow_prevents_traversal(self, tmp_path):
"""Test complete upload flow prevents path traversal."""
from app.utils.filename_utils import sanitize_filename
import os
import uuid
from app.utils.filename_utils import sanitize_filename
# Simulate ui_upload flow
malicious_upload_filename = "../../../etc/passwd"
@@ -472,9 +474,10 @@ class TestEndToEndPathTraversal:
def test_metadata_embedding_flow_prevents_traversal(self, tmp_path):
"""Test metadata embedding flow prevents path traversal."""
from app.utils.filename_utils import sanitize_filename
import os
from app.utils.filename_utils import sanitize_filename
# Simulate GPT returning malicious filename
gpt_metadata = {
"filename": "../../../etc/shadow",
tests/test_rate_limiting.py
+4 -2
View File
@@ -101,9 +101,10 @@ def test_rate_limit_exceeded_returns_429(client):
@pytest.mark.security
def test_rate_limiting_uses_correct_identifier():
"""Test that rate limiting uses IP or user ID as identifier."""
from app.middleware.rate_limit import get_identifier
from fastapi import Request
from app.middleware.rate_limit import get_identifier
# Create a mock request
class MockRequest:
def __init__(self):
@@ -157,9 +158,10 @@ def test_limiter_disabled():
@pytest.mark.integration
def test_rate_limit_exception_handler_registered():
"""Test that rate limit exception handler is registered."""
from app.main import app
from slowapi.errors import RateLimitExceeded
from app.main import app
# Verify exception handler is registered
assert RateLimitExceeded in app.exception_handlers
tests/test_rclone_tasks.py
+3 -1
View File
@@ -1,7 +1,9 @@
"""Tests for app/tasks/upload_with_rclone.py module."""
import os
from unittest.mock import MagicMock, patch
import pytest
from unittest.mock import patch, MagicMock
from app.tasks.upload_with_rclone import upload_with_rclone
tests/test_settings.py
+74 -72
View File
@@ -6,112 +6,112 @@ import pytest
from fastapi.testclient import TestClient
from sqlalchemy.orm import Session
from app.config import Settings
from app.models import ApplicationSettings
from app.utils.config_loader import convert_setting_value, load_settings_from_db
from app.utils.settings_service import (
get_setting_from_db,
save_setting_to_db,
get_all_settings_from_db,
SETTING_METADATA,
delete_setting_from_db,
validate_setting_value,
get_all_settings_from_db,
get_setting_from_db,
get_setting_metadata,
get_settings_by_category,
SETTING_METADATA,
save_setting_to_db,
validate_setting_value,
)
from app.utils.config_loader import convert_setting_value, load_settings_from_db
from app.config import Settings
@pytest.mark.unit
class TestSettingsService:
"""Test settings service functions"""
def test_save_and_get_setting(self, db_session: Session):
"""Test saving and retrieving a setting from database"""
# Save a setting
result = save_setting_to_db(db_session, "test_key", "test_value")
assert result is True
# Retrieve the setting
value = get_setting_from_db(db_session, "test_key")
assert value == "test_value"
def test_update_existing_setting(self, db_session: Session):
"""Test updating an existing setting"""
# Save initial value
save_setting_to_db(db_session, "test_key", "initial_value")
# Update the value
result = save_setting_to_db(db_session, "test_key", "updated_value")
assert result is True
# Verify update
value = get_setting_from_db(db_session, "test_key")
assert value == "updated_value"
def test_get_nonexistent_setting(self, db_session: Session):
"""Test retrieving a setting that doesn't exist"""
value = get_setting_from_db(db_session, "nonexistent_key")
assert value is None
def test_get_all_settings(self, db_session: Session):
"""Test retrieving all settings from database"""
# Save multiple settings
save_setting_to_db(db_session, "key1", "value1")
save_setting_to_db(db_session, "key2", "value2")
save_setting_to_db(db_session, "key3", "value3")
# Get all settings
all_settings = get_all_settings_from_db(db_session)
assert len(all_settings) == 3
assert all_settings["key1"] == "value1"
assert all_settings["key2"] == "value2"
assert all_settings["key3"] == "value3"
def test_delete_setting(self, db_session: Session):
"""Test deleting a setting from database"""
# Save a setting
save_setting_to_db(db_session, "test_key", "test_value")
# Delete the setting
result = delete_setting_from_db(db_session, "test_key")
assert result is True
# Verify deletion
value = get_setting_from_db(db_session, "test_key")
assert value is None
def test_delete_nonexistent_setting(self, db_session: Session):
"""Test deleting a setting that doesn't exist"""
result = delete_setting_from_db(db_session, "nonexistent_key")
assert result is False
def test_validate_setting_value_boolean(self):
"""Test validation of boolean settings"""
# Valid boolean values
is_valid, error = validate_setting_value("debug", "true")
assert is_valid is True
assert error is None
is_valid, error = validate_setting_value("debug", "false")
assert is_valid is True
# Invalid boolean value
is_valid, error = validate_setting_value("debug", "maybe")
assert is_valid is False
assert "boolean" in error.lower()
def test_validate_session_secret_length(self):
"""Test validation of session_secret minimum length"""
# Too short
is_valid, error = validate_setting_value("session_secret", "short")
assert is_valid is False
assert "32 characters" in error
# Long enough
long_secret = "a" * 32
is_valid, error = validate_setting_value("session_secret", long_secret)
assert is_valid is True
def test_get_setting_metadata(self):
"""Test retrieving setting metadata"""
metadata = get_setting_metadata("database_url")
@@ -119,11 +119,11 @@ class TestSettingsService:
assert metadata["type"] == "string"
assert metadata["required"] is True
assert metadata["restart_required"] is True
# Test unknown setting
metadata = get_setting_metadata("unknown_setting")
assert metadata["category"] == "Other"
def test_get_settings_by_category(self):
"""Test getting settings organized by category"""
categories = get_settings_by_category()
@@ -132,17 +132,22 @@ class TestSettingsService:
assert "AI Services" in categories
assert "database_url" in categories["Core"]
assert "auth_enabled" in categories["Authentication"]
def test_setting_metadata_completeness(self):
"""Test that all major settings have metadata"""
# Check that we have a good number of settings defined
assert len(SETTING_METADATA) > 50, "Should have metadata for at least 50 settings"
# Check critical settings are present
critical_settings = [
"database_url", "redis_url", "workdir", "debug",
"openai_api_key", "azure_ai_key",
"auth_enabled", "session_secret"
"database_url",
"redis_url",
"workdir",
"debug",
"openai_api_key",
"azure_ai_key",
"auth_enabled",
"session_secret",
]
for setting in critical_settings:
assert setting in SETTING_METADATA, f"Missing metadata for {setting}"
@@ -151,7 +156,7 @@ class TestSettingsService:
@pytest.mark.unit
class TestConfigLoader:
"""Test configuration loader functions"""
def test_convert_boolean_value(self):
"""Test converting string to boolean"""
assert convert_setting_value("true", bool) is True
@@ -160,27 +165,27 @@ class TestConfigLoader:
assert convert_setting_value("0", bool) is False
assert convert_setting_value("yes", bool) is True
assert convert_setting_value("no", bool) is False
def test_convert_integer_value(self):
"""Test converting string to integer"""
assert convert_setting_value("42", int) == 42
assert convert_setting_value("0", int) == 0
assert convert_setting_value("-5", int) == -5
# Invalid integer
assert convert_setting_value("not_a_number", int) == 0
def test_convert_string_value(self):
"""Test converting to string (default)"""
assert convert_setting_value("hello", str) == "hello"
assert convert_setting_value("123", str) == "123"
def test_convert_none_value(self):
"""Test handling None values"""
assert convert_setting_value(None, str) is None
assert convert_setting_value(None, int) is None
assert convert_setting_value(None, bool) is None
def test_convert_list_value(self):
"""Test converting comma-separated string to list"""
assert convert_setting_value("a,b,c", list) == ["a", "b", "c"]
@@ -192,7 +197,7 @@ class TestConfigLoader:
@pytest.mark.requires_db
class TestSettingsAPI:
"""Test settings API endpoints"""
def test_get_settings_requires_admin(self, client: TestClient):
"""Test that settings endpoint requires admin privileges"""
# With AUTH_ENABLED=False in test environment, this test verifies
@@ -202,7 +207,7 @@ class TestSettingsAPI:
# Should return 403 (no admin session) or redirect
# Note: Test environment has AUTH_ENABLED=False
assert response.status_code in [200, 302, 403]
def test_settings_page_structure(self, client: TestClient):
"""Test that settings page has expected structure"""
# Verify the endpoint exists and returns expected status codes
@@ -212,54 +217,55 @@ class TestSettingsAPI:
@pytest.mark.integration
@pytest.mark.requires_db
@pytest.mark.requires_db
class TestSettingsPrecedence:
"""Test settings precedence (DB > env > defaults)"""
def test_db_overrides_default(self, db_session: Session):
"""Test that database settings override default values"""
# Create a minimal test settings object
from pydantic_settings import BaseSettings
from typing import Optional
from pydantic_settings import BaseSettings
class TestSettings(BaseSettings):
test_value: str = "default"
test_bool: bool = False
class Config:
env_file = None
# Create settings with defaults
test_settings = TestSettings()
assert test_settings.test_value == "default"
assert test_settings.test_bool is False
# Save to database
save_setting_to_db(db_session, "test_value", "from_database")
save_setting_to_db(db_session, "test_bool", "true")
# Load from database
load_settings_from_db(test_settings, db_session)
# Verify database values take precedence
assert test_settings.test_value == "from_database"
assert test_settings.test_bool is True
def test_load_settings_handles_missing_db_settings(self, db_session: Session):
"""Test that loading settings works when no DB settings exist"""
from pydantic_settings import BaseSettings
class TestSettings(BaseSettings):
test_value: str = "default"
class Config:
env_file = None
test_settings = TestSettings()
# Load from empty database - should not crash
load_settings_from_db(test_settings, db_session)
# Should still have default value
assert test_settings.test_value == "default"
@@ -267,16 +273,13 @@ class TestSettingsPrecedence:
@pytest.mark.unit
class TestApplicationSettingsModel:
"""Test the ApplicationSettings database model"""
def test_create_setting_record(self, db_session: Session):
"""Test creating an ApplicationSettings record"""
setting = ApplicationSettings(
key="test_key",
value="test_value"
)
setting = ApplicationSettings(key="test_key", value="test_value")
db_session.add(setting)
db_session.commit()
# Retrieve and verify
retrieved = db_session.query(ApplicationSettings).filter_by(key="test_key").first()
assert retrieved is not None
@@ -284,45 +287,44 @@ class TestApplicationSettingsModel:
assert retrieved.value == "test_value"
assert retrieved.created_at is not None
assert retrieved.updated_at is not None
def test_unique_key_constraint(self, db_session: Session):
"""Test that key field has unique constraint"""
# Create first setting
setting1 = ApplicationSettings(key="unique_key", value="value1")
db_session.add(setting1)
db_session.commit()
# Try to create duplicate - should fail
setting2 = ApplicationSettings(key="unique_key", value="value2")
db_session.add(setting2)
with pytest.raises(Exception): # SQLAlchemy will raise an exception
db_session.commit()
@pytest.mark.skipif(
True, # Skip for all databases - timestamp update behavior varies
reason="Timestamp update behavior varies by database backend"
reason="Timestamp update behavior varies by database backend",
)
def test_update_timestamp(self, db_session: Session):
"""Test that updated_at timestamp is updated on modification"""
import time
# Create setting
setting = ApplicationSettings(key="test_key", value="initial")
db_session.add(setting)
db_session.commit()
initial_updated_at = setting.updated_at
# Small delay to ensure timestamp difference
time.sleep(0.1)
# Update setting
setting.value = "updated"
db_session.commit()
# Verify updated_at changed
# Note: SQLite doesn't automatically update onupdate timestamps
# This test is skipped as behavior varies by database backend
assert setting.updated_at is not None
tests/test_settings_display.py
+3 -1
View File
@@ -1,7 +1,9 @@
"""Tests for app/utils/config_validator/settings_display.py module."""
import pytest
from unittest.mock import patch
import pytest
from app.utils.config_validator.settings_display import dump_all_settings, get_settings_for_display
tests/test_setup_wizard.py
+6 -4
View File
@@ -1,12 +1,14 @@
"""Tests for app/utils/setup_wizard.py module."""
import pytest
from unittest.mock import patch
import pytest
from app.utils.setup_wizard import (
get_required_settings,
is_setup_required,
get_missing_required_settings,
get_required_settings,
get_wizard_steps,
is_setup_required,
)
@@ -115,7 +117,7 @@ class TestGetWizardSteps:
all_step_keys = []
for settings_list in steps.values():
all_step_keys.extend([s["key"] for s in settings_list])
required_keys = [s["key"] for s in get_required_settings()]
for key in required_keys:
assert key in all_step_keys, f"Setting {key} not assigned to any wizard step"
tests/test_step_manager.py
+9 -2
View File
@@ -172,7 +172,12 @@ class TestStepManager:
# Update an existing step
now = datetime.now()
update_step_status(
db_session, file_record.id, "create_file_record", "success", started_at=now - timedelta(seconds=5), completed_at=now
db_session,
file_record.id,
"create_file_record",
"success",
started_at=now - timedelta(seconds=5),
completed_at=now,
)
# Verify step was updated
@@ -201,7 +206,9 @@ class TestStepManager:
now = datetime.now()
update_step_status(db_session, file_record.id, "create_file_record", "success", completed_at=now)
update_step_status(db_session, file_record.id, "check_text", "in_progress", started_at=now)
update_step_status(db_session, file_record.id, "extract_text", "failure", error_message="Failed to extract text")
update_step_status(
db_session, file_record.id, "extract_text", "failure", error_message="Failed to extract text"
)
# Get all step statuses
status_map = get_file_step_status(db_session, file_record.id)
tests/test_storage_reorganization.py
+30 -37
View File
@@ -5,8 +5,8 @@ Tests the new functionality for storing immutable originals and processed copies
collision handling, and forced Cloud OCR reprocessing.
"""
import os
import json
import os
from unittest.mock import MagicMock, patch
import pytest
@@ -23,7 +23,7 @@ class TestImmutableOriginalStorage:
def test_new_file_saves_original_copy(self, db_session, tmp_path):
"""Test that a new file creates an immutable original copy"""
from app.tasks.process_document import process_document
# Create test PDF
test_pdf = tmp_path / "test_input.pdf"
pdf_content = b"""%PDF-1.4
@@ -86,7 +86,7 @@ startxref
%%EOF
"""
test_pdf.write_bytes(pdf_content)
# Setup mocks
with (
patch("app.tasks.process_document.SessionLocal") as mock_session_local,
@@ -98,18 +98,18 @@ startxref
mock_session_local.return_value.__enter__.return_value = db_session
mock_session_local.return_value.__exit__.return_value = None
mock_extract.delay = MagicMock()
# Call process_document
result = process_document(str(test_pdf), original_filename="test_input.pdf")
# Verify original directory was created
original_dir = tmp_path / "original"
assert original_dir.exists()
# Verify an original file was saved
original_files = list(original_dir.glob("*.pdf"))
assert len(original_files) > 0
# Verify database record has original_file_path
file_record = db_session.query(FileRecord).first()
assert file_record is not None
@@ -187,7 +187,7 @@ startxref
original_dir.mkdir()
original_file = original_dir / "existing-original.pdf"
original_file.write_bytes(pdf_content)
# Create existing file record
file_record = FileRecord(
filehash="abc123",
@@ -195,11 +195,11 @@ startxref
local_filename=str(test_pdf),
original_file_path=str(original_file),
file_size=100,
mime_type="application/pdf"
mime_type="application/pdf",
)
db_session.add(file_record)
db_session.commit()
# Setup mocks
with (
patch("app.tasks.process_document.SessionLocal") as mock_session_local,
@@ -211,11 +211,11 @@ startxref
mock_session_local.return_value.__enter__.return_value = db_session
mock_session_local.return_value.__exit__.return_value = None
mock_extract.delay = MagicMock()
# Reprocess with file_id
original_count = len(list(original_dir.glob("*.pdf")))
process_document(str(test_pdf), file_id=file_record.id)
# Should not create new original file
new_count = len(list(original_dir.glob("*.pdf")))
assert new_count == original_count
@@ -228,16 +228,16 @@ class TestCollisionHandling:
def test_collision_handling_in_processed_dir(self, tmp_path):
"""Test that collision handling works in processed directory"""
from app.utils.filename_utils import get_unique_filepath_with_counter
processed_dir = tmp_path / "processed"
processed_dir.mkdir()
# Create first file
(processed_dir / "2024-01-01_Invoice.pdf").touch()
# Get unique path for same filename
result = get_unique_filepath_with_counter(str(processed_dir), "2024-01-01_Invoice")
assert "2024-01-01_Invoice-0001.pdf" in result
assert os.path.exists(str(processed_dir / "2024-01-01_Invoice.pdf"))
@@ -249,34 +249,27 @@ class TestMetadataAugmentation:
def test_metadata_includes_file_paths(self, tmp_path):
"""Test that persisted metadata includes original and processed paths"""
from app.tasks.embed_metadata_into_pdf import persist_metadata
metadata = {
"filename": "2024-01-01_Invoice",
"document_type": "Invoice",
"tags": ["finance", "2024"]
}
metadata = {"filename": "2024-01-01_Invoice", "document_type": "Invoice", "tags": ["finance", "2024"]}
processed_file = tmp_path / "processed" / "2024-01-01_Invoice.pdf"
processed_file.parent.mkdir(parents=True)
processed_file.touch()
original_path = "/workdir/original/abc123.pdf"
processed_path = str(processed_file)
json_path = persist_metadata(
metadata,
str(processed_file),
original_file_path=original_path,
processed_file_path=processed_path
metadata, str(processed_file), original_file_path=original_path, processed_file_path=processed_path
)
# Verify JSON was created
assert os.path.exists(json_path)
# Verify content
with open(json_path, 'r') as f:
with open(json_path, "r") as f:
saved_metadata = json.load(f)
assert "original_file_path" in saved_metadata
assert saved_metadata["original_file_path"] == original_path
assert "processed_file_path" in saved_metadata
@@ -292,7 +285,7 @@ class TestForceCloudOCR:
def test_force_cloud_ocr_parameter(self, db_session, tmp_path):
"""Test that force_cloud_ocr parameter skips local text extraction"""
from app.tasks.process_document import process_document
# Create PDF with embedded text
test_pdf = tmp_path / "with_text.pdf"
pdf_content = b"""%PDF-1.4
@@ -355,7 +348,7 @@ startxref
%%EOF
"""
test_pdf.write_bytes(pdf_content)
with (
patch("app.tasks.process_document.SessionLocal") as mock_session_local,
patch("app.tasks.process_document.settings") as mock_settings,
@@ -366,10 +359,10 @@ startxref
mock_session_local.return_value.__enter__.return_value = db_session
mock_session_local.return_value.__exit__.return_value = None
mock_azure.delay = MagicMock()
# Process with force_cloud_ocr=True
result = process_document(str(test_pdf), force_cloud_ocr=True)
# Should queue Azure OCR, not local extraction
mock_azure.delay.assert_called_once()
assert result["status"] == "Queued for forced OCR"
tests/test_upload_email.py
+4 -1
View File
@@ -1,6 +1,8 @@
"""Tests for app/tasks/upload_to_email.py module."""
from unittest.mock import MagicMock, patch
import pytest
from unittest.mock import patch, MagicMock
@pytest.mark.unit
@@ -10,4 +12,5 @@ class TestUploadToEmail:
def test_module_imports(self):
"""Test that the module can be imported."""
from app.tasks.upload_to_email import upload_to_email
assert callable(upload_to_email)
tests/test_upload_ftp_additional.py
+4 -1
View File
@@ -1,6 +1,8 @@
"""Additional tests for upload_to_ftp task."""
from unittest.mock import MagicMock, patch
import pytest
from unittest.mock import patch, MagicMock
@pytest.mark.unit
@@ -10,4 +12,5 @@ class TestUploadToFtp:
def test_module_imports(self):
"""Test that the module can be imported."""
from app.tasks.upload_to_ftp import upload_to_ftp
assert callable(upload_to_ftp)
tests/test_upload_tasks.py
+6 -4
View File
@@ -3,15 +3,17 @@ Tests for upload tasks including OneDrive, S3, FTP, SFTP, WebDAV, Google Drive,
"""
import os
from unittest.mock import MagicMock, Mock, patch
import pytest
from unittest.mock import Mock, patch, MagicMock
from app.tasks.upload_to_email import upload_to_email
from app.tasks.upload_to_ftp import upload_to_ftp
from app.tasks.upload_to_google_drive import upload_to_google_drive
from app.tasks.upload_to_onedrive import upload_to_onedrive
from app.tasks.upload_to_s3 import upload_to_s3
from app.tasks.upload_to_ftp import upload_to_ftp
from app.tasks.upload_to_sftp import upload_to_sftp
from app.tasks.upload_to_webdav import upload_to_webdav
from app.tasks.upload_to_google_drive import upload_to_google_drive
from app.tasks.upload_to_email import upload_to_email
_TEST_CREDENTIAL = "test_pass" # noqa: S105
tests/test_upload_tasks_additional.py
+4 -1
View File
@@ -1,7 +1,9 @@
"""Additional tests for upload task modules."""
import os
from unittest.mock import MagicMock, patch
import pytest
from unittest.mock import patch, MagicMock
@pytest.mark.unit
@@ -23,6 +25,7 @@ class TestUploadToNextcloud:
mock_settings.workdir = "/tmp"
from app.tasks.upload_to_nextcloud import upload_to_nextcloud
result = upload_to_nextcloud.__wrapped__(mock_self, "/nonexistent/file.pdf")
tests/test_upload_tasks_coverage.py
+7 -5
View File
@@ -1,15 +1,17 @@
"""Tests to increase coverage for upload task modules."""
import os
import pytest
from unittest.mock import patch, MagicMock
from app.tasks.upload_to_paperless import upload_to_paperless
import os
from unittest.mock import MagicMock, patch
import pytest
from app.tasks.upload_to_ftp import upload_to_ftp
from app.tasks.upload_to_nextcloud import upload_to_nextcloud
from app.tasks.upload_to_onedrive import upload_to_onedrive
from app.tasks.upload_to_paperless import upload_to_paperless
from app.tasks.upload_to_s3 import upload_to_s3
from app.tasks.upload_to_sftp import upload_to_sftp
from app.tasks.upload_to_webdav import upload_to_webdav
from app.tasks.upload_to_ftp import upload_to_ftp
@pytest.mark.unit
tests/test_upload_webdav_comprehensive.py
+3 -2
View File
@@ -1,9 +1,10 @@
"""Comprehensive tests for upload_to_webdav task."""
import os
from unittest.mock import MagicMock, Mock, patch
import pytest
from unittest.mock import patch, Mock, MagicMock
from requests.exceptions import ConnectionError, Timeout, RequestException
from requests.exceptions import ConnectionError, RequestException, Timeout
from app.tasks.upload_to_webdav import upload_to_webdav
tests/test_upload_webdav_integration.py
+3 -2
View File
@@ -7,11 +7,12 @@ actual file uploads against it, then verify the files were uploaded successfully
import os
import time
import pytest
import requests
from pathlib import Path
from unittest.mock import patch
import pytest
import requests
from app.tasks.upload_to_webdav import upload_to_webdav
# Import testcontainers - required for these tests
tests/test_uptime_kuma.py
+4 -7
View File
@@ -4,8 +4,9 @@ Tests for app/tasks/uptime_kuma_tasks.py
Tests Uptime Kuma health check ping functionality.
"""
from unittest.mock import MagicMock, Mock, patch
import pytest
from unittest.mock import Mock, patch, MagicMock
import requests
@@ -43,9 +44,7 @@ class TestUptimeKumaTasks:
# Should return True on success
assert result is True
mock_get.assert_called_once_with(
"https://uptime.example.com/ping/123", timeout=10
)
mock_get.assert_called_once_with("https://uptime.example.com/ping/123", timeout=10)
mock_response.raise_for_status.assert_called_once()
@patch("app.tasks.uptime_kuma_tasks.requests.get")
@@ -93,9 +92,7 @@ class TestUptimeKumaTasks:
# Mock HTTP error
mock_response = Mock()
mock_response.status_code = 500
mock_response.raise_for_status.side_effect = requests.exceptions.HTTPError(
"500 Server Error"
)
mock_response.raise_for_status.side_effect = requests.exceptions.HTTPError("500 Server Error")
mock_get.return_value = mock_response
result = ping_uptime_kuma()
tests/test_url_upload.py
+9 -16
View File
@@ -29,9 +29,10 @@ class TestURLUploadValidation:
def test_validate_url_scheme_ftp_rejected(self):
"""Test that FTP URLs are rejected"""
from app.api.url_upload import URLUploadRequest
from pydantic import ValidationError
from app.api.url_upload import URLUploadRequest
with pytest.raises(ValidationError) as exc_info:
URLUploadRequest(url="ftp://example.com/file.pdf")
# Pydantic HttpUrl validates scheme automatically
@@ -39,9 +40,10 @@ class TestURLUploadValidation:
def test_validate_url_scheme_file_rejected(self):
"""Test that file:// URLs are rejected"""
from app.api.url_upload import URLUploadRequest
from pydantic import ValidationError
from app.api.url_upload import URLUploadRequest
with pytest.raises(ValidationError) as exc_info:
URLUploadRequest(url="file:///etc/passwd")
# Pydantic HttpUrl validates scheme automatically
@@ -126,17 +128,14 @@ class TestURLUploadValidation:
# Word
assert validate_file_type("application/msword", "file.doc") is True
assert (
validate_file_type(
"application/vnd.openxmlformats-officedocument.wordprocessingml.document", "file.docx"
)
validate_file_type("application/vnd.openxmlformats-officedocument.wordprocessingml.document", "file.docx")
is True
)
# Excel
assert validate_file_type("application/vnd.ms-excel", "file.xls") is True
assert (
validate_file_type("application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", "file.xlsx")
is True
validate_file_type("application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", "file.xlsx") is True
)
def test_validate_file_type_images_allowed(self):
@@ -248,9 +247,7 @@ class TestURLUploadEndpoint:
@patch("app.api.url_upload.requests.get")
def test_process_url_blocks_metadata_endpoint(self, mock_requests_get, client):
"""Test that cloud metadata endpoints are blocked"""
response = client.post(
"/api/process-url", json={"url": "http://169.254.169.254/latest/meta-data/"}
)
response = client.post("/api/process-url", json={"url": "http://169.254.169.254/latest/meta-data/"})
assert response.status_code == 400
data = response.json()
@@ -341,9 +338,7 @@ class TestURLUploadEndpoint:
@patch("app.api.url_upload.requests.get")
@patch("app.api.url_upload.process_document")
def test_process_url_with_custom_filename(
self, mock_process_document, mock_requests_get, client, tmp_path
):
def test_process_url_with_custom_filename(self, mock_process_document, mock_requests_get, client, tmp_path):
"""Test URL upload with custom filename"""
# Mock successful download
mock_response = Mock()
@@ -369,9 +364,7 @@ class TestURLUploadEndpoint:
@patch("app.api.url_upload.requests.get")
@patch("app.api.url_upload.process_document")
def test_process_url_extracts_filename_from_url(
self, mock_process_document, mock_requests_get, client, tmp_path
):
def test_process_url_extracts_filename_from_url(self, mock_process_document, mock_requests_get, client, tmp_path):
"""Test that filename is extracted from URL when not provided"""
# Mock successful download
mock_response = Mock()
tests/test_views_coverage.py
+2 -1
View File
@@ -1,7 +1,8 @@
"""Additional view tests to increase coverage."""
from unittest.mock import MagicMock, patch
import pytest
from unittest.mock import patch, MagicMock
_TEST_CREDENTIAL = "test" # noqa: S105
tests/test_views_dropbox.py
+1
View File
@@ -1,4 +1,5 @@
"""Tests for app/views/dropbox.py module."""
import pytest
tests/test_views_general.py
+4 -2
View File
@@ -1,7 +1,9 @@
"""Tests for app/views/general.py module."""
import pytest
from unittest.mock import patch, MagicMock
from pathlib import Path
from unittest.mock import MagicMock, patch
import pytest
@pytest.mark.integration
tests/test_views_google_drive.py
+1
View File
@@ -1,4 +1,5 @@
"""Tests for app/views/google_drive.py module."""
import pytest
tests/test_views_license.py
+1
View File
@@ -1,4 +1,5 @@
"""Tests for app/views/license_routes.py module."""
import pytest
tests/test_views_onedrive.py
+1
View File
@@ -1,4 +1,5 @@
"""Tests for app/views/onedrive.py module."""
import pytest
tests/test_views_settings.py
+6 -1
View File
@@ -1,6 +1,8 @@
"""Tests for app/views/settings.py module."""
from unittest.mock import MagicMock, patch
import pytest
from unittest.mock import patch, MagicMock
from app.views.settings import require_admin_access
@@ -12,6 +14,7 @@ class TestRequireAdminAccess:
@pytest.mark.asyncio
async def test_redirects_non_admin_user(self):
"""Test that non-admin users are redirected."""
@require_admin_access
async def dummy_route(request):
return {"success": True}
@@ -25,6 +28,7 @@ class TestRequireAdminAccess:
@pytest.mark.asyncio
async def test_redirects_when_no_user(self):
"""Test that unauthenticated users are redirected."""
@require_admin_access
async def dummy_route(request):
return {"success": True}
@@ -38,6 +42,7 @@ class TestRequireAdminAccess:
@pytest.mark.asyncio
async def test_allows_admin_user(self):
"""Test that admin users can access the route."""
@require_admin_access
async def dummy_route(request):
return {"success": True}
tests/test_views_status.py
+1
View File
@@ -1,4 +1,5 @@
"""Tests for app/views/status.py module."""
import pytest
tests/test_views_wizard.py
+1
View File
@@ -1,4 +1,5 @@
"""Tests for app/views/wizard.py module."""
import pytest