test: Increase coverage for url_upload.py and files.py to 90%+
Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com>
This commit is contained in:
copilot-swe-agent[bot]
@@ -0,0 +1,609 @@
|
||||
"""
|
||||
Extended tests for app/views/files.py to achieve 90%+ coverage.
|
||||
"""
|
||||
|
||||
import json
|
||||
import os
|
||||
from datetime import datetime
|
||||
from unittest.mock import Mock, patch
|
||||
|
||||
import pytest
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
from app.models import FileRecord, ProcessingLog
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
@pytest.mark.requires_db
|
||||
class TestFilesPageExtended:
|
||||
"""Extended tests for the /files page view."""
|
||||
|
||||
def test_files_page_with_search_filter(self, client: TestClient, db_session):
|
||||
"""Test that search filter works correctly"""
|
||||
# Create files with different names
|
||||
for i in range(5):
|
||||
file_record = FileRecord(
|
||||
filehash=f"hash{i}",
|
||||
original_filename=f"invoice_{i}.pdf" if i < 3 else f"report_{i}.pdf",
|
||||
local_filename=f"/tmp/test{i}.pdf",
|
||||
file_size=1024,
|
||||
mime_type="application/pdf",
|
||||
)
|
||||
db_session.add(file_record)
|
||||
db_session.commit()
|
||||
|
||||
# Search for "invoice"
|
||||
response = client.get("/files?search=invoice")
|
||||
assert response.status_code == 200
|
||||
content = response.text
|
||||
assert "invoice" in content
|
||||
# Should not show all files
|
||||
assert "report_4" not in content or content.count("report_") < 2
|
||||
|
||||
def test_files_page_with_mime_type_filter(self, client: TestClient, db_session):
|
||||
"""Test that MIME type filter works correctly"""
|
||||
# Create files with different MIME types
|
||||
file1 = FileRecord(
|
||||
filehash="hash1",
|
||||
original_filename="document.pdf",
|
||||
local_filename="/tmp/test1.pdf",
|
||||
file_size=1024,
|
||||
mime_type="application/pdf",
|
||||
)
|
||||
file2 = FileRecord(
|
||||
filehash="hash2",
|
||||
original_filename="image.jpg",
|
||||
local_filename="/tmp/test2.jpg",
|
||||
file_size=2048,
|
||||
mime_type="image/jpeg",
|
||||
)
|
||||
db_session.add(file1)
|
||||
db_session.add(file2)
|
||||
db_session.commit()
|
||||
|
||||
# Filter by PDF MIME type
|
||||
response = client.get("/files?mime_type=application/pdf")
|
||||
assert response.status_code == 200
|
||||
content = response.text
|
||||
assert "document.pdf" in content
|
||||
|
||||
def test_files_page_with_sorting(self, client: TestClient, db_session):
|
||||
"""Test that sorting works correctly"""
|
||||
# Create files with different sizes
|
||||
for i in range(3):
|
||||
file_record = FileRecord(
|
||||
filehash=f"hash{i}",
|
||||
original_filename=f"file_{i}.pdf",
|
||||
local_filename=f"/tmp/test{i}.pdf",
|
||||
file_size=1024 * (i + 1),
|
||||
mime_type="application/pdf",
|
||||
)
|
||||
db_session.add(file_record)
|
||||
db_session.commit()
|
||||
|
||||
# Sort by file_size ascending
|
||||
response = client.get("/files?sort_by=file_size&sort_order=asc")
|
||||
assert response.status_code == 200
|
||||
|
||||
# Sort by file_size descending
|
||||
response = client.get("/files?sort_by=file_size&sort_order=desc")
|
||||
assert response.status_code == 200
|
||||
|
||||
def test_files_page_pagination(self, client: TestClient, db_session):
|
||||
"""Test pagination with different page sizes"""
|
||||
# Create many files
|
||||
for i in range(25):
|
||||
file_record = FileRecord(
|
||||
filehash=f"hash{i}",
|
||||
original_filename=f"test{i}.pdf",
|
||||
local_filename=f"/tmp/test{i}.pdf",
|
||||
file_size=1024,
|
||||
mime_type="application/pdf",
|
||||
)
|
||||
db_session.add(file_record)
|
||||
db_session.commit()
|
||||
|
||||
# Test page 1
|
||||
response = client.get("/files?page=1&per_page=10")
|
||||
assert response.status_code == 200
|
||||
|
||||
# Test page 2
|
||||
response = client.get("/files?page=2&per_page=10")
|
||||
assert response.status_code == 200
|
||||
|
||||
def test_files_page_error_handling(self, client: TestClient, db_session):
|
||||
"""Test that errors are handled gracefully"""
|
||||
# Test with invalid page number (should default to 1)
|
||||
response = client.get("/files?page=0")
|
||||
# FastAPI query validation should reject page=0
|
||||
assert response.status_code in [200, 422] # Either works or validation error
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
@pytest.mark.requires_db
|
||||
class TestFileDetailPage:
|
||||
"""Tests for the /files/{file_id}/detail page."""
|
||||
|
||||
def test_file_detail_page_with_existing_file(self, client: TestClient, db_session, tmp_path):
|
||||
"""Test file detail page with an existing file"""
|
||||
# Create a test file
|
||||
original_file = tmp_path / "original.pdf"
|
||||
original_file.write_bytes(b"PDF content")
|
||||
|
||||
# Create file record
|
||||
file_record = FileRecord(
|
||||
filehash="testhash",
|
||||
original_filename="test.pdf",
|
||||
local_filename="/tmp/test.pdf",
|
||||
original_file_path=str(original_file),
|
||||
file_size=1024,
|
||||
mime_type="application/pdf",
|
||||
)
|
||||
db_session.add(file_record)
|
||||
db_session.commit()
|
||||
|
||||
# Test file detail page
|
||||
response = client.get(f"/files/{file_record.id}/detail")
|
||||
assert response.status_code == 200
|
||||
content = response.text
|
||||
assert "test.pdf" in content
|
||||
|
||||
def test_file_detail_page_with_missing_file(self, client: TestClient, db_session):
|
||||
"""Test file detail page with non-existent file"""
|
||||
# Try to access non-existent file
|
||||
response = client.get("/files/99999/detail")
|
||||
assert response.status_code == 200
|
||||
content = response.text
|
||||
assert "not found" in content.lower()
|
||||
|
||||
def test_file_detail_page_with_processing_logs(self, client: TestClient, db_session, tmp_path):
|
||||
"""Test file detail page shows processing logs"""
|
||||
# Create a test file
|
||||
original_file = tmp_path / "original.pdf"
|
||||
original_file.write_bytes(b"PDF content")
|
||||
|
||||
# Create file record
|
||||
file_record = FileRecord(
|
||||
filehash="testhash",
|
||||
original_filename="test.pdf",
|
||||
local_filename="/tmp/test.pdf",
|
||||
original_file_path=str(original_file),
|
||||
file_size=1024,
|
||||
mime_type="application/pdf",
|
||||
)
|
||||
db_session.add(file_record)
|
||||
db_session.commit()
|
||||
|
||||
# Add processing logs
|
||||
log1 = ProcessingLog(
|
||||
file_id=file_record.id,
|
||||
step_name="create_file_record",
|
||||
status="success",
|
||||
message="File record created",
|
||||
timestamp=datetime.utcnow(),
|
||||
)
|
||||
log2 = ProcessingLog(
|
||||
file_id=file_record.id,
|
||||
step_name="extract_text",
|
||||
status="success",
|
||||
message="Text extracted",
|
||||
timestamp=datetime.utcnow(),
|
||||
)
|
||||
db_session.add(log1)
|
||||
db_session.add(log2)
|
||||
db_session.commit()
|
||||
|
||||
# Test file detail page
|
||||
response = client.get(f"/files/{file_record.id}/detail")
|
||||
assert response.status_code == 200
|
||||
content = response.text
|
||||
assert "create_file_record" in content
|
||||
assert "extract_text" in content
|
||||
|
||||
def test_file_detail_page_with_metadata(self, client: TestClient, db_session, tmp_path):
|
||||
"""Test file detail page shows metadata when available"""
|
||||
# Create test files
|
||||
original_file = tmp_path / "original.pdf"
|
||||
processed_file = tmp_path / "processed.pdf"
|
||||
metadata_file = tmp_path / "processed.json"
|
||||
|
||||
original_file.write_bytes(b"PDF content")
|
||||
processed_file.write_bytes(b"Processed PDF content")
|
||||
|
||||
# Create metadata JSON
|
||||
metadata = {
|
||||
"document_type": "invoice",
|
||||
"amount": 150.00,
|
||||
"date": "2024-01-15",
|
||||
"vendor": "Test Vendor",
|
||||
}
|
||||
metadata_file.write_text(json.dumps(metadata))
|
||||
|
||||
# Create file record
|
||||
file_record = FileRecord(
|
||||
filehash="testhash",
|
||||
original_filename="test.pdf",
|
||||
local_filename="/tmp/test.pdf",
|
||||
original_file_path=str(original_file),
|
||||
processed_file_path=str(processed_file),
|
||||
file_size=1024,
|
||||
mime_type="application/pdf",
|
||||
)
|
||||
db_session.add(file_record)
|
||||
db_session.commit()
|
||||
|
||||
# Test file detail page
|
||||
response = client.get(f"/files/{file_record.id}/detail")
|
||||
assert response.status_code == 200
|
||||
content = response.text
|
||||
# Should show metadata
|
||||
assert "invoice" in content or "metadata" in content.lower()
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
@pytest.mark.requires_db
|
||||
class TestFilePreviewEndpoints:
|
||||
"""Tests for file preview endpoints."""
|
||||
|
||||
def test_preview_original_file_success(self, client: TestClient, db_session, tmp_path):
|
||||
"""Test successful preview of original file"""
|
||||
# Create a test PDF file
|
||||
original_file = tmp_path / "original.pdf"
|
||||
original_file.write_bytes(b"%PDF-1.4\nTest PDF content")
|
||||
|
||||
# Create file record
|
||||
file_record = FileRecord(
|
||||
filehash="testhash",
|
||||
original_filename="test.pdf",
|
||||
local_filename="/tmp/test.pdf",
|
||||
original_file_path=str(original_file),
|
||||
file_size=1024,
|
||||
mime_type="application/pdf",
|
||||
)
|
||||
db_session.add(file_record)
|
||||
db_session.commit()
|
||||
|
||||
# Test preview endpoint
|
||||
response = client.get(f"/files/{file_record.id}/preview/original")
|
||||
assert response.status_code == 200
|
||||
assert response.headers["content-type"] == "application/pdf"
|
||||
|
||||
def test_preview_original_file_not_found(self, client: TestClient, db_session):
|
||||
"""Test preview when file doesn't exist"""
|
||||
response = client.get("/files/99999/preview/original")
|
||||
assert response.status_code == 404
|
||||
|
||||
def test_preview_original_file_missing_on_disk(self, client: TestClient, db_session):
|
||||
"""Test preview when file record exists but file is missing on disk"""
|
||||
# Create file record with non-existent path
|
||||
file_record = FileRecord(
|
||||
filehash="testhash",
|
||||
original_filename="test.pdf",
|
||||
local_filename="/tmp/test.pdf",
|
||||
original_file_path="/nonexistent/file.pdf",
|
||||
file_size=1024,
|
||||
mime_type="application/pdf",
|
||||
)
|
||||
db_session.add(file_record)
|
||||
db_session.commit()
|
||||
|
||||
# Test preview endpoint
|
||||
response = client.get(f"/files/{file_record.id}/preview/original")
|
||||
assert response.status_code == 404
|
||||
|
||||
def test_preview_processed_file_success(self, client: TestClient, db_session, tmp_path):
|
||||
"""Test successful preview of processed file"""
|
||||
# Create a test PDF file
|
||||
processed_file = tmp_path / "processed.pdf"
|
||||
processed_file.write_bytes(b"%PDF-1.4\nProcessed PDF content")
|
||||
|
||||
# Create file record
|
||||
file_record = FileRecord(
|
||||
filehash="testhash",
|
||||
original_filename="test.pdf",
|
||||
local_filename="/tmp/test.pdf",
|
||||
processed_file_path=str(processed_file),
|
||||
file_size=1024,
|
||||
mime_type="application/pdf",
|
||||
)
|
||||
db_session.add(file_record)
|
||||
db_session.commit()
|
||||
|
||||
# Test preview endpoint
|
||||
response = client.get(f"/files/{file_record.id}/preview/processed")
|
||||
assert response.status_code == 200
|
||||
assert response.headers["content-type"] == "application/pdf"
|
||||
|
||||
def test_preview_processed_file_not_found(self, client: TestClient, db_session):
|
||||
"""Test preview when file doesn't exist"""
|
||||
response = client.get("/files/99999/preview/processed")
|
||||
assert response.status_code == 404
|
||||
|
||||
def test_preview_processed_file_missing_on_disk(self, client: TestClient, db_session):
|
||||
"""Test preview when file record exists but file is missing on disk"""
|
||||
# Create file record with non-existent path
|
||||
file_record = FileRecord(
|
||||
filehash="testhash",
|
||||
original_filename="test.pdf",
|
||||
local_filename="/tmp/test.pdf",
|
||||
processed_file_path="/nonexistent/processed.pdf",
|
||||
file_size=1024,
|
||||
mime_type="application/pdf",
|
||||
)
|
||||
db_session.add(file_record)
|
||||
db_session.commit()
|
||||
|
||||
# Test preview endpoint
|
||||
response = client.get(f"/files/{file_record.id}/preview/processed")
|
||||
assert response.status_code == 404
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
@pytest.mark.requires_db
|
||||
class TestTextExtractionEndpoints:
|
||||
"""Tests for text extraction endpoints."""
|
||||
|
||||
def test_get_original_text_success(self, client: TestClient, db_session, sample_pdf_file):
|
||||
"""Test successful text extraction from original file"""
|
||||
# Create file record
|
||||
file_record = FileRecord(
|
||||
filehash="testhash",
|
||||
original_filename="test.pdf",
|
||||
local_filename="/tmp/test.pdf",
|
||||
original_file_path=str(sample_pdf_file),
|
||||
file_size=1024,
|
||||
mime_type="application/pdf",
|
||||
)
|
||||
db_session.add(file_record)
|
||||
db_session.commit()
|
||||
|
||||
# Test text extraction endpoint
|
||||
response = client.get(f"/files/{file_record.id}/text/original")
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert "text" in data
|
||||
assert "page_count" in data
|
||||
|
||||
def test_get_original_text_file_not_found(self, client: TestClient, db_session):
|
||||
"""Test text extraction when file doesn't exist"""
|
||||
response = client.get("/files/99999/text/original")
|
||||
assert response.status_code == 404
|
||||
|
||||
def test_get_original_text_file_missing_on_disk(self, client: TestClient, db_session):
|
||||
"""Test text extraction when file is missing on disk"""
|
||||
# Create file record with non-existent path
|
||||
file_record = FileRecord(
|
||||
filehash="testhash",
|
||||
original_filename="test.pdf",
|
||||
local_filename="/tmp/test.pdf",
|
||||
original_file_path="/nonexistent/file.pdf",
|
||||
file_size=1024,
|
||||
mime_type="application/pdf",
|
||||
)
|
||||
db_session.add(file_record)
|
||||
db_session.commit()
|
||||
|
||||
# Test text extraction endpoint
|
||||
response = client.get(f"/files/{file_record.id}/text/original")
|
||||
assert response.status_code == 404
|
||||
|
||||
def test_get_original_text_extraction_error(self, client: TestClient, db_session, tmp_path):
|
||||
"""Test text extraction when PDF is invalid"""
|
||||
# Create an invalid PDF file
|
||||
invalid_pdf = tmp_path / "invalid.pdf"
|
||||
invalid_pdf.write_bytes(b"Not a valid PDF")
|
||||
|
||||
# Create file record
|
||||
file_record = FileRecord(
|
||||
filehash="testhash",
|
||||
original_filename="test.pdf",
|
||||
local_filename="/tmp/test.pdf",
|
||||
original_file_path=str(invalid_pdf),
|
||||
file_size=1024,
|
||||
mime_type="application/pdf",
|
||||
)
|
||||
db_session.add(file_record)
|
||||
db_session.commit()
|
||||
|
||||
# Test text extraction endpoint
|
||||
response = client.get(f"/files/{file_record.id}/text/original")
|
||||
assert response.status_code == 500
|
||||
|
||||
def test_get_processed_text_success(self, client: TestClient, db_session, sample_pdf_file):
|
||||
"""Test successful text extraction from processed file"""
|
||||
# Create file record
|
||||
file_record = FileRecord(
|
||||
filehash="testhash",
|
||||
original_filename="test.pdf",
|
||||
local_filename="/tmp/test.pdf",
|
||||
processed_file_path=str(sample_pdf_file),
|
||||
file_size=1024,
|
||||
mime_type="application/pdf",
|
||||
)
|
||||
db_session.add(file_record)
|
||||
db_session.commit()
|
||||
|
||||
# Test text extraction endpoint
|
||||
response = client.get(f"/files/{file_record.id}/text/processed")
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert "text" in data
|
||||
assert "page_count" in data
|
||||
|
||||
def test_get_processed_text_file_not_found(self, client: TestClient, db_session):
|
||||
"""Test text extraction when file doesn't exist"""
|
||||
response = client.get("/files/99999/text/processed")
|
||||
assert response.status_code == 404
|
||||
|
||||
def test_get_processed_text_file_missing_on_disk(self, client: TestClient, db_session):
|
||||
"""Test text extraction when file is missing on disk"""
|
||||
# Create file record with non-existent path
|
||||
file_record = FileRecord(
|
||||
filehash="testhash",
|
||||
original_filename="test.pdf",
|
||||
local_filename="/tmp/test.pdf",
|
||||
processed_file_path="/nonexistent/processed.pdf",
|
||||
file_size=1024,
|
||||
mime_type="application/pdf",
|
||||
)
|
||||
db_session.add(file_record)
|
||||
db_session.commit()
|
||||
|
||||
# Test text extraction endpoint
|
||||
response = client.get(f"/files/{file_record.id}/text/processed")
|
||||
assert response.status_code == 404
|
||||
|
||||
def test_get_processed_text_extraction_error(self, client: TestClient, db_session, tmp_path):
|
||||
"""Test text extraction when PDF is invalid"""
|
||||
# Create an invalid PDF file
|
||||
invalid_pdf = tmp_path / "invalid.pdf"
|
||||
invalid_pdf.write_bytes(b"Not a valid PDF")
|
||||
|
||||
# Create file record
|
||||
file_record = FileRecord(
|
||||
filehash="testhash",
|
||||
original_filename="test.pdf",
|
||||
local_filename="/tmp/test.pdf",
|
||||
processed_file_path=str(invalid_pdf),
|
||||
file_size=1024,
|
||||
mime_type="application/pdf",
|
||||
)
|
||||
db_session.add(file_record)
|
||||
db_session.commit()
|
||||
|
||||
# Test text extraction endpoint
|
||||
response = client.get(f"/files/{file_record.id}/text/processed")
|
||||
assert response.status_code == 500
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestProcessingFlowFunctions:
|
||||
"""Unit tests for processing flow helper functions."""
|
||||
|
||||
def test_compute_processing_flow_basic(self):
|
||||
"""Test _compute_processing_flow with basic logs"""
|
||||
from app.views.files import _compute_processing_flow
|
||||
|
||||
# Create mock logs
|
||||
logs = [
|
||||
Mock(
|
||||
step_name="create_file_record",
|
||||
status="success",
|
||||
message="File created",
|
||||
timestamp=datetime.utcnow(),
|
||||
task_id="task1",
|
||||
),
|
||||
Mock(
|
||||
step_name="extract_text",
|
||||
status="success",
|
||||
message="Text extracted",
|
||||
timestamp=datetime.utcnow(),
|
||||
task_id="task2",
|
||||
),
|
||||
]
|
||||
|
||||
flow = _compute_processing_flow(logs)
|
||||
assert isinstance(flow, list)
|
||||
assert len(flow) > 0
|
||||
# Check that stages have expected keys
|
||||
for stage in flow:
|
||||
assert "key" in stage
|
||||
assert "label" in stage
|
||||
assert "status" in stage
|
||||
|
||||
def test_compute_processing_flow_with_uploads(self):
|
||||
"""Test _compute_processing_flow with upload branches"""
|
||||
from app.views.files import _compute_processing_flow
|
||||
|
||||
# Create mock logs including upload tasks
|
||||
logs = [
|
||||
Mock(
|
||||
step_name="send_to_all_destinations",
|
||||
status="success",
|
||||
message="Distribution queued",
|
||||
timestamp=datetime.utcnow(),
|
||||
task_id="task1",
|
||||
),
|
||||
Mock(
|
||||
step_name="upload_to_dropbox",
|
||||
status="success",
|
||||
message="Uploaded to Dropbox",
|
||||
timestamp=datetime.utcnow(),
|
||||
task_id="task2",
|
||||
),
|
||||
Mock(
|
||||
step_name="upload_to_google_drive",
|
||||
status="failure",
|
||||
message="Failed to upload",
|
||||
timestamp=datetime.utcnow(),
|
||||
task_id="task3",
|
||||
),
|
||||
]
|
||||
|
||||
flow = _compute_processing_flow(logs)
|
||||
assert isinstance(flow, list)
|
||||
|
||||
# Find the upload stage
|
||||
upload_stage = next((s for s in flow if s.get("is_branch_parent")), None)
|
||||
if upload_stage:
|
||||
assert "branches" in upload_stage
|
||||
assert len(upload_stage["branches"]) > 0
|
||||
|
||||
def test_compute_step_summary_basic(self):
|
||||
"""Test _compute_step_summary with basic logs"""
|
||||
from app.views.files import _compute_step_summary
|
||||
|
||||
# Create mock logs
|
||||
logs = [
|
||||
Mock(
|
||||
step_name="create_file_record",
|
||||
status="success",
|
||||
timestamp=datetime.utcnow(),
|
||||
),
|
||||
Mock(
|
||||
step_name="extract_text",
|
||||
status="success",
|
||||
timestamp=datetime.utcnow(),
|
||||
),
|
||||
Mock(
|
||||
step_name="upload_to_dropbox",
|
||||
status="failure",
|
||||
timestamp=datetime.utcnow(),
|
||||
),
|
||||
]
|
||||
|
||||
summary = _compute_step_summary(logs)
|
||||
assert isinstance(summary, dict)
|
||||
assert "main" in summary
|
||||
assert "uploads" in summary
|
||||
assert "total_main_steps" in summary
|
||||
assert "total_upload_tasks" in summary
|
||||
|
||||
# Check that counts are correct
|
||||
assert summary["main"]["success"] >= 2
|
||||
assert summary["uploads"]["failure"] >= 1
|
||||
|
||||
def test_compute_step_summary_order_independent(self):
|
||||
"""Test that _compute_step_summary is order-independent"""
|
||||
from app.views.files import _compute_step_summary
|
||||
|
||||
# Create logs with same step appearing multiple times
|
||||
earlier_time = datetime(2024, 1, 1, 10, 0, 0)
|
||||
later_time = datetime(2024, 1, 1, 11, 0, 0)
|
||||
|
||||
logs_forward = [
|
||||
Mock(step_name="extract_text", status="queued", timestamp=earlier_time),
|
||||
Mock(step_name="extract_text", status="success", timestamp=later_time),
|
||||
]
|
||||
|
||||
logs_backward = [
|
||||
Mock(step_name="extract_text", status="success", timestamp=later_time),
|
||||
Mock(step_name="extract_text", status="queued", timestamp=earlier_time),
|
||||
]
|
||||
|
||||
summary_forward = _compute_step_summary(logs_forward)
|
||||
summary_backward = _compute_step_summary(logs_backward)
|
||||
|
||||
# Should be the same regardless of log order
|
||||
assert summary_forward["main"]["success"] == summary_backward["main"]["success"]
|
||||
assert summary_forward["main"]["queued"] == summary_backward["main"]["queued"]
|
||||
@@ -411,3 +411,179 @@ class TestURLUploadEndpoint:
|
||||
|
||||
# Should not process document
|
||||
mock_process_document.delay.assert_not_called()
|
||||
|
||||
@patch("app.api.url_upload.requests.get")
|
||||
def test_process_url_request_exception(self, mock_requests_get, client):
|
||||
"""Test handling of generic RequestException"""
|
||||
mock_requests_get.side_effect = requests.exceptions.RequestException("Generic request error")
|
||||
|
||||
response = client.post("/api/process-url", json={"url": "https://example.com/file.pdf"})
|
||||
|
||||
assert response.status_code == 500
|
||||
data = response.json()
|
||||
assert "Failed to download file" in data["detail"]
|
||||
|
||||
@patch("app.api.url_upload.requests.get")
|
||||
def test_process_url_oserror_during_save(self, mock_requests_get, client, tmp_path, monkeypatch):
|
||||
"""Test handling of OSError when saving file"""
|
||||
# Mock successful download
|
||||
mock_response = Mock()
|
||||
mock_response.status_code = 200
|
||||
mock_response.headers = {"Content-Type": "application/pdf", "Content-Length": "100"}
|
||||
mock_response.iter_content = Mock(return_value=[b"PDF"])
|
||||
mock_response.raise_for_status = Mock()
|
||||
mock_requests_get.return_value = mock_response
|
||||
|
||||
# Mock workdir to a non-existent path to trigger OSError
|
||||
from app.config import settings
|
||||
|
||||
original_workdir = settings.workdir
|
||||
monkeypatch.setattr(settings, "workdir", "/nonexistent/path/that/does/not/exist")
|
||||
|
||||
try:
|
||||
response = client.post("/api/process-url", json={"url": "https://example.com/file.pdf"})
|
||||
|
||||
assert response.status_code == 500
|
||||
data = response.json()
|
||||
assert "Failed to save file" in data["detail"]
|
||||
finally:
|
||||
# Restore original workdir
|
||||
monkeypatch.setattr(settings, "workdir", original_workdir)
|
||||
|
||||
@patch("app.api.url_upload.requests.get")
|
||||
@patch("app.api.url_upload.process_document")
|
||||
def test_process_url_unexpected_exception(self, mock_process_document, mock_requests_get, client):
|
||||
"""Test handling of unexpected exceptions"""
|
||||
# Mock successful download but process_document.delay raises unexpected error
|
||||
mock_response = Mock()
|
||||
mock_response.status_code = 200
|
||||
mock_response.headers = {"Content-Type": "application/pdf", "Content-Length": "100"}
|
||||
mock_response.iter_content = Mock(return_value=[b"PDF"])
|
||||
mock_response.raise_for_status = Mock()
|
||||
mock_requests_get.return_value = mock_response
|
||||
|
||||
# Mock process_document.delay to raise an unexpected exception
|
||||
mock_process_document.delay.side_effect = RuntimeError("Unexpected processing error")
|
||||
|
||||
response = client.post("/api/process-url", json={"url": "https://example.com/file.pdf"})
|
||||
|
||||
assert response.status_code == 500
|
||||
data = response.json()
|
||||
assert "Unexpected error" in data["detail"]
|
||||
|
||||
@patch("app.api.url_upload.requests.get")
|
||||
@patch("app.api.url_upload.process_document")
|
||||
def test_process_url_filename_without_extension(self, mock_process_document, mock_requests_get, client):
|
||||
"""Test that files without extensions are handled correctly"""
|
||||
# Mock successful download
|
||||
mock_response = Mock()
|
||||
mock_response.status_code = 200
|
||||
mock_response.headers = {"Content-Type": "application/pdf", "Content-Length": "100"}
|
||||
mock_response.iter_content = Mock(return_value=[b"PDF"])
|
||||
mock_response.raise_for_status = Mock()
|
||||
mock_requests_get.return_value = mock_response
|
||||
|
||||
# Mock Celery task
|
||||
mock_task = Mock()
|
||||
mock_task.id = "test-task-id"
|
||||
mock_process_document.delay.return_value = mock_task
|
||||
|
||||
# URL with no extension in path
|
||||
response = client.post("/api/process-url", json={"url": "https://example.com/document", "filename": "noext"})
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
# Should still work, just without extension
|
||||
assert data["task_id"] == "test-task-id"
|
||||
|
||||
@patch("app.api.url_upload.requests.get")
|
||||
@patch("app.api.url_upload.process_document")
|
||||
def test_process_url_empty_path_uses_download(self, mock_process_document, mock_requests_get, client):
|
||||
"""Test that empty URL path defaults to 'download' filename"""
|
||||
# Mock successful download
|
||||
mock_response = Mock()
|
||||
mock_response.status_code = 200
|
||||
mock_response.headers = {"Content-Type": "application/pdf", "Content-Length": "100"}
|
||||
mock_response.iter_content = Mock(return_value=[b"PDF"])
|
||||
mock_response.raise_for_status = Mock()
|
||||
mock_requests_get.return_value = mock_response
|
||||
|
||||
# Mock Celery task
|
||||
mock_task = Mock()
|
||||
mock_task.id = "test-task-id"
|
||||
mock_process_document.delay.return_value = mock_task
|
||||
|
||||
# URL with no path (will default to "download")
|
||||
response = client.post("/api/process-url", json={"url": "https://example.com"})
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["task_id"] == "test-task-id"
|
||||
# Filename should start with "document" when no path is provided (sanitize_filename adds timestamp)
|
||||
assert "document" in data["filename"]
|
||||
|
||||
def test_validate_url_no_hostname(self):
|
||||
"""Test that URLs without hostname are rejected"""
|
||||
from fastapi import HTTPException
|
||||
|
||||
from app.api.url_upload import validate_url_safety
|
||||
|
||||
with pytest.raises(HTTPException) as exc_info:
|
||||
validate_url_safety("http://")
|
||||
assert exc_info.value.status_code == 400
|
||||
assert "no hostname" in exc_info.value.detail
|
||||
|
||||
def test_validate_file_type_by_extension_fallback(self):
|
||||
"""Test that file type validation falls back to extension when content-type is empty"""
|
||||
from app.api.url_upload import validate_file_type
|
||||
|
||||
# Empty content-type but valid extension
|
||||
assert validate_file_type("", "document.pdf") is True
|
||||
assert validate_file_type("", "image.jpg") is True
|
||||
assert validate_file_type("", "spreadsheet.xlsx") is True
|
||||
|
||||
# Empty content-type and invalid extension
|
||||
assert validate_file_type("", "malware.exe") is False
|
||||
assert validate_file_type("", "script.sh") is False
|
||||
|
||||
def test_is_private_ip_ipv6_loopback(self):
|
||||
"""Test that IPv6 loopback is detected as private"""
|
||||
from app.api.url_upload import is_private_ip
|
||||
|
||||
# IPv6 loopback (::1)
|
||||
assert is_private_ip("::1") is True
|
||||
|
||||
def test_is_private_ip_link_local(self):
|
||||
"""Test that link-local addresses are detected as private"""
|
||||
from app.api.url_upload import is_private_ip
|
||||
|
||||
# Link-local address
|
||||
assert is_private_ip("169.254.1.1") is True
|
||||
|
||||
@patch("app.api.url_upload.requests.get")
|
||||
@patch("app.api.url_upload.process_document")
|
||||
def test_process_url_sanitizes_dangerous_filename(self, mock_process_document, mock_requests_get, client):
|
||||
"""Test that dangerous filenames are sanitized"""
|
||||
# Mock successful download
|
||||
mock_response = Mock()
|
||||
mock_response.status_code = 200
|
||||
mock_response.headers = {"Content-Type": "application/pdf", "Content-Length": "100"}
|
||||
mock_response.iter_content = Mock(return_value=[b"PDF"])
|
||||
mock_response.raise_for_status = Mock()
|
||||
mock_requests_get.return_value = mock_response
|
||||
|
||||
# Mock Celery task
|
||||
mock_task = Mock()
|
||||
mock_task.id = "test-task-id"
|
||||
mock_process_document.delay.return_value = mock_task
|
||||
|
||||
# Dangerous filename with path traversal
|
||||
response = client.post(
|
||||
"/api/process-url", json={"url": "https://example.com/file.pdf", "filename": "../../../etc/passwd"}
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
# Filename should be sanitized (no path traversal)
|
||||
assert ".." not in data["filename"]
|
||||
assert "/" not in data["filename"]
|
||||
|
||||
Reference in New Issue
Block a user