diff --git a/app/api/diagnostic.py b/app/api/diagnostic.py index 400b2f9f..7fffbcbd 100644 --- a/app/api/diagnostic.py +++ b/app/api/diagnostic.py @@ -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 diff --git a/app/api/dropbox.py b/app/api/dropbox.py index 711055cd..fde72f16 100644 --- a/app/api/dropbox.py +++ b/app/api/dropbox.py @@ -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 diff --git a/app/api/files.py b/app/api/files.py index 0e81f43b..abdc4ed5 100644 --- a/app/api/files.py +++ b/app/api/files.py @@ -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. diff --git a/app/api/google_drive.py b/app/api/google_drive.py index c2f74fab..e1942df4 100644 --- a/app/api/google_drive.py +++ b/app/api/google_drive.py @@ -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 diff --git a/app/api/logs.py b/app/api/logs.py index 7ef0c434..caeb1225 100644 --- a/app/api/logs.py +++ b/app/api/logs.py @@ -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). diff --git a/app/api/onedrive.py b/app/api/onedrive.py index bbaaef96..1b226768 100644 --- a/app/api/onedrive.py +++ b/app/api/onedrive.py @@ -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) diff --git a/app/api/settings.py b/app/api/settings.py index 5e9bd4af..7cf74a9e 100644 --- a/app/api/settings.py +++ b/app/api/settings.py @@ -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. diff --git a/requirements-dev.txt b/requirements-dev.txt index b20fdb6b..765e2ae7 100644 --- a/requirements-dev.txt +++ b/requirements-dev.txt @@ -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 diff --git a/tests/conftest.py b/tests/conftest.py index 03d8b6f4..20f395b3 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -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.""" diff --git a/tests/fixtures_integration.py b/tests/fixtures_integration.py index 94704eab..f1c534e9 100644 --- a/tests/fixtures_integration.py +++ b/tests/fixtures_integration.py @@ -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: diff --git a/tests/test_auth.py b/tests/test_auth.py index aaa6f389..76bb9a4e 100644 --- a/tests/test_auth.py +++ b/tests/test_auth.py @@ -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 diff --git a/tests/test_e2e_full_stack.py b/tests/test_e2e_full_stack.py index 3bf673f0..842cdb7c 100644 --- a/tests/test_e2e_full_stack.py +++ b/tests/test_e2e_full_stack.py @@ -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:

Integration Test

This is a test document.

""") - + # 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() diff --git a/tests/test_external_integrations.py b/tests/test_external_integrations.py new file mode 100644 index 00000000..669fd9aa --- /dev/null +++ b/tests/test_external_integrations.py @@ -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) diff --git a/tests/test_imap_tasks.py b/tests/test_imap_tasks.py index ce39d145..1dce4994 100644 --- a/tests/test_imap_tasks.py +++ b/tests/test_imap_tasks.py @@ -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, ) diff --git a/tests/test_notification.py b/tests/test_notification.py index 758b4766..095d21ea 100644 --- a/tests/test_notification.py +++ b/tests/test_notification.py @@ -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 diff --git a/tests/test_path_traversal_security.py b/tests/test_path_traversal_security.py index 878b547d..0bb44178 100644 --- a/tests/test_path_traversal_security.py +++ b/tests/test_path_traversal_security.py @@ -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 diff --git a/tests/test_rate_limiting.py b/tests/test_rate_limiting.py index 8b1827a6..05421764 100644 --- a/tests/test_rate_limiting.py +++ b/tests/test_rate_limiting.py @@ -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) diff --git a/tests/test_upload_tasks.py b/tests/test_upload_tasks.py index db621933..e9175585 100644 --- a/tests/test_upload_tasks.py +++ b/tests/test_upload_tasks.py @@ -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 = "Test email" 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): diff --git a/tests/test_upload_webdav_comprehensive.py b/tests/test_upload_webdav_comprehensive.py index 80646c6e..131ec5cd 100644 --- a/tests/test_upload_webdav_comprehensive.py +++ b/tests/test_upload_webdav_comprehensive.py @@ -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 diff --git a/tests/test_upload_webdav_integration.py b/tests/test_upload_webdav_integration.py index 36174319..f2bc1bdb 100644 --- a/tests/test_upload_webdav_integration.py +++ b/tests/test_upload_webdav_integration.py @@ -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] diff --git a/tests/test_views_coverage.py b/tests/test_views_coverage.py index 81797e25..1021db33 100644 --- a/tests/test_views_coverage.py +++ b/tests/test_views_coverage.py @@ -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)