diff --git a/requirements-dev.txt b/requirements-dev.txt index 15cbb280..5df362e4 100644 --- a/requirements-dev.txt +++ b/requirements-dev.txt @@ -7,6 +7,7 @@ pytest-cov>=4.1.0 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 # Code quality flake8>=7.0.0 diff --git a/tests/conftest.py b/tests/conftest.py index dfe5fc50..e3b6e15c 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -177,3 +177,4 @@ def pytest_configure(config): config.addinivalue_line("markers", "requires_external: Tests requiring external services") config.addinivalue_line("markers", "requires_db: Tests requiring database") config.addinivalue_line("markers", "requires_redis: Tests requiring Redis") + config.addinivalue_line("markers", "requires_docker: Tests requiring Docker") diff --git a/tests/test_upload_webdav_integration.py b/tests/test_upload_webdav_integration.py new file mode 100644 index 00000000..36174319 --- /dev/null +++ b/tests/test_upload_webdav_integration.py @@ -0,0 +1,427 @@ +""" +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 +import requests +from pathlib import Path +from unittest.mock import patch + +from app.tasks.upload_to_webdav import upload_to_webdav + +# Import testcontainers - required for these tests +pytest.importorskip("testcontainers", reason="testcontainers not installed") +from testcontainers.core.container import DockerContainer + + +@pytest.mark.integration +@pytest.mark.requires_docker +class TestWebDAVIntegration: + """Integration tests with real WebDAV server.""" + + @pytest.fixture(scope="class") + 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 + container = DockerContainer("bytemark/webdav:latest") + container.with_exposed_ports(80) + container.with_env("AUTH_TYPE", "Basic") + container.with_env("USERNAME", "testuser") + container.with_env("PASSWORD", "testpass") + + # 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" + } + + # Verify server is accessible + try: + response = requests.get( + 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}" + 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"): + + # Configure settings to point to real WebDAV server + 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 # 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() + + # 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}" + + # 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" + + 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"): + + # 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 + ) + + # Configure settings + 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 = 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() + + # 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}" + + 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"): + + 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() + + # 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 + ) + + 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" + + 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"): + + mock_settings.webdav_url = webdav_server["url"] + "/" + mock_settings.webdav_username = "wronguser" + mock_settings.webdav_password = "wrongpass" + 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() + + 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"): + + 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() + + 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 + ) + 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"): + + 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() + 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 + ) + 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() + assert result2["status"] == "Completed" + + # Verify file was overwritten + 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"): + + 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() + + 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 + ) + + assert response.status_code == 200 + assert len(response.content) == 1024 * 1024, \ + f"File size mismatch: expected 1MB, got {len(response.content)} bytes" + + +@pytest.mark.integration +@pytest.mark.requires_docker +class TestWebDAVServerVerification: + """Tests to verify WebDAV server behavior.""" + + @pytest.fixture(scope="class") + def webdav_server(self): + """Start WebDAV server for verification tests.""" + container = DockerContainer("bytemark/webdav:latest") + container.with_exposed_ports(80) + 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" + } + + yield server_info + container.stop() + + def test_webdav_server_basic_auth(self, webdav_server): + """Verify WebDAV server requires authentication.""" + # 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 + ) + assert response.status_code in [200, 301, 302] + + def test_webdav_put_method(self, webdav_server, tmp_path): + """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 + ) + + 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 may return 207 Multi-Status, 200 OK, or 403 Forbidden + # depending on server configuration + assert response.status_code in [200, 207, 403]