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

Merge pull request #239 from christianlouis/copilot/update-tests-external-apis

Browse Source 
Fix DeepSource issues: lazy logging, static methods, unused vars, reimports
This commit is contained in:
Christian Krakau-Louis
2026-02-11 18:56:48 +01:00
committed by GitHub
parent ef97746a5f bf0151340d
commit bb1370d46e
21 changed files with 1718 additions and 792 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/api/diagnostic.py
+4 -1
View File
@@ -3,6 +3,7 @@ Diagnostic API endpoints
"""
import logging
from typing import Annotated
from fastapi import APIRouter, Depends, Request
@@ -14,10 +15,12 @@ logger = logging.getLogger(__name__)
router = APIRouter()
CurrentUser = Annotated[dict, Depends(get_current_user)]
@router.get("/diagnostic/settings")
@require_login
async def diagnostic_settings(request: Request, current_user: dict = Depends(get_current_user)):
async def diagnostic_settings(request: Request, current_user: CurrentUser):
"""
API endpoint to dump settings to the log and view basic config information
This endpoint doesn't expose sensitive information like passwords or tokens
app/api/dropbox.py
+14 -13
View File
@@ -4,6 +4,7 @@ Dropbox API endpoints
import logging
import os
from typing import Annotated, Optional
import requests
from fastapi import APIRouter, Form, HTTPException, Request, status
@@ -22,11 +23,11 @@ router = APIRouter()
@require_login
async def exchange_dropbox_token(
request: Request,
client_id: str = Form(...),
client_secret: str = Form(...),
redirect_uri: str = Form(...),
code: str = Form(...),
folder_path: str = Form(None),
client_id: Annotated[str, Form(...)],
client_secret: Annotated[str, Form(...)],
redirect_uri: Annotated[str, Form(...)],
code: Annotated[str, Form(...)],
folder_path: Annotated[Optional[str], Form()] = None,
):
"""
Exchange an authorization code for a refresh token from Dropbox.
@@ -58,10 +59,10 @@ async def exchange_dropbox_token(
@require_login
async def update_dropbox_settings(
request: Request,
app_key: str = Form(None),
app_secret: str = Form(None),
refresh_token: str = Form(...),
folder_path: str = Form(None),
refresh_token: Annotated[str, Form(...)],
app_key: Annotated[Optional[str], Form()] = None,
app_secret: Annotated[Optional[str], Form()] = None,
folder_path: Annotated[Optional[str], Form()] = None,
):
"""
Update Dropbox settings in memory
@@ -182,10 +183,10 @@ async def test_dropbox_token(request: Request):
@require_login
async def save_dropbox_settings(
request: Request,
app_key: str = Form(None),
app_secret: str = Form(None),
refresh_token: str = Form(...),
folder_path: str = Form(None),
refresh_token: Annotated[str, Form(...)],
app_key: Annotated[Optional[str], Form()] = None,
app_secret: Annotated[Optional[str], Form()] = None,
folder_path: Annotated[Optional[str], Form()] = None,
):
"""
Save Dropbox settings to the .env file
app/api/files.py
+12 -10
View File
@@ -6,7 +6,7 @@ import logging
import mimetypes
import os
import uuid
from typing import List, Optional
from typing import Annotated, List, Optional
from fastapi import APIRouter, Depends, File, HTTPException, Query, Request, UploadFile
from sqlalchemy import asc, desc
@@ -27,6 +27,8 @@ logger = logging.getLogger(__name__)
router = APIRouter()
DbSession = Annotated[Session, Depends(get_db)]
def get_limiter():
"""Get the limiter from app state."""
@@ -39,7 +41,7 @@ def get_limiter():
@require_login
def list_files_api(
request: Request,
db: Session = Depends(get_db),
db: DbSession,
page: int = Query(1, ge=1, description="Page number"),
per_page: int = Query(50, ge=1, le=200, description="Items per page"),
sort_by: str = Query(
@@ -152,7 +154,7 @@ def _get_file_processing_status(db: Session, file_id: int) -> dict:
@router.get("/files/{file_id}")
@require_login
def get_file_details(request: Request, file_id: int, db: Session = Depends(get_db)):
def get_file_details(request: Request, file_id: int, db: DbSession):
"""
Get detailed information about a specific file including processing history.
"""
@@ -205,7 +207,7 @@ def get_file_details(request: Request, file_id: int, db: Session = Depends(get_d
@router.delete("/files/{file_id}")
@require_login
def delete_file_record(request: Request, file_id: int, db: Session = Depends(get_db)):
def delete_file_record(request: Request, file_id: int, db: DbSession):
"""
Delete a file record from the database.
This only removes the database entry, not the actual file.
@@ -240,7 +242,7 @@ def delete_file_record(request: Request, file_id: int, db: Session = Depends(get
@router.post("/files/bulk-delete")
@require_login
def bulk_delete_files(request: Request, file_ids: List[int], db: Session = Depends(get_db)):
def bulk_delete_files(request: Request, file_ids: List[int], db: DbSession):
"""
Delete multiple file records from the database.
This only removes the database entries, not the actual files.
@@ -284,7 +286,7 @@ def bulk_delete_files(request: Request, file_ids: List[int], db: Session = Depen
@router.post("/files/bulk-reprocess")
@require_login
def bulk_reprocess_files(request: Request, file_ids: List[int], db: Session = Depends(get_db)):
def bulk_reprocess_files(request: Request, file_ids: List[int], db: DbSession):
"""
Reprocess multiple files by queuing them for processing.
"""
@@ -345,7 +347,7 @@ def bulk_reprocess_files(request: Request, file_ids: List[int], db: Session = De
@router.post("/files/{file_id}/reprocess")
@require_login
def reprocess_single_file(request: Request, file_id: int, db: Session = Depends(get_db)):
def reprocess_single_file(request: Request, file_id: int, db: DbSession):
"""
Reprocess a single file by queuing it for processing again.
@@ -487,10 +489,10 @@ def _retry_pipeline_step(file_record: FileRecord, step_name: str, db: Session) -
def retry_subtask(
request: Request,
file_id: int,
db: DbSession,
subtask_name: str = Query(
..., description="Name of the subtask to retry (e.g., 'upload_to_dropbox', 'extract_metadata_with_gpt')"
),
db: Session = Depends(get_db),
):
"""
Retry a specific failed subtask for a file.
@@ -603,8 +605,8 @@ def retry_subtask(
def get_file_preview(
request: Request,
file_id: int,
db: DbSession,
version: str = Query("original", description="original or processed"),
db: Session = Depends(get_db),
):
"""
Get file content for preview (original or processed version).
@@ -675,8 +677,8 @@ def get_file_preview(
def download_file(
request: Request,
file_id: int,
db: DbSession,
version: str = Query("original", description="original or processed"),
db: Session = Depends(get_db),
):
"""
Download file (original or processed version) as attachment.
app/api/google_drive.py
+16 -16
View File
@@ -5,7 +5,7 @@ Google Drive API endpoints
import logging
import os
from datetime import datetime
from typing import Optional
from typing import Annotated, Optional
from fastapi import APIRouter, Form, HTTPException, Request, status
@@ -23,11 +23,11 @@ router = APIRouter()
@require_login
async def exchange_google_drive_token(
request: Request,
client_id: str = Form(...),
client_secret: str = Form(...),
redirect_uri: str = Form(...),
code: str = Form(...),
folder_id: Optional[str] = Form(None),
client_id: Annotated[str, Form(...)],
client_secret: Annotated[str, Form(...)],
redirect_uri: Annotated[str, Form(...)],
code: Annotated[str, Form(...)],
folder_id: Annotated[Optional[str], Form()] = None,
):
"""
Exchange an authorization code for refresh and access tokens from Google.
@@ -59,11 +59,11 @@ async def exchange_google_drive_token(
@require_login
async def update_google_drive_settings(
request: Request,
client_id: str = Form(None),
client_secret: str = Form(None),
refresh_token: str = Form(...),
folder_id: str = Form(None),
use_oauth: str = Form("true"),
refresh_token: Annotated[str, Form(...)],
client_id: Annotated[Optional[str], Form()] = None,
client_secret: Annotated[Optional[str], Form()] = None,
folder_id: Annotated[Optional[str], Form()] = None,
use_oauth: Annotated[str, Form()] = "true",
):
"""
Update Google Drive settings in memory
@@ -328,11 +328,11 @@ def format_time_remaining(time_delta):
@require_login
async def save_dropbox_settings(
request: Request,
client_id: str = Form(None),
client_secret: str = Form(None),
refresh_token: str = Form(...),
folder_id: str = Form(None),
use_oauth: str = Form("true"),
refresh_token: Annotated[str, Form(...)],
client_id: Annotated[Optional[str], Form()] = None,
client_secret: Annotated[Optional[str], Form()] = None,
folder_id: Annotated[Optional[str], Form()] = None,
use_oauth: Annotated[str, Form()] = "true",
):
"""
Save Google Drive settings to the .env file
app/api/logs.py
+6 -4
View File
@@ -3,7 +3,7 @@ Processing logs API endpoints
"""
import logging
from typing import Optional
from typing import Annotated, Optional
from fastapi import APIRouter, Depends, HTTPException, Query, Request
from sqlalchemy import desc
@@ -18,12 +18,14 @@ logger = logging.getLogger(__name__)
router = APIRouter()
DbSession = Annotated[Session, Depends(get_db)]
@router.get("/logs")
@require_login
def list_processing_logs(
request: Request,
db: Session = Depends(get_db),
db: DbSession,
file_id: Optional[int] = Query(None, description="Filter by file ID"),
task_id: Optional[str] = Query(None, description="Filter by task ID"),
limit: int = Query(100, ge=1, le=1000, description="Number of logs to return"),
@@ -81,7 +83,7 @@ def list_processing_logs(
@router.get("/logs/file/{file_id}")
@require_login
def get_file_processing_logs(request: Request, file_id: int, db: Session = Depends(get_db)):
def get_file_processing_logs(request: Request, file_id: int, db: DbSession):
"""
Get all processing logs for a specific file.
Returns logs ordered by timestamp (oldest first to show processing flow).
@@ -125,7 +127,7 @@ def get_file_processing_logs(request: Request, file_id: int, db: Session = Depen
@router.get("/logs/task/{task_id}")
@require_login
def get_task_processing_logs(request: Request, task_id: str, db: Session = Depends(get_db)):
def get_task_processing_logs(request: Request, task_id: str, db: DbSession):
"""
Get all processing logs for a specific task.
Returns logs ordered by timestamp (oldest first to show processing flow).
app/api/onedrive.py
+16 -15
View File
@@ -5,6 +5,7 @@ OneDrive API endpoints
import logging
import os
from datetime import datetime, timedelta
from typing import Annotated, Optional
import requests
from fastapi import APIRouter, Form, HTTPException, Request, status
@@ -23,11 +24,11 @@ router = APIRouter()
@require_login
async def exchange_onedrive_token(
request: Request,
client_id: str = Form(...),
client_secret: str = Form(...),
redirect_uri: str = Form(...),
code: str = Form(...),
tenant_id: str = Form(...),
client_id: Annotated[str, Form(...)],
client_secret: Annotated[str, Form(...)],
redirect_uri: Annotated[str, Form(...)],
code: Annotated[str, Form(...)],
tenant_id: Annotated[str, Form(...)],
):
"""
Exchange an authorization code for a refresh token.
@@ -197,11 +198,11 @@ def format_time_remaining(time_delta):
@require_login
async def save_onedrive_settings(
request: Request,
client_id: str = Form(None),
client_secret: str = Form(None),
refresh_token: str = Form(...),
tenant_id: str = Form("common"),
folder_path: str = Form(None),
refresh_token: Annotated[str, Form(...)],
client_id: Annotated[Optional[str], Form()] = None,
client_secret: Annotated[Optional[str], Form()] = None,
tenant_id: Annotated[str, Form()] = "common",
folder_path: Annotated[Optional[str], Form()] = None,
):
"""
Save OneDrive settings to the .env file
@@ -292,11 +293,11 @@ async def save_onedrive_settings(
@require_login
async def update_onedrive_settings(
request: Request,
client_id: str = Form(None),
client_secret: str = Form(None),
refresh_token: str = Form(...),
tenant_id: str = Form("common"),
folder_path: str = Form(None),
refresh_token: Annotated[str, Form(...)],
client_id: Annotated[Optional[str], Form()] = None,
client_secret: Annotated[Optional[str], Form()] = None,
tenant_id: Annotated[str, Form()] = "common",
folder_path: Annotated[Optional[str], Form()] = None,
):
"""
Update OneDrive settings in memory (without modifying .env file)
app/api/settings.py
+11 -11
View File
@@ -3,7 +3,7 @@ API endpoints for managing application settings.
"""
import logging
from typing import Any, Dict, Optional
from typing import Annotated, Any, Dict, Optional
from fastapi import APIRouter, Depends, HTTPException, Request, status
from pydantic import BaseModel, Field
@@ -39,6 +39,10 @@ def require_admin(request: Request) -> dict:
return user
DbSession = Annotated[Session, Depends(get_db)]
AdminUser = Annotated[dict, Depends(require_admin)]
class SettingUpdate(BaseModel):
"""Model for updating a setting"""
@@ -63,7 +67,7 @@ class SettingsListResponse(BaseModel):
@router.get("/", response_model=SettingsListResponse)
async def get_settings(request: Request, db: Session = Depends(get_db), admin: dict = Depends(require_admin)):
async def get_settings(request: Request, db: DbSession, admin: AdminUser):
"""
Get all application settings with metadata.
Admin only.
@@ -89,7 +93,7 @@ async def get_settings(request: Request, db: Session = Depends(get_db), admin: d
@router.get("/{key}", response_model=SettingResponse)
async def get_setting(key: str, request: Request, db: Session = Depends(get_db), admin: dict = Depends(require_admin)):
async def get_setting(key: str, request: Request, db: DbSession, admin: AdminUser):
"""
Get a specific setting by key.
Admin only.
@@ -114,8 +118,8 @@ async def update_setting(
key: str,
setting: SettingUpdate,
request: Request,
db: Session = Depends(get_db),
admin: dict = Depends(require_admin),
db: DbSession,
admin: AdminUser,
):
"""
Update a specific setting.
@@ -156,9 +160,7 @@ async def update_setting(
@router.delete("/{key}")
async def delete_setting(
key: str, request: Request, db: Session = Depends(get_db), admin: dict = Depends(require_admin)
):
async def delete_setting(key: str, request: Request, db: DbSession, admin: AdminUser):
"""
Delete a setting from the database (reverts to environment variable or default).
Admin only.
@@ -182,9 +184,7 @@ async def delete_setting(
@router.post("/bulk-update")
async def bulk_update_settings(
updates: list[SettingUpdate], request: Request, db: Session = Depends(get_db), admin: dict = Depends(require_admin)
):
async def bulk_update_settings(updates: list[SettingUpdate], request: Request, db: DbSession, admin: AdminUser):
"""
Update multiple settings at once.
Admin only.
requirements-dev.txt
+1
View File
@@ -8,6 +8,7 @@ pytest-asyncio>=0.23.0
pytest-mock>=3.12.0
httpx>=0.26.0 # For async test client
testcontainers>=3.7.1 # For integration tests with real containers
fpdf2>=2.8.0 # For generating test PDF documents in integration tests
minio>=7.1.0 # For MinIO/S3 integration tests
redis>=4.5.0 # For Redis integration tests
boto3>=1.26.0 # For S3 integration tests
tests/conftest.py
+52 -1
View File
@@ -4,7 +4,7 @@ Pytest configuration and shared fixtures for DocuElevate tests.
import os
import tempfile
from typing import Generator
from typing import Dict, Generator, Optional
import pytest
from fastapi.testclient import TestClient
@@ -12,6 +12,37 @@ from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.pool import StaticPool
# Capture original environment variables before overriding with test defaults.
# This allows integration tests to detect when real API credentials are available
# (e.g., injected via GitHub Actions secrets) and run live API verification.
_EXTERNAL_API_ENV_KEYS = [
"OPENAI_API_KEY",
"OPENAI_BASE_URL",
"AZURE_AI_KEY",
"AZURE_ENDPOINT",
"AZURE_REGION",
"AWS_ACCESS_KEY_ID",
"AWS_SECRET_ACCESS_KEY",
"S3_BUCKET_NAME",
"S3_FOLDER_PREFIX",
"DROPBOX_APP_KEY",
"DROPBOX_APP_SECRET",
"DROPBOX_REFRESH_TOKEN",
"ONEDRIVE_CLIENT_ID",
"ONEDRIVE_CLIENT_SECRET",
"ONEDRIVE_REFRESH_TOKEN",
"ONEDRIVE_TENANT_ID",
"ONEDRIVE_FOLDER_PATH",
"GOOGLE_DRIVE_CREDENTIALS_JSON",
"GOOGLE_DRIVE_FOLDER_ID",
"AUTHENTIK_CLIENT_ID",
"AUTHENTIK_CLIENT_SECRET",
"AUTHENTIK_CONFIG_URL",
"SESSION_SECRET",
]
_PLACEHOLDER_VALUES = {"test-key", "test", "", "NOT_SET"}
_original_env: Dict[str, Optional[str]] = {key: os.environ.get(key) for key in _EXTERNAL_API_ENV_KEYS}
# Set test environment variables before importing app
os.environ["DATABASE_URL"] = "sqlite:///:memory:"
os.environ["REDIS_URL"] = "redis://localhost:6379/1"
@@ -167,6 +198,26 @@ def mock_azure_response():
return {"analyzeResult": {"content": "Test document content extracted by OCR", "pages": [{"pageNumber": 1}]}}
def has_real_env(*keys: str) -> bool:
"""Check if real (non-placeholder) environment variables were set before test overrides.
Returns True only if ALL specified keys had non-placeholder values in the
original environment. Used by integration tests to decide whether to skip
when real credentials are unavailable.
"""
for key in keys:
value = _original_env.get(key)
if value is None or value in _PLACEHOLDER_VALUES:
return False
return True
@pytest.fixture(scope="session")
def original_env() -> Dict[str, Optional[str]]:
"""Provide access to the original environment variables captured before test overrides."""
return dict(_original_env)
# Markers for categorizing tests
def pytest_configure(config):
"""Configure custom pytest markers."""
tests/fixtures_integration.py
+51 -50
View File
@@ -11,6 +11,7 @@ This module provides fixtures for spinning up real infrastructure components:
These tests exercise the full application stack end-to-end.
"""
import os
import time
import pytest
@@ -24,21 +25,23 @@ from testcontainers.postgres import PostgresContainer
from testcontainers.redis import RedisContainer
from testcontainers.minio import MinioContainer
_TEST_CREDENTIAL = "testpass" # noqa: S105
@pytest.fixture(scope="session")
def postgres_container() -> Generator:
"""
Start a real PostgreSQL database container for testing.
This replaces the in-memory SQLite used in unit tests.
"""
with PostgresContainer("postgres:15-alpine") as postgres:
# Wait for PostgreSQL to be ready
time.sleep(2)
# Set environment variable for the app to use
os.environ["DATABASE_URL"] = postgres.get_connection_url()
yield {
"container": postgres,
"url": postgres.get_connection_url(),
@@ -54,23 +57,23 @@ def postgres_container() -> Generator:
def redis_container() -> Generator:
"""
Start a real Redis container for Celery broker/backend.
This provides actual message queueing and task result storage.
"""
with RedisContainer("redis:7-alpine") as redis:
# Wait for Redis to be ready
time.sleep(2)
# Build Redis URL manually
host = redis.get_container_host_ip()
port = redis.get_exposed_port(6379)
redis_url = f"redis://{host}:{port}/0"
# Set environment variables for the app
os.environ["REDIS_URL"] = redis_url
os.environ["CELERY_BROKER_URL"] = redis_url
os.environ["CELERY_RESULT_BACKEND"] = redis_url
yield {
"container": redis,
"url": redis_url,
@@ -83,32 +86,30 @@ def redis_container() -> Generator:
def gotenberg_container() -> Generator:
"""
Start a real Gotenberg container for PDF conversion.
This provides actual document conversion capabilities.
"""
container = DockerContainer("gotenberg/gotenberg:8")
container.with_exposed_ports(3000)
container.with_command(
"gotenberg --chromium-disable-javascript=false --chromium-allow-list=file:///.*"
)
container.with_command("gotenberg --chromium-disable-javascript=false --chromium-allow-list=file:///.*")
container.start()
time.sleep(5) # Gotenberg takes a bit longer to start
host = container.get_container_host_ip()
port = container.get_exposed_port(3000)
gotenberg_url = f"http://{host}:{port}"
# Set environment variable
os.environ["GOTENBERG_URL"] = gotenberg_url
yield {
"container": container,
"url": gotenberg_url,
"host": host,
"port": port,
}
container.stop()
@@ -121,23 +122,23 @@ def webdav_container() -> Generator:
container.with_exposed_ports(80)
container.with_env("AUTH_TYPE", "Basic")
container.with_env("USERNAME", "testuser")
container.with_env("PASSWORD", "testpass")
container.with_env("PASSWORD", _TEST_CREDENTIAL)
container.start()
time.sleep(2)
host = container.get_container_host_ip()
port = container.get_exposed_port(80)
yield {
"container": container,
"url": f"http://{host}:{port}",
"host": host,
"port": port,
"username": "testuser",
"password": "testpass",
"password": _TEST_CREDENTIAL,
}
container.stop()
@@ -145,29 +146,29 @@ def webdav_container() -> Generator:
def sftp_container() -> Generator:
"""
Start a real SFTP server for upload testing.
Uses atmoz/sftp which provides a simple SSH/SFTP server.
"""
container = DockerContainer("atmoz/sftp:latest")
container.with_exposed_ports(22)
# Create user: username:password:uid:gid:directory
container.with_command("testuser:testpass:1001:1001:upload")
container.with_command(f"testuser:{_TEST_CREDENTIAL}:1001:1001:upload")
container.start()
time.sleep(3) # SFTP server needs time to initialize
host = container.get_container_host_ip()
port = container.get_exposed_port(22)
yield {
"container": container,
"host": host,
"port": port,
"username": "testuser",
"password": "testpass",
"password": _TEST_CREDENTIAL,
"folder": "/home/testuser/upload",
}
container.stop()
@@ -175,16 +176,16 @@ def sftp_container() -> Generator:
def minio_container() -> Generator:
"""
Start a real MinIO container (S3-compatible storage).
MinIO provides S3-compatible API for testing S3 uploads.
"""
with MinioContainer() as minio:
time.sleep(2)
# MinIO uses random credentials, get them
access_key = minio.access_key
secret_key = minio.secret_key
yield {
"container": minio,
"url": minio.get_config()["endpoint"],
@@ -198,31 +199,31 @@ def minio_container() -> Generator:
def ftp_container() -> Generator:
"""
Start a real FTP server for upload testing.
Uses stilliard/pure-ftpd which provides a simple FTP server.
"""
container = DockerContainer("stilliard/pure-ftpd:latest")
container.with_exposed_ports(21, 30000, 30001, 30002, 30003, 30004)
container.with_env("PUBLICHOST", "localhost")
container.with_env("FTP_USER_NAME", "testuser")
container.with_env("FTP_USER_PASS", "testpass")
container.with_env("FTP_USER_PASS", _TEST_CREDENTIAL)
container.with_env("FTP_USER_HOME", "/home/testuser")
container.start()
time.sleep(3)
host = container.get_container_host_ip()
port = container.get_exposed_port(21)
yield {
"container": container,
"host": host,
"port": port,
"username": "testuser",
"password": "testpass",
"password": _TEST_CREDENTIAL,
"folder": "/",
}
container.stop()
@@ -237,7 +238,7 @@ def full_infrastructure(
):
"""
Combined fixture that provides all infrastructure components.
Use this fixture when you need the complete application stack.
"""
return {
@@ -254,11 +255,11 @@ def full_infrastructure(
def celery_app(redis_container):
"""
Create a Celery app configured to use the real Redis container.
This allows testing actual task queueing and execution.
"""
from app.celery_app import celery
# Update Celery configuration to use test Redis
celery.conf.update(
broker_url=redis_container["url"],
@@ -267,7 +268,7 @@ def celery_app(redis_container):
task_eager_propagates=True,
result_expires=3600,
)
return celery
@@ -275,11 +276,11 @@ def celery_app(redis_container):
def celery_worker(celery_app, redis_container):
"""
Start a real Celery worker for processing tasks.
This runs tasks asynchronously like in production.
"""
from celery.contrib.testing.worker import start_worker
# Start worker in test mode
with start_worker(
celery_app,
@@ -294,23 +295,23 @@ def celery_worker(celery_app, redis_container):
def db_session_real(postgres_container):
"""
Create a database session using the real PostgreSQL container.
This replaces the in-memory SQLite session for integration tests.
"""
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from app.database import Base
# Create engine using PostgreSQL container
engine = create_engine(postgres_container["url"])
# Create all tables
Base.metadata.create_all(bind=engine)
# Create session
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
session = SessionLocal()
try:
yield session
finally:
tests/test_auth.py
+3 -1
View File
@@ -9,6 +9,8 @@ from starlette.responses import RedirectResponse
from app.auth import get_current_user, get_gravatar_url, require_login
_TEST_CREDENTIAL = "test" # noqa: S105
@pytest.mark.unit
class TestGetCurrentUser:
@@ -217,7 +219,7 @@ class TestAuthEndpoints:
def test_auth_post_not_available_when_auth_disabled(self, client):
"""Test that POST /auth returns 404 when auth is disabled."""
response = client.post("/auth", data={"username": "admin", "password": "test"})
response = client.post("/auth", data={"username": "admin", "password": _TEST_CREDENTIAL})
assert response.status_code == 404
tests/test_e2e_full_stack.py
+137 -133
View File
@@ -4,6 +4,7 @@ End-to-end integration tests using real infrastructure.
These tests spin up actual services (PostgreSQL, Redis, Gotenberg, WebDAV, SFTP, MinIO)
and test the complete application workflow from API request to file upload.
"""
import os
import time
import pytest
@@ -13,6 +14,13 @@ from unittest.mock import patch
# Import testcontainers requirement
pytest.importorskip("testcontainers", reason="testcontainers not installed")
try:
import psycopg2 # noqa: F401
_has_psycopg2 = True
except ModuleNotFoundError:
_has_psycopg2 = False
from tests.fixtures_integration import (
postgres_container,
redis_container,
@@ -26,6 +34,8 @@ from tests.fixtures_integration import (
db_session_real,
)
_TEST_CREDENTIAL = "pass" # noqa: S105
@pytest.mark.integration
@pytest.mark.requires_docker
@@ -33,7 +43,7 @@ from tests.fixtures_integration import (
class TestEndToEndWithRedis:
"""
End-to-end tests with real Redis and Celery workers.
These tests verify the complete task queueing and execution workflow.
"""
@@ -47,14 +57,16 @@ class TestEndToEndWithRedis:
):
"""
Test complete workflow: Queue task in Redis → Celery worker executes → Upload to WebDAV.
This is the closest to production - actual message queueing and async execution.
"""
from app.tasks.upload_to_webdav import upload_to_webdav
with patch("app.tasks.upload_to_webdav.settings") as mock_settings, \
patch("app.tasks.upload_to_webdav.log_task_progress"):
with (
patch("app.tasks.upload_to_webdav.settings") as mock_settings,
patch("app.tasks.upload_to_webdav.log_task_progress"),
):
# Configure to use real WebDAV server
mock_settings.webdav_url = webdav_container["url"] + "/"
mock_settings.webdav_username = webdav_container["username"]
@@ -62,10 +74,10 @@ class TestEndToEndWithRedis:
mock_settings.webdav_folder = ""
mock_settings.webdav_verify_ssl = False
mock_settings.http_request_timeout = 30
# Queue the task (it goes to Redis)
result = upload_to_webdav.delay(sample_text_file, file_id=1)
# Wait for task to complete (worker picks it up from Redis)
timeout = 30
start_time = time.time()
@@ -73,26 +85,24 @@ class TestEndToEndWithRedis:
if time.time() - start_time > timeout:
pytest.fail(f"Task did not complete within {timeout} seconds")
time.sleep(0.5)
# Get the result
task_result = result.get(timeout=10)
# Verify task completed successfully
assert task_result["status"] == "Completed"
assert task_result["file"] == sample_text_file
# Verify file was actually uploaded to WebDAV server
filename = os.path.basename(sample_text_file)
file_url = f"{webdav_container['url']}/{filename}"
response = requests.get(
file_url,
auth=(webdav_container["username"], webdav_container["password"]),
timeout=5
file_url, auth=(webdav_container["username"], webdav_container["password"]), timeout=5
)
assert response.status_code == 200
# Verify content matches
with open(sample_text_file, "rb") as f:
assert response.content == f.read()
@@ -104,33 +114,30 @@ class TestEndToEndWithRedis:
):
"""
Test that tasks are properly queued in Redis.
This verifies the Redis broker is working correctly.
"""
from app.tasks.upload_to_webdav import upload_to_webdav
import redis
# Connect to Redis directly
r = redis.from_url(redis_container["url"])
# Check Redis is accessible
assert r.ping()
# Get current queue length
initial_queue_length = r.llen("celery")
# Queue a task (don't execute, just verify queueing)
with patch("app.tasks.upload_to_webdav.settings") as mock_settings:
mock_settings.webdav_url = "http://test.com"
mock_settings.webdav_username = "user"
mock_settings.webdav_password = "pass"
mock_settings.webdav_password = _TEST_CREDENTIAL
# This will queue the task in Redis
result = upload_to_webdav.apply_async(
args=["/tmp/test.txt"],
kwargs={"file_id": 1}
)
result = upload_to_webdav.apply_async(args=["/tmp/test.txt"], kwargs={"file_id": 1})
# Verify task ID was generated
assert result.id is not None
@@ -144,67 +151,64 @@ class TestEndToEndWithRedis:
):
"""
Test multiple tasks executing in parallel through Redis/Celery.
This tests concurrent task processing.
"""
from app.tasks.upload_to_webdav import upload_to_webdav
# Create multiple test files
files = []
for i in range(5):
test_file = tmp_path / f"test_{i}.txt"
test_file.write_text(f"Test file {i}")
files.append(str(test_file))
with patch("app.tasks.upload_to_webdav.settings") as mock_settings, \
patch("app.tasks.upload_to_webdav.log_task_progress"):
with (
patch("app.tasks.upload_to_webdav.settings") as mock_settings,
patch("app.tasks.upload_to_webdav.log_task_progress"),
):
mock_settings.webdav_url = webdav_container["url"] + "/"
mock_settings.webdav_username = webdav_container["username"]
mock_settings.webdav_password = webdav_container["password"]
mock_settings.webdav_folder = "parallel-test"
mock_settings.webdav_verify_ssl = False
mock_settings.http_request_timeout = 30
# Create folder on WebDAV server
folder_url = f"{webdav_container['url']}/parallel-test"
requests.request(
"MKCOL",
folder_url,
auth=(webdav_container["username"], webdav_container["password"]),
timeout=5
"MKCOL", folder_url, auth=(webdav_container["username"], webdav_container["password"]), timeout=5
)
# Queue all tasks
results = []
for idx, file_path in enumerate(files):
result = upload_to_webdav.delay(file_path, file_id=idx + 100)
results.append((result, file_path))
# Wait for all tasks to complete
timeout = 60
start_time = time.time()
all_ready = False
while not all_ready:
if time.time() - start_time > timeout:
pytest.fail("Tasks did not complete within timeout")
all_ready = all(r.ready() for r, _ in results)
time.sleep(0.5)
# Verify all tasks succeeded
for result, file_path in results:
task_result = result.get(timeout=5)
assert task_result["status"] == "Completed"
# Verify file on server
filename = os.path.basename(file_path)
file_url = f"{webdav_container['url']}/parallel-test/{filename}"
response = requests.get(
file_url,
auth=(webdav_container["username"], webdav_container["password"]),
timeout=5
file_url, auth=(webdav_container["username"], webdav_container["password"]), timeout=5
)
assert response.status_code == 200
@@ -217,37 +221,39 @@ class TestEndToEndWithRedis:
):
"""
Test that tasks retry on failure using Redis.
This verifies the retry mechanism works with real broker.
"""
from app.tasks.upload_to_webdav import upload_to_webdav
with patch("app.tasks.upload_to_webdav.settings") as mock_settings, \
patch("app.tasks.upload_to_webdav.log_task_progress"), \
patch("app.tasks.upload_to_webdav.requests.put") as mock_put:
with (
patch("app.tasks.upload_to_webdav.settings") as mock_settings,
patch("app.tasks.upload_to_webdav.log_task_progress"),
patch("app.tasks.upload_to_webdav.requests.put") as mock_put,
):
mock_settings.webdav_url = "http://test.com/"
mock_settings.webdav_username = "user"
mock_settings.webdav_password = "pass"
mock_settings.webdav_password = _TEST_CREDENTIAL
mock_settings.webdav_folder = ""
mock_settings.webdav_verify_ssl = False
mock_settings.http_request_timeout = 30
# First attempt fails with 500
mock_response_fail = requests.Response()
mock_response_fail.status_code = 500
mock_response_fail._content = b"Server Error"
# Second attempt succeeds
mock_response_success = requests.Response()
mock_response_success.status_code = 201
# Configure mock to fail once, then succeed
mock_put.side_effect = [mock_response_fail, mock_response_success]
# Queue task
result = upload_to_webdav.delay(sample_text_file, file_id=1)
# Wait for completion (including retry)
timeout = 30
start_time = time.time()
@@ -263,7 +269,7 @@ class TestEndToEndWithRedis:
class TestFullInfrastructure:
"""
Tests using the complete infrastructure stack.
PostgreSQL + Redis + Gotenberg + Upload targets (WebDAV/SFTP/MinIO)
"""
@@ -272,38 +278,43 @@ class TestFullInfrastructure:
Verify all infrastructure components are running.
"""
infra = full_infrastructure
# Check PostgreSQL
assert infra["postgres"]["url"] is not None
assert "postgresql" in infra["postgres"]["url"]
# Check Redis
assert infra["redis"]["url"] is not None
import redis
r = redis.from_url(infra["redis"]["url"])
assert r.ping()
# Check Gotenberg
assert infra["gotenberg"]["url"] is not None
response = requests.get(f"{infra['gotenberg']['url']}/health", timeout=5)
assert response.status_code == 200
# Check WebDAV
assert infra["webdav"]["url"] is not None
# Check SFTP
assert infra["sftp"]["host"] is not None
assert infra["sftp"]["port"] is not None
# Check MinIO
assert infra["minio"]["access_key"] is not None
@pytest.mark.skipif(
not _has_psycopg2,
reason="psycopg2 not installed",
)
def test_database_with_real_postgres(self, postgres_container, db_session_real):
"""
Test database operations with real PostgreSQL instead of SQLite.
"""
from app.models import FileRecord
# Create a file record
file_record = FileRecord(
filename="test.pdf",
@@ -311,13 +322,13 @@ class TestFullInfrastructure:
file_size=1024,
mime_type="application/pdf",
)
db_session_real.add(file_record)
db_session_real.commit()
# Verify it was saved
assert file_record.id is not None
# Query it back
queried = db_session_real.query(FileRecord).filter_by(filename="test.pdf").first()
assert queried is not None
@@ -333,26 +344,28 @@ class TestFullInfrastructure:
):
"""
Test uploading to multiple targets in parallel (WebDAV + SFTP).
This simulates the send_to_all_destinations workflow.
"""
from app.tasks.upload_to_webdav import upload_to_webdav
infra = full_infrastructure
with patch("app.tasks.upload_to_webdav.settings") as mock_settings, \
patch("app.tasks.upload_to_webdav.log_task_progress"):
with (
patch("app.tasks.upload_to_webdav.settings") as mock_settings,
patch("app.tasks.upload_to_webdav.log_task_progress"),
):
mock_settings.webdav_url = infra["webdav"]["url"] + "/"
mock_settings.webdav_username = infra["webdav"]["username"]
mock_settings.webdav_password = infra["webdav"]["password"]
mock_settings.webdav_folder = ""
mock_settings.webdav_verify_ssl = False
mock_settings.http_request_timeout = 30
# Upload to WebDAV
webdav_result = upload_to_webdav.delay(sample_text_file, file_id=1)
# Wait for completion
timeout = 30
start_time = time.time()
@@ -360,25 +373,23 @@ class TestFullInfrastructure:
if time.time() - start_time > timeout:
pytest.fail("Task timeout")
time.sleep(0.5)
# Verify WebDAV upload
result = webdav_result.get(timeout=10)
assert result["status"] == "Completed"
# Verify file on WebDAV server
filename = os.path.basename(sample_text_file)
file_url = f"{infra['webdav']['url']}/{filename}"
response = requests.get(
file_url,
auth=(infra["webdav"]["username"], infra["webdav"]["password"]),
timeout=5
file_url, auth=(infra["webdav"]["username"], infra["webdav"]["password"]), timeout=5
)
assert response.status_code == 200
def test_gotenberg_pdf_conversion(self, gotenberg_container, tmp_path):
"""
Test PDF conversion using real Gotenberg service.
This verifies document processing capabilities.
"""
# Create a simple HTML file
@@ -390,16 +401,14 @@ class TestFullInfrastructure:
<body><h1>Integration Test</h1><p>This is a test document.</p></body>
</html>
""")
# Convert to PDF using Gotenberg
with open(html_file, "rb") as f:
files = {"files": f}
response = requests.post(
f"{gotenberg_container['url']}/forms/chromium/convert/html",
files=files,
timeout=30
f"{gotenberg_container['url']}/forms/chromium/convert/html", files=files, timeout=30
)
assert response.status_code == 200
assert response.headers["Content-Type"] == "application/pdf"
assert len(response.content) > 0
@@ -408,12 +417,12 @@ class TestFullInfrastructure:
def test_minio_s3_upload(self, minio_container, sample_text_file):
"""
Test S3-compatible upload using real MinIO.
This tests S3 upload functionality with actual storage.
"""
import boto3
from botocore.client import Config
# Create S3 client configured for MinIO
s3_client = boto3.client(
"s3",
@@ -423,28 +432,27 @@ class TestFullInfrastructure:
config=Config(signature_version="s3v4"),
region_name=minio_container["region"],
)
# Create bucket
bucket_name = "test-bucket"
s3_client.create_bucket(Bucket=bucket_name)
# Upload file
filename = os.path.basename(sample_text_file)
with open(sample_text_file, "rb") as f:
s3_client.upload_fileobj(f, bucket_name, filename)
# Verify upload
response = s3_client.list_objects_v2(Bucket=bucket_name)
assert "Contents" in response
assert len(response["Contents"]) == 1
assert response["Contents"][0]["Key"] == filename
# Download and verify content
download_path = os.path.join(os.path.dirname(sample_text_file), "downloaded.txt")
s3_client.download_file(bucket_name, filename, download_path)
with open(sample_text_file, "rb") as original, \
open(download_path, "rb") as downloaded:
with open(sample_text_file, "rb") as original, open(download_path, "rb") as downloaded:
assert original.read() == downloaded.read()
def test_sftp_upload(self, sftp_container, sample_text_file):
@@ -452,11 +460,11 @@ class TestFullInfrastructure:
Test SFTP upload using real SFTP server.
"""
import paramiko
# Create SFTP client
ssh = paramiko.SSHClient()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
# Connect to SFTP server
ssh.connect(
hostname=sftp_container["host"],
@@ -465,35 +473,34 @@ class TestFullInfrastructure:
password=sftp_container["password"],
timeout=10,
)
sftp = ssh.open_sftp()
try:
# Upload file
filename = os.path.basename(sample_text_file)
remote_path = f"{sftp_container['folder']}/{filename}"
sftp.put(sample_text_file, remote_path)
# Verify upload
stat = sftp.stat(remote_path)
assert stat.st_size == os.path.getsize(sample_text_file)
# Download and verify content
download_path = os.path.join(os.path.dirname(sample_text_file), "sftp_downloaded.txt")
sftp.get(remote_path, download_path)
with open(sample_text_file, "rb") as original, \
open(download_path, "rb") as downloaded:
with open(sample_text_file, "rb") as original, open(download_path, "rb") as downloaded:
assert original.read() == downloaded.read()
finally:
sftp.close()
ssh.close()
@pytest.mark.integration
@pytest.mark.requires_docker
@pytest.mark.requires_docker
@pytest.mark.e2e
@pytest.mark.slow
class TestProductionLikeScenarios:
@@ -517,16 +524,16 @@ class TestProductionLikeScenarios:
4. Process document (mock OCR/metadata extraction)
5. Upload to WebDAV via Celery
6. Verify all steps completed
This is the closest to real production usage.
"""
from app.models import FileRecord
from app.tasks.upload_to_webdav import upload_to_webdav
# Create test document
test_doc = tmp_path / "invoice.pdf"
test_doc.write_bytes(b"%PDF-1.4\n%Test PDF\n%%EOF")
# Step 1: Store in database
file_record = FileRecord(
filename="invoice.pdf",
@@ -536,35 +543,34 @@ class TestProductionLikeScenarios:
)
db_session_real.add(file_record)
db_session_real.commit()
assert file_record.id is not None
db_file_id = file_record.id
# Step 2: Queue upload task
infra = full_infrastructure
with patch("app.tasks.upload_to_webdav.settings") as mock_settings, \
patch("app.tasks.upload_to_webdav.log_task_progress"):
with (
patch("app.tasks.upload_to_webdav.settings") as mock_settings,
patch("app.tasks.upload_to_webdav.log_task_progress"),
):
mock_settings.webdav_url = infra["webdav"]["url"] + "/"
mock_settings.webdav_username = infra["webdav"]["username"]
mock_settings.webdav_password = infra["webdav"]["password"]
mock_settings.webdav_folder = "processed"
mock_settings.webdav_verify_ssl = False
mock_settings.http_request_timeout = 30
# Create folder
folder_url = f"{infra['webdav']['url']}/processed"
requests.request(
"MKCOL",
folder_url,
auth=(infra["webdav"]["username"], infra["webdav"]["password"]),
timeout=5
"MKCOL", folder_url, auth=(infra["webdav"]["username"], infra["webdav"]["password"]), timeout=5
)
# Step 3: Queue upload
result = upload_to_webdav.delay(str(test_doc), file_id=db_file_id)
# Step 4: Wait for processing
timeout = 30
start_time = time.time()
@@ -572,17 +578,15 @@ class TestProductionLikeScenarios:
if time.time() - start_time > timeout:
pytest.fail("Pipeline timeout")
time.sleep(0.5)
# Step 5: Verify completion
task_result = result.get(timeout=10)
assert task_result["status"] == "Completed"
# Step 6: Verify file on WebDAV
file_url = f"{infra['webdav']['url']}/processed/invoice.pdf"
response = requests.get(
file_url,
auth=(infra["webdav"]["username"], infra["webdav"]["password"]),
timeout=5
file_url, auth=(infra["webdav"]["username"], infra["webdav"]["password"]), timeout=5
)
assert response.status_code == 200
assert response.content == test_doc.read_bytes()
tests/test_external_integrations.py
+822
View File
@@ -0,0 +1,822 @@
"""
Integration tests for external API services using real credentials.
These tests verify that external API integrations work correctly when real
credentials are provided via environment variables (e.g., GitHub Actions secrets).
Each test is guarded by ``pytest.mark.skipif`` so it is skipped automatically
when the required environment variables are absent or still set to placeholder
values. All tests carry the ``@pytest.mark.requires_external`` marker so they
can be run (or excluded) with::
pytest -m requires_external # run only external tests
pytest -m "not requires_external" # skip external tests
**Pipeline coverage:**
The tests exercise real end-to-end flows wherever credentials allow:
- *OpenAI*: key validation **and** metadata extraction via chat completion.
- *Azure Document Intelligence*: admin connectivity **and** OCR of a generated PDF.
- *S3*: bucket access, file upload, download verification, and cleanup.
- *Dropbox*: token refresh, file upload, download verification, and cleanup.
- *OneDrive*: token refresh, file upload, download verification, and cleanup.
- *Authentik/OIDC*: discovery endpoint and credential consistency.
Each pipeline test dynamically generates a unique PDF with ``fpdf2`` so every
run operates on fresh data. Uploaded test files are cleaned up in ``finally``
blocks to avoid polluting external storage.
"""
import json
import logging
import os
import tempfile
import uuid
from typing import Optional
import pytest
from tests.conftest import has_real_env
logger = logging.getLogger(__name__)
# ---------------------------------------------------------------------------
# Test-PDF generator helper
# ---------------------------------------------------------------------------
_TEST_PREFIX = "docuelevate_test_"
def generate_test_pdf(
content: Optional[str] = None,
filename_prefix: str = _TEST_PREFIX,
) -> str:
"""Generate a unique test PDF with embedded text.
Creates a one-page PDF containing *content* (or a random invoice stub)
and returns the path to the temporary file. The caller is responsible
for deleting the file when done.
Args:
content: Optional text to embed. When ``None`` a realistic
invoice-style document is generated.
filename_prefix: Prefix for the temp filename.
Returns:
Absolute path to the generated PDF file.
"""
from fpdf import FPDF
unique_id = uuid.uuid4().hex[:8]
if content is None:
content = (
f"Invoice #{unique_id}\n"
f"Date: 2024-06-15\n"
f"From: Acme Integration Testing GmbH\n"
f"To: DocuElevate QA Department\n"
f"Amount: EUR 1,234.56\n\n"
f"Description: Annual subscription renewal for cloud document\n"
f"processing services. Reference: REF-{unique_id}.\n\n"
f"Payment terms: Net 30 days.\n"
f"Bank: Deutsche Bank, IBAN: DE89 3704 0044 0532 0130 00\n"
)
pdf = FPDF()
pdf.add_page()
pdf.set_font("Helvetica", size=11)
pdf.multi_cell(0, 7, text=content)
fd, path = tempfile.mkstemp(prefix=filename_prefix, suffix=".pdf")
os.close(fd)
pdf.output(path)
return path
# ---------------------------------------------------------------------------
# OpenAI
# ---------------------------------------------------------------------------
@pytest.mark.requires_external
@pytest.mark.skipif(
not has_real_env("OPENAI_API_KEY"),
reason="Real OPENAI_API_KEY not available",
)
class TestOpenAIIntegration:
"""Verify OpenAI API connectivity and metadata extraction with real credentials."""
def test_openai_api_key_is_valid(self, original_env: dict) -> None:
"""Validate that the configured OpenAI API key can list models."""
import openai
api_key = original_env["OPENAI_API_KEY"]
base_url = original_env.get("OPENAI_BASE_URL") or "https://api.openai.com/v1"
oai = openai.OpenAI(api_key=api_key, base_url=base_url)
models = oai.models.list()
assert hasattr(models, "data"), "Expected models response to have 'data' attribute"
assert len(models.data) > 0, "Expected at least one model to be available"
def test_openai_test_endpoint_with_real_key(self, client, original_env: dict, monkeypatch) -> None:
"""Test the /api/openai/test endpoint returns success with a real API key."""
monkeypatch.setattr("app.config.settings.openai_api_key", original_env["OPENAI_API_KEY"])
if original_env.get("OPENAI_BASE_URL"):
monkeypatch.setattr("app.config.settings.openai_base_url", original_env["OPENAI_BASE_URL"])
response = client.get("/api/openai/test")
assert response.status_code == 200
data = response.json()
assert data["status"] == "success", f"OpenAI test endpoint failed: {data.get('message')}"
assert data.get("models_available", 0) > 0
def test_openai_metadata_extraction(self, original_env: dict) -> None:
"""End-to-end: send document text to OpenAI and receive structured metadata."""
import openai
api_key = original_env["OPENAI_API_KEY"]
base_url = original_env.get("OPENAI_BASE_URL") or "https://api.openai.com/v1"
# Use the same prompt structure as extract_metadata_with_gpt task
unique_id = uuid.uuid4().hex[:8]
sample_text = (
f"Invoice #{unique_id}\n"
f"Date: 2024-06-15\n"
f"From: Acme Integration Testing GmbH\n"
f"To: DocuElevate QA Department\n"
f"Amount: EUR 1,234.56\n"
)
prompt = (
"You are a specialized document analyzer. Analyze the given text and return a JSON object with:\n"
'- "document_type": precise classification (e.g., Invoice, Contract)\n'
'- "language": ISO 639-1 code\n'
'- "tags": list of up to 4 keywords\n'
'- "absender": sender name\n'
'- "empfaenger": recipient name\n\n'
f"Text:\n{sample_text}\n\n"
"Return only valid JSON."
)
oai = openai.OpenAI(api_key=api_key, base_url=base_url)
completion = oai.chat.completions.create(
model="gpt-4o-mini",
messages=[
{"role": "system", "content": "You are an intelligent document classifier."},
{"role": "user", "content": prompt},
],
temperature=0,
)
content = completion.choices[0].message.content
assert content, "OpenAI returned empty content"
# Extract JSON from response (may be wrapped in markdown fences)
import re
json_match = re.search(r"\{.*\}", content, re.DOTALL)
assert json_match, f"No JSON found in OpenAI response: {content[:200]}"
metadata = json.loads(json_match.group())
assert "document_type" in metadata, "Missing document_type in extracted metadata"
assert "language" in metadata, "Missing language in extracted metadata"
# ---------------------------------------------------------------------------
# Azure Document Intelligence
# ---------------------------------------------------------------------------
@pytest.mark.requires_external
@pytest.mark.skipif(
not has_real_env("AZURE_AI_KEY", "AZURE_ENDPOINT"),
reason="Real AZURE_AI_KEY and AZURE_ENDPOINT not available",
)
class TestAzureDocumentIntelligenceIntegration:
"""Verify Azure Document Intelligence connectivity and OCR with real credentials."""
def test_azure_admin_client_connects(self, original_env: dict) -> None:
"""Validate that the Azure admin client can list operations."""
from azure.ai.documentintelligence import DocumentIntelligenceAdministrationClient
from azure.core.credentials import AzureKeyCredential
admin_client = DocumentIntelligenceAdministrationClient(
endpoint=original_env["AZURE_ENDPOINT"],
credential=AzureKeyCredential(original_env["AZURE_AI_KEY"]),
)
operations = list(admin_client.list_operations())
assert isinstance(operations, list)
def test_azure_test_endpoint_with_real_credentials(self, client, original_env: dict, monkeypatch) -> None:
"""Test the /api/azure/test endpoint returns success with real credentials."""
monkeypatch.setattr("app.config.settings.azure_ai_key", original_env["AZURE_AI_KEY"])
monkeypatch.setattr("app.config.settings.azure_endpoint", original_env["AZURE_ENDPOINT"])
if original_env.get("AZURE_REGION"):
monkeypatch.setattr("app.config.settings.azure_region", original_env["AZURE_REGION"])
response = client.get("/api/azure/test")
assert response.status_code == 200
data = response.json()
assert data["status"] == "success", f"Azure test endpoint failed: {data.get('message')}"
def test_azure_ocr_on_generated_pdf(self, original_env: dict) -> None:
"""End-to-end: send a generated PDF to Azure and receive OCR text back."""
from azure.ai.documentintelligence import DocumentIntelligenceClient
from azure.ai.documentintelligence.models import AnalyzeOutputOption
from azure.core.credentials import AzureKeyCredential
pdf_path = generate_test_pdf()
try:
doc_client = DocumentIntelligenceClient(
endpoint=original_env["AZURE_ENDPOINT"],
credential=AzureKeyCredential(original_env["AZURE_AI_KEY"]),
)
with open(pdf_path, "rb") as f:
poller = doc_client.begin_analyze_document(
"prebuilt-read",
body=f,
output=[AnalyzeOutputOption.PDF],
)
result = poller.result()
assert result.content, "Azure OCR returned no content"
assert len(result.content) > 10, f"OCR text too short: {result.content[:50]}"
# Verify the generated text is recognizable
assert (
"Acme" in result.content or "Invoice" in result.content
), f"OCR text does not contain expected keywords: {result.content[:200]}"
# Retrieve the searchable PDF output
operation_id = poller.details["operation_id"]
pdf_response = doc_client.get_analyze_result_pdf(
model_id=result.model_id,
result_id=operation_id,
)
searchable_bytes = b"".join(pdf_response)
assert len(searchable_bytes) > 0, "Searchable PDF output is empty"
assert searchable_bytes[:5] == b"%PDF-", "Output is not a valid PDF"
finally:
os.unlink(pdf_path)
# ---------------------------------------------------------------------------
# AWS S3 – full upload/download/delete pipeline
# ---------------------------------------------------------------------------
@pytest.mark.requires_external
@pytest.mark.skipif(
not has_real_env("AWS_ACCESS_KEY_ID", "AWS_SECRET_ACCESS_KEY", "S3_BUCKET_NAME"),
reason="Real AWS credentials and S3_BUCKET_NAME not available",
)
class TestS3Integration:
"""Verify AWS S3 connectivity and upload/download pipeline with real credentials."""
def test_s3_bucket_accessible(self, original_env: dict) -> None:
"""Validate that the S3 bucket exists and credentials are accepted."""
import boto3
s3_client = boto3.client(
"s3",
region_name=original_env.get("AWS_REGION", "us-east-1"),
aws_access_key_id=original_env["AWS_ACCESS_KEY_ID"],
aws_secret_access_key=original_env["AWS_SECRET_ACCESS_KEY"],
)
response = s3_client.head_bucket(Bucket=original_env["S3_BUCKET_NAME"])
assert response["ResponseMetadata"]["HTTPStatusCode"] == 200
def test_s3_upload_download_delete(self, original_env: dict) -> None:
"""End-to-end: upload a generated PDF to S3, download and verify, then delete."""
import boto3
pdf_path = generate_test_pdf()
s3_key = None
try:
s3_client = boto3.client(
"s3",
region_name=original_env.get("AWS_REGION", "us-east-1"),
aws_access_key_id=original_env["AWS_ACCESS_KEY_ID"],
aws_secret_access_key=original_env["AWS_SECRET_ACCESS_KEY"],
)
bucket = original_env["S3_BUCKET_NAME"]
prefix = original_env.get("S3_FOLDER_PREFIX", "")
if prefix and not prefix.endswith("/"):
prefix += "/"
s3_key = f"{prefix}{_TEST_PREFIX}{uuid.uuid4().hex[:8]}.pdf"
# Upload
s3_client.upload_file(pdf_path, bucket, s3_key)
# Download and verify
download_path = pdf_path + ".downloaded"
s3_client.download_file(bucket, s3_key, download_path)
with open(pdf_path, "rb") as orig, open(download_path, "rb") as dl:
assert orig.read() == dl.read(), "Downloaded file does not match uploaded file"
os.unlink(download_path)
finally:
os.unlink(pdf_path)
# Cleanup: delete the test object from S3
if s3_key:
try:
s3_client.delete_object(Bucket=bucket, Key=s3_key)
except Exception as exc:
logger.warning("Failed to clean up S3 test object %s: %s", s3_key, exc)
# ---------------------------------------------------------------------------
# Dropbox – full upload/download/delete pipeline
# ---------------------------------------------------------------------------
@pytest.mark.requires_external
@pytest.mark.skipif(
not has_real_env("DROPBOX_APP_KEY", "DROPBOX_APP_SECRET", "DROPBOX_REFRESH_TOKEN"),
reason="Real Dropbox credentials not available",
)
class TestDropboxIntegration:
"""Verify Dropbox token validity and upload/download pipeline with real credentials."""
def test_dropbox_token_refresh_and_account_info(self, original_env: dict) -> None:
"""Validate token refresh and account info retrieval in one go."""
import dropbox as dbx_lib
dbx = dbx_lib.Dropbox(
app_key=original_env["DROPBOX_APP_KEY"],
app_secret=original_env["DROPBOX_APP_SECRET"],
oauth2_refresh_token=original_env["DROPBOX_REFRESH_TOKEN"],
)
account = dbx.users_get_current_account()
assert account.email, "Dropbox account missing email"
def test_dropbox_upload_download_delete(self, original_env: dict) -> None:
"""End-to-end: upload a generated PDF to Dropbox, download it, then delete it."""
import dropbox as dbx_lib
pdf_path = generate_test_pdf()
remote_path = f"/{_TEST_PREFIX}{uuid.uuid4().hex[:8]}.pdf"
dbx = None
try:
dbx = dbx_lib.Dropbox(
app_key=original_env["DROPBOX_APP_KEY"],
app_secret=original_env["DROPBOX_APP_SECRET"],
oauth2_refresh_token=original_env["DROPBOX_REFRESH_TOKEN"],
)
# Upload
with open(pdf_path, "rb") as f:
dbx.files_upload(
f.read(),
remote_path,
mode=dbx_lib.files.WriteMode.overwrite,
)
# Download and verify
_, response = dbx.files_download(remote_path)
downloaded = response.content
with open(pdf_path, "rb") as f:
assert f.read() == downloaded, "Downloaded Dropbox file does not match uploaded file"
finally:
os.unlink(pdf_path)
# Cleanup
if dbx:
try:
dbx.files_delete_v2(remote_path)
except Exception as exc:
logger.warning("Failed to clean up Dropbox test file %s: %s", remote_path, exc)
# ---------------------------------------------------------------------------
# OneDrive – full upload/download/delete pipeline
# ---------------------------------------------------------------------------
@pytest.mark.requires_external
@pytest.mark.skipif(
not has_real_env("ONEDRIVE_CLIENT_ID", "ONEDRIVE_CLIENT_SECRET", "ONEDRIVE_REFRESH_TOKEN"),
reason="Real OneDrive credentials not available",
)
class TestOneDriveIntegration:
"""Verify OneDrive token validity and upload/download pipeline with real credentials."""
@staticmethod
def _get_access_token(env: dict) -> str:
"""Obtain a fresh OneDrive access token via MSAL."""
import msal
tenant = env.get("ONEDRIVE_TENANT_ID") or "common"
app = msal.ConfidentialClientApplication(
client_id=env["ONEDRIVE_CLIENT_ID"],
client_credential=env["ONEDRIVE_CLIENT_SECRET"],
authority=f"https://login.microsoftonline.com/{tenant}",
)
result = app.acquire_token_by_refresh_token(
refresh_token=env["ONEDRIVE_REFRESH_TOKEN"],
scopes=["https://graph.microsoft.com/.default"],
)
assert "access_token" in result, f"OneDrive token acquisition failed: {result.get('error_description')}"
return result["access_token"]
def test_onedrive_token_refresh_and_user_info(self, original_env: dict) -> None:
"""Validate token refresh and user info retrieval."""
import requests
token = self._get_access_token(original_env)
resp = requests.get(
"https://graph.microsoft.com/v1.0/me",
headers={"Authorization": f"Bearer {token}"},
timeout=30,
)
assert resp.status_code == 200, f"OneDrive user info failed: {resp.text}"
def test_onedrive_upload_download_delete(self, original_env: dict) -> None:
"""End-to-end: upload a generated PDF to OneDrive, download it, then delete it."""
import requests
pdf_path = generate_test_pdf()
filename = f"{_TEST_PREFIX}{uuid.uuid4().hex[:8]}.pdf"
folder = original_env.get("ONEDRIVE_FOLDER_PATH", "").strip("/")
item_id = None
token = None
try:
token = self._get_access_token(original_env)
headers = {"Authorization": f"Bearer {token}"}
# Upload (simple upload for small files)
if folder:
upload_url = f"https://graph.microsoft.com/v1.0/me/drive/root:/{folder}/{filename}:/content"
else:
upload_url = f"https://graph.microsoft.com/v1.0/me/drive/root:/{filename}:/content"
with open(pdf_path, "rb") as f:
upload_resp = requests.put(
upload_url,
headers={**headers, "Content-Type": "application/pdf"},
data=f.read(),
timeout=60,
)
assert upload_resp.status_code in (200, 201), f"OneDrive upload failed: {upload_resp.text}"
item_id = upload_resp.json().get("id")
# Download and verify
download_url = f"https://graph.microsoft.com/v1.0/me/drive/items/{item_id}/content"
dl_resp = requests.get(download_url, headers=headers, timeout=60)
assert dl_resp.status_code == 200, f"OneDrive download failed: {dl_resp.status_code}"
with open(pdf_path, "rb") as f:
assert f.read() == dl_resp.content, "Downloaded OneDrive file does not match uploaded file"
finally:
os.unlink(pdf_path)
# Cleanup
if item_id and token:
try:
requests.delete(
f"https://graph.microsoft.com/v1.0/me/drive/items/{item_id}",
headers={"Authorization": f"Bearer {token}"},
timeout=30,
)
except Exception as exc:
logger.warning("Failed to clean up OneDrive test file %s: %s", filename, exc)
# ---------------------------------------------------------------------------
# Authentik / OpenID Connect – discovery endpoint validation
# ---------------------------------------------------------------------------
@pytest.mark.requires_external
@pytest.mark.skipif(
not has_real_env("AUTHENTIK_CONFIG_URL"),
reason="Real AUTHENTIK_CONFIG_URL not available",
)
class TestAuthentikIntegration:
"""Verify Authentik / OpenID Connect discovery endpoint is reachable."""
@staticmethod
def test_oidc_discovery_endpoint(original_env: dict) -> None:
"""Validate that the OIDC discovery URL returns a valid JSON document."""
import requests
config_url = original_env["AUTHENTIK_CONFIG_URL"]
response = requests.get(config_url, timeout=30)
assert response.status_code == 200, f"OIDC discovery failed: {response.status_code}"
data = response.json()
assert "issuer" in data, "OIDC response missing 'issuer'"
assert "authorization_endpoint" in data, "OIDC response missing 'authorization_endpoint'"
assert "token_endpoint" in data, "OIDC response missing 'token_endpoint'"
@staticmethod
def test_authentik_client_credentials_present(original_env: dict) -> None:
"""Validate that Authentik client credentials are configured alongside the config URL."""
client_id = original_env.get("AUTHENTIK_CLIENT_ID")
client_secret = original_env.get("AUTHENTIK_CLIENT_SECRET")
if client_id and client_secret:
assert len(client_id) > 0, "AUTHENTIK_CLIENT_ID should not be empty"
assert len(client_secret) > 0, "AUTHENTIK_CLIENT_SECRET should not be empty"
else:
pytest.skip("AUTHENTIK_CLIENT_ID and/or AUTHENTIK_CLIENT_SECRET not set")
# ---------------------------------------------------------------------------
# Full pipeline: Azure OCR → OpenAI metadata extraction (end-to-end)
# ---------------------------------------------------------------------------
@pytest.mark.requires_external
@pytest.mark.skipif(
not has_real_env("AZURE_AI_KEY", "AZURE_ENDPOINT", "OPENAI_API_KEY"),
reason="Real Azure + OpenAI credentials required for full pipeline test",
)
class TestFullOCRMetadataPipeline:
"""End-to-end pipeline: generate PDF → Azure OCR → OpenAI metadata extraction.
This replicates the core DocuElevate processing flow without requiring
Celery or Redis, by calling the service APIs directly.
"""
@staticmethod
def test_ocr_then_metadata_extraction(original_env: dict) -> None:
"""Generate a PDF, OCR it with Azure, then extract metadata with OpenAI."""
import re
import openai
from azure.ai.documentintelligence import DocumentIntelligenceClient
from azure.ai.documentintelligence.models import AnalyzeOutputOption
from azure.core.credentials import AzureKeyCredential
pdf_path = generate_test_pdf()
try:
# --- Step 1: Azure OCR ---
doc_client = DocumentIntelligenceClient(
endpoint=original_env["AZURE_ENDPOINT"],
credential=AzureKeyCredential(original_env["AZURE_AI_KEY"]),
)
with open(pdf_path, "rb") as f:
poller = doc_client.begin_analyze_document(
"prebuilt-read",
body=f,
output=[AnalyzeOutputOption.PDF],
)
result = poller.result()
extracted_text = result.content
assert extracted_text and len(extracted_text) > 10, "OCR produced insufficient text"
# --- Step 2: OpenAI metadata extraction ---
api_key = original_env["OPENAI_API_KEY"]
base_url = original_env.get("OPENAI_BASE_URL") or "https://api.openai.com/v1"
oai = openai.OpenAI(api_key=api_key, base_url=base_url)
prompt = (
"You are a specialized document analyzer. Analyze the following OCR-extracted text "
"and return a JSON object with these fields:\n"
'- "document_type": classification (e.g. Invoice, Contract, Letter)\n'
'- "language": ISO 639-1 code\n'
'- "absender": sender\n'
'- "empfaenger": recipient\n'
'- "tags": up to 4 keywords\n'
'- "confidence_score": 0-100\n\n'
f"OCR text:\n{extracted_text}\n\n"
"Return only valid JSON."
)
completion = oai.chat.completions.create(
model="gpt-4o-mini",
messages=[
{"role": "system", "content": "You are an intelligent document classifier."},
{"role": "user", "content": prompt},
],
temperature=0,
)
content = completion.choices[0].message.content
assert content, "OpenAI returned empty response"
json_match = re.search(r"\{.*\}", content, re.DOTALL)
assert json_match, f"No JSON found in response: {content[:200]}"
metadata = json.loads(json_match.group())
# Validate key metadata fields
assert "document_type" in metadata, "Missing document_type"
assert "language" in metadata, "Missing language"
assert "absender" in metadata, "Missing absender"
assert "empfaenger" in metadata, "Missing empfaenger"
# The generated invoice should be classified reasonably
doc_type = metadata["document_type"].lower()
assert any(
kw in doc_type for kw in ("invoice", "rechnung", "bill")
), f"Unexpected document_type: {metadata['document_type']}"
finally:
os.unlink(pdf_path)
# ---------------------------------------------------------------------------
# Configuration: verify that Settings correctly loads external env vars
# ---------------------------------------------------------------------------
@pytest.mark.unit
class TestExternalEnvVarConfiguration:
"""Verify that the Settings class correctly reads external API environment variables.
These tests do NOT call external APIs; they only validate that the configuration
layer correctly maps environment variables to Settings fields.
"""
@staticmethod
def test_settings_reads_openai_base_url() -> None:
"""Test that OPENAI_BASE_URL is correctly loaded into Settings."""
from app.config import Settings
custom_url = "https://custom-openai.example.com/v1"
config = Settings(
database_url="sqlite:///test.db",
redis_url="redis://localhost:6379",
openai_api_key="test",
openai_base_url=custom_url,
azure_ai_key="test",
azure_region="test",
azure_endpoint="https://test.example.com",
gotenberg_url="http://localhost:3000",
workdir="/tmp",
auth_enabled=False,
)
assert config.openai_base_url == custom_url
@staticmethod
def test_settings_reads_s3_configuration() -> None:
"""Test that S3-related settings are correctly loaded."""
from app.config import Settings
config = Settings(
database_url="sqlite:///test.db",
redis_url="redis://localhost:6379",
openai_api_key="test",
azure_ai_key="test",
azure_region="test",
azure_endpoint="https://test.example.com",
gotenberg_url="http://localhost:3000",
workdir="/tmp",
auth_enabled=False,
aws_access_key_id="AKIAEXAMPLE",
aws_secret_access_key="secretkey",
s3_bucket_name="my-bucket",
s3_folder_prefix="uploads/",
)
assert config.aws_access_key_id == "AKIAEXAMPLE"
assert config.aws_secret_access_key == "secretkey"
assert config.s3_bucket_name == "my-bucket"
assert config.s3_folder_prefix == "uploads/"
@staticmethod
def test_settings_reads_onedrive_configuration() -> None:
"""Test that OneDrive-related settings are correctly loaded."""
from app.config import Settings
config = Settings(
database_url="sqlite:///test.db",
redis_url="redis://localhost:6379",
openai_api_key="test",
azure_ai_key="test",
azure_region="test",
azure_endpoint="https://test.example.com",
gotenberg_url="http://localhost:3000",
workdir="/tmp",
auth_enabled=False,
onedrive_client_id="client-123",
onedrive_client_secret="secret-456",
onedrive_tenant_id="tenant-789",
onedrive_refresh_token="refresh-abc",
onedrive_folder_path="Documents/Test",
)
assert config.onedrive_client_id == "client-123"
assert config.onedrive_client_secret == "secret-456"
assert config.onedrive_tenant_id == "tenant-789"
assert config.onedrive_refresh_token == "refresh-abc"
assert config.onedrive_folder_path == "Documents/Test"
@staticmethod
def test_settings_reads_dropbox_configuration() -> None:
"""Test that Dropbox-related settings are correctly loaded."""
from app.config import Settings
config = Settings(
database_url="sqlite:///test.db",
redis_url="redis://localhost:6379",
openai_api_key="test",
azure_ai_key="test",
azure_region="test",
azure_endpoint="https://test.example.com",
gotenberg_url="http://localhost:3000",
workdir="/tmp",
auth_enabled=False,
dropbox_app_key="dbx-key",
dropbox_app_secret="dbx-secret",
dropbox_refresh_token="dbx-refresh",
)
assert config.dropbox_app_key == "dbx-key"
assert config.dropbox_app_secret == "dbx-secret"
assert config.dropbox_refresh_token == "dbx-refresh"
@staticmethod
def test_settings_reads_authentik_configuration() -> None:
"""Test that Authentik/OIDC settings are correctly loaded."""
from app.config import Settings
config = Settings(
database_url="sqlite:///test.db",
redis_url="redis://localhost:6379",
openai_api_key="test",
azure_ai_key="test",
azure_region="test",
azure_endpoint="https://test.example.com",
gotenberg_url="http://localhost:3000",
workdir="/tmp",
auth_enabled=False,
authentik_client_id="auth-client",
authentik_client_secret="auth-secret",
authentik_config_url="https://auth.example.com/.well-known/openid-configuration",
)
assert config.authentik_client_id == "auth-client"
assert config.authentik_client_secret == "auth-secret"
assert config.authentik_config_url == "https://auth.example.com/.well-known/openid-configuration"
@staticmethod
def test_settings_optional_services_default_to_none() -> None:
"""Test that optional external service settings default to None when not provided."""
from app.config import Settings
config = Settings(
database_url="sqlite:///test.db",
redis_url="redis://localhost:6379",
openai_api_key="test",
azure_ai_key="test",
azure_region="test",
azure_endpoint="https://test.example.com",
gotenberg_url="http://localhost:3000",
workdir="/tmp",
auth_enabled=False,
)
assert config.aws_access_key_id is None
assert config.aws_secret_access_key is None
assert config.s3_bucket_name is None
assert config.onedrive_client_id is None
assert config.onedrive_client_secret is None
assert config.onedrive_refresh_token is None
assert config.dropbox_app_key is None
assert config.dropbox_app_secret is None
assert config.dropbox_refresh_token is None
assert config.authentik_client_id is None
assert config.authentik_client_secret is None
assert config.authentik_config_url is None
# ---------------------------------------------------------------------------
# Test-PDF generator unit tests
# ---------------------------------------------------------------------------
@pytest.mark.unit
class TestPdfGenerator:
"""Verify that the test PDF generator produces valid PDFs with extractable text."""
@staticmethod
def test_generate_default_pdf() -> None:
"""Test that generate_test_pdf creates a valid PDF with embedded text."""
import PyPDF2
path = generate_test_pdf()
try:
assert os.path.exists(path)
assert os.path.getsize(path) > 100
with open(path, "rb") as f:
reader = PyPDF2.PdfReader(f)
assert len(reader.pages) >= 1
text = reader.pages[0].extract_text()
assert "Invoice" in text
assert "Acme" in text
finally:
os.unlink(path)
@staticmethod
def test_generate_custom_content_pdf() -> None:
"""Test that generate_test_pdf accepts custom content."""
import PyPDF2
custom = "Custom test content for verification"
path = generate_test_pdf(content=custom)
try:
with open(path, "rb") as f:
reader = PyPDF2.PdfReader(f)
text = reader.pages[0].extract_text()
assert "Custom test content" in text
finally:
os.unlink(path)
@staticmethod
def test_generated_pdfs_are_unique() -> None:
"""Test that consecutive calls produce different PDFs."""
path1 = generate_test_pdf()
path2 = generate_test_pdf()
try:
with open(path1, "rb") as f1, open(path2, "rb") as f2:
assert f1.read() != f2.read(), "Two generated PDFs should differ"
finally:
os.unlink(path1)
os.unlink(path2)
tests/test_imap_tasks.py
+5 -2
View File
@@ -1,4 +1,5 @@
"""Tests for app/tasks/imap_tasks.py module."""
import os
import json
import pytest
@@ -19,6 +20,8 @@ from app.tasks.imap_tasks import (
get_capabilities,
)
_TEST_CREDENTIAL = "pass" # noqa: S105
@pytest.mark.unit
class TestCleanupOldEntries:
@@ -84,7 +87,7 @@ class TestCheckAndPullMailbox:
host=None,
port=993,
username="user",
password="pass",
password=_TEST_CREDENTIAL,
use_ssl=True,
delete_after_process=False,
)
@@ -112,7 +115,7 @@ class TestCheckAndPullMailbox:
host="imap.example.com",
port=993,
username="user",
password="pass",
password=_TEST_CREDENTIAL,
use_ssl=True,
delete_after_process=False,
)
tests/test_notification.py
+5 -2
View File
@@ -7,6 +7,9 @@ Tests notification utilities and URL masking.
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
@pytest.mark.unit
class TestNotificationUrlMasking:
@@ -16,7 +19,7 @@ class TestNotificationUrlMasking:
"""Test masking of basic auth URLs"""
from app.utils.notification import _mask_sensitive_url
url = "https://user:password@example.com/notify"
url = _TEST_CREDENTIAL_URL
masked = _mask_sensitive_url(url)
# Password should be masked
@@ -75,7 +78,7 @@ class TestNotificationUrlMasking:
"""Test masking with multiple sensitive parameters"""
from app.utils.notification import _mask_sensitive_url
url = "https://example.com/api?key=secret1&password=secret2&public=visible"
url = _TEST_QUERY_URL
masked = _mask_sensitive_url(url)
# Sensitive params should be masked
tests/test_path_traversal_security.py
+67 -60
View File
@@ -45,7 +45,7 @@ class TestFilenameSanitization:
result = sanitize_filename("/etc/passwd")
assert "/" not in result
assert result == "_etc_passwd"
assert result == "etc_passwd"
def test_sanitize_removes_windows_path_separators(self):
"""Test that Windows path separators are removed."""
@@ -83,9 +83,9 @@ class TestFilenameSanitization:
from app.utils.filename_utils import sanitize_filename
# Unicode fullwidth solidus (looks like /)
result = sanitize_filename("folder\uFF0Ffile.pdf")
result = sanitize_filename("folder\uff0ffile.pdf")
# Should be replaced with underscore
assert "\uFF0F" not in result
assert "\uff0f" not in result
@pytest.mark.security
@@ -100,10 +100,10 @@ class TestEmbedMetadataPathTraversal:
# Simulate malicious metadata from GPT
malicious_filename = "../../etc/passwd"
# This should be sanitized before being used
sanitized = sanitize_filename(malicious_filename)
# Verify sanitization removes path traversal
assert ".." not in sanitized
assert "/" not in sanitized
@@ -112,7 +112,7 @@ class TestEmbedMetadataPathTraversal:
# Verify unique_filepath with sanitized name stays in directory
result = unique_filepath(str(tmp_path), sanitized, ".pdf")
result_path = Path(result)
# Ensure result is within tmp_path
assert result_path.parent == tmp_path
@@ -120,7 +120,7 @@ class TestEmbedMetadataPathTraversal:
"""Test that embed_metadata_into_pdf sanitizes the filename from metadata."""
from app.tasks.embed_metadata_into_pdf import unique_filepath
from app.utils.filename_utils import sanitize_filename
# Test various malicious filenames
malicious_filenames = [
"../../../etc/passwd",
@@ -130,15 +130,15 @@ class TestEmbedMetadataPathTraversal:
"folder/../file",
"folder\\..\\file",
]
for malicious in malicious_filenames:
# Sanitize as the task should do
sanitized = sanitize_filename(malicious)
# Verify no path traversal is possible
result = unique_filepath(str(tmp_path), sanitized, ".pdf")
result_path = Path(result)
# Result must be direct child of tmp_path
assert result_path.parent == tmp_path, f"Failed for: {malicious}"
@@ -163,16 +163,16 @@ class TestEmbedMetadataPathTraversal:
# Setup
mock_settings.workdir = str(tmp_path)
# Create a temporary PDF file
test_pdf = tmp_path / "test.pdf"
test_pdf.write_bytes(b"%PDF-1.4\n")
# Mock PDF operations
mock_reader_instance = MagicMock()
mock_reader_instance.pages = []
mock_pdf_reader.return_value = mock_reader_instance
mock_writer_instance = MagicMock()
mock_pdf_writer.return_value = mock_writer_instance
@@ -187,12 +187,8 @@ class TestEmbedMetadataPathTraversal:
processed_dir = tmp_path / "processed"
processed_dir.mkdir()
# Execute task
task_mock = MagicMock()
task_mock.request.id = "test-task-id"
# Execute task (called directly, Celery injects 'self' automatically)
result = embed_metadata_into_pdf(
task_mock,
str(test_pdf),
"test text",
malicious_metadata,
@@ -219,11 +215,11 @@ class TestExtractMetadataFilenameValidation:
def test_validates_filename_format(self):
"""Test that invalid filename formats are rejected."""
import re
# Valid pattern from extract_metadata_with_gpt.py
# TODO: Consider extracting this to a shared constant to avoid duplication
valid_pattern = r'^[\w\-\. ]+$'
valid_pattern = r"^[\w\-\. ]+$"
# Test valid filenames
valid_filenames = [
"2024-01-15_Invoice.pdf",
@@ -231,12 +227,12 @@ class TestExtractMetadataFilenameValidation:
"My Document 2024.pdf",
"file-name_123.pdf",
]
for filename in valid_filenames:
# Remove extension for test
name_only = filename.rsplit(".", 1)[0]
assert re.match(valid_pattern, name_only), f"Valid filename rejected: {filename}"
# Test invalid filenames
invalid_filenames = [
"../../../etc/passwd",
@@ -247,7 +243,7 @@ class TestExtractMetadataFilenameValidation:
"file|name.pdf",
"file<>name.pdf",
]
for filename in invalid_filenames:
assert not re.match(valid_pattern, filename), f"Invalid filename accepted: {filename}"
@@ -259,7 +255,7 @@ class TestExtractMetadataFilenameValidation:
"/etc/shadow",
"folder/../file",
]
for filename in malicious_filenames:
# Check for path traversal indicators
has_traversal = ".." in filename or "/" in filename or "\\" in filename
@@ -275,18 +271,18 @@ class TestPathValidationSecurity:
"""Test that is_relative_to prevents directory traversal."""
base_dir = tmp_path / "workdir"
base_dir.mkdir()
# Create a file outside base_dir
outside_dir = tmp_path / "outside"
outside_dir.mkdir()
outside_file = outside_dir / "file.txt"
outside_file.write_text("test")
# Attempt to access file outside base_dir
try:
outside_resolved = outside_file.resolve()
base_resolved = base_dir.resolve()
# Should return False (file is not relative to base_dir)
is_safe = outside_resolved.is_relative_to(base_resolved)
assert not is_safe, "Path traversal not detected"
@@ -299,21 +295,21 @@ class TestPathValidationSecurity:
"""Test that resolve() handles symlink attacks."""
base_dir = tmp_path / "workdir"
base_dir.mkdir()
# Create target outside base_dir
outside_dir = tmp_path / "outside"
outside_dir.mkdir()
target_file = outside_dir / "secret.txt"
target_file.write_text("secret")
# Create symlink inside base_dir pointing outside
symlink_path = base_dir / "link.txt"
symlink_path.symlink_to(target_file)
# Resolve should give us the real path
resolved = symlink_path.resolve()
base_resolved = base_dir.resolve()
# The resolved path should NOT be relative to base_dir
try:
is_safe = resolved.is_relative_to(base_resolved)
@@ -326,17 +322,17 @@ class TestPathValidationSecurity:
"""Demonstrate why string-based path validation is insecure."""
base_dir = tmp_path / "workdir"
base_dir.mkdir()
# Create a similar-named directory
fake_dir = tmp_path / "workdir-fake"
fake_dir.mkdir()
fake_file = fake_dir / "file.txt"
fake_file.write_text("content")
# String-based check (insecure)
base_str = str(base_dir)
fake_str = str(fake_file)
# This would INCORRECTLY pass string.startswith() if not careful
# because "workdir-fake" starts with "workdir"
if base_str.endswith("/") or base_str.endswith("\\"):
@@ -345,7 +341,7 @@ class TestPathValidationSecurity:
else:
# Without separator, vulnerable to partial matches
string_check_unsafe = fake_str.startswith(base_str)
# Pathlib-based check (secure)
try:
pathlib_check = fake_file.resolve().is_relative_to(base_dir.resolve())
@@ -364,39 +360,50 @@ class TestFileUploadSecurity:
def test_ui_upload_uses_basename(self):
"""Test that ui_upload extracts basename to prevent path traversal."""
import os
from app.utils.filename_utils import sanitize_filename
# Simulate malicious filenames
malicious_filenames = [
"../../../etc/passwd",
"..\\..\\windows\\system32",
"/etc/shadow",
"folder/../file.pdf",
]
for malicious in malicious_filenames:
# os.path.basename should extract just the filename
basename = os.path.basename(malicious)
# Verify no path traversal remains in basename
assert ".." not in basename, f"Path traversal not removed: {malicious} -> {basename}"
assert "/" not in basename, f"Path separator not removed: {malicious} -> {basename}"
assert "\\" not in basename, f"Path separator not removed: {malicious} -> {basename}"
# Windows-style backslash paths: os.path.basename on Linux does NOT
# split on backslash, so the application also uses sanitize_filename
# to handle these. Verify the combined approach is safe.
windows_paths = [
"..\\..\\windows\\system32",
]
for malicious in windows_paths:
sanitized = sanitize_filename(os.path.basename(malicious))
assert ".." not in sanitized, f"Path traversal not removed after sanitize: {malicious} -> {sanitized}"
assert "\\" not in sanitized, f"Backslash not removed after sanitize: {malicious} -> {sanitized}"
def test_sanitize_after_basename(self):
"""Test that sanitization happens after basename extraction."""
from app.utils.filename_utils import sanitize_filename
import os
malicious = "../../../passwd.pdf"
# Step 1: Extract basename (as ui_upload does)
basename = os.path.basename(malicious)
assert basename == "passwd.pdf"
# Step 2: Sanitize (as ui_upload does)
sanitized = sanitize_filename(basename)
assert sanitized == "passwd.pdf"
# Final result is safe
assert ".." not in sanitized
assert "/" not in sanitized
@@ -410,11 +417,11 @@ class TestFileHashSecurity:
def test_hash_file_with_absolute_path_only(self, tmp_path):
"""Test that hash_file should only accept absolute paths."""
from app.utils.file_operations import hash_file
# Create a test file
test_file = tmp_path / "test.pdf"
test_file.write_bytes(b"test content")
# Should work with absolute path
result = hash_file(str(test_file))
assert isinstance(result, str)
@@ -423,7 +430,7 @@ class TestFileHashSecurity:
def test_hash_file_rejects_path_traversal(self):
"""Test that hash_file doesn't allow path traversal."""
from app.utils.file_operations import hash_file
# Attempt to hash a file using path traversal
# This should fail because the file doesn't exist
with pytest.raises(FileNotFoundError):
@@ -440,25 +447,25 @@ class TestEndToEndPathTraversal:
from app.utils.filename_utils import sanitize_filename
import os
import uuid
# Simulate ui_upload flow
malicious_upload_filename = "../../../etc/passwd"
# Step 1: Extract basename
base_filename = os.path.basename(malicious_upload_filename)
assert base_filename == "passwd"
# Step 2: Sanitize
safe_filename = sanitize_filename(base_filename)
assert safe_filename == "passwd"
# Step 3: Add UUID (as ui_upload does)
unique_id = str(uuid.uuid4())
target_filename = f"{unique_id}.{safe_filename}"
# Step 4: Join with workdir
target_path = os.path.join(str(tmp_path), target_filename)
# Verify final path is safe
final_path = Path(target_path)
assert final_path.parent == tmp_path
@@ -469,27 +476,27 @@ class TestEndToEndPathTraversal:
"""Test metadata embedding flow prevents path traversal."""
from app.utils.filename_utils import sanitize_filename
import os
# Simulate GPT returning malicious filename
gpt_metadata = {
"filename": "../../../etc/shadow",
"document_type": "Invoice",
}
# Step 1: Extract filename from metadata
suggested_filename = gpt_metadata.get("filename", "fallback")
# Step 2: Sanitize (as embed_metadata_into_pdf should do)
suggested_filename = sanitize_filename(suggested_filename)
# Step 3: Remove extension
suggested_filename = os.path.splitext(suggested_filename)[0]
# Step 4: Build final path
processed_dir = tmp_path / "processed"
processed_dir.mkdir()
final_path = os.path.join(str(processed_dir), f"{suggested_filename}.pdf")
# Verify final path is safe
result_path = Path(final_path)
assert result_path.parent == processed_dir
tests/test_rate_limiting.py
-4
View File
@@ -32,7 +32,6 @@ def test_rate_limit_configuration():
assert hasattr(settings, "rate_limiting_enabled")
assert hasattr(settings, "rate_limit_default")
assert hasattr(settings, "rate_limit_upload")
assert hasattr(settings, "rate_limit_process")
assert hasattr(settings, "rate_limit_auth")
# Verify that settings are strings in correct format
@@ -40,8 +39,6 @@ def test_rate_limit_configuration():
assert "/" in settings.rate_limit_default # Should be like "100/minute"
assert isinstance(settings.rate_limit_upload, str)
assert "/" in settings.rate_limit_upload
assert isinstance(settings.rate_limit_process, str)
assert "/" in settings.rate_limit_process
assert isinstance(settings.rate_limit_auth, str)
assert "/" in settings.rate_limit_auth
@@ -224,7 +221,6 @@ def test_rate_limit_format_validation():
assert validate_rate_limit(settings.rate_limit_default)
assert validate_rate_limit(settings.rate_limit_upload)
assert validate_rate_limit(settings.rate_limit_process)
assert validate_rate_limit(settings.rate_limit_auth)
tests/test_upload_tasks.py
+144 -117
View File
@@ -13,13 +13,16 @@ 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
@pytest.fixture
def mock_settings():
"""Mock settings for upload tests."""
with patch("app.tasks.upload_to_onedrive.settings") as onedrive_settings, patch(
"app.tasks.upload_to_s3.settings"
) as s3_settings:
with (
patch("app.tasks.upload_to_onedrive.settings") as onedrive_settings,
patch("app.tasks.upload_to_s3.settings") as s3_settings,
):
# OneDrive settings
onedrive_settings.onedrive_client_id = "test_client_id"
onedrive_settings.onedrive_client_secret = "test_secret"
@@ -42,10 +45,11 @@ def mock_settings():
@pytest.mark.unit
def test_upload_to_onedrive_accepts_file_id(sample_text_file, mock_settings):
"""Test that upload_to_onedrive accepts file_id parameter."""
with patch("app.tasks.upload_to_onedrive.get_onedrive_token") as mock_token, patch(
"app.tasks.upload_to_onedrive.create_upload_session"
) as mock_session, patch("app.tasks.upload_to_onedrive.upload_large_file") as mock_upload, patch(
"app.tasks.upload_to_onedrive.log_task_progress"
with (
patch("app.tasks.upload_to_onedrive.get_onedrive_token") as mock_token,
patch("app.tasks.upload_to_onedrive.create_upload_session") as mock_session,
patch("app.tasks.upload_to_onedrive.upload_large_file") as mock_upload,
patch("app.tasks.upload_to_onedrive.log_task_progress"),
):
# Setup mocks
@@ -65,10 +69,11 @@ def test_upload_to_onedrive_accepts_file_id(sample_text_file, mock_settings):
@pytest.mark.unit
def test_upload_to_onedrive_without_file_id(sample_text_file, mock_settings):
"""Test that upload_to_onedrive works without file_id parameter."""
with patch("app.tasks.upload_to_onedrive.get_onedrive_token") as mock_token, patch(
"app.tasks.upload_to_onedrive.create_upload_session"
) as mock_session, patch("app.tasks.upload_to_onedrive.upload_large_file") as mock_upload, patch(
"app.tasks.upload_to_onedrive.log_task_progress"
with (
patch("app.tasks.upload_to_onedrive.get_onedrive_token") as mock_token,
patch("app.tasks.upload_to_onedrive.create_upload_session") as mock_session,
patch("app.tasks.upload_to_onedrive.upload_large_file") as mock_upload,
patch("app.tasks.upload_to_onedrive.log_task_progress"),
):
# Setup mocks
@@ -86,8 +91,9 @@ def test_upload_to_onedrive_without_file_id(sample_text_file, mock_settings):
@pytest.mark.unit
def test_upload_to_s3_accepts_file_id(sample_text_file, mock_settings):
"""Test that upload_to_s3 accepts file_id parameter."""
with patch("app.tasks.upload_to_s3.boto3.client") as mock_boto_client, patch(
"app.tasks.upload_to_s3.log_task_progress"
with (
patch("app.tasks.upload_to_s3.boto3.client") as mock_boto_client,
patch("app.tasks.upload_to_s3.log_task_progress"),
):
# Setup mock S3 client
@@ -107,8 +113,9 @@ def test_upload_to_s3_accepts_file_id(sample_text_file, mock_settings):
@pytest.mark.unit
def test_upload_to_s3_without_file_id(sample_text_file, mock_settings):
"""Test that upload_to_s3 works without file_id parameter."""
with patch("app.tasks.upload_to_s3.boto3.client") as mock_boto_client, patch(
"app.tasks.upload_to_s3.log_task_progress"
with (
patch("app.tasks.upload_to_s3.boto3.client") as mock_boto_client,
patch("app.tasks.upload_to_s3.log_task_progress"),
):
# Setup mock S3 client
@@ -142,11 +149,12 @@ def test_upload_to_s3_file_not_found(mock_settings):
@pytest.mark.unit
def test_upload_to_onedrive_logs_with_file_id(sample_text_file, mock_settings):
"""Test that upload_to_onedrive properly logs with file_id."""
with patch("app.tasks.upload_to_onedrive.get_onedrive_token") as mock_token, patch(
"app.tasks.upload_to_onedrive.create_upload_session"
) as mock_session, patch("app.tasks.upload_to_onedrive.upload_large_file") as mock_upload, patch(
"app.tasks.upload_to_onedrive.log_task_progress"
) as mock_log:
with (
patch("app.tasks.upload_to_onedrive.get_onedrive_token") as mock_token,
patch("app.tasks.upload_to_onedrive.create_upload_session") as mock_session,
patch("app.tasks.upload_to_onedrive.upload_large_file") as mock_upload,
patch("app.tasks.upload_to_onedrive.log_task_progress") as mock_log,
):
# Setup mocks
mock_token.return_value = "test_access_token"
@@ -168,9 +176,10 @@ def test_upload_to_onedrive_logs_with_file_id(sample_text_file, mock_settings):
@pytest.mark.unit
def test_upload_to_s3_logs_with_file_id(sample_text_file, mock_settings):
"""Test that upload_to_s3 properly logs with file_id."""
with patch("app.tasks.upload_to_s3.boto3.client") as mock_boto_client, patch(
"app.tasks.upload_to_s3.log_task_progress"
) as mock_log:
with (
patch("app.tasks.upload_to_s3.boto3.client") as mock_boto_client,
patch("app.tasks.upload_to_s3.log_task_progress") as mock_log,
):
# Setup mock S3 client
mock_s3 = Mock()
@@ -191,29 +200,32 @@ def test_upload_to_s3_logs_with_file_id(sample_text_file, mock_settings):
# Tests for newly standardized upload tasks
@pytest.mark.unit
def test_upload_to_ftp_accepts_file_id(sample_text_file):
"""Test that upload_to_ftp accepts file_id parameter."""
with patch("app.tasks.upload_to_ftp.settings") as mock_settings, \
patch("app.tasks.upload_to_ftp.ftplib.FTP") as mock_ftp, \
patch("app.tasks.upload_to_ftp.log_task_progress"):
with (
patch("app.tasks.upload_to_ftp.settings") as mock_settings,
patch("app.tasks.upload_to_ftp.ftplib.FTP") as mock_ftp,
patch("app.tasks.upload_to_ftp.log_task_progress"),
):
# Setup settings
mock_settings.ftp_host = "ftp.example.com"
mock_settings.ftp_port = 21
mock_settings.ftp_username = "test_user"
mock_settings.ftp_password = "test_pass"
mock_settings.ftp_password = _TEST_CREDENTIAL
mock_settings.ftp_folder = "uploads"
mock_settings.ftp_use_tls = False
mock_settings.ftp_allow_plaintext = True
# Setup mock FTP
mock_ftp_instance = Mock()
mock_ftp.return_value = mock_ftp_instance
# Call with file_id parameter
result = upload_to_ftp.apply(args=[sample_text_file], kwargs={"file_id": 100}).get()
assert result["status"] == "Completed"
assert result["file"] == sample_text_file
assert result["ftp_host"] == "ftp.example.com"
@@ -222,45 +234,49 @@ def test_upload_to_ftp_accepts_file_id(sample_text_file):
@pytest.mark.unit
def test_upload_to_ftp_without_file_id(sample_text_file):
"""Test that upload_to_ftp works without file_id parameter."""
with patch("app.tasks.upload_to_ftp.settings") as mock_settings, \
patch("app.tasks.upload_to_ftp.ftplib.FTP") as mock_ftp, \
patch("app.tasks.upload_to_ftp.log_task_progress"):
with (
patch("app.tasks.upload_to_ftp.settings") as mock_settings,
patch("app.tasks.upload_to_ftp.ftplib.FTP") as mock_ftp,
patch("app.tasks.upload_to_ftp.log_task_progress"),
):
# Setup settings
mock_settings.ftp_host = "ftp.example.com"
mock_settings.ftp_username = "test_user"
mock_settings.ftp_password = "test_pass"
mock_settings.ftp_password = _TEST_CREDENTIAL
mock_settings.ftp_folder = None
mock_settings.ftp_use_tls = False
mock_settings.ftp_allow_plaintext = True
# Setup mock FTP
mock_ftp_instance = Mock()
mock_ftp.return_value = mock_ftp_instance
# Call without file_id parameter
result = upload_to_ftp.apply(args=[sample_text_file]).get()
assert result["status"] == "Completed"
@pytest.mark.unit
def test_upload_to_sftp_accepts_file_id(sample_text_file):
"""Test that upload_to_sftp accepts file_id parameter."""
with patch("app.tasks.upload_to_sftp.settings") as mock_settings, \
patch("app.tasks.upload_to_sftp.paramiko.SSHClient") as mock_ssh, \
patch("app.tasks.upload_to_sftp.log_task_progress"), \
patch("app.tasks.upload_to_sftp.extract_remote_path") as mock_extract, \
patch("app.tasks.upload_to_sftp.get_unique_filename") as mock_unique:
with (
patch("app.tasks.upload_to_sftp.settings") as mock_settings,
patch("app.tasks.upload_to_sftp.paramiko.SSHClient") as mock_ssh,
patch("app.tasks.upload_to_sftp.log_task_progress"),
patch("app.tasks.upload_to_sftp.extract_remote_path") as mock_extract,
patch("app.tasks.upload_to_sftp.get_unique_filename") as mock_unique,
):
# Setup settings
mock_settings.sftp_host = "sftp.example.com"
mock_settings.sftp_port = 22
mock_settings.sftp_username = "test_user"
mock_settings.sftp_password = "test_pass"
mock_settings.sftp_password = _TEST_CREDENTIAL
mock_settings.sftp_folder = "/uploads"
mock_settings.workdir = "/tmp"
# Setup mocks
mock_ssh_instance = Mock()
mock_sftp = Mock()
@@ -268,10 +284,10 @@ def test_upload_to_sftp_accepts_file_id(sample_text_file):
mock_ssh_instance.open_sftp.return_value = mock_sftp
mock_extract.return_value = "/uploads/test.txt"
mock_unique.return_value = "/uploads/test.txt"
# Call with file_id parameter
result = upload_to_sftp.apply(args=[sample_text_file], kwargs={"file_id": 200}).get()
assert result["status"] == "Completed"
assert result["file_path"] == sample_text_file
assert "sftp_path" in result
@@ -280,25 +296,27 @@ def test_upload_to_sftp_accepts_file_id(sample_text_file):
@pytest.mark.unit
def test_upload_to_webdav_accepts_file_id(sample_text_file):
"""Test that upload_to_webdav accepts file_id parameter."""
with patch("app.tasks.upload_to_webdav.settings") as mock_settings, \
patch("app.tasks.upload_to_webdav.requests.put") as mock_put, \
patch("app.tasks.upload_to_webdav.log_task_progress"):
with (
patch("app.tasks.upload_to_webdav.settings") as mock_settings,
patch("app.tasks.upload_to_webdav.requests.put") as mock_put,
patch("app.tasks.upload_to_webdav.log_task_progress"),
):
# Setup settings
mock_settings.webdav_url = "https://webdav.example.com/"
mock_settings.webdav_username = "test_user"
mock_settings.webdav_password = "test_pass"
mock_settings.webdav_password = _TEST_CREDENTIAL
mock_settings.webdav_folder = "uploads"
mock_settings.webdav_verify_ssl = True
# Setup mock response
mock_response = Mock()
mock_response.status_code = 201
mock_put.return_value = mock_response
# Call with file_id parameter
result = upload_to_webdav.apply(args=[sample_text_file], kwargs={"file_id": 300}).get()
assert result["status"] == "Completed"
assert result["file"] == sample_text_file
assert "url" in result
@@ -307,20 +325,22 @@ def test_upload_to_webdav_accepts_file_id(sample_text_file):
@pytest.mark.unit
def test_upload_to_google_drive_accepts_file_id(sample_text_file):
"""Test that upload_to_google_drive accepts file_id parameter."""
with patch("app.tasks.upload_to_google_drive.get_google_drive_service") as mock_service, \
patch("app.tasks.upload_to_google_drive.MediaFileUpload") as mock_media, \
patch("app.tasks.upload_to_google_drive.extract_metadata_from_file") as mock_metadata, \
patch("app.tasks.upload_to_google_drive.settings") as mock_settings, \
patch("app.tasks.upload_to_google_drive.log_task_progress"):
with (
patch("app.tasks.upload_to_google_drive.get_google_drive_service") as mock_service,
patch("app.tasks.upload_to_google_drive.MediaFileUpload"),
patch("app.tasks.upload_to_google_drive.extract_metadata_from_file") as mock_metadata,
patch("app.tasks.upload_to_google_drive.settings") as mock_settings,
patch("app.tasks.upload_to_google_drive.log_task_progress"),
):
# Setup settings
mock_settings.google_drive_folder_id = "test_folder_id"
# Setup mocks
mock_drive_service = Mock()
mock_service.return_value = mock_drive_service
mock_metadata.return_value = {}
mock_files = Mock()
mock_drive_service.files.return_value = mock_files
mock_create = Mock()
@@ -328,12 +348,12 @@ def test_upload_to_google_drive_accepts_file_id(sample_text_file):
mock_create.execute.return_value = {
"id": "file123",
"name": "test.txt",
"webViewLink": "https://drive.google.com/file/d/file123"
"webViewLink": "https://drive.google.com/file/d/file123",
}
# Call with file_id parameter
result = upload_to_google_drive.apply(args=[sample_text_file], kwargs={"file_id": 400}).get()
assert result["status"] == "Completed"
assert result["file_path"] == sample_text_file
assert "google_drive_file_id" in result
@@ -342,37 +362,39 @@ def test_upload_to_google_drive_accepts_file_id(sample_text_file):
@pytest.mark.unit
def test_upload_to_email_accepts_file_id(sample_text_file):
"""Test that upload_to_email accepts file_id parameter."""
with patch("app.tasks.upload_to_email.settings") as mock_settings, \
patch("app.tasks.upload_to_email.smtplib.SMTP") as mock_smtp, \
patch("app.tasks.upload_to_email.get_email_template") as mock_template, \
patch("app.tasks.upload_to_email.extract_metadata_from_file") as mock_metadata, \
patch("app.tasks.upload_to_email.log_task_progress"), \
patch("app.tasks.upload_to_email._prepare_recipients") as mock_recipients, \
patch("app.tasks.upload_to_email._send_email_with_smtp") as mock_send, \
patch("app.tasks.upload_to_email.attach_logo") as mock_logo:
with (
patch("app.tasks.upload_to_email.settings") as mock_settings,
patch("app.tasks.upload_to_email.smtplib.SMTP"),
patch("app.tasks.upload_to_email.get_email_template") as mock_template,
patch("app.tasks.upload_to_email.extract_metadata_from_file") as mock_metadata,
patch("app.tasks.upload_to_email.log_task_progress"),
patch("app.tasks.upload_to_email._prepare_recipients") as mock_recipients,
patch("app.tasks.upload_to_email._send_email_with_smtp") as mock_send,
patch("app.tasks.upload_to_email.attach_logo") as mock_logo,
):
# Setup settings
mock_settings.email_host = "smtp.example.com"
mock_settings.email_port = 587
mock_settings.email_username = "test@example.com"
mock_settings.email_password = "test_pass"
mock_settings.email_password = _TEST_CREDENTIAL
mock_settings.email_use_tls = True
mock_settings.email_sender = "sender@example.com"
mock_settings.external_hostname = "docuelevate.example.com"
# Setup mocks
mock_recipients.return_value = (["recipient@example.com"], None)
mock_send.return_value = None
mock_metadata.return_value = {}
mock_logo.return_value = False
mock_template_obj = Mock()
mock_template_obj.render.return_value = "<html>Test email</html>"
mock_template.return_value = mock_template_obj
# Call with file_id parameter
result = upload_to_email.apply(args=[sample_text_file], kwargs={"file_id": 500}).get()
assert result["status"] == "Completed"
assert result["file"] == sample_text_file
assert "recipients" in result
@@ -381,11 +403,10 @@ def test_upload_to_email_accepts_file_id(sample_text_file):
@pytest.mark.unit
def test_upload_to_ftp_file_not_found():
"""Test that upload_to_ftp raises error for missing file."""
with patch("app.tasks.upload_to_ftp.settings") as mock_settings, \
patch("app.tasks.upload_to_ftp.log_task_progress"):
with patch("app.tasks.upload_to_ftp.settings") as mock_settings, patch("app.tasks.upload_to_ftp.log_task_progress"):
mock_settings.ftp_host = "ftp.example.com"
with pytest.raises(FileNotFoundError):
upload_to_ftp.apply(args=["/nonexistent/file.pdf"], kwargs={"file_id": 1}).get()
@@ -393,13 +414,15 @@ def test_upload_to_ftp_file_not_found():
@pytest.mark.unit
def test_upload_to_sftp_file_not_found():
"""Test that upload_to_sftp raises error for missing file."""
with patch("app.tasks.upload_to_sftp.settings") as mock_settings, \
patch("app.tasks.upload_to_sftp.log_task_progress"):
with (
patch("app.tasks.upload_to_sftp.settings") as mock_settings,
patch("app.tasks.upload_to_sftp.log_task_progress"),
):
mock_settings.sftp_host = "sftp.example.com"
mock_settings.sftp_port = 22
mock_settings.sftp_username = "test_user"
with pytest.raises(FileNotFoundError):
upload_to_sftp.apply(args=["/nonexistent/file.pdf"], kwargs={"file_id": 1}).get()
@@ -407,11 +430,13 @@ def test_upload_to_sftp_file_not_found():
@pytest.mark.unit
def test_upload_to_webdav_file_not_found():
"""Test that upload_to_webdav raises error for missing file."""
with patch("app.tasks.upload_to_webdav.settings") as mock_settings, \
patch("app.tasks.upload_to_webdav.log_task_progress"):
with (
patch("app.tasks.upload_to_webdav.settings") as mock_settings,
patch("app.tasks.upload_to_webdav.log_task_progress"),
):
mock_settings.webdav_url = "https://webdav.example.com/"
with pytest.raises(FileNotFoundError):
upload_to_webdav.apply(args=["/nonexistent/file.pdf"], kwargs={"file_id": 1}).get()
@@ -419,15 +444,15 @@ def test_upload_to_webdav_file_not_found():
@pytest.mark.unit
def test_all_upload_tasks_have_consistent_signature(sample_text_file):
"""Test that all upload tasks accept file_id as a keyword parameter.
This test verifies that all upload tasks can be called with the same signature
as used in send_to_all.py: task.delay(file_path, file_id=file_id)
Note: We use task.run to inspect the actual function signature because
Celery tasks wrap the original function, and .run provides access to
the unwrapped callable's signature.
"""
upload_tasks = [
(upload_to_s3, "app.tasks.upload_to_s3"),
(upload_to_ftp, "app.tasks.upload_to_ftp"),
@@ -436,19 +461,19 @@ def test_all_upload_tasks_have_consistent_signature(sample_text_file):
(upload_to_google_drive, "app.tasks.upload_to_google_drive"),
(upload_to_email, "app.tasks.upload_to_email"),
]
import inspect
for task, module_path in upload_tasks:
# Use task.run to inspect the actual wrapped function's signature
# This is necessary because Celery's task decorator wraps the original function
sig = inspect.signature(task.run)
params = list(sig.parameters.keys())
# Should have at least file_path and file_id parameters
assert "file_path" in params, f"{task.name} missing file_path parameter"
assert "file_id" in params, f"{task.name} missing file_id parameter"
# file_id should have a default value (None)
assert sig.parameters["file_id"].default is None, f"{task.name} file_id should default to None"
@@ -456,29 +481,31 @@ def test_all_upload_tasks_have_consistent_signature(sample_text_file):
@pytest.mark.unit
def test_send_to_all_calls_upload_tasks_with_keyword_argument():
"""Test that send_to_all_destinations calls upload tasks with file_id as keyword argument.
Regression test for issue: upload_to_s3() takes 1 positional argument but 2 were given.
This ensures that file_id is always passed as a keyword argument, not positional.
"""
from app.tasks.send_to_all import send_to_all_destinations
test_file = "/tmp/test_file.pdf"
# Create the test file
with open(test_file, "w") as f:
f.write("test content")
try:
# Mock all the upload functions and settings
with patch("app.tasks.send_to_all.upload_to_s3") as mock_s3, \
patch("app.tasks.send_to_all.settings") as mock_settings, \
patch("app.tasks.send_to_all.log_task_progress"), \
patch("app.tasks.send_to_all.SessionLocal"), \
patch("app.tasks.send_to_all.get_configured_services_from_validator") as mock_validator:
with (
patch("app.tasks.send_to_all.upload_to_s3") as mock_s3,
patch("app.tasks.send_to_all.settings") as mock_settings,
patch("app.tasks.send_to_all.log_task_progress"),
patch("app.tasks.send_to_all.SessionLocal"),
patch("app.tasks.send_to_all.get_configured_services_from_validator") as mock_validator,
):
# Configure validator to return S3 as configured
mock_validator.return_value = {"s3": True}
# Configure settings to enable only S3
mock_settings.s3_bucket_name = "test-bucket"
mock_settings.aws_access_key_id = "test-key"
@@ -493,26 +520,26 @@ def test_send_to_all_calls_upload_tasks_with_keyword_argument():
mock_settings.email_host = None
mock_settings.onedrive_client_id = None
mock_settings.workdir = "/tmp"
# Mock the delay method to track how it's called
mock_s3_task = Mock()
mock_s3_task.id = "test-task-id"
mock_s3.delay.return_value = mock_s3_task
# Call send_to_all_destinations with file_id
result = send_to_all_destinations.apply(args=[test_file], kwargs={"file_id": 123}).get()
# Verify that upload_to_s3.delay was called with file_id as keyword argument
mock_s3.delay.assert_called_once()
call_args, call_kwargs = mock_s3.delay.call_args
# The call should be: delay(file_path, file_id=file_id)
# So we expect 1 positional arg (file_path) and file_id in kwargs
assert len(call_args) == 1, "Should have exactly 1 positional argument (file_path)"
assert call_args[0] == test_file, "First positional arg should be file_path"
assert "file_id" in call_kwargs, "file_id should be passed as keyword argument"
assert call_kwargs["file_id"] == 123, "file_id value should be correct"
finally:
# Clean up test file
if os.path.exists(test_file):
tests/test_upload_webdav_comprehensive.py
+211 -177
View File
@@ -1,4 +1,5 @@
"""Comprehensive tests for upload_to_webdav task."""
import os
import pytest
from unittest.mock import patch, Mock, MagicMock
@@ -6,6 +7,9 @@ from requests.exceptions import ConnectionError, Timeout, RequestException
from app.tasks.upload_to_webdav import upload_to_webdav
_TEST_CREDENTIAL = "test_pass" # noqa: S105
_TEST_CUSTOM_CREDENTIAL = "custom_password123" # noqa: S105
@pytest.mark.unit
class TestUploadToWebDAV:
@@ -13,233 +17,252 @@ class TestUploadToWebDAV:
def test_upload_success_with_file_id(self, sample_text_file):
"""Test successful upload with file_id parameter."""
with patch("app.tasks.upload_to_webdav.settings") as mock_settings, \
patch("app.tasks.upload_to_webdav.requests.put") as mock_put, \
patch("app.tasks.upload_to_webdav.log_task_progress") as mock_log:
with (
patch("app.tasks.upload_to_webdav.settings") as mock_settings,
patch("app.tasks.upload_to_webdav.requests.put") as mock_put,
patch("app.tasks.upload_to_webdav.log_task_progress") as mock_log,
):
# Setup settings
mock_settings.webdav_url = "https://webdav.example.com/"
mock_settings.webdav_username = "test_user"
mock_settings.webdav_password = "test_pass"
mock_settings.webdav_password = _TEST_CREDENTIAL
mock_settings.webdav_folder = "uploads"
mock_settings.webdav_verify_ssl = True
mock_settings.http_request_timeout = 30
# Setup mock response - 201 Created
mock_response = Mock()
mock_response.status_code = 201
mock_put.return_value = mock_response
# Execute upload
result = upload_to_webdav.apply(args=[sample_text_file], kwargs={"file_id": 100}).get()
# Verify result
assert result["status"] == "Completed"
assert result["file"] == sample_text_file
assert "url" in result
assert "webdav.example.com" in result["url"]
# Verify requests.put was called correctly
assert mock_put.called
call_args = mock_put.call_args
assert call_args[1]["auth"] == ("test_user", "test_pass")
assert call_args[1]["auth"] == ("test_user", _TEST_CREDENTIAL)
assert call_args[1]["verify"] is True
assert call_args[1]["timeout"] == 30
# Verify logging was called with file_id
assert mock_log.called
log_calls_with_file_id = [
call for call in mock_log.call_args_list
if call[1].get("file_id") == 100
]
log_calls_with_file_id = [call for call in mock_log.call_args_list if call[1].get("file_id") == 100]
assert len(log_calls_with_file_id) > 0
def test_upload_success_without_file_id(self, sample_text_file):
"""Test successful upload without file_id parameter."""
with patch("app.tasks.upload_to_webdav.settings") as mock_settings, \
patch("app.tasks.upload_to_webdav.requests.put") as mock_put, \
patch("app.tasks.upload_to_webdav.log_task_progress"):
with (
patch("app.tasks.upload_to_webdav.settings") as mock_settings,
patch("app.tasks.upload_to_webdav.requests.put") as mock_put,
patch("app.tasks.upload_to_webdav.log_task_progress"),
):
mock_settings.webdav_url = "https://webdav.example.com/"
mock_settings.webdav_username = "test_user"
mock_settings.webdav_password = "test_pass"
mock_settings.webdav_password = _TEST_CREDENTIAL
mock_settings.webdav_folder = ""
mock_settings.webdav_verify_ssl = True
mock_settings.http_request_timeout = 30
mock_response = Mock()
mock_response.status_code = 200 # 200 OK is also valid
mock_put.return_value = mock_response
result = upload_to_webdav.apply(args=[sample_text_file]).get()
assert result["status"] == "Completed"
assert result["file"] == sample_text_file
def test_upload_success_status_204(self, sample_text_file):
"""Test successful upload with 204 No Content status."""
with patch("app.tasks.upload_to_webdav.settings") as mock_settings, \
patch("app.tasks.upload_to_webdav.requests.put") as mock_put, \
patch("app.tasks.upload_to_webdav.log_task_progress"):
with (
patch("app.tasks.upload_to_webdav.settings") as mock_settings,
patch("app.tasks.upload_to_webdav.requests.put") as mock_put,
patch("app.tasks.upload_to_webdav.log_task_progress"),
):
mock_settings.webdav_url = "https://webdav.example.com/"
mock_settings.webdav_username = "test_user"
mock_settings.webdav_password = "test_pass"
mock_settings.webdav_password = _TEST_CREDENTIAL
mock_settings.webdav_folder = None
mock_settings.webdav_verify_ssl = True
mock_settings.http_request_timeout = 30
mock_response = Mock()
mock_response.status_code = 204 # 204 No Content
mock_put.return_value = mock_response
result = upload_to_webdav.apply(args=[sample_text_file]).get()
assert result["status"] == "Completed"
def test_missing_webdav_url(self, sample_text_file):
"""Test that missing WebDAV URL raises ValueError."""
with patch("app.tasks.upload_to_webdav.settings") as mock_settings, \
patch("app.tasks.upload_to_webdav.log_task_progress"):
with (
patch("app.tasks.upload_to_webdav.settings") as mock_settings,
patch("app.tasks.upload_to_webdav.log_task_progress"),
):
mock_settings.webdav_url = None
mock_settings.webdav_username = "test_user"
mock_settings.webdav_password = "test_pass"
mock_settings.webdav_password = _TEST_CREDENTIAL
with pytest.raises(ValueError, match="WebDAV URL is not configured"):
upload_to_webdav.apply(args=[sample_text_file]).get()
def test_file_not_found(self):
"""Test that missing file raises FileNotFoundError."""
with patch("app.tasks.upload_to_webdav.settings") as mock_settings, \
patch("app.tasks.upload_to_webdav.log_task_progress"):
with (
patch("app.tasks.upload_to_webdav.settings") as mock_settings,
patch("app.tasks.upload_to_webdav.log_task_progress"),
):
mock_settings.webdav_url = "https://webdav.example.com/"
with pytest.raises(FileNotFoundError, match="File not found"):
upload_to_webdav.apply(args=["/nonexistent/file.pdf"]).get()
def test_http_error_response(self, sample_text_file):
"""Test handling of HTTP error responses (4xx, 5xx)."""
with patch("app.tasks.upload_to_webdav.settings") as mock_settings, \
patch("app.tasks.upload_to_webdav.requests.put") as mock_put, \
patch("app.tasks.upload_to_webdav.log_task_progress"):
with (
patch("app.tasks.upload_to_webdav.settings") as mock_settings,
patch("app.tasks.upload_to_webdav.requests.put") as mock_put,
patch("app.tasks.upload_to_webdav.log_task_progress"),
):
mock_settings.webdav_url = "https://webdav.example.com/"
mock_settings.webdav_username = "test_user"
mock_settings.webdav_password = "test_pass"
mock_settings.webdav_password = _TEST_CREDENTIAL
mock_settings.webdav_folder = "uploads"
mock_settings.webdav_verify_ssl = True
mock_settings.http_request_timeout = 30
# Simulate 401 Unauthorized
mock_response = Mock()
mock_response.status_code = 401
mock_response.text = "Unauthorized"
mock_put.return_value = mock_response
with pytest.raises(Exception, match="Failed to upload.*401"):
upload_to_webdav.apply(args=[sample_text_file]).get()
def test_http_404_not_found(self, sample_text_file):
"""Test handling of 404 Not Found response."""
with patch("app.tasks.upload_to_webdav.settings") as mock_settings, \
patch("app.tasks.upload_to_webdav.requests.put") as mock_put, \
patch("app.tasks.upload_to_webdav.log_task_progress"):
with (
patch("app.tasks.upload_to_webdav.settings") as mock_settings,
patch("app.tasks.upload_to_webdav.requests.put") as mock_put,
patch("app.tasks.upload_to_webdav.log_task_progress"),
):
mock_settings.webdav_url = "https://webdav.example.com/"
mock_settings.webdav_username = "test_user"
mock_settings.webdav_password = "test_pass"
mock_settings.webdav_password = _TEST_CREDENTIAL
mock_settings.webdav_folder = "nonexistent"
mock_settings.webdav_verify_ssl = True
mock_settings.http_request_timeout = 30
mock_response = Mock()
mock_response.status_code = 404
mock_response.text = "Not Found"
mock_put.return_value = mock_response
with pytest.raises(Exception, match="Failed to upload.*404"):
upload_to_webdav.apply(args=[sample_text_file]).get()
def test_http_500_server_error(self, sample_text_file):
"""Test handling of 500 Internal Server Error."""
with patch("app.tasks.upload_to_webdav.settings") as mock_settings, \
patch("app.tasks.upload_to_webdav.requests.put") as mock_put, \
patch("app.tasks.upload_to_webdav.log_task_progress"):
with (
patch("app.tasks.upload_to_webdav.settings") as mock_settings,
patch("app.tasks.upload_to_webdav.requests.put") as mock_put,
patch("app.tasks.upload_to_webdav.log_task_progress"),
):
mock_settings.webdav_url = "https://webdav.example.com/"
mock_settings.webdav_username = "test_user"
mock_settings.webdav_password = "test_pass"
mock_settings.webdav_password = _TEST_CREDENTIAL
mock_settings.webdav_folder = "uploads"
mock_settings.webdav_verify_ssl = True
mock_settings.http_request_timeout = 30
mock_response = Mock()
mock_response.status_code = 500
mock_response.text = "Internal Server Error"
mock_put.return_value = mock_response
with pytest.raises(Exception, match="Failed to upload.*500"):
upload_to_webdav.apply(args=[sample_text_file]).get()
def test_connection_error(self, sample_text_file):
"""Test handling of connection errors."""
with patch("app.tasks.upload_to_webdav.settings") as mock_settings, \
patch("app.tasks.upload_to_webdav.requests.put") as mock_put, \
patch("app.tasks.upload_to_webdav.log_task_progress"):
with (
patch("app.tasks.upload_to_webdav.settings") as mock_settings,
patch("app.tasks.upload_to_webdav.requests.put") as mock_put,
patch("app.tasks.upload_to_webdav.log_task_progress"),
):
mock_settings.webdav_url = "https://webdav.example.com/"
mock_settings.webdav_username = "test_user"
mock_settings.webdav_password = "test_pass"
mock_settings.webdav_password = _TEST_CREDENTIAL
mock_settings.webdav_folder = "uploads"
mock_settings.webdav_verify_ssl = True
mock_settings.http_request_timeout = 30
# Simulate connection error
mock_put.side_effect = ConnectionError("Connection refused")
with pytest.raises(Exception, match="Error uploading.*Connection refused"):
upload_to_webdav.apply(args=[sample_text_file]).get()
def test_timeout_error(self, sample_text_file):
"""Test handling of timeout errors."""
with patch("app.tasks.upload_to_webdav.settings") as mock_settings, \
patch("app.tasks.upload_to_webdav.requests.put") as mock_put, \
patch("app.tasks.upload_to_webdav.log_task_progress"):
with (
patch("app.tasks.upload_to_webdav.settings") as mock_settings,
patch("app.tasks.upload_to_webdav.requests.put") as mock_put,
patch("app.tasks.upload_to_webdav.log_task_progress"),
):
mock_settings.webdav_url = "https://webdav.example.com/"
mock_settings.webdav_username = "test_user"
mock_settings.webdav_password = "test_pass"
mock_settings.webdav_password = _TEST_CREDENTIAL
mock_settings.webdav_folder = "uploads"
mock_settings.webdav_verify_ssl = True
mock_settings.http_request_timeout = 30
# Simulate timeout
mock_put.side_effect = Timeout("Request timed out")
with pytest.raises(Exception, match="Error uploading.*timed out"):
upload_to_webdav.apply(args=[sample_text_file]).get()
def test_url_construction_with_trailing_slash(self, sample_text_file):
"""Test URL construction when base URL has trailing slash."""
with patch("app.tasks.upload_to_webdav.settings") as mock_settings, \
patch("app.tasks.upload_to_webdav.requests.put") as mock_put, \
patch("app.tasks.upload_to_webdav.log_task_progress"):
with (
patch("app.tasks.upload_to_webdav.settings") as mock_settings,
patch("app.tasks.upload_to_webdav.requests.put") as mock_put,
patch("app.tasks.upload_to_webdav.log_task_progress"),
):
mock_settings.webdav_url = "https://webdav.example.com/"
mock_settings.webdav_username = "test_user"
mock_settings.webdav_password = "test_pass"
mock_settings.webdav_password = _TEST_CREDENTIAL
mock_settings.webdav_folder = "uploads"
mock_settings.webdav_verify_ssl = True
mock_settings.http_request_timeout = 30
mock_response = Mock()
mock_response.status_code = 201
mock_put.return_value = mock_response
result = upload_to_webdav.apply(args=[sample_text_file]).get()
# Verify URL construction
called_url = mock_put.call_args[0][0]
assert called_url.startswith("https://webdav.example.com/")
@@ -248,23 +271,25 @@ class TestUploadToWebDAV:
def test_url_construction_without_trailing_slash(self, sample_text_file):
"""Test URL construction when base URL has no trailing slash."""
with patch("app.tasks.upload_to_webdav.settings") as mock_settings, \
patch("app.tasks.upload_to_webdav.requests.put") as mock_put, \
patch("app.tasks.upload_to_webdav.log_task_progress"):
with (
patch("app.tasks.upload_to_webdav.settings") as mock_settings,
patch("app.tasks.upload_to_webdav.requests.put") as mock_put,
patch("app.tasks.upload_to_webdav.log_task_progress"),
):
mock_settings.webdav_url = "https://webdav.example.com"
mock_settings.webdav_username = "test_user"
mock_settings.webdav_password = "test_pass"
mock_settings.webdav_password = _TEST_CREDENTIAL
mock_settings.webdav_folder = "documents"
mock_settings.webdav_verify_ssl = True
mock_settings.http_request_timeout = 30
mock_response = Mock()
mock_response.status_code = 201
mock_put.return_value = mock_response
result = upload_to_webdav.apply(args=[sample_text_file]).get()
# Verify URL construction
called_url = mock_put.call_args[0][0]
assert "webdav.example.com" in called_url
@@ -272,23 +297,25 @@ class TestUploadToWebDAV:
def test_folder_path_with_leading_slash(self, sample_text_file):
"""Test folder path normalization when it starts with /."""
with patch("app.tasks.upload_to_webdav.settings") as mock_settings, \
patch("app.tasks.upload_to_webdav.requests.put") as mock_put, \
patch("app.tasks.upload_to_webdav.log_task_progress"):
with (
patch("app.tasks.upload_to_webdav.settings") as mock_settings,
patch("app.tasks.upload_to_webdav.requests.put") as mock_put,
patch("app.tasks.upload_to_webdav.log_task_progress"),
):
mock_settings.webdav_url = "https://webdav.example.com/"
mock_settings.webdav_username = "test_user"
mock_settings.webdav_password = "test_pass"
mock_settings.webdav_password = _TEST_CREDENTIAL
mock_settings.webdav_folder = "/uploads/documents" # Leading slash
mock_settings.webdav_verify_ssl = True
mock_settings.http_request_timeout = 30
mock_response = Mock()
mock_response.status_code = 201
mock_put.return_value = mock_response
result = upload_to_webdav.apply(args=[sample_text_file]).get()
# Verify the leading slash was removed in URL construction
called_url = mock_put.call_args[0][0]
# Should not have double slashes like //uploads
@@ -296,170 +323,178 @@ class TestUploadToWebDAV:
def test_empty_folder_path(self, sample_text_file):
"""Test upload with empty folder path (root directory)."""
with patch("app.tasks.upload_to_webdav.settings") as mock_settings, \
patch("app.tasks.upload_to_webdav.requests.put") as mock_put, \
patch("app.tasks.upload_to_webdav.log_task_progress"):
with (
patch("app.tasks.upload_to_webdav.settings") as mock_settings,
patch("app.tasks.upload_to_webdav.requests.put") as mock_put,
patch("app.tasks.upload_to_webdav.log_task_progress"),
):
mock_settings.webdav_url = "https://webdav.example.com/"
mock_settings.webdav_username = "test_user"
mock_settings.webdav_password = "test_pass"
mock_settings.webdav_password = _TEST_CREDENTIAL
mock_settings.webdav_folder = "" # Empty folder
mock_settings.webdav_verify_ssl = True
mock_settings.http_request_timeout = 30
mock_response = Mock()
mock_response.status_code = 201
mock_put.return_value = mock_response
result = upload_to_webdav.apply(args=[sample_text_file]).get()
assert result["status"] == "Completed"
def test_ssl_verification_enabled(self, sample_text_file):
"""Test that SSL verification is enabled when configured."""
with patch("app.tasks.upload_to_webdav.settings") as mock_settings, \
patch("app.tasks.upload_to_webdav.requests.put") as mock_put, \
patch("app.tasks.upload_to_webdav.log_task_progress"):
with (
patch("app.tasks.upload_to_webdav.settings") as mock_settings,
patch("app.tasks.upload_to_webdav.requests.put") as mock_put,
patch("app.tasks.upload_to_webdav.log_task_progress"),
):
mock_settings.webdav_url = "https://webdav.example.com/"
mock_settings.webdav_username = "test_user"
mock_settings.webdav_password = "test_pass"
mock_settings.webdav_password = _TEST_CREDENTIAL
mock_settings.webdav_folder = "uploads"
mock_settings.webdav_verify_ssl = True
mock_settings.http_request_timeout = 30
mock_response = Mock()
mock_response.status_code = 201
mock_put.return_value = mock_response
upload_to_webdav.apply(args=[sample_text_file]).get()
# Verify SSL verification was enabled
call_kwargs = mock_put.call_args[1]
assert call_kwargs["verify"] is True
def test_ssl_verification_disabled(self, sample_text_file):
"""Test that SSL verification can be disabled."""
with patch("app.tasks.upload_to_webdav.settings") as mock_settings, \
patch("app.tasks.upload_to_webdav.requests.put") as mock_put, \
patch("app.tasks.upload_to_webdav.log_task_progress"):
with (
patch("app.tasks.upload_to_webdav.settings") as mock_settings,
patch("app.tasks.upload_to_webdav.requests.put") as mock_put,
patch("app.tasks.upload_to_webdav.log_task_progress"),
):
mock_settings.webdav_url = "https://webdav.example.com/"
mock_settings.webdav_username = "test_user"
mock_settings.webdav_password = "test_pass"
mock_settings.webdav_password = _TEST_CREDENTIAL
mock_settings.webdav_folder = "uploads"
mock_settings.webdav_verify_ssl = False
mock_settings.http_request_timeout = 30
mock_response = Mock()
mock_response.status_code = 201
mock_put.return_value = mock_response
upload_to_webdav.apply(args=[sample_text_file]).get()
# Verify SSL verification was disabled
call_kwargs = mock_put.call_args[1]
assert call_kwargs["verify"] is False
def test_authentication_credentials(self, sample_text_file):
"""Test that authentication credentials are properly passed."""
with patch("app.tasks.upload_to_webdav.settings") as mock_settings, \
patch("app.tasks.upload_to_webdav.requests.put") as mock_put, \
patch("app.tasks.upload_to_webdav.log_task_progress"):
with (
patch("app.tasks.upload_to_webdav.settings") as mock_settings,
patch("app.tasks.upload_to_webdav.requests.put") as mock_put,
patch("app.tasks.upload_to_webdav.log_task_progress"),
):
mock_settings.webdav_url = "https://webdav.example.com/"
mock_settings.webdav_username = "custom_user"
mock_settings.webdav_password = "custom_password123"
mock_settings.webdav_password = _TEST_CUSTOM_CREDENTIAL
mock_settings.webdav_folder = "uploads"
mock_settings.webdav_verify_ssl = True
mock_settings.http_request_timeout = 30
mock_response = Mock()
mock_response.status_code = 201
mock_put.return_value = mock_response
upload_to_webdav.apply(args=[sample_text_file]).get()
# Verify correct credentials were used
call_kwargs = mock_put.call_args[1]
assert call_kwargs["auth"] == ("custom_user", "custom_password123")
assert call_kwargs["auth"] == ("custom_user", _TEST_CUSTOM_CREDENTIAL)
def test_logging_on_success(self, sample_text_file):
"""Test that progress is logged on successful upload."""
with patch("app.tasks.upload_to_webdav.settings") as mock_settings, \
patch("app.tasks.upload_to_webdav.requests.put") as mock_put, \
patch("app.tasks.upload_to_webdav.log_task_progress") as mock_log:
with (
patch("app.tasks.upload_to_webdav.settings") as mock_settings,
patch("app.tasks.upload_to_webdav.requests.put") as mock_put,
patch("app.tasks.upload_to_webdav.log_task_progress") as mock_log,
):
mock_settings.webdav_url = "https://webdav.example.com/"
mock_settings.webdav_username = "test_user"
mock_settings.webdav_password = "test_pass"
mock_settings.webdav_password = _TEST_CREDENTIAL
mock_settings.webdav_folder = "uploads"
mock_settings.webdav_verify_ssl = True
mock_settings.http_request_timeout = 30
mock_response = Mock()
mock_response.status_code = 201
mock_put.return_value = mock_response
upload_to_webdav.apply(args=[sample_text_file], kwargs={"file_id": 42}).get()
# Verify logging calls
assert mock_log.call_count >= 2 # At least in_progress and success
# Check for success log
success_calls = [
call for call in mock_log.call_args_list
if call[0][2] == "success"
]
success_calls = [call for call in mock_log.call_args_list if call[0][2] == "success"]
assert len(success_calls) >= 1
def test_logging_on_failure(self, sample_text_file):
"""Test that progress is logged on failed upload."""
with patch("app.tasks.upload_to_webdav.settings") as mock_settings, \
patch("app.tasks.upload_to_webdav.requests.put") as mock_put, \
patch("app.tasks.upload_to_webdav.log_task_progress") as mock_log:
with (
patch("app.tasks.upload_to_webdav.settings") as mock_settings,
patch("app.tasks.upload_to_webdav.requests.put") as mock_put,
patch("app.tasks.upload_to_webdav.log_task_progress") as mock_log,
):
mock_settings.webdav_url = "https://webdav.example.com/"
mock_settings.webdav_username = "test_user"
mock_settings.webdav_password = "test_pass"
mock_settings.webdav_password = _TEST_CREDENTIAL
mock_settings.webdav_folder = "uploads"
mock_settings.webdav_verify_ssl = True
mock_settings.http_request_timeout = 30
mock_response = Mock()
mock_response.status_code = 403
mock_response.text = "Forbidden"
mock_put.return_value = mock_response
with pytest.raises(Exception):
upload_to_webdav.apply(args=[sample_text_file], kwargs={"file_id": 42}).get()
# Check for failure log
failure_calls = [
call for call in mock_log.call_args_list
if call[0][2] == "failure"
]
failure_calls = [call for call in mock_log.call_args_list if call[0][2] == "failure"]
assert len(failure_calls) >= 1
def test_file_content_uploaded(self, sample_text_file):
"""Test that file content is actually read and uploaded."""
with patch("app.tasks.upload_to_webdav.settings") as mock_settings, \
patch("app.tasks.upload_to_webdav.requests.put") as mock_put, \
patch("app.tasks.upload_to_webdav.log_task_progress"):
with (
patch("app.tasks.upload_to_webdav.settings") as mock_settings,
patch("app.tasks.upload_to_webdav.requests.put") as mock_put,
patch("app.tasks.upload_to_webdav.log_task_progress"),
):
mock_settings.webdav_url = "https://webdav.example.com/"
mock_settings.webdav_username = "test_user"
mock_settings.webdav_password = "test_pass"
mock_settings.webdav_password = _TEST_CREDENTIAL
mock_settings.webdav_folder = "uploads"
mock_settings.webdav_verify_ssl = True
mock_settings.http_request_timeout = 30
mock_response = Mock()
mock_response.status_code = 201
mock_put.return_value = mock_response
upload_to_webdav.apply(args=[sample_text_file]).get()
# Verify data was passed to PUT request
call_kwargs = mock_put.call_args[1]
assert "data" in call_kwargs
@@ -468,23 +503,25 @@ class TestUploadToWebDAV:
def test_return_value_structure(self, sample_text_file):
"""Test that return value has correct structure."""
with patch("app.tasks.upload_to_webdav.settings") as mock_settings, \
patch("app.tasks.upload_to_webdav.requests.put") as mock_put, \
patch("app.tasks.upload_to_webdav.log_task_progress"):
with (
patch("app.tasks.upload_to_webdav.settings") as mock_settings,
patch("app.tasks.upload_to_webdav.requests.put") as mock_put,
patch("app.tasks.upload_to_webdav.log_task_progress"),
):
mock_settings.webdav_url = "https://webdav.example.com/"
mock_settings.webdav_username = "test_user"
mock_settings.webdav_password = "test_pass"
mock_settings.webdav_password = _TEST_CREDENTIAL
mock_settings.webdav_folder = "uploads"
mock_settings.webdav_verify_ssl = True
mock_settings.http_request_timeout = 30
mock_response = Mock()
mock_response.status_code = 201
mock_put.return_value = mock_response
result = upload_to_webdav.apply(args=[sample_text_file]).get()
# Check return value structure
assert isinstance(result, dict)
assert "status" in result
@@ -495,14 +532,11 @@ class TestUploadToWebDAV:
def test_module_importable(self):
"""Test that upload_to_webdav module is importable."""
from app.tasks.upload_to_webdav import upload_to_webdav
assert callable(upload_to_webdav)
def test_task_has_retry_configuration(self):
"""Test that the task has retry configuration from BaseTaskWithRetry."""
from app.tasks.upload_to_webdav import upload_to_webdav
# BaseTaskWithRetry should provide retry configuration
assert hasattr(upload_to_webdav, 'max_retries')
assert hasattr(upload_to_webdav, "max_retries")
# BaseTaskWithRetry configures 3 retries
assert upload_to_webdav.max_retries == 3
tests/test_upload_webdav_integration.py
+132 -174
View File
@@ -4,6 +4,7 @@ Integration tests for WebDAV upload with real server.
These tests spin up a real WebDAV server in a Docker container and test
actual file uploads against it, then verify the files were uploaded successfully.
"""
import os
import time
import pytest
@@ -17,6 +18,9 @@ from app.tasks.upload_to_webdav import upload_to_webdav
pytest.importorskip("testcontainers", reason="testcontainers not installed")
from testcontainers.core.container import DockerContainer
_TEST_CREDENTIAL = "testpass" # noqa: S105
_TEST_WRONG_CREDENTIAL = "wrongpass" # noqa: S105
@pytest.mark.integration
@pytest.mark.requires_docker
@@ -27,7 +31,7 @@ class TestWebDAVIntegration:
def webdav_server(self):
"""
Start a real WebDAV server in a Docker container.
Uses bytemark/webdav image which provides a simple WebDAV server.
"""
# Start WebDAV container
@@ -35,50 +39,54 @@ class TestWebDAVIntegration:
container.with_exposed_ports(80)
container.with_env("AUTH_TYPE", "Basic")
container.with_env("USERNAME", "testuser")
container.with_env("PASSWORD", "testpass")
container.with_env("PASSWORD", _TEST_CREDENTIAL)
# Start the container
container.start()
# Wait for server to be ready
time.sleep(2)
# Get the mapped port
host = container.get_container_host_ip()
port = container.get_exposed_port(80)
server_info = {
"container": container,
"host": host,
"port": port,
"url": f"http://{host}:{port}",
"username": "testuser",
"password": "testpass"
"password": _TEST_CREDENTIAL,
}
# Verify server is accessible
try:
response = requests.get(
server_info["url"],
auth=(server_info["username"], server_info["password"]),
timeout=5
server_info["url"], auth=(server_info["username"], server_info["password"]), timeout=5
)
assert response.status_code in [200, 301, 302, 401], \
f"WebDAV server not ready, got status {response.status_code}"
assert response.status_code in [
200,
301,
302,
401,
], f"WebDAV server not ready, got status {response.status_code}"
except Exception as e:
container.stop()
pytest.fail(f"Failed to connect to WebDAV server: {e}")
yield server_info
# Cleanup
container.stop()
def test_upload_file_to_real_webdav_server(self, webdav_server, sample_text_file):
"""Test uploading a file to a real WebDAV server."""
with patch("app.tasks.upload_to_webdav.settings") as mock_settings, \
patch("app.tasks.upload_to_webdav.log_task_progress"):
with (
patch("app.tasks.upload_to_webdav.settings") as mock_settings,
patch("app.tasks.upload_to_webdav.log_task_progress"),
):
# Configure settings to point to real WebDAV server
mock_settings.webdav_url = webdav_server["url"] + "/"
mock_settings.webdav_username = webdav_server["username"]
@@ -86,55 +94,45 @@ class TestWebDAVIntegration:
mock_settings.webdav_folder = ""
mock_settings.webdav_verify_ssl = False # HTTP server
mock_settings.http_request_timeout = 30
# Upload the file
result = upload_to_webdav.apply(
args=[sample_text_file],
kwargs={"file_id": 1}
).get()
result = upload_to_webdav.apply(args=[sample_text_file], kwargs={"file_id": 1}).get()
# Verify upload succeeded
assert result["status"] == "Completed"
assert result["file"] == sample_text_file
assert "url" in result
# Verify the file actually exists on the server
filename = os.path.basename(sample_text_file)
file_url = f"{webdav_server['url']}/{filename}"
response = requests.get(
file_url,
auth=(webdav_server["username"], webdav_server["password"]),
timeout=5
)
assert response.status_code == 200, \
f"File not found on server: {response.status_code}"
response = requests.get(file_url, auth=(webdav_server["username"], webdav_server["password"]), timeout=5)
assert response.status_code == 200, f"File not found on server: {response.status_code}"
# Verify file content matches
with open(sample_text_file, "rb") as f:
expected_content = f.read()
assert response.content == expected_content, \
"Uploaded file content does not match original"
assert response.content == expected_content, "Uploaded file content does not match original"
def test_upload_to_subfolder(self, webdav_server, sample_text_file):
"""Test uploading a file to a subfolder on WebDAV server."""
with patch("app.tasks.upload_to_webdav.settings") as mock_settings, \
patch("app.tasks.upload_to_webdav.log_task_progress"):
with (
patch("app.tasks.upload_to_webdav.settings") as mock_settings,
patch("app.tasks.upload_to_webdav.log_task_progress"),
):
# Create a test folder first
folder_name = "test-uploads"
folder_url = f"{webdav_server['url']}/{folder_name}"
# Create folder using MKCOL method
requests.request(
"MKCOL",
folder_url,
auth=(webdav_server["username"], webdav_server["password"]),
timeout=5
"MKCOL", folder_url, auth=(webdav_server["username"], webdav_server["password"]), timeout=5
)
# Configure settings
mock_settings.webdav_url = webdav_server["url"] + "/"
mock_settings.webdav_username = webdav_server["username"]
@@ -142,216 +140,181 @@ class TestWebDAVIntegration:
mock_settings.webdav_folder = folder_name
mock_settings.webdav_verify_ssl = False
mock_settings.http_request_timeout = 30
# Upload the file
result = upload_to_webdav.apply(
args=[sample_text_file],
kwargs={"file_id": 2}
).get()
result = upload_to_webdav.apply(args=[sample_text_file], kwargs={"file_id": 2}).get()
# Verify upload succeeded
assert result["status"] == "Completed"
# Verify the file exists in the subfolder
filename = os.path.basename(sample_text_file)
file_url = f"{webdav_server['url']}/{folder_name}/{filename}"
response = requests.get(
file_url,
auth=(webdav_server["username"], webdav_server["password"]),
timeout=5
)
assert response.status_code == 200, \
f"File not found in subfolder: {response.status_code}"
response = requests.get(file_url, auth=(webdav_server["username"], webdav_server["password"]), timeout=5)
assert response.status_code == 200, f"File not found in subfolder: {response.status_code}"
def test_upload_pdf_file(self, webdav_server, sample_pdf_path):
"""Test uploading a PDF file to WebDAV server."""
with patch("app.tasks.upload_to_webdav.settings") as mock_settings, \
patch("app.tasks.upload_to_webdav.log_task_progress"):
with (
patch("app.tasks.upload_to_webdav.settings") as mock_settings,
patch("app.tasks.upload_to_webdav.log_task_progress"),
):
mock_settings.webdav_url = webdav_server["url"] + "/"
mock_settings.webdav_username = webdav_server["username"]
mock_settings.webdav_password = webdav_server["password"]
mock_settings.webdav_folder = ""
mock_settings.webdav_verify_ssl = False
mock_settings.http_request_timeout = 30
# Upload the PDF
result = upload_to_webdav.apply(
args=[sample_pdf_path],
kwargs={"file_id": 3}
).get()
result = upload_to_webdav.apply(args=[sample_pdf_path], kwargs={"file_id": 3}).get()
# Verify upload succeeded
assert result["status"] == "Completed"
# Verify the file exists
filename = os.path.basename(sample_pdf_path)
file_url = f"{webdav_server['url']}/{filename}"
response = requests.get(
file_url,
auth=(webdav_server["username"], webdav_server["password"]),
timeout=5
)
response = requests.get(file_url, auth=(webdav_server["username"], webdav_server["password"]), timeout=5)
assert response.status_code == 200
# Verify it's a PDF (check magic bytes)
assert response.content.startswith(b'%PDF'), \
"Uploaded file is not a valid PDF"
assert response.content.startswith(b"%PDF"), "Uploaded file is not a valid PDF"
def test_upload_with_wrong_credentials(self, webdav_server, sample_text_file):
"""Test that upload fails with wrong credentials."""
with patch("app.tasks.upload_to_webdav.settings") as mock_settings, \
patch("app.tasks.upload_to_webdav.log_task_progress"):
with (
patch("app.tasks.upload_to_webdav.settings") as mock_settings,
patch("app.tasks.upload_to_webdav.log_task_progress"),
):
mock_settings.webdav_url = webdav_server["url"] + "/"
mock_settings.webdav_username = "wronguser"
mock_settings.webdav_password = "wrongpass"
mock_settings.webdav_password = _TEST_WRONG_CREDENTIAL
mock_settings.webdav_folder = ""
mock_settings.webdav_verify_ssl = False
mock_settings.http_request_timeout = 30
# Upload should fail with 401 Unauthorized
with pytest.raises(Exception, match="Failed to upload.*401"):
upload_to_webdav.apply(
args=[sample_text_file],
kwargs={"file_id": 4}
).get()
upload_to_webdav.apply(args=[sample_text_file], kwargs={"file_id": 4}).get()
def test_upload_multiple_files(self, webdav_server, sample_text_file, tmp_path):
"""Test uploading multiple files sequentially."""
with patch("app.tasks.upload_to_webdav.settings") as mock_settings, \
patch("app.tasks.upload_to_webdav.log_task_progress"):
with (
patch("app.tasks.upload_to_webdav.settings") as mock_settings,
patch("app.tasks.upload_to_webdav.log_task_progress"),
):
mock_settings.webdav_url = webdav_server["url"] + "/"
mock_settings.webdav_username = webdav_server["username"]
mock_settings.webdav_password = webdav_server["password"]
mock_settings.webdav_folder = ""
mock_settings.webdav_verify_ssl = False
mock_settings.http_request_timeout = 30
# Create additional test files
file1 = tmp_path / "test1.txt"
file1.write_text("Test file 1")
file2 = tmp_path / "test2.txt"
file2.write_text("Test file 2")
file3 = tmp_path / "test3.txt"
file3.write_text("Test file 3")
# Upload all files
files = [str(file1), str(file2), str(file3)]
uploaded_files = []
for idx, file_path in enumerate(files, start=1):
result = upload_to_webdav.apply(
args=[file_path],
kwargs={"file_id": idx + 10}
).get()
result = upload_to_webdav.apply(args=[file_path], kwargs={"file_id": idx + 10}).get()
assert result["status"] == "Completed"
uploaded_files.append(os.path.basename(file_path))
# Verify all files exist on server
for filename in uploaded_files:
file_url = f"{webdav_server['url']}/{filename}"
response = requests.get(
file_url,
auth=(webdav_server["username"], webdav_server["password"]),
timeout=5
file_url, auth=(webdav_server["username"], webdav_server["password"]), timeout=5
)
assert response.status_code == 200, \
f"File {filename} not found on server"
assert response.status_code == 200, f"File {filename} not found on server"
def test_overwrite_existing_file(self, webdav_server, sample_text_file, tmp_path):
"""Test that uploading a file with the same name overwrites the existing one."""
with patch("app.tasks.upload_to_webdav.settings") as mock_settings, \
patch("app.tasks.upload_to_webdav.log_task_progress"):
with (
patch("app.tasks.upload_to_webdav.settings") as mock_settings,
patch("app.tasks.upload_to_webdav.log_task_progress"),
):
mock_settings.webdav_url = webdav_server["url"] + "/"
mock_settings.webdav_username = webdav_server["username"]
mock_settings.webdav_password = webdav_server["password"]
mock_settings.webdav_folder = ""
mock_settings.webdav_verify_ssl = False
mock_settings.http_request_timeout = 30
# Create two files with same name but different content
file1 = tmp_path / "duplicate.txt"
file1.write_text("Original content")
# Upload first version
result1 = upload_to_webdav.apply(
args=[str(file1)],
kwargs={"file_id": 20}
).get()
result1 = upload_to_webdav.apply(args=[str(file1)], kwargs={"file_id": 20}).get()
assert result1["status"] == "Completed"
# Verify first version
file_url = f"{webdav_server['url']}/duplicate.txt"
response = requests.get(
file_url,
auth=(webdav_server["username"], webdav_server["password"]),
timeout=5
)
response = requests.get(file_url, auth=(webdav_server["username"], webdav_server["password"]), timeout=5)
assert response.text == "Original content"
# Update file content
file1.write_text("Updated content - version 2")
# Upload second version
result2 = upload_to_webdav.apply(
args=[str(file1)],
kwargs={"file_id": 21}
).get()
result2 = upload_to_webdav.apply(args=[str(file1)], kwargs={"file_id": 21}).get()
assert result2["status"] == "Completed"
# Verify file was overwritten
response = requests.get(
file_url,
auth=(webdav_server["username"], webdav_server["password"]),
timeout=5
)
response = requests.get(file_url, auth=(webdav_server["username"], webdav_server["password"]), timeout=5)
assert response.text == "Updated content - version 2"
def test_large_file_upload(self, webdav_server, tmp_path):
"""Test uploading a larger file (1MB)."""
with patch("app.tasks.upload_to_webdav.settings") as mock_settings, \
patch("app.tasks.upload_to_webdav.log_task_progress"):
with (
patch("app.tasks.upload_to_webdav.settings") as mock_settings,
patch("app.tasks.upload_to_webdav.log_task_progress"),
):
mock_settings.webdav_url = webdav_server["url"] + "/"
mock_settings.webdav_username = webdav_server["username"]
mock_settings.webdav_password = webdav_server["password"]
mock_settings.webdav_folder = ""
mock_settings.webdav_verify_ssl = False
mock_settings.http_request_timeout = 60 # Longer timeout for large file
# Create a 1MB test file
large_file = tmp_path / "large_file.bin"
large_file.write_bytes(b"X" * (1024 * 1024)) # 1MB of X's
# Upload the large file
result = upload_to_webdav.apply(
args=[str(large_file)],
kwargs={"file_id": 30}
).get()
result = upload_to_webdav.apply(args=[str(large_file)], kwargs={"file_id": 30}).get()
assert result["status"] == "Completed"
# Verify file exists and has correct size
file_url = f"{webdav_server['url']}/large_file.bin"
response = requests.get(
file_url,
auth=(webdav_server["username"], webdav_server["password"]),
timeout=10
)
response = requests.get(file_url, auth=(webdav_server["username"], webdav_server["password"]), timeout=10)
assert response.status_code == 200
assert len(response.content) == 1024 * 1024, \
f"File size mismatch: expected 1MB, got {len(response.content)} bytes"
assert (
len(response.content) == 1024 * 1024
), f"File size mismatch: expected 1MB, got {len(response.content)} bytes"
@pytest.mark.integration
@@ -367,20 +330,20 @@ class TestWebDAVServerVerification:
container.with_env("AUTH_TYPE", "Basic")
container.with_env("USERNAME", "admin")
container.with_env("PASSWORD", "admin123")
container.start()
time.sleep(2)
host = container.get_container_host_ip()
port = container.get_exposed_port(80)
server_info = {
"container": container,
"url": f"http://{host}:{port}",
"username": "admin",
"password": "admin123"
"password": "admin123",
}
yield server_info
container.stop()
@@ -389,12 +352,10 @@ class TestWebDAVServerVerification:
# Request without auth should fail
response = requests.get(webdav_server["url"], timeout=5)
assert response.status_code == 401
# Request with auth should succeed
response = requests.get(
webdav_server["url"],
auth=(webdav_server["username"], webdav_server["password"]),
timeout=5
webdav_server["url"], auth=(webdav_server["username"], webdav_server["password"]), timeout=5
)
assert response.status_code in [200, 301, 302]
@@ -402,26 +363,23 @@ class TestWebDAVServerVerification:
"""Verify WebDAV server accepts PUT requests."""
test_file = tmp_path / "put_test.txt"
test_file.write_text("PUT method test")
with open(test_file, "rb") as f:
response = requests.put(
f"{webdav_server['url']}/put_test.txt",
auth=(webdav_server["username"], webdav_server["password"]),
data=f,
timeout=5
timeout=5,
)
assert response.status_code in [200, 201, 204]
def test_webdav_propfind_method(self, webdav_server):
"""Verify WebDAV server supports PROPFIND (directory listing)."""
response = requests.request(
"PROPFIND",
webdav_server["url"],
auth=(webdav_server["username"], webdav_server["password"]),
timeout=5
"PROPFIND", webdav_server["url"], auth=(webdav_server["username"], webdav_server["password"]), timeout=5
)
# PROPFIND may return 207 Multi-Status, 200 OK, or 403 Forbidden
# depending on server configuration
assert response.status_code in [200, 207, 403]
tests/test_views_coverage.py
+9 -1
View File
@@ -1,7 +1,10 @@
"""Additional view tests to increase coverage."""
import pytest
from unittest.mock import patch, MagicMock
_TEST_CREDENTIAL = "test" # noqa: S105
@pytest.mark.integration
class TestWizardPost:
@@ -31,7 +34,12 @@ class TestWizardPost:
"""Test POST wizard step 2."""
response = client.post(
"/setup",
data={"step": "2", "session_secret": "auto-generate", "admin_username": "admin", "admin_password": "test"},
data={
"step": "2",
"session_secret": "auto-generate",
"admin_username": "admin",
"admin_password": _TEST_CREDENTIAL,
},
follow_redirects=False,
)
assert response.status_code in (200, 303)