style: fix linting issues (whitespace and formatting)
- Remove trailing whitespace from blank lines - Apply black formatting to test file - All tests still pass Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com>
This commit is contained in:
@@ -88,11 +88,11 @@ def sanitize_filename(filename):
|
|||||||
"""
|
"""
|
||||||
# First, replace all path separators (both Unix and Windows style) with underscores
|
# First, replace all path separators (both Unix and Windows style) with underscores
|
||||||
sanitized = filename.replace("/", "_").replace("\\", "_")
|
sanitized = filename.replace("/", "_").replace("\\", "_")
|
||||||
|
|
||||||
# Replace characters that are problematic in various filesystems
|
# Replace characters that are problematic in various filesystems
|
||||||
# Keep only alphanumeric, dash, underscore, period, and space
|
# Keep only alphanumeric, dash, underscore, period, and space
|
||||||
sanitized = re.sub(r"[^\w\-\. ]", "_", sanitized)
|
sanitized = re.sub(r"[^\w\-\. ]", "_", sanitized)
|
||||||
|
|
||||||
# Remove or replace path traversal patterns
|
# Remove or replace path traversal patterns
|
||||||
# Replace specifically '..' to prevent path traversal while preserving single dots
|
# Replace specifically '..' to prevent path traversal while preserving single dots
|
||||||
sanitized = sanitized.replace("..", "_")
|
sanitized = sanitized.replace("..", "_")
|
||||||
|
|||||||
+111
-141
@@ -9,6 +9,7 @@ Tests cover:
|
|||||||
- Security issues (path traversal)
|
- Security issues (path traversal)
|
||||||
- Error handling
|
- Error handling
|
||||||
"""
|
"""
|
||||||
|
|
||||||
import io
|
import io
|
||||||
import os
|
import os
|
||||||
import pytest
|
import pytest
|
||||||
@@ -21,68 +22,63 @@ from fastapi.testclient import TestClient
|
|||||||
def mock_celery_tasks():
|
def mock_celery_tasks():
|
||||||
"""Mock all Celery tasks to prevent execution."""
|
"""Mock all Celery tasks to prevent execution."""
|
||||||
# Patch the entire task object where it's used (in app.api.files)
|
# Patch the entire task object where it's used (in app.api.files)
|
||||||
with patch("app.api.files.process_document") as mock_process_task, \
|
with (
|
||||||
patch("app.api.files.convert_to_pdf") as mock_convert_task:
|
patch("app.api.files.process_document") as mock_process_task,
|
||||||
|
patch("app.api.files.convert_to_pdf") as mock_convert_task,
|
||||||
|
):
|
||||||
|
|
||||||
# Setup default return values for .delay()
|
# Setup default return values for .delay()
|
||||||
mock_task = MagicMock()
|
mock_task = MagicMock()
|
||||||
mock_task.id = "test-task-id-123"
|
mock_task.id = "test-task-id-123"
|
||||||
mock_process_task.delay.return_value = mock_task
|
mock_process_task.delay.return_value = mock_task
|
||||||
mock_convert_task.delay.return_value = mock_task
|
mock_convert_task.delay.return_value = mock_task
|
||||||
|
|
||||||
yield {
|
yield {"process_document": mock_process_task.delay, "convert_to_pdf": mock_convert_task.delay}
|
||||||
"process_document": mock_process_task.delay,
|
|
||||||
"convert_to_pdf": mock_convert_task.delay
|
|
||||||
}
|
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.integration
|
@pytest.mark.integration
|
||||||
class TestValidFileUploads:
|
class TestValidFileUploads:
|
||||||
"""Tests for successful file uploads with various valid file types."""
|
"""Tests for successful file uploads with various valid file types."""
|
||||||
|
|
||||||
def test_upload_valid_pdf(self, client: TestClient, sample_pdf_path: str, mock_celery_tasks):
|
def test_upload_valid_pdf(self, client: TestClient, sample_pdf_path: str, mock_celery_tasks):
|
||||||
"""Test uploading a valid PDF file."""
|
"""Test uploading a valid PDF file."""
|
||||||
with open(sample_pdf_path, "rb") as f:
|
with open(sample_pdf_path, "rb") as f:
|
||||||
response = client.post(
|
response = client.post("/api/ui-upload", files={"file": ("document.pdf", f, "application/pdf")})
|
||||||
"/api/ui-upload",
|
|
||||||
files={"file": ("document.pdf", f, "application/pdf")}
|
|
||||||
)
|
|
||||||
|
|
||||||
assert response.status_code == 200
|
assert response.status_code == 200
|
||||||
data = response.json()
|
data = response.json()
|
||||||
|
|
||||||
# Verify response structure
|
# Verify response structure
|
||||||
assert "task_id" in data
|
assert "task_id" in data
|
||||||
assert "status" in data
|
assert "status" in data
|
||||||
assert "original_filename" in data
|
assert "original_filename" in data
|
||||||
assert "stored_filename" in data
|
assert "stored_filename" in data
|
||||||
|
|
||||||
# Verify response values
|
# Verify response values
|
||||||
assert data["status"] == "queued"
|
assert data["status"] == "queued"
|
||||||
assert data["original_filename"] == "document.pdf"
|
assert data["original_filename"] == "document.pdf"
|
||||||
assert data["stored_filename"].endswith(".pdf")
|
assert data["stored_filename"].endswith(".pdf")
|
||||||
|
|
||||||
# Verify the processing task was called
|
# Verify the processing task was called
|
||||||
mock_celery_tasks["process_document"].assert_called_once()
|
mock_celery_tasks["process_document"].assert_called_once()
|
||||||
call_args = mock_celery_tasks["process_document"].call_args
|
call_args = mock_celery_tasks["process_document"].call_args
|
||||||
assert call_args.kwargs["original_filename"] == "document.pdf"
|
assert call_args.kwargs["original_filename"] == "document.pdf"
|
||||||
|
|
||||||
def test_upload_valid_text_file(self, client: TestClient, mock_celery_tasks):
|
def test_upload_valid_text_file(self, client: TestClient, mock_celery_tasks):
|
||||||
"""Test uploading a valid text file."""
|
"""Test uploading a valid text file."""
|
||||||
text_content = b"This is a test text file.\nWith multiple lines."
|
text_content = b"This is a test text file.\nWith multiple lines."
|
||||||
response = client.post(
|
response = client.post(
|
||||||
"/api/ui-upload",
|
"/api/ui-upload", files={"file": ("document.txt", io.BytesIO(text_content), "text/plain")}
|
||||||
files={"file": ("document.txt", io.BytesIO(text_content), "text/plain")}
|
|
||||||
)
|
)
|
||||||
|
|
||||||
assert response.status_code == 200
|
assert response.status_code == 200
|
||||||
data = response.json()
|
data = response.json()
|
||||||
assert data["original_filename"] == "document.txt"
|
assert data["original_filename"] == "document.txt"
|
||||||
assert data["stored_filename"].endswith(".txt")
|
assert data["stored_filename"].endswith(".txt")
|
||||||
|
|
||||||
# Text files should be converted to PDF
|
# Text files should be converted to PDF
|
||||||
mock_celery_tasks["convert_to_pdf"].assert_called_once()
|
mock_celery_tasks["convert_to_pdf"].assert_called_once()
|
||||||
|
|
||||||
def test_upload_valid_image_jpeg(self, client: TestClient, mock_celery_tasks):
|
def test_upload_valid_image_jpeg(self, client: TestClient, mock_celery_tasks):
|
||||||
"""Test uploading a valid JPEG image."""
|
"""Test uploading a valid JPEG image."""
|
||||||
# Create a minimal valid JPEG
|
# Create a minimal valid JPEG
|
||||||
@@ -91,19 +87,16 @@ class TestValidFileUploads:
|
|||||||
b"\xff\xdb\x00C\x00\x08\x06\x06\x07\x06\x05\x08\x07\x07\x07\t\t\x08\n"
|
b"\xff\xdb\x00C\x00\x08\x06\x06\x07\x06\x05\x08\x07\x07\x07\t\t\x08\n"
|
||||||
b"\xff\xd9"
|
b"\xff\xd9"
|
||||||
)
|
)
|
||||||
|
|
||||||
response = client.post(
|
response = client.post("/api/ui-upload", files={"file": ("image.jpg", io.BytesIO(jpeg_content), "image/jpeg")})
|
||||||
"/api/ui-upload",
|
|
||||||
files={"file": ("image.jpg", io.BytesIO(jpeg_content), "image/jpeg")}
|
|
||||||
)
|
|
||||||
|
|
||||||
assert response.status_code == 200
|
assert response.status_code == 200
|
||||||
data = response.json()
|
data = response.json()
|
||||||
assert data["original_filename"] == "image.jpg"
|
assert data["original_filename"] == "image.jpg"
|
||||||
|
|
||||||
# Images should be converted to PDF
|
# Images should be converted to PDF
|
||||||
mock_celery_tasks["convert_to_pdf"].assert_called_once()
|
mock_celery_tasks["convert_to_pdf"].assert_called_once()
|
||||||
|
|
||||||
def test_upload_valid_png_image(self, client: TestClient, mock_celery_tasks):
|
def test_upload_valid_png_image(self, client: TestClient, mock_celery_tasks):
|
||||||
"""Test uploading a valid PNG image."""
|
"""Test uploading a valid PNG image."""
|
||||||
# Create a minimal valid PNG (1x1 transparent pixel)
|
# Create a minimal valid PNG (1x1 transparent pixel)
|
||||||
@@ -112,51 +105,47 @@ class TestValidFileUploads:
|
|||||||
b"\x08\x06\x00\x00\x00\x1f\x15\xc4\x89\x00\x00\x00\nIDATx\x9cc\x00\x01"
|
b"\x08\x06\x00\x00\x00\x1f\x15\xc4\x89\x00\x00\x00\nIDATx\x9cc\x00\x01"
|
||||||
b"\x00\x00\x05\x00\x01\r\n-\xb4\x00\x00\x00\x00IEND\xaeB`\x82"
|
b"\x00\x00\x05\x00\x01\r\n-\xb4\x00\x00\x00\x00IEND\xaeB`\x82"
|
||||||
)
|
)
|
||||||
|
|
||||||
response = client.post(
|
response = client.post(
|
||||||
"/api/ui-upload",
|
"/api/ui-upload", files={"file": ("screenshot.png", io.BytesIO(png_content), "image/png")}
|
||||||
files={"file": ("screenshot.png", io.BytesIO(png_content), "image/png")}
|
|
||||||
)
|
)
|
||||||
|
|
||||||
assert response.status_code == 200
|
assert response.status_code == 200
|
||||||
data = response.json()
|
data = response.json()
|
||||||
assert data["original_filename"] == "screenshot.png"
|
assert data["original_filename"] == "screenshot.png"
|
||||||
assert data["stored_filename"].endswith(".png")
|
assert data["stored_filename"].endswith(".png")
|
||||||
mock_celery_tasks["convert_to_pdf"].assert_called_once()
|
mock_celery_tasks["convert_to_pdf"].assert_called_once()
|
||||||
|
|
||||||
def test_upload_office_document_docx(self, client: TestClient, mock_celery_tasks):
|
def test_upload_office_document_docx(self, client: TestClient, mock_celery_tasks):
|
||||||
"""Test uploading a Word document."""
|
"""Test uploading a Word document."""
|
||||||
# Create minimal DOCX content (ZIP file with proper structure)
|
# Create minimal DOCX content (ZIP file with proper structure)
|
||||||
docx_content = (
|
docx_content = b"PK\x03\x04\x14\x00\x00\x00\x08\x00" + b"\x00" * 50
|
||||||
b"PK\x03\x04\x14\x00\x00\x00\x08\x00" + b"\x00" * 50
|
|
||||||
)
|
|
||||||
|
|
||||||
response = client.post(
|
response = client.post(
|
||||||
"/api/ui-upload",
|
"/api/ui-upload",
|
||||||
files={"file": (
|
files={
|
||||||
"report.docx",
|
"file": (
|
||||||
io.BytesIO(docx_content),
|
"report.docx",
|
||||||
"application/vnd.openxmlformats-officedocument.wordprocessingml.document"
|
io.BytesIO(docx_content),
|
||||||
)}
|
"application/vnd.openxmlformats-officedocument.wordprocessingml.document",
|
||||||
|
)
|
||||||
|
},
|
||||||
)
|
)
|
||||||
|
|
||||||
assert response.status_code == 200
|
assert response.status_code == 200
|
||||||
data = response.json()
|
data = response.json()
|
||||||
assert data["original_filename"] == "report.docx"
|
assert data["original_filename"] == "report.docx"
|
||||||
assert data["stored_filename"].endswith(".docx")
|
assert data["stored_filename"].endswith(".docx")
|
||||||
|
|
||||||
# Office documents should be converted to PDF
|
# Office documents should be converted to PDF
|
||||||
mock_celery_tasks["convert_to_pdf"].assert_called_once()
|
mock_celery_tasks["convert_to_pdf"].assert_called_once()
|
||||||
|
|
||||||
def test_upload_csv_file(self, client: TestClient, mock_celery_tasks):
|
def test_upload_csv_file(self, client: TestClient, mock_celery_tasks):
|
||||||
"""Test uploading a CSV file."""
|
"""Test uploading a CSV file."""
|
||||||
csv_content = b"name,age,city\nJohn,30,NYC\nJane,25,LA\n"
|
csv_content = b"name,age,city\nJohn,30,NYC\nJane,25,LA\n"
|
||||||
|
|
||||||
response = client.post(
|
response = client.post("/api/ui-upload", files={"file": ("data.csv", io.BytesIO(csv_content), "text/csv")})
|
||||||
"/api/ui-upload",
|
|
||||||
files={"file": ("data.csv", io.BytesIO(csv_content), "text/csv")}
|
|
||||||
)
|
|
||||||
|
|
||||||
assert response.status_code == 200
|
assert response.status_code == 200
|
||||||
data = response.json()
|
data = response.json()
|
||||||
assert data["original_filename"] == "data.csv"
|
assert data["original_filename"] == "data.csv"
|
||||||
@@ -166,54 +155,49 @@ class TestValidFileUploads:
|
|||||||
@pytest.mark.integration
|
@pytest.mark.integration
|
||||||
class TestInvalidFileUploads:
|
class TestInvalidFileUploads:
|
||||||
"""Tests for handling invalid or problematic file uploads."""
|
"""Tests for handling invalid or problematic file uploads."""
|
||||||
|
|
||||||
def test_upload_file_too_large(self, client: TestClient):
|
def test_upload_file_too_large(self, client: TestClient):
|
||||||
"""Test that files over 500MB are rejected."""
|
"""Test that files over 500MB are rejected."""
|
||||||
# Create a large file content (mock it to avoid memory issues)
|
# Create a large file content (mock it to avoid memory issues)
|
||||||
large_content = b"x" * 1024 # 1KB for testing
|
large_content = b"x" * 1024 # 1KB for testing
|
||||||
|
|
||||||
with patch("os.path.getsize") as mock_getsize:
|
with patch("os.path.getsize") as mock_getsize:
|
||||||
# Mock the file size to be over 500MB
|
# Mock the file size to be over 500MB
|
||||||
mock_getsize.return_value = 501 * 1024 * 1024 # 501MB
|
mock_getsize.return_value = 501 * 1024 * 1024 # 501MB
|
||||||
|
|
||||||
response = client.post(
|
response = client.post(
|
||||||
"/api/ui-upload",
|
"/api/ui-upload", files={"file": ("huge.pdf", io.BytesIO(large_content), "application/pdf")}
|
||||||
files={"file": ("huge.pdf", io.BytesIO(large_content), "application/pdf")}
|
|
||||||
)
|
)
|
||||||
|
|
||||||
assert response.status_code == 413 # Request Entity Too Large
|
assert response.status_code == 413 # Request Entity Too Large
|
||||||
assert "too large" in response.json()["detail"].lower()
|
assert "too large" in response.json()["detail"].lower()
|
||||||
|
|
||||||
def test_upload_executable_file(self, client: TestClient, mock_celery_tasks):
|
def test_upload_executable_file(self, client: TestClient, mock_celery_tasks):
|
||||||
"""Test that executable files are handled (attempted conversion)."""
|
"""Test that executable files are handled (attempted conversion)."""
|
||||||
exe_content = b"MZ\x90\x00" # PE header
|
exe_content = b"MZ\x90\x00" # PE header
|
||||||
|
|
||||||
response = client.post(
|
response = client.post(
|
||||||
"/api/ui-upload",
|
"/api/ui-upload", files={"file": ("program.exe", io.BytesIO(exe_content), "application/x-msdownload")}
|
||||||
files={"file": ("program.exe", io.BytesIO(exe_content), "application/x-msdownload")}
|
|
||||||
)
|
)
|
||||||
|
|
||||||
# Per the code, unsupported types get a warning but are still processed
|
# Per the code, unsupported types get a warning but are still processed
|
||||||
assert response.status_code == 200
|
assert response.status_code == 200
|
||||||
# The system attempts conversion even for unsupported types
|
# The system attempts conversion even for unsupported types
|
||||||
mock_celery_tasks["convert_to_pdf"].assert_called_once()
|
mock_celery_tasks["convert_to_pdf"].assert_called_once()
|
||||||
|
|
||||||
def test_upload_empty_file(self, client: TestClient, mock_celery_tasks):
|
def test_upload_empty_file(self, client: TestClient, mock_celery_tasks):
|
||||||
"""Test uploading an empty file."""
|
"""Test uploading an empty file."""
|
||||||
response = client.post(
|
response = client.post("/api/ui-upload", files={"file": ("empty.txt", io.BytesIO(b""), "text/plain")})
|
||||||
"/api/ui-upload",
|
|
||||||
files={"file": ("empty.txt", io.BytesIO(b""), "text/plain")}
|
|
||||||
)
|
|
||||||
|
|
||||||
# Empty files are accepted and queued for processing
|
# Empty files are accepted and queued for processing
|
||||||
assert response.status_code == 200
|
assert response.status_code == 200
|
||||||
data = response.json()
|
data = response.json()
|
||||||
assert data["original_filename"] == "empty.txt"
|
assert data["original_filename"] == "empty.txt"
|
||||||
|
|
||||||
def test_upload_without_file(self, client: TestClient):
|
def test_upload_without_file(self, client: TestClient):
|
||||||
"""Test POST request without a file."""
|
"""Test POST request without a file."""
|
||||||
response = client.post("/api/ui-upload")
|
response = client.post("/api/ui-upload")
|
||||||
|
|
||||||
# FastAPI should return a validation error
|
# FastAPI should return a validation error
|
||||||
assert response.status_code == 422 # Unprocessable Entity
|
assert response.status_code == 422 # Unprocessable Entity
|
||||||
|
|
||||||
@@ -222,56 +206,53 @@ class TestInvalidFileUploads:
|
|||||||
@pytest.mark.security
|
@pytest.mark.security
|
||||||
class TestUploadSecurity:
|
class TestUploadSecurity:
|
||||||
"""Tests for security aspects of file uploads."""
|
"""Tests for security aspects of file uploads."""
|
||||||
|
|
||||||
def test_path_traversal_prevention_dotdot(self, client: TestClient, mock_celery_tasks):
|
def test_path_traversal_prevention_dotdot(self, client: TestClient, mock_celery_tasks):
|
||||||
"""Test that path traversal attempts are prevented."""
|
"""Test that path traversal attempts are prevented."""
|
||||||
# Try to upload a file with path traversal in filename
|
# Try to upload a file with path traversal in filename
|
||||||
malicious_filename = "../../etc/passwd.pdf"
|
malicious_filename = "../../etc/passwd.pdf"
|
||||||
pdf_content = b"%PDF-1.4\n%EOF"
|
pdf_content = b"%PDF-1.4\n%EOF"
|
||||||
|
|
||||||
response = client.post(
|
response = client.post(
|
||||||
"/api/ui-upload",
|
"/api/ui-upload", files={"file": (malicious_filename, io.BytesIO(pdf_content), "application/pdf")}
|
||||||
files={"file": (malicious_filename, io.BytesIO(pdf_content), "application/pdf")}
|
|
||||||
)
|
)
|
||||||
|
|
||||||
assert response.status_code == 200
|
assert response.status_code == 200
|
||||||
data = response.json()
|
data = response.json()
|
||||||
|
|
||||||
# The original filename should be sanitized (only basename)
|
# The original filename should be sanitized (only basename)
|
||||||
assert data["original_filename"] == "passwd.pdf"
|
assert data["original_filename"] == "passwd.pdf"
|
||||||
assert ".." not in data["stored_filename"]
|
assert ".." not in data["stored_filename"]
|
||||||
assert "/" not in data["stored_filename"]
|
assert "/" not in data["stored_filename"]
|
||||||
|
|
||||||
def test_path_traversal_prevention_absolute(self, client: TestClient, mock_celery_tasks):
|
def test_path_traversal_prevention_absolute(self, client: TestClient, mock_celery_tasks):
|
||||||
"""Test that absolute path attempts are prevented."""
|
"""Test that absolute path attempts are prevented."""
|
||||||
malicious_filename = "/etc/shadow.pdf"
|
malicious_filename = "/etc/shadow.pdf"
|
||||||
pdf_content = b"%PDF-1.4\n%EOF"
|
pdf_content = b"%PDF-1.4\n%EOF"
|
||||||
|
|
||||||
response = client.post(
|
response = client.post(
|
||||||
"/api/ui-upload",
|
"/api/ui-upload", files={"file": (malicious_filename, io.BytesIO(pdf_content), "application/pdf")}
|
||||||
files={"file": (malicious_filename, io.BytesIO(pdf_content), "application/pdf")}
|
|
||||||
)
|
)
|
||||||
|
|
||||||
assert response.status_code == 200
|
assert response.status_code == 200
|
||||||
data = response.json()
|
data = response.json()
|
||||||
|
|
||||||
# Only the basename should be kept
|
# Only the basename should be kept
|
||||||
assert data["original_filename"] == "shadow.pdf"
|
assert data["original_filename"] == "shadow.pdf"
|
||||||
assert not data["stored_filename"].startswith("/")
|
assert not data["stored_filename"].startswith("/")
|
||||||
|
|
||||||
def test_filename_with_special_characters(self, client: TestClient, mock_celery_tasks):
|
def test_filename_with_special_characters(self, client: TestClient, mock_celery_tasks):
|
||||||
"""Test handling of filenames with special characters."""
|
"""Test handling of filenames with special characters."""
|
||||||
special_filename = "file name with spaces & special!@#chars.pdf"
|
special_filename = "file name with spaces & special!@#chars.pdf"
|
||||||
pdf_content = b"%PDF-1.4\n%EOF"
|
pdf_content = b"%PDF-1.4\n%EOF"
|
||||||
|
|
||||||
response = client.post(
|
response = client.post(
|
||||||
"/api/ui-upload",
|
"/api/ui-upload", files={"file": (special_filename, io.BytesIO(pdf_content), "application/pdf")}
|
||||||
files={"file": (special_filename, io.BytesIO(pdf_content), "application/pdf")}
|
|
||||||
)
|
)
|
||||||
|
|
||||||
assert response.status_code == 200
|
assert response.status_code == 200
|
||||||
data = response.json()
|
data = response.json()
|
||||||
|
|
||||||
# Original filename should be sanitized (special chars replaced with underscores)
|
# Original filename should be sanitized (special chars replaced with underscores)
|
||||||
assert "!" not in data["original_filename"]
|
assert "!" not in data["original_filename"]
|
||||||
assert "@" not in data["original_filename"]
|
assert "@" not in data["original_filename"]
|
||||||
@@ -281,21 +262,20 @@ class TestUploadSecurity:
|
|||||||
assert ".pdf" in data["original_filename"]
|
assert ".pdf" in data["original_filename"]
|
||||||
# Stored filename should have UUID and extension
|
# Stored filename should have UUID and extension
|
||||||
assert data["stored_filename"].endswith(".pdf")
|
assert data["stored_filename"].endswith(".pdf")
|
||||||
|
|
||||||
def test_path_traversal_prevention_windows_style(self, client: TestClient, mock_celery_tasks):
|
def test_path_traversal_prevention_windows_style(self, client: TestClient, mock_celery_tasks):
|
||||||
"""Test that Windows-style path traversal attempts are prevented."""
|
"""Test that Windows-style path traversal attempts are prevented."""
|
||||||
# Try Windows-style path with backslashes
|
# Try Windows-style path with backslashes
|
||||||
malicious_filename = "..\\..\\..\\windows\\system32\\config.pdf"
|
malicious_filename = "..\\..\\..\\windows\\system32\\config.pdf"
|
||||||
pdf_content = b"%PDF-1.4\n%EOF"
|
pdf_content = b"%PDF-1.4\n%EOF"
|
||||||
|
|
||||||
response = client.post(
|
response = client.post(
|
||||||
"/api/ui-upload",
|
"/api/ui-upload", files={"file": (malicious_filename, io.BytesIO(pdf_content), "application/pdf")}
|
||||||
files={"file": (malicious_filename, io.BytesIO(pdf_content), "application/pdf")}
|
|
||||||
)
|
)
|
||||||
|
|
||||||
assert response.status_code == 200
|
assert response.status_code == 200
|
||||||
data = response.json()
|
data = response.json()
|
||||||
|
|
||||||
# On Unix systems, os.path.basename doesn't recognize Windows backslashes,
|
# On Unix systems, os.path.basename doesn't recognize Windows backslashes,
|
||||||
# so the full string goes to sanitize_filename which then:
|
# so the full string goes to sanitize_filename which then:
|
||||||
# 1. Replaces backslashes with underscores
|
# 1. Replaces backslashes with underscores
|
||||||
@@ -308,20 +288,19 @@ class TestUploadSecurity:
|
|||||||
assert "\\" not in data["original_filename"]
|
assert "\\" not in data["original_filename"]
|
||||||
assert ".." not in data["original_filename"]
|
assert ".." not in data["original_filename"]
|
||||||
assert "/" not in data["original_filename"]
|
assert "/" not in data["original_filename"]
|
||||||
|
|
||||||
def test_path_traversal_prevention_mixed_separators(self, client: TestClient, mock_celery_tasks):
|
def test_path_traversal_prevention_mixed_separators(self, client: TestClient, mock_celery_tasks):
|
||||||
"""Test handling of filenames with mixed path separators."""
|
"""Test handling of filenames with mixed path separators."""
|
||||||
malicious_filename = "../path\\to/file.pdf"
|
malicious_filename = "../path\\to/file.pdf"
|
||||||
pdf_content = b"%PDF-1.4\n%EOF"
|
pdf_content = b"%PDF-1.4\n%EOF"
|
||||||
|
|
||||||
response = client.post(
|
response = client.post(
|
||||||
"/api/ui-upload",
|
"/api/ui-upload", files={"file": (malicious_filename, io.BytesIO(pdf_content), "application/pdf")}
|
||||||
files={"file": (malicious_filename, io.BytesIO(pdf_content), "application/pdf")}
|
|
||||||
)
|
)
|
||||||
|
|
||||||
assert response.status_code == 200
|
assert response.status_code == 200
|
||||||
data = response.json()
|
data = response.json()
|
||||||
|
|
||||||
# os.path.basename recognizes Unix / separators and extracts "file.pdf"
|
# os.path.basename recognizes Unix / separators and extracts "file.pdf"
|
||||||
# sanitize_filename then ensures it's safe (already is in this case)
|
# sanitize_filename then ensures it's safe (already is in this case)
|
||||||
expected_filename = "file.pdf"
|
expected_filename = "file.pdf"
|
||||||
@@ -335,75 +314,68 @@ class TestUploadSecurity:
|
|||||||
@pytest.mark.integration
|
@pytest.mark.integration
|
||||||
class TestUploadErrorHandling:
|
class TestUploadErrorHandling:
|
||||||
"""Tests for error handling during file uploads."""
|
"""Tests for error handling during file uploads."""
|
||||||
|
|
||||||
def test_upload_disk_write_failure(self, client: TestClient):
|
def test_upload_disk_write_failure(self, client: TestClient):
|
||||||
"""Test handling of disk write failures."""
|
"""Test handling of disk write failures."""
|
||||||
with patch("builtins.open", side_effect=IOError("Disk full")):
|
with patch("builtins.open", side_effect=IOError("Disk full")):
|
||||||
pdf_content = b"%PDF-1.4\n%EOF"
|
pdf_content = b"%PDF-1.4\n%EOF"
|
||||||
|
|
||||||
response = client.post(
|
response = client.post(
|
||||||
"/api/ui-upload",
|
"/api/ui-upload", files={"file": ("test.pdf", io.BytesIO(pdf_content), "application/pdf")}
|
||||||
files={"file": ("test.pdf", io.BytesIO(pdf_content), "application/pdf")}
|
|
||||||
)
|
)
|
||||||
|
|
||||||
assert response.status_code == 500
|
assert response.status_code == 500
|
||||||
assert "Failed to save file" in response.json()["detail"]
|
assert "Failed to save file" in response.json()["detail"]
|
||||||
|
|
||||||
def test_upload_celery_task_failure(self, client: TestClient, mock_celery_tasks):
|
def test_upload_celery_task_failure(self, client: TestClient, mock_celery_tasks):
|
||||||
"""Test handling when Celery task queueing fails."""
|
"""Test handling when Celery task queueing fails."""
|
||||||
mock_celery_tasks["process_document"].side_effect = Exception("Celery connection failed")
|
mock_celery_tasks["process_document"].side_effect = Exception("Celery connection failed")
|
||||||
|
|
||||||
pdf_content = b"%PDF-1.4\n%EOF"
|
pdf_content = b"%PDF-1.4\n%EOF"
|
||||||
|
|
||||||
# The endpoint should still handle the error gracefully
|
# The endpoint should still handle the error gracefully
|
||||||
# In this case, the exception will propagate
|
# In this case, the exception will propagate
|
||||||
with pytest.raises(Exception):
|
with pytest.raises(Exception):
|
||||||
client.post(
|
client.post("/api/ui-upload", files={"file": ("test.pdf", io.BytesIO(pdf_content), "application/pdf")})
|
||||||
"/api/ui-upload",
|
|
||||||
files={"file": ("test.pdf", io.BytesIO(pdf_content), "application/pdf")}
|
|
||||||
)
|
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.integration
|
@pytest.mark.integration
|
||||||
class TestUploadFilenameHandling:
|
class TestUploadFilenameHandling:
|
||||||
"""Tests for filename handling and UUID generation."""
|
"""Tests for filename handling and UUID generation."""
|
||||||
|
|
||||||
def test_unique_filename_generation(self, client: TestClient, mock_celery_tasks):
|
def test_unique_filename_generation(self, client: TestClient, mock_celery_tasks):
|
||||||
"""Test that uploaded files get unique UUIDs."""
|
"""Test that uploaded files get unique UUIDs."""
|
||||||
pdf_content = b"%PDF-1.4\n%EOF"
|
pdf_content = b"%PDF-1.4\n%EOF"
|
||||||
|
|
||||||
# Upload same file twice
|
# Upload same file twice
|
||||||
response1 = client.post(
|
response1 = client.post(
|
||||||
"/api/ui-upload",
|
"/api/ui-upload", files={"file": ("same.pdf", io.BytesIO(pdf_content), "application/pdf")}
|
||||||
files={"file": ("same.pdf", io.BytesIO(pdf_content), "application/pdf")}
|
|
||||||
)
|
)
|
||||||
|
|
||||||
response2 = client.post(
|
response2 = client.post(
|
||||||
"/api/ui-upload",
|
"/api/ui-upload", files={"file": ("same.pdf", io.BytesIO(pdf_content), "application/pdf")}
|
||||||
files={"file": ("same.pdf", io.BytesIO(pdf_content), "application/pdf")}
|
|
||||||
)
|
)
|
||||||
|
|
||||||
assert response1.status_code == 200
|
assert response1.status_code == 200
|
||||||
assert response2.status_code == 200
|
assert response2.status_code == 200
|
||||||
|
|
||||||
data1 = response1.json()
|
data1 = response1.json()
|
||||||
data2 = response2.json()
|
data2 = response2.json()
|
||||||
|
|
||||||
# Original filenames should be the same
|
# Original filenames should be the same
|
||||||
assert data1["original_filename"] == data2["original_filename"] == "same.pdf"
|
assert data1["original_filename"] == data2["original_filename"] == "same.pdf"
|
||||||
|
|
||||||
# But stored filenames should be different (unique UUIDs)
|
# But stored filenames should be different (unique UUIDs)
|
||||||
assert data1["stored_filename"] != data2["stored_filename"]
|
assert data1["stored_filename"] != data2["stored_filename"]
|
||||||
|
|
||||||
def test_filename_without_extension(self, client: TestClient, mock_celery_tasks):
|
def test_filename_without_extension(self, client: TestClient, mock_celery_tasks):
|
||||||
"""Test handling of files without extensions."""
|
"""Test handling of files without extensions."""
|
||||||
content = b"Some content"
|
content = b"Some content"
|
||||||
|
|
||||||
response = client.post(
|
response = client.post(
|
||||||
"/api/ui-upload",
|
"/api/ui-upload", files={"file": ("NOEXTENSION", io.BytesIO(content), "application/octet-stream")}
|
||||||
files={"file": ("NOEXTENSION", io.BytesIO(content), "application/octet-stream")}
|
|
||||||
)
|
)
|
||||||
|
|
||||||
assert response.status_code == 200
|
assert response.status_code == 200
|
||||||
data = response.json()
|
data = response.json()
|
||||||
assert data["original_filename"] == "NOEXTENSION"
|
assert data["original_filename"] == "NOEXTENSION"
|
||||||
@@ -414,30 +386,28 @@ class TestUploadFilenameHandling:
|
|||||||
@pytest.mark.integration
|
@pytest.mark.integration
|
||||||
class TestUploadMimeTypeDetection:
|
class TestUploadMimeTypeDetection:
|
||||||
"""Tests for MIME type detection and routing."""
|
"""Tests for MIME type detection and routing."""
|
||||||
|
|
||||||
def test_pdf_by_extension_only(self, client: TestClient, mock_celery_tasks):
|
def test_pdf_by_extension_only(self, client: TestClient, mock_celery_tasks):
|
||||||
"""Test PDF detection by file extension when MIME type is generic."""
|
"""Test PDF detection by file extension when MIME type is generic."""
|
||||||
pdf_content = b"%PDF-1.4\n%EOF"
|
pdf_content = b"%PDF-1.4\n%EOF"
|
||||||
|
|
||||||
response = client.post(
|
response = client.post(
|
||||||
"/api/ui-upload",
|
"/api/ui-upload", files={"file": ("doc.pdf", io.BytesIO(pdf_content), "application/octet-stream")}
|
||||||
files={"file": ("doc.pdf", io.BytesIO(pdf_content), "application/octet-stream")}
|
|
||||||
)
|
)
|
||||||
|
|
||||||
assert response.status_code == 200
|
assert response.status_code == 200
|
||||||
# Should route to process_document (not convert_to_pdf) based on extension
|
# Should route to process_document (not convert_to_pdf) based on extension
|
||||||
mock_celery_tasks["process_document"].assert_called_once()
|
mock_celery_tasks["process_document"].assert_called_once()
|
||||||
|
|
||||||
def test_image_by_extension(self, client: TestClient, mock_celery_tasks):
|
def test_image_by_extension(self, client: TestClient, mock_celery_tasks):
|
||||||
"""Test image detection by file extension."""
|
"""Test image detection by file extension."""
|
||||||
# Generic binary content with image extension
|
# Generic binary content with image extension
|
||||||
image_content = b"\x00\x01\x02\x03"
|
image_content = b"\x00\x01\x02\x03"
|
||||||
|
|
||||||
response = client.post(
|
response = client.post(
|
||||||
"/api/ui-upload",
|
"/api/ui-upload", files={"file": ("photo.jpg", io.BytesIO(image_content), "application/octet-stream")}
|
||||||
files={"file": ("photo.jpg", io.BytesIO(image_content), "application/octet-stream")}
|
|
||||||
)
|
)
|
||||||
|
|
||||||
assert response.status_code == 200
|
assert response.status_code == 200
|
||||||
# Should route to convert_to_pdf based on .jpg extension
|
# Should route to convert_to_pdf based on .jpg extension
|
||||||
mock_celery_tasks["convert_to_pdf"].assert_called_once()
|
mock_celery_tasks["convert_to_pdf"].assert_called_once()
|
||||||
|
|||||||
Reference in New Issue
Block a user