copilot-swe-agent[bot]
a644efe016
The file detail page (/files/{id}/detail) and file view page (/files/{id})
previously showed no information about which processing pipeline was used.
Changes:
- _STEP_TYPE_TO_STAGES mapping: pipeline step_type → Celery log stage keys
(with maintenance comment requiring updates when new step types are added)
- _ALWAYS_SHOW_STAGES: stages always visible regardless of pipeline
- _resolve_pipeline(db, file_record): resolves the pipeline for a file —
uses explicit pipeline_id when set, falls back to active system default
- _compute_processing_flow: new pipeline_steps parameter; when provided,
filters flow graph to only show stages for the pipeline's enabled steps
(+ always-show stages + any stage that actually ran). Also adds
convert_to_pdf to the flow stage catalogue.
- file_detail_page: passes pipeline_info + pipeline-filtered flow_data
- file_view_page: passes pipeline_info
Templates:
- file_detail.html: 'Processing Pipeline' detail row with name link and
colour-coded badge (System Default / System / Custom)
- file_view.html: 'Pipeline' info row in sidebar with (default)/(custom) tag
Tests:
- TestPipelineInfoInViews with 14 tests covering _resolve_pipeline,
_compute_processing_flow filtering, completeness assertion for
_STEP_TYPE_TO_STAGES, and HTTP-level view tests
Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com>
1984 lines
73 KiB
Python
1984 lines
73 KiB
Python
"""
|
|
Comprehensive unit tests for app/views/files.py
|
|
|
|
Tests all view endpoints with success and error cases, proper mocking, and edge cases.
|
|
Target: Bring coverage from 8.77% to 70%+
|
|
"""
|
|
|
|
import json
|
|
from datetime import datetime, timedelta
|
|
from unittest.mock import Mock, patch
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
from app.models import FileProcessingStep, FileRecord, ProcessingLog
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestFilesPage:
|
|
"""Tests for GET /files endpoint."""
|
|
|
|
def test_files_page_empty_database(self, client: TestClient, db_session):
|
|
"""Test files page with no files in database."""
|
|
response = client.get("/files")
|
|
assert response.status_code == 200
|
|
assert b"files.html" in response.content or b"Files" in response.content
|
|
|
|
def test_files_page_with_data(self, client: TestClient, db_session):
|
|
"""Test files page with existing files."""
|
|
file1 = FileRecord(
|
|
filehash="hash1",
|
|
original_filename="test1.pdf",
|
|
local_filename="/tmp/test1.pdf",
|
|
file_size=1024,
|
|
mime_type="application/pdf",
|
|
)
|
|
file2 = FileRecord(
|
|
filehash="hash2",
|
|
original_filename="test2.pdf",
|
|
local_filename="/tmp/test2.pdf",
|
|
file_size=2048,
|
|
mime_type="application/pdf",
|
|
)
|
|
db_session.add(file1)
|
|
db_session.add(file2)
|
|
db_session.commit()
|
|
|
|
response = client.get("/files")
|
|
assert response.status_code == 200
|
|
|
|
def test_files_page_with_pagination(self, client: TestClient, db_session):
|
|
"""Test pagination on files page."""
|
|
# Create 10 files
|
|
for i in range(10):
|
|
file = 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)
|
|
db_session.commit()
|
|
|
|
# Test page 1
|
|
response = client.get("/files?page=1&per_page=5")
|
|
assert response.status_code == 200
|
|
|
|
# Test page 2
|
|
response = client.get("/files?page=2&per_page=5")
|
|
assert response.status_code == 200
|
|
|
|
def test_files_page_with_search_filter(self, client: TestClient, db_session):
|
|
"""Test search filtering."""
|
|
file1 = FileRecord(
|
|
filehash="hash1",
|
|
original_filename="invoice.pdf",
|
|
local_filename="/tmp/invoice.pdf",
|
|
file_size=1024,
|
|
mime_type="application/pdf",
|
|
)
|
|
file2 = FileRecord(
|
|
filehash="hash2",
|
|
original_filename="receipt.pdf",
|
|
local_filename="/tmp/receipt.pdf",
|
|
file_size=2048,
|
|
mime_type="application/pdf",
|
|
)
|
|
db_session.add(file1)
|
|
db_session.add(file2)
|
|
db_session.commit()
|
|
|
|
response = client.get("/files?search=invoice")
|
|
assert response.status_code == 200
|
|
|
|
def test_files_page_with_mime_type_filter(self, client: TestClient, db_session):
|
|
"""Test MIME type filtering."""
|
|
file1 = FileRecord(
|
|
filehash="hash1",
|
|
original_filename="doc.pdf",
|
|
local_filename="/tmp/doc.pdf",
|
|
file_size=1024,
|
|
mime_type="application/pdf",
|
|
)
|
|
file2 = FileRecord(
|
|
filehash="hash2",
|
|
original_filename="image.jpg",
|
|
local_filename="/tmp/image.jpg",
|
|
file_size=2048,
|
|
mime_type="image/jpeg",
|
|
)
|
|
db_session.add(file1)
|
|
db_session.add(file2)
|
|
db_session.commit()
|
|
|
|
response = client.get("/files?mime_type=application/pdf")
|
|
assert response.status_code == 200
|
|
|
|
def test_files_page_with_status_filter(self, client: TestClient, db_session):
|
|
"""Test status filtering."""
|
|
file = FileRecord(
|
|
filehash="hash1",
|
|
original_filename="test.pdf",
|
|
local_filename="/tmp/test.pdf",
|
|
file_size=1024,
|
|
mime_type="application/pdf",
|
|
)
|
|
db_session.add(file)
|
|
db_session.commit()
|
|
|
|
response = client.get("/files?status=completed")
|
|
assert response.status_code == 200
|
|
|
|
def test_files_page_sorting_by_filename_asc(self, client: TestClient, db_session):
|
|
"""Test sorting by filename ascending."""
|
|
file1 = FileRecord(
|
|
filehash="hash1",
|
|
original_filename="aaa.pdf",
|
|
local_filename="/tmp/aaa.pdf",
|
|
file_size=1024,
|
|
mime_type="application/pdf",
|
|
)
|
|
file2 = FileRecord(
|
|
filehash="hash2",
|
|
original_filename="zzz.pdf",
|
|
local_filename="/tmp/zzz.pdf",
|
|
file_size=2048,
|
|
mime_type="application/pdf",
|
|
)
|
|
db_session.add(file1)
|
|
db_session.add(file2)
|
|
db_session.commit()
|
|
|
|
response = client.get("/files?sort_by=original_filename&sort_order=asc")
|
|
assert response.status_code == 200
|
|
|
|
def test_files_page_sorting_by_size_desc(self, client: TestClient, db_session):
|
|
"""Test sorting by file size descending."""
|
|
file1 = FileRecord(
|
|
filehash="hash1",
|
|
original_filename="small.pdf",
|
|
local_filename="/tmp/small.pdf",
|
|
file_size=100,
|
|
mime_type="application/pdf",
|
|
)
|
|
file2 = FileRecord(
|
|
filehash="hash2",
|
|
original_filename="large.pdf",
|
|
local_filename="/tmp/large.pdf",
|
|
file_size=10000,
|
|
mime_type="application/pdf",
|
|
)
|
|
db_session.add(file1)
|
|
db_session.add(file2)
|
|
db_session.commit()
|
|
|
|
response = client.get("/files?sort_by=file_size&sort_order=desc")
|
|
assert response.status_code == 200
|
|
|
|
def test_files_page_error_handling(self, client: TestClient, db_session):
|
|
"""Test error handling in files page."""
|
|
# This test would require mocking the internal query which is complex
|
|
# The error handling is verified by the other tests that handle errors gracefully
|
|
# Skip this test as error path is already covered
|
|
pytest.skip("Error path covered by other test scenarios")
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestFileDetailPage:
|
|
"""Tests for GET /files/{file_id}/detail endpoint."""
|
|
|
|
def test_file_detail_page_success(self, client: TestClient, db_session, tmp_path):
|
|
"""Test file detail page with existing file."""
|
|
# Create file with paths that exist
|
|
file_path = tmp_path / "test.pdf"
|
|
file_path.write_bytes(b"%PDF-1.4")
|
|
|
|
file = FileRecord(
|
|
filehash="hash1",
|
|
original_filename="test.pdf",
|
|
local_filename=str(file_path),
|
|
original_file_path=str(file_path),
|
|
file_size=1024,
|
|
mime_type="application/pdf",
|
|
)
|
|
db_session.add(file)
|
|
db_session.commit()
|
|
|
|
response = client.get(f"/files/{file.id}/detail")
|
|
assert response.status_code == 200
|
|
|
|
def test_file_detail_page_not_found(self, client: TestClient, db_session):
|
|
"""Test file detail page for non-existent file."""
|
|
response = client.get("/files/99999/detail")
|
|
assert response.status_code == 200 # Still renders template with error
|
|
|
|
def test_file_detail_page_with_processing_logs(self, client: TestClient, db_session, tmp_path):
|
|
"""Test file detail page includes processing logs."""
|
|
file_path = tmp_path / "test.pdf"
|
|
file_path.write_bytes(b"%PDF-1.4")
|
|
|
|
file = FileRecord(
|
|
filehash="hash1",
|
|
original_filename="test.pdf",
|
|
local_filename=str(file_path),
|
|
file_size=1024,
|
|
mime_type="application/pdf",
|
|
)
|
|
db_session.add(file)
|
|
db_session.commit()
|
|
|
|
# Add processing logs
|
|
log1 = ProcessingLog(
|
|
file_id=file.id,
|
|
task_id="task1",
|
|
step_name="create_file_record",
|
|
status="success",
|
|
message="File record created",
|
|
)
|
|
log2 = ProcessingLog(
|
|
file_id=file.id, task_id="task2", step_name="extract_text", status="success", message="Text extracted"
|
|
)
|
|
db_session.add(log1)
|
|
db_session.add(log2)
|
|
db_session.commit()
|
|
|
|
response = client.get(f"/files/{file.id}/detail")
|
|
assert response.status_code == 200
|
|
|
|
def test_file_detail_page_with_metadata_json(self, client: TestClient, db_session, tmp_path):
|
|
"""Test file detail page loads GPT metadata from JSON file."""
|
|
file_path = tmp_path / "test.pdf"
|
|
file_path.write_bytes(b"%PDF-1.4")
|
|
|
|
# Create processed file path
|
|
processed_path = tmp_path / "test_processed.pdf"
|
|
processed_path.write_bytes(b"%PDF-1.4")
|
|
|
|
# Create metadata JSON file
|
|
metadata_path = tmp_path / "test_processed.json"
|
|
metadata = {"document_type": "invoice", "amount": 100.00}
|
|
metadata_path.write_text(json.dumps(metadata))
|
|
|
|
file = FileRecord(
|
|
filehash="hash1",
|
|
original_filename="test.pdf",
|
|
local_filename=str(file_path),
|
|
processed_file_path=str(processed_path),
|
|
file_size=1024,
|
|
mime_type="application/pdf",
|
|
)
|
|
db_session.add(file)
|
|
db_session.commit()
|
|
|
|
response = client.get(f"/files/{file.id}/detail")
|
|
assert response.status_code == 200
|
|
|
|
def test_file_detail_checks_original_file_exists(self, client: TestClient, db_session, tmp_path):
|
|
"""Test that file detail checks if original file exists on disk."""
|
|
# File without existing path
|
|
file = FileRecord(
|
|
filehash="hash1",
|
|
original_filename="test.pdf",
|
|
local_filename="/nonexistent/local.pdf", # Required field
|
|
original_file_path="/nonexistent/test.pdf",
|
|
file_size=1024,
|
|
mime_type="application/pdf",
|
|
)
|
|
db_session.add(file)
|
|
db_session.commit()
|
|
|
|
response = client.get(f"/files/{file.id}/detail")
|
|
assert response.status_code == 200
|
|
|
|
def test_file_detail_error_handling(self, client: TestClient, db_session):
|
|
"""Test error handling in file detail page."""
|
|
# Error handling path is already covered by other tests
|
|
# Skip to avoid complex database mocking
|
|
pytest.skip("Error path covered by not_found test")
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestComputeProcessingFlow:
|
|
"""Tests for _compute_processing_flow helper function."""
|
|
|
|
@patch("app.config.settings.enable_deduplication", False)
|
|
def test_compute_processing_flow_basic(self, db_session):
|
|
"""Test basic processing flow computation."""
|
|
from app.views.files import _compute_processing_flow
|
|
|
|
logs = [
|
|
Mock(
|
|
step_name="create_file_record", status="success", message="Created", timestamp=Mock(), task_id="task1"
|
|
),
|
|
Mock(step_name="check_text", status="success", message="Checked", timestamp=Mock(), task_id="task2"),
|
|
]
|
|
|
|
flow = _compute_processing_flow(logs)
|
|
assert isinstance(flow, list)
|
|
assert len(flow) > 0
|
|
|
|
@patch("app.config.settings.enable_deduplication", True)
|
|
@patch("app.config.settings.show_deduplication_step", True)
|
|
def test_compute_processing_flow_with_deduplication(self, db_session):
|
|
"""Test flow includes deduplication when enabled."""
|
|
from app.views.files import _compute_processing_flow
|
|
|
|
logs = [
|
|
Mock(
|
|
step_name="check_for_duplicates",
|
|
status="success",
|
|
message="No duplicates",
|
|
timestamp=Mock(),
|
|
task_id="task1",
|
|
)
|
|
]
|
|
|
|
flow = _compute_processing_flow(logs)
|
|
# Should include deduplication step
|
|
step_keys = [step["key"] for step in flow]
|
|
assert "check_for_duplicates" in step_keys
|
|
|
|
def test_compute_processing_flow_with_upload_branches(self, db_session):
|
|
"""Test flow includes upload branches."""
|
|
from app.views.files import _compute_processing_flow
|
|
|
|
logs = [
|
|
Mock(
|
|
step_name="send_to_all_destinations",
|
|
status="success",
|
|
message="Sent",
|
|
timestamp=Mock(),
|
|
task_id="task1",
|
|
),
|
|
Mock(
|
|
step_name="upload_to_dropbox", status="success", message="Uploaded", timestamp=Mock(), task_id="task2"
|
|
),
|
|
Mock(
|
|
step_name="upload_to_google_drive",
|
|
status="failure",
|
|
message="Failed",
|
|
timestamp=Mock(),
|
|
task_id="task3",
|
|
),
|
|
]
|
|
|
|
flow = _compute_processing_flow(logs)
|
|
# 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_processing_flow_handles_failure_status(self, db_session):
|
|
"""Test flow correctly identifies failed steps."""
|
|
from app.views.files import _compute_processing_flow
|
|
|
|
logs = [
|
|
Mock(
|
|
step_name="extract_metadata_with_gpt",
|
|
status="failure",
|
|
message="Failed to extract",
|
|
timestamp=Mock(),
|
|
task_id="task1",
|
|
)
|
|
]
|
|
|
|
flow = _compute_processing_flow(logs)
|
|
failed_steps = [s for s in flow if s["status"] == "failure"]
|
|
# Should have at least the failed step we added
|
|
assert len(failed_steps) >= 1
|
|
assert failed_steps[0]["can_retry"] is True
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestComputeStepSummary:
|
|
"""Tests for _compute_step_summary helper function."""
|
|
|
|
@patch("app.config.settings.enable_deduplication", False)
|
|
def test_compute_step_summary_basic(self):
|
|
"""Test basic step summary computation."""
|
|
from app.views.files import _compute_step_summary
|
|
|
|
logs = [
|
|
Mock(step_name="create_file_record", status="success", timestamp=Mock()),
|
|
Mock(step_name="check_text", status="success", timestamp=Mock()),
|
|
Mock(step_name="extract_text", status="success", timestamp=Mock()),
|
|
]
|
|
|
|
summary = _compute_step_summary(logs)
|
|
assert "main" in summary
|
|
assert "uploads" in summary
|
|
assert isinstance(summary["main"]["success"], int)
|
|
|
|
def test_compute_step_summary_with_uploads(self):
|
|
"""Test summary includes upload task counts."""
|
|
from app.views.files import _compute_step_summary
|
|
|
|
logs = [
|
|
Mock(step_name="create_file_record", status="success", timestamp=Mock()),
|
|
Mock(step_name="upload_to_dropbox", status="success", timestamp=Mock()),
|
|
Mock(step_name="upload_to_google_drive", status="failure", timestamp=Mock()),
|
|
]
|
|
|
|
summary = _compute_step_summary(logs)
|
|
assert summary["uploads"]["success"] >= 1
|
|
assert summary["uploads"]["failure"] >= 1
|
|
|
|
def test_compute_step_summary_normalizes_pending_status(self):
|
|
"""Test that 'pending' status is normalized to 'queued'."""
|
|
from app.views.files import _compute_step_summary
|
|
|
|
logs = [Mock(step_name="create_file_record", status="pending", timestamp=Mock())]
|
|
|
|
summary = _compute_step_summary(logs)
|
|
# Should count as queued, not pending
|
|
assert summary["main"]["queued"] >= 1
|
|
|
|
def test_compute_step_summary_order_independent(self):
|
|
"""Test that summary is order-independent (uses latest timestamp)."""
|
|
from datetime import datetime, timedelta
|
|
|
|
from app.views.files import _compute_step_summary
|
|
|
|
now = datetime.now()
|
|
logs = [
|
|
Mock(step_name="create_file_record", status="queued", timestamp=now),
|
|
Mock(step_name="create_file_record", status="success", timestamp=now + timedelta(seconds=10)),
|
|
]
|
|
|
|
summary = _compute_step_summary(logs)
|
|
# Should count success (latest) not queued
|
|
assert summary["main"]["success"] >= 1
|
|
assert summary["main"]["queued"] == 0
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestPreviewOriginalFile:
|
|
"""Tests for GET /files/{file_id}/preview/original endpoint."""
|
|
|
|
def test_preview_original_file_success(self, client: TestClient, db_session, tmp_path):
|
|
"""Test preview of original file."""
|
|
file_path = tmp_path / "test.pdf"
|
|
file_path.write_bytes(b"%PDF-1.4")
|
|
|
|
file = FileRecord(
|
|
filehash="hash1",
|
|
original_filename="test.pdf",
|
|
local_filename=str(file_path), # Required field
|
|
original_file_path=str(file_path),
|
|
file_size=1024,
|
|
mime_type="application/pdf",
|
|
)
|
|
db_session.add(file)
|
|
db_session.commit()
|
|
|
|
response = client.get(f"/files/{file.id}/preview/original")
|
|
assert response.status_code == 200
|
|
assert response.headers["content-type"] == "application/pdf"
|
|
assert "inline" in response.headers.get("content-disposition", "")
|
|
|
|
def test_preview_original_file_not_found(self, client: TestClient, db_session):
|
|
"""Test preview when file record 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 doesn't exist on disk."""
|
|
file = FileRecord(
|
|
filehash="hash1",
|
|
original_filename="test.pdf",
|
|
local_filename="/nonexistent/local.pdf", # Required field
|
|
original_file_path="/nonexistent/test.pdf",
|
|
file_size=1024,
|
|
mime_type="application/pdf",
|
|
)
|
|
db_session.add(file)
|
|
db_session.commit()
|
|
|
|
response = client.get(f"/files/{file.id}/preview/original")
|
|
assert response.status_code == 404
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestPreviewProcessedFile:
|
|
"""Tests for GET /files/{file_id}/preview/processed endpoint."""
|
|
|
|
def test_preview_processed_file_success(self, client: TestClient, db_session, tmp_path):
|
|
"""Test preview of processed file."""
|
|
processed_path = tmp_path / "test_processed.pdf"
|
|
processed_path.write_bytes(b"%PDF-1.4")
|
|
|
|
file = FileRecord(
|
|
filehash="hash1",
|
|
original_filename="test.pdf",
|
|
local_filename=str(processed_path), # Required field
|
|
processed_file_path=str(processed_path),
|
|
file_size=1024,
|
|
mime_type="application/pdf",
|
|
)
|
|
db_session.add(file)
|
|
db_session.commit()
|
|
|
|
response = client.get(f"/files/{file.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 record 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 processed file doesn't exist on disk."""
|
|
file = FileRecord(
|
|
filehash="hash1",
|
|
original_filename="test.pdf",
|
|
local_filename="/nonexistent/local.pdf", # Required field
|
|
processed_file_path="/nonexistent/test_processed.pdf",
|
|
file_size=1024,
|
|
mime_type="application/pdf",
|
|
)
|
|
db_session.add(file)
|
|
db_session.commit()
|
|
|
|
response = client.get(f"/files/{file.id}/preview/processed")
|
|
assert response.status_code == 404
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestGetOriginalText:
|
|
"""Tests for GET /files/{file_id}/text/original endpoint."""
|
|
|
|
def test_get_original_text_success(self, client: TestClient, db_session, tmp_path):
|
|
"""Test extracting text from original PDF."""
|
|
# Create a minimal PDF
|
|
pdf_path = tmp_path / "test.pdf"
|
|
pdf_content = b"""%PDF-1.4
|
|
1 0 obj
|
|
<< /Type /Catalog /Pages 2 0 R >>
|
|
endobj
|
|
2 0 obj
|
|
<< /Type /Pages /Kids [3 0 R] /Count 1 >>
|
|
endobj
|
|
3 0 obj
|
|
<< /Type /Page /Parent 2 0 R /MediaBox [0 0 612 792] >>
|
|
endobj
|
|
xref
|
|
0 4
|
|
0000000000 65535 f
|
|
0000000009 00000 n
|
|
0000000058 00000 n
|
|
0000000115 00000 n
|
|
trailer
|
|
<< /Size 4 /Root 1 0 R >>
|
|
startxref
|
|
197
|
|
%%EOF
|
|
"""
|
|
pdf_path.write_bytes(pdf_content)
|
|
|
|
file = FileRecord(
|
|
filehash="hash1",
|
|
original_filename="test.pdf",
|
|
local_filename=str(pdf_path), # Required field
|
|
original_file_path=str(pdf_path),
|
|
file_size=1024,
|
|
mime_type="application/pdf",
|
|
)
|
|
db_session.add(file)
|
|
db_session.commit()
|
|
|
|
response = client.get(f"/files/{file.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 for non-existent file."""
|
|
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 doesn't exist on disk."""
|
|
file = FileRecord(
|
|
filehash="hash1",
|
|
original_filename="test.pdf",
|
|
local_filename="/nonexistent/local.pdf", # Required field
|
|
original_file_path="/nonexistent/test.pdf",
|
|
file_size=1024,
|
|
mime_type="application/pdf",
|
|
)
|
|
db_session.add(file)
|
|
db_session.commit()
|
|
|
|
response = client.get(f"/files/{file.id}/text/original")
|
|
assert response.status_code == 404
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestGetProcessedText:
|
|
"""Tests for GET /files/{file_id}/text/processed endpoint."""
|
|
|
|
def test_get_processed_text_success(self, client: TestClient, db_session, tmp_path):
|
|
"""Test extracting text from processed PDF."""
|
|
pdf_path = tmp_path / "test_processed.pdf"
|
|
pdf_content = b"""%PDF-1.4
|
|
1 0 obj
|
|
<< /Type /Catalog /Pages 2 0 R >>
|
|
endobj
|
|
2 0 obj
|
|
<< /Type /Pages /Kids [3 0 R] /Count 1 >>
|
|
endobj
|
|
3 0 obj
|
|
<< /Type /Page /Parent 2 0 R /MediaBox [0 0 612 792] >>
|
|
endobj
|
|
xref
|
|
0 4
|
|
0000000000 65535 f
|
|
0000000009 00000 n
|
|
0000000058 00000 n
|
|
0000000115 00000 n
|
|
trailer
|
|
<< /Size 4 /Root 1 0 R >>
|
|
startxref
|
|
197
|
|
%%EOF
|
|
"""
|
|
pdf_path.write_bytes(pdf_content)
|
|
|
|
file = FileRecord(
|
|
filehash="hash1",
|
|
original_filename="test.pdf",
|
|
local_filename=str(pdf_path), # local_filename is NOT NULL
|
|
processed_file_path=str(pdf_path),
|
|
file_size=1024,
|
|
mime_type="application/pdf",
|
|
)
|
|
db_session.add(file)
|
|
db_session.commit()
|
|
|
|
response = client.get(f"/files/{file.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 for non-existent file."""
|
|
response = client.get("/files/99999/text/processed")
|
|
assert response.status_code == 404
|
|
|
|
def test_get_processed_text_no_text_extracted(self, client: TestClient, db_session, tmp_path):
|
|
"""Test when no text can be extracted from PDF."""
|
|
# Create a valid but minimal PDF with no text
|
|
pdf_path = tmp_path / "empty.pdf"
|
|
pdf_content = b"""%PDF-1.4
|
|
1 0 obj
|
|
<< /Type /Catalog /Pages 2 0 R >>
|
|
endobj
|
|
2 0 obj
|
|
<< /Type /Pages /Kids [3 0 R] /Count 1 >>
|
|
endobj
|
|
3 0 obj
|
|
<< /Type /Page /Parent 2 0 R /MediaBox [0 0 612 792] >>
|
|
endobj
|
|
xref
|
|
0 4
|
|
0000000000 65535 f
|
|
0000000009 00000 n
|
|
0000000058 00000 n
|
|
0000000115 00000 n
|
|
trailer
|
|
<< /Size 4 /Root 1 0 R >>
|
|
startxref
|
|
197
|
|
%%EOF
|
|
"""
|
|
pdf_path.write_bytes(pdf_content)
|
|
|
|
file = FileRecord(
|
|
filehash="hash1",
|
|
original_filename="empty.pdf",
|
|
local_filename=str(pdf_path), # local_filename is NOT NULL
|
|
processed_file_path=str(pdf_path),
|
|
file_size=100,
|
|
mime_type="application/pdf",
|
|
)
|
|
db_session.add(file)
|
|
db_session.commit()
|
|
|
|
response = client.get(f"/files/{file.id}/text/processed")
|
|
# Should return 200 with empty text or message about no text
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "text" in data
|
|
# Text should be empty or contain "No text" message
|
|
assert data["text"] == "" or "No text" in data["text"]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Additional tests to improve coverage of uncovered lines
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestFilesPageAdditionalFilters:
|
|
"""Additional filter coverage for GET /files endpoint."""
|
|
|
|
def test_files_page_invalid_date_from_does_not_raise_error(self, client: TestClient, db_session):
|
|
"""Test that an invalid date_from does not crash the endpoint (line 64-68)."""
|
|
response = client.get("/files?date_from=not-a-valid-date")
|
|
assert response.status_code == 200
|
|
|
|
def test_files_page_invalid_date_to_does_not_raise_error(self, client: TestClient, db_session):
|
|
"""Test that an invalid date_to does not crash the endpoint (line 71-75)."""
|
|
response = client.get("/files?date_to=not-a-valid-date")
|
|
assert response.status_code == 200
|
|
|
|
def test_files_page_valid_date_from_filter(self, client: TestClient, db_session):
|
|
"""Test that a valid date_from actually filters (line 65-66)."""
|
|
file = FileRecord(
|
|
filehash="hash_date_from",
|
|
original_filename="dated.pdf",
|
|
local_filename="/tmp/dated.pdf",
|
|
file_size=1024,
|
|
mime_type="application/pdf",
|
|
)
|
|
db_session.add(file)
|
|
db_session.commit()
|
|
|
|
response = client.get("/files?date_from=2000-01-01")
|
|
assert response.status_code == 200
|
|
|
|
def test_files_page_valid_date_to_filter(self, client: TestClient, db_session):
|
|
"""Test that a valid date_to actually filters (line 72-73)."""
|
|
response = client.get("/files?date_to=2099-12-31")
|
|
assert response.status_code == 200
|
|
|
|
def test_files_page_storage_provider_filter(self, client: TestClient, db_session):
|
|
"""Test storage provider filter creates correct subquery (lines 79-89)."""
|
|
file = FileRecord(
|
|
filehash="hash_storage",
|
|
original_filename="stored.pdf",
|
|
local_filename="/tmp/stored.pdf",
|
|
file_size=1024,
|
|
mime_type="application/pdf",
|
|
)
|
|
db_session.add(file)
|
|
db_session.commit()
|
|
|
|
# Add a successful upload step
|
|
step = FileProcessingStep(
|
|
file_id=file.id,
|
|
step_name="upload_to_dropbox",
|
|
status="success",
|
|
)
|
|
db_session.add(step)
|
|
db_session.commit()
|
|
|
|
response = client.get("/files?storage_provider=dropbox")
|
|
assert response.status_code == 200
|
|
|
|
def test_files_page_storage_provider_no_match(self, client: TestClient, db_session):
|
|
"""Test storage provider filter with no matching files (lines 79-89)."""
|
|
response = client.get("/files?storage_provider=s3")
|
|
assert response.status_code == 200
|
|
|
|
def test_files_page_tags_filter_single_tag(self, client: TestClient, db_session):
|
|
"""Test tags filter with a single tag (lines 93-97)."""
|
|
file = FileRecord(
|
|
filehash="hash_tagged",
|
|
original_filename="tagged.pdf",
|
|
local_filename="/tmp/tagged.pdf",
|
|
file_size=1024,
|
|
mime_type="application/pdf",
|
|
ai_metadata='{"tags": ["invoice"]}',
|
|
)
|
|
db_session.add(file)
|
|
db_session.commit()
|
|
|
|
response = client.get("/files?tags=invoice")
|
|
assert response.status_code == 200
|
|
|
|
def test_files_page_tags_filter_multiple_tags(self, client: TestClient, db_session):
|
|
"""Test tags filter with multiple comma-separated tags (lines 93-97)."""
|
|
response = client.get("/files?tags=invoice, receipt")
|
|
assert response.status_code == 200
|
|
|
|
def test_files_page_tags_filter_with_sql_wildcards(self, client: TestClient, db_session):
|
|
"""Test tags filter escapes SQL wildcards (line 96)."""
|
|
response = client.get("/files?tags=test%tag,with_underscore")
|
|
assert response.status_code == 200
|
|
|
|
def test_files_page_ocr_quality_poor(self, client: TestClient, db_session):
|
|
"""Test OCR quality 'poor' filter (lines 100-106)."""
|
|
file = FileRecord(
|
|
filehash="hash_poor_ocr",
|
|
original_filename="poor.pdf",
|
|
local_filename="/tmp/poor.pdf",
|
|
file_size=1024,
|
|
mime_type="application/pdf",
|
|
ocr_quality_score=0.3,
|
|
)
|
|
db_session.add(file)
|
|
db_session.commit()
|
|
|
|
response = client.get("/files?ocr_quality=poor")
|
|
assert response.status_code == 200
|
|
|
|
def test_files_page_ocr_quality_good(self, client: TestClient, db_session):
|
|
"""Test OCR quality 'good' filter (lines 107-112)."""
|
|
file = FileRecord(
|
|
filehash="hash_good_ocr",
|
|
original_filename="good.pdf",
|
|
local_filename="/tmp/good.pdf",
|
|
file_size=1024,
|
|
mime_type="application/pdf",
|
|
ocr_quality_score=0.95,
|
|
)
|
|
db_session.add(file)
|
|
db_session.commit()
|
|
|
|
response = client.get("/files?ocr_quality=good")
|
|
assert response.status_code == 200
|
|
|
|
def test_files_page_ocr_quality_unchecked(self, client: TestClient, db_session):
|
|
"""Test OCR quality 'unchecked' filter (lines 113-114)."""
|
|
file = FileRecord(
|
|
filehash="hash_unchecked_ocr",
|
|
original_filename="unchecked.pdf",
|
|
local_filename="/tmp/unchecked.pdf",
|
|
file_size=1024,
|
|
mime_type="application/pdf",
|
|
ocr_quality_score=None,
|
|
)
|
|
db_session.add(file)
|
|
db_session.commit()
|
|
|
|
response = client.get("/files?ocr_quality=unchecked")
|
|
assert response.status_code == 200
|
|
|
|
def test_files_page_mime_types_cache_hit(self, client: TestClient, db_session):
|
|
"""Test that the mime_types cache is used when cache_get returns a non-None value."""
|
|
# This test validates the basic two-request flow; the effective cache-hit
|
|
# assertion is in TestFilesPageCacheHit.test_files_page_mime_types_from_cache.
|
|
file = FileRecord(
|
|
filehash="hash_cache",
|
|
original_filename="cached.pdf",
|
|
local_filename="/tmp/cached.pdf",
|
|
file_size=1024,
|
|
mime_type="application/pdf",
|
|
)
|
|
db_session.add(file)
|
|
db_session.commit()
|
|
|
|
response1 = client.get("/files")
|
|
assert response1.status_code == 200
|
|
|
|
response2 = client.get("/files")
|
|
assert response2.status_code == 200
|
|
|
|
def test_files_page_error_handling(self, client: TestClient, db_session):
|
|
"""Test that the files page handles query errors gracefully (lines 190-204)."""
|
|
|
|
with patch("app.views.files.get_files_processing_status", side_effect=RuntimeError("DB error")):
|
|
response = client.get("/files")
|
|
# Should return 200 with error message in template context
|
|
assert response.status_code == 200
|
|
|
|
def test_files_page_sorting_by_id(self, client: TestClient, db_session):
|
|
"""Test sorting by id (covers sort_column dict lookup)."""
|
|
for i in range(3):
|
|
file = FileRecord(
|
|
filehash=f"hash_sort_id_{i}",
|
|
original_filename=f"sort_id_{i}.pdf",
|
|
local_filename=f"/tmp/sort_id_{i}.pdf",
|
|
file_size=1024,
|
|
mime_type="application/pdf",
|
|
)
|
|
db_session.add(file)
|
|
db_session.commit()
|
|
|
|
response = client.get("/files?sort_by=id&sort_order=asc")
|
|
assert response.status_code == 200
|
|
|
|
def test_files_page_sorting_by_mime_type(self, client: TestClient, db_session):
|
|
"""Test sorting by mime_type (covers sort_column dict lookup)."""
|
|
response = client.get("/files?sort_by=mime_type&sort_order=asc")
|
|
assert response.status_code == 200
|
|
|
|
def test_files_page_unknown_sort_column(self, client: TestClient, db_session):
|
|
"""Test sorting by unknown column falls back to created_at."""
|
|
response = client.get("/files?sort_by=nonexistent_column&sort_order=desc")
|
|
assert response.status_code == 200
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestFileViewPage:
|
|
"""Tests for GET /files/{file_id} endpoint (file_view_page)."""
|
|
|
|
def test_file_view_page_not_found(self, client: TestClient, db_session):
|
|
"""Test file view page when file doesn't exist (line 223)."""
|
|
response = client.get("/files/99999")
|
|
assert response.status_code == 200
|
|
assert "not found" in response.text.lower() or "error" in response.text.lower()
|
|
|
|
def test_file_view_page_success(self, client: TestClient, db_session, tmp_path):
|
|
"""Test file view page with an existing file."""
|
|
pdf_path = tmp_path / "view_test.pdf"
|
|
pdf_path.write_bytes(b"%PDF-1.4 test")
|
|
|
|
file = FileRecord(
|
|
filehash="hash_view_success",
|
|
original_filename="view_test.pdf",
|
|
local_filename=str(pdf_path),
|
|
original_file_path=str(pdf_path),
|
|
file_size=1024,
|
|
mime_type="application/pdf",
|
|
)
|
|
db_session.add(file)
|
|
db_session.commit()
|
|
|
|
response = client.get(f"/files/{file.id}")
|
|
assert response.status_code == 200
|
|
|
|
def test_file_view_page_with_metadata_sidecar(self, client: TestClient, db_session, tmp_path):
|
|
"""Test file view page loads metadata from JSON sidecar file (lines 251-257)."""
|
|
processed_path = tmp_path / "processed.pdf"
|
|
processed_path.write_bytes(b"%PDF-1.4 processed")
|
|
|
|
metadata = {"document_type": "invoice", "amount": 150.00, "currency": "USD"}
|
|
metadata_path = tmp_path / "processed.json"
|
|
metadata_path.write_text(json.dumps(metadata))
|
|
|
|
file = FileRecord(
|
|
filehash="hash_view_sidecar",
|
|
original_filename="sidecar_test.pdf",
|
|
local_filename=str(processed_path),
|
|
processed_file_path=str(processed_path),
|
|
file_size=1024,
|
|
mime_type="application/pdf",
|
|
)
|
|
db_session.add(file)
|
|
db_session.commit()
|
|
|
|
response = client.get(f"/files/{file.id}")
|
|
assert response.status_code == 200
|
|
|
|
def test_file_view_page_with_bad_metadata_sidecar(self, client: TestClient, db_session, tmp_path):
|
|
"""Test file view page handles broken JSON sidecar gracefully (lines 256-257)."""
|
|
processed_path = tmp_path / "processed_bad.pdf"
|
|
processed_path.write_bytes(b"%PDF-1.4 processed")
|
|
|
|
# Create a broken JSON sidecar
|
|
metadata_path = tmp_path / "processed_bad.json"
|
|
metadata_path.write_text("{this is: not valid json}")
|
|
|
|
file = FileRecord(
|
|
filehash="hash_view_bad_sidecar",
|
|
original_filename="bad_sidecar_test.pdf",
|
|
local_filename=str(processed_path),
|
|
processed_file_path=str(processed_path),
|
|
file_size=1024,
|
|
mime_type="application/pdf",
|
|
)
|
|
db_session.add(file)
|
|
db_session.commit()
|
|
|
|
response = client.get(f"/files/{file.id}")
|
|
assert response.status_code == 200
|
|
|
|
def test_file_view_page_with_ai_metadata_from_db(self, client: TestClient, db_session, tmp_path):
|
|
"""Test file view page loads metadata from DB column (lines 260-263)."""
|
|
pdf_path = tmp_path / "ai_meta.pdf"
|
|
pdf_path.write_bytes(b"%PDF-1.4 test")
|
|
|
|
metadata = {"document_type": "receipt", "total": 42.00}
|
|
|
|
file = FileRecord(
|
|
filehash="hash_view_ai_meta",
|
|
original_filename="ai_meta.pdf",
|
|
local_filename=str(pdf_path),
|
|
original_file_path=str(pdf_path),
|
|
file_size=1024,
|
|
mime_type="application/pdf",
|
|
ai_metadata=json.dumps(metadata),
|
|
)
|
|
db_session.add(file)
|
|
db_session.commit()
|
|
|
|
response = client.get(f"/files/{file.id}")
|
|
assert response.status_code == 200
|
|
|
|
def test_file_view_page_with_bad_ai_metadata_in_db(self, client: TestClient, db_session, tmp_path):
|
|
"""Test file view page handles invalid ai_metadata JSON in DB (lines 262-263)."""
|
|
pdf_path = tmp_path / "bad_ai_meta.pdf"
|
|
pdf_path.write_bytes(b"%PDF-1.4 test")
|
|
|
|
file = FileRecord(
|
|
filehash="hash_view_bad_ai_meta",
|
|
original_filename="bad_ai_meta.pdf",
|
|
local_filename=str(pdf_path),
|
|
original_file_path=str(pdf_path),
|
|
file_size=1024,
|
|
mime_type="application/pdf",
|
|
ai_metadata="{invalid json}",
|
|
)
|
|
db_session.add(file)
|
|
db_session.commit()
|
|
|
|
response = client.get(f"/files/{file.id}")
|
|
assert response.status_code == 200
|
|
|
|
def test_file_view_page_step_summary_exception_fallback(self, client: TestClient, db_session, tmp_path):
|
|
"""Test file view page falls back gracefully when step_manager raises (lines 270-271)."""
|
|
pdf_path = tmp_path / "step_fail.pdf"
|
|
pdf_path.write_bytes(b"%PDF-1.4 test")
|
|
|
|
file = FileRecord(
|
|
filehash="hash_view_step_fail",
|
|
original_filename="step_fail.pdf",
|
|
local_filename=str(pdf_path),
|
|
original_file_path=str(pdf_path),
|
|
file_size=1024,
|
|
mime_type="application/pdf",
|
|
)
|
|
db_session.add(file)
|
|
db_session.commit()
|
|
|
|
with patch("app.utils.step_manager.get_step_summary", side_effect=Exception("Step manager down")):
|
|
response = client.get(f"/files/{file.id}")
|
|
assert response.status_code == 200
|
|
|
|
def test_file_view_page_error_handling(self, client: TestClient, db_session, tmp_path):
|
|
"""Test file view page returns error template on unexpected exception (lines 284-286)."""
|
|
pdf_path = tmp_path / "err_view.pdf"
|
|
pdf_path.write_bytes(b"%PDF-1.4 test")
|
|
|
|
file = FileRecord(
|
|
filehash="hash_view_err",
|
|
original_filename="err_view.pdf",
|
|
local_filename=str(pdf_path),
|
|
original_file_path=str(pdf_path),
|
|
file_size=1024,
|
|
mime_type="application/pdf",
|
|
)
|
|
db_session.add(file)
|
|
db_session.commit()
|
|
|
|
# Directly trigger error by patching templates to raise on the inner path
|
|
with patch("app.views.files.logger") as _mock_logger:
|
|
with patch("app.views.files.templates") as mock_templates:
|
|
mock_templates.TemplateResponse.side_effect = [
|
|
RuntimeError("Template failure"),
|
|
Mock(status_code=200),
|
|
]
|
|
# This will hit the outer except and call TemplateResponse a second time
|
|
try:
|
|
response = client.get(f"/files/{file.id}")
|
|
# If it gets here, the error was caught
|
|
assert response.status_code in (200, 500)
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestFileDetailPageAdditional:
|
|
"""Additional tests for file_detail_page to cover uncovered lines."""
|
|
|
|
def test_file_detail_metadata_json_load_error(self, client: TestClient, db_session, tmp_path):
|
|
"""Test file detail handles metadata JSON parse error gracefully (lines 337-338)."""
|
|
file_path = tmp_path / "detail_test.pdf"
|
|
file_path.write_bytes(b"%PDF-1.4")
|
|
|
|
processed_path = tmp_path / "detail_processed.pdf"
|
|
processed_path.write_bytes(b"%PDF-1.4 processed")
|
|
|
|
# Write invalid JSON to the metadata sidecar
|
|
bad_json_path = tmp_path / "detail_processed.json"
|
|
bad_json_path.write_text("{not: valid json content}")
|
|
|
|
file = FileRecord(
|
|
filehash="hash_detail_bad_json",
|
|
original_filename="detail_bad_json.pdf",
|
|
local_filename=str(file_path),
|
|
original_file_path=str(file_path),
|
|
processed_file_path=str(processed_path),
|
|
file_size=1024,
|
|
mime_type="application/pdf",
|
|
)
|
|
db_session.add(file)
|
|
db_session.commit()
|
|
|
|
response = client.get(f"/files/{file.id}/detail")
|
|
assert response.status_code == 200
|
|
|
|
def test_file_detail_step_summary_fallback(self, client: TestClient, db_session, tmp_path):
|
|
"""Test file detail falls back to log-based step summary when step_manager raises (lines 348-350)."""
|
|
file_path = tmp_path / "detail_fallback.pdf"
|
|
file_path.write_bytes(b"%PDF-1.4")
|
|
|
|
file = FileRecord(
|
|
filehash="hash_detail_fallback",
|
|
original_filename="detail_fallback.pdf",
|
|
local_filename=str(file_path),
|
|
original_file_path=str(file_path),
|
|
file_size=1024,
|
|
mime_type="application/pdf",
|
|
)
|
|
db_session.add(file)
|
|
db_session.commit()
|
|
|
|
# Add a processing log so _compute_step_summary has data
|
|
log = ProcessingLog(
|
|
file_id=file.id,
|
|
task_id="task_fallback",
|
|
step_name="create_file_record",
|
|
status="success",
|
|
message="Done",
|
|
)
|
|
db_session.add(log)
|
|
db_session.commit()
|
|
|
|
with patch("app.utils.step_manager.get_step_summary", side_effect=Exception("Table not found")):
|
|
response = client.get(f"/files/{file.id}/detail")
|
|
assert response.status_code == 200
|
|
|
|
def test_file_detail_error_handling(self, client: TestClient, db_session):
|
|
"""Test file detail page returns error template on unexpected exception (lines 365-367)."""
|
|
with patch("app.views.files.templates") as mock_templates:
|
|
mock_templates.TemplateResponse.side_effect = [
|
|
RuntimeError("Template failure"),
|
|
Mock(status_code=200),
|
|
]
|
|
try:
|
|
response = client.get("/files/1/detail")
|
|
assert response.status_code in (200, 500)
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestComputeProcessingFlowAdditional:
|
|
"""Additional tests for _compute_processing_flow to cover uncovered lines."""
|
|
|
|
def test_compute_processing_flow_legacy_ocr_step_name(self):
|
|
"""Test that 'process_with_azure_document_intelligence' is normalized to 'process_with_ocr' (line 452)."""
|
|
from app.views.files import _compute_processing_flow
|
|
|
|
logs = [
|
|
Mock(
|
|
step_name="process_with_azure_document_intelligence",
|
|
status="success",
|
|
message="OCR done",
|
|
timestamp=Mock(),
|
|
task_id="task_legacy",
|
|
),
|
|
]
|
|
|
|
flow = _compute_processing_flow(logs)
|
|
# The step should appear as process_with_ocr in the flow
|
|
step_keys = [step["key"] for step in flow]
|
|
assert "process_with_ocr" in step_keys
|
|
ocr_step = next(s for s in flow if s["key"] == "process_with_ocr")
|
|
assert ocr_step["status"] == "success"
|
|
|
|
def test_compute_processing_flow_duplicate_upload_branches(self):
|
|
"""Test upload branches handle multiple logs for the same upload task (line 444->446)."""
|
|
from app.views.files import _compute_processing_flow
|
|
|
|
logs = [
|
|
Mock(
|
|
step_name="upload_to_dropbox",
|
|
status="failure",
|
|
message="First attempt failed",
|
|
timestamp=Mock(),
|
|
task_id="task_upload_1",
|
|
),
|
|
Mock(
|
|
step_name="upload_to_dropbox",
|
|
status="success",
|
|
message="Retry succeeded",
|
|
timestamp=Mock(),
|
|
task_id="task_upload_2",
|
|
),
|
|
]
|
|
|
|
flow = _compute_processing_flow(logs)
|
|
upload_stage = next((s for s in flow if s.get("is_branch_parent")), None)
|
|
if upload_stage and "branches" in upload_stage:
|
|
dropbox_branch = next(
|
|
(b for b in upload_stage["branches"] if b["key"] == "upload_to_dropbox"),
|
|
None,
|
|
)
|
|
assert dropbox_branch is not None
|
|
|
|
def test_compute_processing_flow_duplicate_regular_steps(self):
|
|
"""Test regular steps handle multiple log entries (line 454->456)."""
|
|
from app.views.files import _compute_processing_flow
|
|
|
|
logs = [
|
|
Mock(
|
|
step_name="create_file_record",
|
|
status="in_progress",
|
|
message="Starting",
|
|
timestamp=Mock(),
|
|
task_id="task_a",
|
|
),
|
|
Mock(
|
|
step_name="create_file_record",
|
|
status="success",
|
|
message="Done",
|
|
timestamp=Mock(),
|
|
task_id="task_b",
|
|
),
|
|
]
|
|
|
|
flow = _compute_processing_flow(logs)
|
|
create_step = next((s for s in flow if s["key"] == "create_file_record"), None)
|
|
assert create_step is not None
|
|
assert create_step["status"] == "success"
|
|
|
|
def test_compute_processing_flow_queue_prefix_upload_tasks(self):
|
|
"""Test that queue_ prefix upload tasks are normalized to upload_to_ (lines 443, 444-446)."""
|
|
from app.views.files import _compute_processing_flow
|
|
|
|
logs = [
|
|
Mock(
|
|
step_name="queue_dropbox",
|
|
status="success",
|
|
message="Queued",
|
|
timestamp=Mock(),
|
|
task_id="task_queue_1",
|
|
),
|
|
Mock(
|
|
step_name="queue_dropbox",
|
|
status="success",
|
|
message="Queued again",
|
|
timestamp=Mock(),
|
|
task_id="task_queue_2",
|
|
),
|
|
]
|
|
|
|
flow = _compute_processing_flow(logs)
|
|
# The upload stage should have a branch for dropbox
|
|
upload_stage = next((s for s in flow if s.get("is_branch_parent")), None)
|
|
if upload_stage and "branches" in upload_stage:
|
|
assert len(upload_stage["branches"]) >= 1
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestComputeStepSummaryAdditional:
|
|
"""Additional tests for _compute_step_summary to cover edge cases."""
|
|
|
|
def test_compute_step_summary_legacy_ocr_step(self):
|
|
"""Test that 'process_with_azure_document_intelligence' is normalized (line 561)."""
|
|
from app.views.files import _compute_step_summary
|
|
|
|
now = datetime.now()
|
|
logs = [
|
|
Mock(
|
|
step_name="process_with_azure_document_intelligence",
|
|
status="success",
|
|
timestamp=now,
|
|
),
|
|
]
|
|
|
|
summary = _compute_step_summary(logs)
|
|
# Should count as a main step with process_with_ocr normalization
|
|
assert summary["main"]["success"] >= 1
|
|
|
|
def test_compute_step_summary_upload_seen_with_older_timestamp(self):
|
|
"""Test upload task already seen with newer timestamp is not overwritten (line 568->551)."""
|
|
from app.views.files import _compute_step_summary
|
|
|
|
now = datetime.now()
|
|
old_time = now - timedelta(seconds=10)
|
|
|
|
logs = [
|
|
# First log: newer timestamp with success
|
|
Mock(step_name="upload_to_dropbox", status="success", timestamp=now),
|
|
# Second log: older timestamp with failure - should NOT overwrite
|
|
Mock(step_name="upload_to_dropbox", status="failure", timestamp=old_time),
|
|
]
|
|
|
|
summary = _compute_step_summary(logs)
|
|
# The success status (newer timestamp) should win
|
|
assert summary["uploads"]["success"] == 1
|
|
assert summary["uploads"]["failure"] == 0
|
|
|
|
def test_compute_step_summary_step_not_in_main_steps(self):
|
|
"""Test that steps not in main_steps are ignored (line 570->551)."""
|
|
from app.views.files import _compute_step_summary
|
|
|
|
now = datetime.now()
|
|
logs = [
|
|
Mock(step_name="unknown_custom_step", status="success", timestamp=now),
|
|
Mock(step_name="create_file_record", status="success", timestamp=now),
|
|
]
|
|
|
|
summary = _compute_step_summary(logs)
|
|
# Only create_file_record should be counted
|
|
assert summary["main"]["success"] == 1
|
|
assert summary["total_main_steps"] == 1
|
|
|
|
def test_compute_step_summary_unknown_status_not_counted(self):
|
|
"""Test that unknown statuses are not counted in main_counts (lines 577->576)."""
|
|
from app.views.files import _compute_step_summary
|
|
|
|
now = datetime.now()
|
|
logs = [
|
|
Mock(step_name="create_file_record", status="skipped", timestamp=now),
|
|
]
|
|
|
|
summary = _compute_step_summary(logs)
|
|
# "skipped" is not in main_counts, so total counts should stay 0
|
|
total = sum(summary["main"].values())
|
|
assert total == 0
|
|
|
|
def test_compute_step_summary_unknown_upload_status_not_counted(self):
|
|
"""Test that unknown statuses in upload counts are not counted (lines 582->581)."""
|
|
from app.views.files import _compute_step_summary
|
|
|
|
now = datetime.now()
|
|
logs = [
|
|
Mock(step_name="upload_to_dropbox", status="retrying", timestamp=now),
|
|
]
|
|
|
|
summary = _compute_step_summary(logs)
|
|
# "retrying" is not in upload_counts, so total upload counts should stay 0
|
|
total = sum(summary["uploads"].values())
|
|
assert total == 0
|
|
|
|
def test_compute_step_summary_main_step_seen_with_older_timestamp(self):
|
|
"""Test main step already seen with newer timestamp is not overwritten (line 572->551)."""
|
|
from app.views.files import _compute_step_summary
|
|
|
|
now = datetime.now()
|
|
old_time = now - timedelta(seconds=5)
|
|
|
|
logs = [
|
|
# Newer: success
|
|
Mock(step_name="check_text", status="success", timestamp=now),
|
|
# Older: failure - should NOT overwrite
|
|
Mock(step_name="check_text", status="failure", timestamp=old_time),
|
|
]
|
|
|
|
summary = _compute_step_summary(logs)
|
|
assert summary["main"]["success"] == 1
|
|
assert summary["main"]["failure"] == 0
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestGetOriginalTextError:
|
|
"""Tests for error handling in get_original_text."""
|
|
|
|
def test_get_original_text_extraction_error(self, client: TestClient, db_session, tmp_path):
|
|
"""Test that PDF extraction errors return HTTP 500 (lines 680-684)."""
|
|
# Create a file that exists but is NOT a valid PDF
|
|
bad_pdf_path = tmp_path / "bad.pdf"
|
|
bad_pdf_path.write_bytes(b"this is not a pdf at all")
|
|
|
|
file = FileRecord(
|
|
filehash="hash_bad_pdf_orig",
|
|
original_filename="bad.pdf",
|
|
local_filename=str(bad_pdf_path),
|
|
original_file_path=str(bad_pdf_path),
|
|
file_size=len(b"this is not a pdf at all"),
|
|
mime_type="application/pdf",
|
|
)
|
|
db_session.add(file)
|
|
db_session.commit()
|
|
|
|
response = client.get(f"/files/{file.id}/text/original")
|
|
assert response.status_code == 500
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestGetProcessedTextAdditional:
|
|
"""Additional tests for get_processed_text."""
|
|
|
|
def test_get_processed_text_file_missing_on_disk(self, client: TestClient, db_session):
|
|
"""Test processed text when file is not on disk (lines 704-705)."""
|
|
file = FileRecord(
|
|
filehash="hash_proc_missing",
|
|
original_filename="proc_missing.pdf",
|
|
local_filename="/nonexistent/local.pdf",
|
|
processed_file_path="/nonexistent/processed.pdf",
|
|
file_size=1024,
|
|
mime_type="application/pdf",
|
|
)
|
|
db_session.add(file)
|
|
db_session.commit()
|
|
|
|
response = client.get(f"/files/{file.id}/text/processed")
|
|
assert response.status_code == 404
|
|
|
|
def test_get_processed_text_extraction_error(self, client: TestClient, db_session, tmp_path):
|
|
"""Test that PDF extraction errors return HTTP 500 (lines 720-724)."""
|
|
# Create a file that exists but is NOT a valid PDF
|
|
bad_pdf_path = tmp_path / "bad_proc.pdf"
|
|
bad_pdf_path.write_bytes(b"this is not a valid pdf content")
|
|
|
|
file = FileRecord(
|
|
filehash="hash_bad_pdf_proc",
|
|
original_filename="bad_proc.pdf",
|
|
local_filename=str(bad_pdf_path),
|
|
processed_file_path=str(bad_pdf_path),
|
|
file_size=len(b"this is not a valid pdf content"),
|
|
mime_type="application/pdf",
|
|
)
|
|
db_session.add(file)
|
|
db_session.commit()
|
|
|
|
response = client.get(f"/files/{file.id}/text/processed")
|
|
assert response.status_code == 500
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestDuplicatesPage:
|
|
"""Tests for GET /duplicates endpoint."""
|
|
|
|
def test_duplicates_page_empty(self, client: TestClient, db_session):
|
|
"""Test duplicates page with no duplicates (lines 740-807)."""
|
|
response = client.get("/duplicates")
|
|
assert response.status_code == 200
|
|
|
|
def test_duplicates_page_with_duplicates(self, client: TestClient, db_session):
|
|
"""Test duplicates page with actual duplicate files."""
|
|
# Create original file
|
|
original = FileRecord(
|
|
filehash="dup_hash_1",
|
|
original_filename="original.pdf",
|
|
local_filename="/tmp/original.pdf",
|
|
file_size=1024,
|
|
mime_type="application/pdf",
|
|
is_duplicate=False,
|
|
)
|
|
db_session.add(original)
|
|
db_session.commit()
|
|
|
|
# Create duplicate
|
|
duplicate = FileRecord(
|
|
filehash="dup_hash_1",
|
|
original_filename="duplicate.pdf",
|
|
local_filename="/tmp/duplicate.pdf",
|
|
file_size=1024,
|
|
mime_type="application/pdf",
|
|
is_duplicate=True,
|
|
duplicate_of_id=original.id,
|
|
)
|
|
db_session.add(duplicate)
|
|
db_session.commit()
|
|
|
|
response = client.get("/duplicates")
|
|
assert response.status_code == 200
|
|
|
|
def test_duplicates_page_pagination(self, client: TestClient, db_session):
|
|
"""Test duplicates page pagination parameters."""
|
|
response = client.get("/duplicates?page=1&per_page=10")
|
|
assert response.status_code == 200
|
|
|
|
def test_duplicates_page_group_without_original(self, client: TestClient, db_session):
|
|
"""Test duplicates page when a group has no original file."""
|
|
# Create only a duplicate without an original in the DB
|
|
duplicate = FileRecord(
|
|
filehash="orphan_dup_hash",
|
|
original_filename="orphan_dup.pdf",
|
|
local_filename="/tmp/orphan_dup.pdf",
|
|
file_size=1024,
|
|
mime_type="application/pdf",
|
|
is_duplicate=True,
|
|
duplicate_of_id=None,
|
|
)
|
|
db_session.add(duplicate)
|
|
db_session.commit()
|
|
|
|
response = client.get("/duplicates")
|
|
assert response.status_code == 200
|
|
|
|
def test_duplicates_page_error_handling(self, client: TestClient, db_session):
|
|
"""Test duplicates page returns fallback template on error (lines 808-821)."""
|
|
with patch("app.views.files.templates") as mock_templates:
|
|
mock_templates.TemplateResponse.side_effect = [
|
|
RuntimeError("DB failure"),
|
|
Mock(status_code=200),
|
|
]
|
|
try:
|
|
response = client.get("/duplicates")
|
|
assert response.status_code in (200, 500)
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestSimilarityDashboardPage:
|
|
"""Tests for GET /similarity endpoint."""
|
|
|
|
def test_similarity_dashboard_page_success(self, client: TestClient, db_session):
|
|
"""Test similarity dashboard renders successfully (lines 836-857)."""
|
|
# Create some files with and without embeddings
|
|
file1 = FileRecord(
|
|
filehash="sim_hash_1",
|
|
original_filename="sim1.pdf",
|
|
local_filename="/tmp/sim1.pdf",
|
|
file_size=1024,
|
|
mime_type="application/pdf",
|
|
embedding="[0.1, 0.2, 0.3]",
|
|
ocr_text="Some OCR text here",
|
|
)
|
|
file2 = FileRecord(
|
|
filehash="sim_hash_2",
|
|
original_filename="sim2.pdf",
|
|
local_filename="/tmp/sim2.pdf",
|
|
file_size=2048,
|
|
mime_type="application/pdf",
|
|
embedding=None,
|
|
ocr_text="Another OCR text",
|
|
)
|
|
file3 = FileRecord(
|
|
filehash="sim_hash_3",
|
|
original_filename="sim3.pdf",
|
|
local_filename="/tmp/sim3.pdf",
|
|
file_size=512,
|
|
mime_type="application/pdf",
|
|
embedding=None,
|
|
ocr_text=None,
|
|
)
|
|
db_session.add(file1)
|
|
db_session.add(file2)
|
|
db_session.add(file3)
|
|
db_session.commit()
|
|
|
|
response = client.get("/similarity")
|
|
assert response.status_code == 200
|
|
|
|
def test_similarity_dashboard_page_empty_database(self, client: TestClient, db_session):
|
|
"""Test similarity dashboard with no files (lines 836-857)."""
|
|
response = client.get("/similarity")
|
|
assert response.status_code == 200
|
|
|
|
def test_similarity_dashboard_page_error_handling(self, client: TestClient, db_session):
|
|
"""Test similarity dashboard returns fallback template on error (lines 858-872)."""
|
|
with patch("app.views.files.templates") as mock_templates:
|
|
mock_templates.TemplateResponse.side_effect = [
|
|
RuntimeError("Query failure"),
|
|
Mock(status_code=200),
|
|
]
|
|
try:
|
|
response = client.get("/similarity")
|
|
assert response.status_code in (200, 500)
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestFilesPageCacheHit:
|
|
"""Test the MIME types cache hit path."""
|
|
|
|
def test_files_page_mime_types_from_cache(self, client: TestClient, db_session):
|
|
"""Test that cached mime_types are used when available (line 155->161)."""
|
|
cached_mime_types = ["application/pdf", "image/jpeg"]
|
|
|
|
# Patch the name as bound in the files view module so the cache hit is triggered
|
|
with patch("app.views.files.cache_get", return_value=cached_mime_types):
|
|
response = client.get("/files")
|
|
assert response.status_code == 200
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestFileViewPageNoJsonSidecar:
|
|
"""Test file_view_page when processed_file_path has no JSON sidecar (line 252->259)."""
|
|
|
|
def test_file_view_page_processed_path_no_sidecar(self, client: TestClient, db_session, tmp_path):
|
|
"""Test that _safe_exists returns False when no JSON sidecar exists (line 252->259)."""
|
|
processed_path = tmp_path / "no_sidecar.pdf"
|
|
processed_path.write_bytes(b"%PDF-1.4 processed")
|
|
# Deliberately do NOT create a .json sidecar
|
|
|
|
file = FileRecord(
|
|
filehash="hash_no_sidecar",
|
|
original_filename="no_sidecar.pdf",
|
|
local_filename=str(processed_path),
|
|
processed_file_path=str(processed_path),
|
|
file_size=1024,
|
|
mime_type="application/pdf",
|
|
)
|
|
db_session.add(file)
|
|
db_session.commit()
|
|
|
|
response = client.get(f"/files/{file.id}")
|
|
assert response.status_code == 200
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestFileDetailNoJsonSidecar:
|
|
"""Test file_detail_page when processed_file_path has no JSON sidecar (line 332->341)."""
|
|
|
|
def test_file_detail_processed_path_no_sidecar(self, client: TestClient, db_session, tmp_path):
|
|
"""Test that os.path.exists returns False when no JSON sidecar exists (line 332->341)."""
|
|
file_path = tmp_path / "detail_no_sidecar.pdf"
|
|
file_path.write_bytes(b"%PDF-1.4")
|
|
processed_path = tmp_path / "detail_no_sidecar_processed.pdf"
|
|
processed_path.write_bytes(b"%PDF-1.4 processed")
|
|
# Deliberately do NOT create a .json sidecar
|
|
|
|
file = FileRecord(
|
|
filehash="hash_detail_no_sidecar",
|
|
original_filename="detail_no_sidecar.pdf",
|
|
local_filename=str(file_path),
|
|
original_file_path=str(file_path),
|
|
processed_file_path=str(processed_path),
|
|
file_size=1024,
|
|
mime_type="application/pdf",
|
|
)
|
|
db_session.add(file)
|
|
db_session.commit()
|
|
|
|
response = client.get(f"/files/{file.id}/detail")
|
|
assert response.status_code == 200
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestGetTextWithContent:
|
|
"""Tests for text extraction when PDFs contain actual text (lines 676->679 and 716->719)."""
|
|
|
|
# A minimal but valid PDF that embeds the string "Hello World" as extractable text
|
|
_PDF_WITH_TEXT = (
|
|
b"%PDF-1.4\n"
|
|
b"1 0 obj\n<< /Type /Catalog /Pages 2 0 R >>\nendobj\n"
|
|
b"2 0 obj\n<< /Type /Pages /Kids [3 0 R] /Count 1 >>\nendobj\n"
|
|
b"3 0 obj\n<< /Type /Page /Parent 2 0 R /MediaBox [0 0 612 792]\n"
|
|
b" /Contents 4 0 R /Resources << /Font << /F1 5 0 R >> >> >>\nendobj\n"
|
|
b"4 0 obj\n<< /Length 44 >>\nstream\n"
|
|
b"BT /F1 12 Tf 100 700 Td (Hello World) Tj ET\n"
|
|
b"endstream\nendobj\n"
|
|
b"5 0 obj\n<< /Type /Font /Subtype /Type1 /BaseFont /Helvetica >>\nendobj\n"
|
|
b"xref\n0 6\n"
|
|
b"0000000000 65535 f \n"
|
|
b"0000000009 00000 n \n"
|
|
b"0000000058 00000 n \n"
|
|
b"0000000115 00000 n \n"
|
|
b"0000000266 00000 n \n"
|
|
b"0000000360 00000 n \n"
|
|
b"trailer\n<< /Size 6 /Root 1 0 R >>\n"
|
|
b"startxref\n441\n%%EOF"
|
|
)
|
|
|
|
def test_get_original_text_with_content(self, client: TestClient, db_session, tmp_path):
|
|
"""Test text extraction when original PDF has extractable text (line 676->679)."""
|
|
pdf_path = tmp_path / "text_original.pdf"
|
|
pdf_path.write_bytes(self._PDF_WITH_TEXT)
|
|
|
|
file = FileRecord(
|
|
filehash="hash_text_orig",
|
|
original_filename="text_original.pdf",
|
|
local_filename=str(pdf_path),
|
|
original_file_path=str(pdf_path),
|
|
file_size=len(self._PDF_WITH_TEXT),
|
|
mime_type="application/pdf",
|
|
)
|
|
db_session.add(file)
|
|
db_session.commit()
|
|
|
|
response = client.get(f"/files/{file.id}/text/original")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "text" in data
|
|
assert data["text"] # Should have non-empty text
|
|
assert "No text" not in data["text"]
|
|
|
|
def test_get_processed_text_with_content(self, client: TestClient, db_session, tmp_path):
|
|
"""Test text extraction when processed PDF has extractable text (line 716->719)."""
|
|
pdf_path = tmp_path / "text_processed.pdf"
|
|
pdf_path.write_bytes(self._PDF_WITH_TEXT)
|
|
|
|
file = FileRecord(
|
|
filehash="hash_text_proc",
|
|
original_filename="text_processed.pdf",
|
|
local_filename=str(pdf_path),
|
|
processed_file_path=str(pdf_path),
|
|
file_size=len(self._PDF_WITH_TEXT),
|
|
mime_type="application/pdf",
|
|
)
|
|
db_session.add(file)
|
|
db_session.commit()
|
|
|
|
response = client.get(f"/files/{file.id}/text/processed")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "text" in data
|
|
assert data["text"] # Should have non-empty text
|
|
assert "No text" not in data["text"]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Pipeline-info tests: file_detail and file_view views
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestPipelineInfoInViews:
|
|
"""Tests that pipeline information is correctly resolved and passed to templates."""
|
|
|
|
# ------------------------------------------------------------------
|
|
# Helpers
|
|
# ------------------------------------------------------------------
|
|
|
|
@staticmethod
|
|
def _make_file(db_session, pipeline_id=None):
|
|
from app.models import FileRecord
|
|
|
|
f = FileRecord(
|
|
filehash="ph_" + str(pipeline_id),
|
|
original_filename="doc.pdf",
|
|
local_filename="/tmp/doc.pdf",
|
|
file_size=1024,
|
|
mime_type="application/pdf",
|
|
pipeline_id=pipeline_id,
|
|
)
|
|
db_session.add(f)
|
|
db_session.commit()
|
|
return f
|
|
|
|
@staticmethod
|
|
def _make_system_pipeline(db_session, is_default=True):
|
|
from app.models import Pipeline, PipelineStep
|
|
|
|
p = Pipeline(
|
|
owner_id=None,
|
|
name="Standard Processing Pipeline",
|
|
description="System default",
|
|
is_default=is_default,
|
|
is_active=True,
|
|
)
|
|
db_session.add(p)
|
|
db_session.flush()
|
|
|
|
for pos, (step_type, label) in enumerate(
|
|
[
|
|
("convert_to_pdf", "Convert to PDF"),
|
|
("ocr", "OCR"),
|
|
("send_to_destinations", "Send"),
|
|
]
|
|
):
|
|
db_session.add(PipelineStep(pipeline_id=p.id, position=pos, step_type=step_type, label=label, enabled=True))
|
|
|
|
db_session.commit()
|
|
return p
|
|
|
|
@staticmethod
|
|
def _make_custom_pipeline(db_session):
|
|
from app.models import Pipeline, PipelineStep
|
|
|
|
p = Pipeline(
|
|
owner_id="user1",
|
|
name="My Custom Pipeline",
|
|
is_default=False,
|
|
is_active=True,
|
|
)
|
|
db_session.add(p)
|
|
db_session.flush()
|
|
db_session.add(PipelineStep(pipeline_id=p.id, position=0, step_type="ocr", label="OCR", enabled=True))
|
|
db_session.commit()
|
|
return p
|
|
|
|
# ------------------------------------------------------------------
|
|
# _resolve_pipeline unit tests
|
|
# ------------------------------------------------------------------
|
|
|
|
def test_resolve_pipeline_explicit_assignment(self, db_session):
|
|
"""File with an explicit pipeline_id resolves to that pipeline."""
|
|
from app.views.files import _resolve_pipeline
|
|
|
|
pipeline = self._make_system_pipeline(db_session)
|
|
file_rec = self._make_file(db_session, pipeline_id=pipeline.id)
|
|
|
|
info = _resolve_pipeline(db_session, file_rec)
|
|
|
|
assert info is not None
|
|
assert info["id"] == pipeline.id
|
|
assert info["is_explicit"] is True
|
|
assert info["is_system"] is True
|
|
|
|
def test_resolve_pipeline_fallback_to_system_default(self, db_session):
|
|
"""File without pipeline_id falls back to system-default pipeline."""
|
|
from app.views.files import _resolve_pipeline
|
|
|
|
pipeline = self._make_system_pipeline(db_session)
|
|
file_rec = self._make_file(db_session, pipeline_id=None)
|
|
|
|
info = _resolve_pipeline(db_session, file_rec)
|
|
|
|
assert info is not None
|
|
assert info["id"] == pipeline.id
|
|
assert info["is_explicit"] is False
|
|
assert info["is_system"] is True
|
|
assert info["is_default"] is True
|
|
|
|
def test_resolve_pipeline_returns_none_when_no_pipeline_in_db(self, db_session):
|
|
"""Returns None when no pipeline exists (empty database)."""
|
|
from app.views.files import _resolve_pipeline
|
|
|
|
file_rec = self._make_file(db_session, pipeline_id=None)
|
|
info = _resolve_pipeline(db_session, file_rec)
|
|
|
|
assert info is None
|
|
|
|
def test_resolve_pipeline_includes_steps(self, db_session):
|
|
"""Returned dict contains the pipeline's steps in order."""
|
|
from app.views.files import _resolve_pipeline
|
|
|
|
pipeline = self._make_system_pipeline(db_session)
|
|
file_rec = self._make_file(db_session, pipeline_id=pipeline.id)
|
|
|
|
info = _resolve_pipeline(db_session, file_rec)
|
|
|
|
assert info is not None
|
|
assert len(info["steps"]) == 3
|
|
assert info["steps"][0].step_type == "convert_to_pdf"
|
|
|
|
def test_resolve_pipeline_custom_pipeline(self, db_session):
|
|
"""File with an explicit custom pipeline resolves correctly."""
|
|
from app.views.files import _resolve_pipeline
|
|
|
|
pipeline = self._make_custom_pipeline(db_session)
|
|
file_rec = self._make_file(db_session, pipeline_id=pipeline.id)
|
|
|
|
info = _resolve_pipeline(db_session, file_rec)
|
|
|
|
assert info is not None
|
|
assert info["id"] == pipeline.id
|
|
assert info["name"] == "My Custom Pipeline"
|
|
assert info["is_system"] is False
|
|
assert info["is_explicit"] is True
|
|
|
|
# ------------------------------------------------------------------
|
|
# _compute_processing_flow pipeline filtering
|
|
# ------------------------------------------------------------------
|
|
|
|
def test_compute_flow_without_pipeline_shows_all_stages(self):
|
|
"""Without a pipeline, all hardcoded stages are included."""
|
|
from app.views.files import _compute_processing_flow
|
|
|
|
flow = _compute_processing_flow([], pipeline_steps=None)
|
|
# Should include standard stages (create_file_record, check_text, etc.)
|
|
keys = {s["key"] for s in flow}
|
|
assert "create_file_record" in keys
|
|
assert "extract_metadata_with_gpt" in keys
|
|
|
|
def test_step_type_mapping_is_complete(self):
|
|
"""Every step type in PIPELINE_STEP_TYPES has an entry in _STEP_TYPE_TO_STAGES."""
|
|
from app.api.pipelines import PIPELINE_STEP_TYPES
|
|
from app.views.files import _STEP_TYPE_TO_STAGES
|
|
|
|
missing = set(PIPELINE_STEP_TYPES.keys()) - set(_STEP_TYPE_TO_STAGES.keys())
|
|
assert not missing, (
|
|
f"The following pipeline step types are missing from _STEP_TYPE_TO_STAGES "
|
|
f"in app/views/files.py: {missing}. "
|
|
"Add them with their corresponding Celery log stage key(s) (use [] if none yet)."
|
|
)
|
|
|
|
def test_compute_flow_with_pipeline_filters_stages(self, db_session):
|
|
"""With a pipeline, only mapped stages are shown (plus always-show and ran stages)."""
|
|
|
|
from app.models import PipelineStep
|
|
from app.views.files import _compute_processing_flow
|
|
|
|
pipeline = self._make_system_pipeline(db_session)
|
|
# pipeline has: convert_to_pdf, ocr, send_to_destinations
|
|
|
|
# Query steps explicitly (no SQLAlchemy relationship defined on Pipeline)
|
|
steps = db_session.query(PipelineStep).filter(PipelineStep.pipeline_id == pipeline.id).all()
|
|
|
|
flow = _compute_processing_flow([], pipeline_steps=steps)
|
|
# When no logs and pipeline steps provided, only pipeline-mapped + always-show stages appear
|
|
keys = {s["key"] for s in flow}
|
|
# Always show
|
|
assert "create_file_record" in keys
|
|
# ocr maps to check_text / extract_text / process_with_ocr
|
|
assert "check_text" in keys or "extract_text" in keys
|
|
# embed_metadata not in pipeline → should be absent (no logs ran it)
|
|
assert "embed_metadata_into_pdf" not in keys
|
|
|
|
def test_compute_flow_with_pipeline_always_shows_ran_stages(self, db_session):
|
|
"""Stages that actually ran are always shown even if not in the pipeline."""
|
|
from unittest.mock import Mock
|
|
|
|
from app.models import PipelineStep
|
|
from app.views.files import _compute_processing_flow
|
|
|
|
pipeline = self._make_system_pipeline(db_session)
|
|
steps = db_session.query(PipelineStep).filter(PipelineStep.pipeline_id == pipeline.id).all()
|
|
|
|
# Simulate a log entry for embed_metadata_into_pdf (not in this pipeline)
|
|
ran_log = Mock(
|
|
step_name="embed_metadata_into_pdf",
|
|
status="success",
|
|
message="Done",
|
|
timestamp=Mock(),
|
|
task_id="t1",
|
|
)
|
|
|
|
flow = _compute_processing_flow([ran_log], pipeline_steps=steps)
|
|
keys = {s["key"] for s in flow}
|
|
assert "embed_metadata_into_pdf" in keys
|
|
|
|
# ------------------------------------------------------------------
|
|
# Integration: view endpoints pass pipeline_info to template
|
|
# ------------------------------------------------------------------
|
|
|
|
def test_file_detail_page_includes_pipeline_name(self, client, db_session):
|
|
"""GET /files/{id}/detail response body contains the pipeline name."""
|
|
pipeline = self._make_system_pipeline(db_session)
|
|
file_rec = self._make_file(db_session, pipeline_id=None)
|
|
|
|
response = client.get(f"/files/{file_rec.id}/detail")
|
|
|
|
assert response.status_code == 200
|
|
assert b"Standard Processing Pipeline" in response.content
|
|
|
|
def test_file_detail_page_shows_system_default_badge(self, client, db_session):
|
|
"""File without explicit pipeline shows 'System Default' badge in detail view."""
|
|
self._make_system_pipeline(db_session)
|
|
file_rec = self._make_file(db_session, pipeline_id=None)
|
|
|
|
response = client.get(f"/files/{file_rec.id}/detail")
|
|
|
|
assert response.status_code == 200
|
|
assert b"System Default" in response.content
|
|
|
|
def test_file_detail_page_shows_custom_badge_for_custom_pipeline(self, client, db_session):
|
|
"""File with a custom (non-system) pipeline shows 'Custom' badge."""
|
|
pipeline = self._make_custom_pipeline(db_session)
|
|
file_rec = self._make_file(db_session, pipeline_id=pipeline.id)
|
|
|
|
response = client.get(f"/files/{file_rec.id}/detail")
|
|
|
|
assert response.status_code == 200
|
|
assert b"My Custom Pipeline" in response.content
|
|
assert b"Custom" in response.content
|
|
|
|
def test_file_view_page_includes_pipeline_name(self, client, db_session):
|
|
"""GET /files/{id} response body contains the pipeline name in the sidebar."""
|
|
pipeline = self._make_system_pipeline(db_session)
|
|
file_rec = self._make_file(db_session, pipeline_id=None)
|
|
|
|
response = client.get(f"/files/{file_rec.id}")
|
|
|
|
assert response.status_code == 200
|
|
assert b"Standard Processing Pipeline" in response.content
|
|
|
|
def test_file_view_page_no_pipeline_shows_standard(self, client, db_session):
|
|
"""When no pipeline exists, file view shows 'Standard' fallback text."""
|
|
# No pipeline in DB
|
|
file_rec = self._make_file(db_session, pipeline_id=None)
|
|
|
|
response = client.get(f"/files/{file_rec.id}")
|
|
|
|
assert response.status_code == 200
|
|
assert b"Standard" in response.content
|