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/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/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/tests/fixtures_integration.py b/tests/fixtures_integration.py index a36e9a32..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 @@ -31,16 +32,16 @@ _TEST_CREDENTIAL = "testpass" # noqa: S105 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(), @@ -56,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, @@ -85,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() @@ -124,13 +123,13 @@ def webdav_container() -> Generator: container.with_env("AUTH_TYPE", "Basic") container.with_env("USERNAME", "testuser") 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}", @@ -139,7 +138,7 @@ def webdav_container() -> Generator: "username": "testuser", "password": _TEST_CREDENTIAL, } - + container.stop() @@ -147,20 +146,20 @@ 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(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, @@ -169,7 +168,7 @@ def sftp_container() -> Generator: "password": _TEST_CREDENTIAL, "folder": "/home/testuser/upload", } - + container.stop() @@ -177,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"], @@ -200,7 +199,7 @@ 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") @@ -209,13 +208,13 @@ def ftp_container() -> Generator: container.with_env("FTP_USER_NAME", "testuser") 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, @@ -224,7 +223,7 @@ def ftp_container() -> Generator: "password": _TEST_CREDENTIAL, "folder": "/", } - + container.stop() @@ -239,7 +238,7 @@ def full_infrastructure( ): """ Combined fixture that provides all infrastructure components. - + Use this fixture when you need the complete application stack. """ return { @@ -256,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"], @@ -269,7 +268,7 @@ def celery_app(redis_container): task_eager_propagates=True, result_expires=3600, ) - + return celery @@ -277,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, @@ -296,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_imap_tasks.py b/tests/test_imap_tasks.py index 57ab4120..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 diff --git a/tests/test_upload_tasks.py b/tests/test_upload_tasks.py index 5af4caeb..2de25a9b 100644 --- a/tests/test_upload_tasks.py +++ b/tests/test_upload_tasks.py @@ -19,9 +19,10 @@ _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" @@ -44,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 @@ -67,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 @@ -88,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 @@ -109,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 @@ -144,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" @@ -170,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() @@ -193,13 +200,16 @@ 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 @@ -208,14 +218,14 @@ def test_upload_to_ftp_accepts_file_id(sample_text_file): 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" @@ -224,10 +234,12 @@ 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" @@ -235,26 +247,28 @@ def test_upload_to_ftp_without_file_id(sample_text_file): 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 @@ -262,7 +276,7 @@ def test_upload_to_sftp_accepts_file_id(sample_text_file): mock_settings.sftp_password = _TEST_CREDENTIAL mock_settings.sftp_folder = "/uploads" mock_settings.workdir = "/tmp" - + # Setup mocks mock_ssh_instance = Mock() mock_sftp = Mock() @@ -270,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 @@ -282,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_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 @@ -309,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") 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"), + ): + # 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() @@ -330,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 @@ -344,15 +362,17 @@ 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") 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, + ): + # Setup settings mock_settings.email_host = "smtp.example.com" mock_settings.email_port = 587 @@ -361,20 +381,20 @@ def test_upload_to_email_accepts_file_id(sample_text_file): 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 @@ -383,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() @@ -395,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() @@ -409,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() @@ -421,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"), @@ -438,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" @@ -458,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" @@ -495,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 77a28c6f..2a78d1e7 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 @@ -16,10 +17,12 @@ 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" @@ -27,222 +30,239 @@ class TestUploadToWebDAV: 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_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_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_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_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_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_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_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_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_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_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/") @@ -251,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_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 @@ -275,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_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 @@ -299,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_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_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_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 = _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", _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_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_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_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 @@ -471,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_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 @@ -499,13 +533,14 @@ 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 04be8f40..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 @@ -30,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 @@ -39,49 +40,53 @@ class TestWebDAVIntegration: container.with_env("AUTH_TYPE", "Basic") container.with_env("USERNAME", "testuser") 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": _TEST_CREDENTIAL + "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"] @@ -89,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"] @@ -145,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 = _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 @@ -370,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() @@ -392,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] @@ -405,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 62a3bf9e..1021db33 100644 --- a/tests/test_views_coverage.py +++ b/tests/test_views_coverage.py @@ -1,4 +1,5 @@ """Additional view tests to increase coverage.""" + import pytest from unittest.mock import patch, MagicMock @@ -33,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_CREDENTIAL}, + data={ + "step": "2", + "session_secret": "auto-generate", + "admin_username": "admin", + "admin_password": _TEST_CREDENTIAL, + }, follow_redirects=False, ) assert response.status_code in (200, 303)