refactor: extract shared file-status query logic into utility function

- Created app/utils/file_queries.py with apply_status_filter function - Updated app/api/files.py to use shared function - Updated app/views/files.py to use shared function - Removed unused 'or_' import from app/api/files.py - Added comprehensive tests in tests/test_file_queries.py - All tests pass (10 new tests, 14 existing tests verified) Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com>
2026-02-10 20:40:11 +00:00
parent eb666d8325
commit 8edc11da04
4 changed files with 368 additions and 61 deletions
@@ -0,0 +1,291 @@
+"""
+Tests for file query utilities.
+
+This module tests the shared file status filtering logic in app/utils/file_queries.py.
+"""
+
+import pytest
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
+
+from app.database import Base
+from app.models import FileRecord, ProcessingLog
+from app.utils.file_queries import apply_status_filter
+
+
+@pytest.fixture
+def db_session():
+    """Create an in-memory SQLite database for testing."""
+    engine = create_engine("sqlite:///:memory:")
+    Base.metadata.create_all(engine)
+    SessionLocal = sessionmaker(bind=engine)
+    session = SessionLocal()
+    yield session
+    session.close()
+    Base.metadata.drop_all(engine)
+
+
+@pytest.fixture
+def sample_files(db_session):
+    """Create sample files with different processing statuses."""
+    # File 1: pending (no logs)
+    file1 = FileRecord(
+        filehash="hash1",
+        original_filename="pending.pdf",
+        local_filename="/tmp/pending.pdf",
+        file_size=1024,
+        mime_type="application/pdf",
+    )
+    db_session.add(file1)
+    db_session.flush()
+
+    # File 2: processing (has in_progress log)
+    file2 = FileRecord(
+        filehash="hash2",
+        original_filename="processing.pdf",
+        local_filename="/tmp/processing.pdf",
+        file_size=2048,
+        mime_type="application/pdf",
+    )
+    db_session.add(file2)
+    db_session.flush()
+
+    log2 = ProcessingLog(
+        file_id=file2.id,
+        task_id="task2",
+        step_name="extract_text",
+        status="in_progress",
+        message="Processing...",
+    )
+    db_session.add(log2)
+
+    # File 3: failed (has failure log)
+    file3 = FileRecord(
+        filehash="hash3",
+        original_filename="failed.pdf",
+        local_filename="/tmp/failed.pdf",
+        file_size=3072,
+        mime_type="application/pdf",
+    )
+    db_session.add(file3)
+    db_session.flush()
+
+    log3 = ProcessingLog(
+        file_id=file3.id,
+        task_id="task3",
+        step_name="extract_text",
+        status="failure",
+        message="Error occurred",
+    )
+    db_session.add(log3)
+
+    # File 4: completed (has success log, no failures)
+    file4 = FileRecord(
+        filehash="hash4",
+        original_filename="completed.pdf",
+        local_filename="/tmp/completed.pdf",
+        file_size=4096,
+        mime_type="application/pdf",
+    )
+    db_session.add(file4)
+    db_session.flush()
+
+    log4 = ProcessingLog(file_id=file4.id, task_id="task4", step_name="extract_text", status="success", message="Done")
+    db_session.add(log4)
+
+    # File 5: completed with multiple success logs
+    file5 = FileRecord(
+        filehash="hash5",
+        original_filename="completed2.pdf",
+        local_filename="/tmp/completed2.pdf",
+        file_size=5120,
+        mime_type="application/pdf",
+    )
+    db_session.add(file5)
+    db_session.flush()
+
+    log5a = ProcessingLog(
+        file_id=file5.id,
+        task_id="task5a",
+        step_name="extract_text",
+        status="success",
+        message="Step 1 done",
+    )
+    log5b = ProcessingLog(
+        file_id=file5.id,
+        task_id="task5b",
+        step_name="extract_metadata_with_gpt",
+        status="success",
+        message="Step 2 done",
+    )
+    db_session.add_all([log5a, log5b])
+
+    # File 6: has success but also failure (should be filtered out from completed)
+    file6 = FileRecord(
+        filehash="hash6",
+        original_filename="mixed.pdf",
+        local_filename="/tmp/mixed.pdf",
+        file_size=6144,
+        mime_type="application/pdf",
+    )
+    db_session.add(file6)
+    db_session.flush()
+
+    log6a = ProcessingLog(
+        file_id=file6.id,
+        task_id="task6a",
+        step_name="extract_text",
+        status="success",
+        message="Step 1 done",
+    )
+    log6b = ProcessingLog(
+        file_id=file6.id,
+        task_id="task6b",
+        step_name="upload_to_s3",
+        status="failure",
+        message="Upload failed",
+    )
+    db_session.add_all([log6a, log6b])
+
+    db_session.commit()
+
+    return {
+        "pending": file1,
+        "processing": file2,
+        "failed": file3,
+        "completed": file4,
+        "completed2": file5,
+        "mixed": file6,
+    }
+
+
+@pytest.mark.unit
+def test_apply_status_filter_none(db_session, sample_files):
+    """Test that no filter is applied when status is None."""
+    query = db_session.query(FileRecord)
+    filtered_query = apply_status_filter(query, db_session, None)
+    results = filtered_query.all()
+
+    # Should return all files
+    assert len(results) == 6
+
+
+@pytest.mark.unit
+def test_apply_status_filter_pending(db_session, sample_files):
+    """Test filtering for pending files (no logs)."""
+    query = db_session.query(FileRecord)
+    filtered_query = apply_status_filter(query, db_session, "pending")
+    results = filtered_query.all()
+
+    # Should return only file1 (no logs)
+    assert len(results) == 1
+    assert results[0].filehash == "hash1"
+    assert results[0].original_filename == "pending.pdf"
+
+
+@pytest.mark.unit
+def test_apply_status_filter_processing(db_session, sample_files):
+    """Test filtering for files in processing state."""
+    query = db_session.query(FileRecord)
+    filtered_query = apply_status_filter(query, db_session, "processing")
+    results = filtered_query.all()
+
+    # Should return only file2 (has in_progress log)
+    assert len(results) == 1
+    assert results[0].filehash == "hash2"
+    assert results[0].original_filename == "processing.pdf"
+
+
+@pytest.mark.unit
+def test_apply_status_filter_failed(db_session, sample_files):
+    """Test filtering for failed files."""
+    query = db_session.query(FileRecord)
+    filtered_query = apply_status_filter(query, db_session, "failed")
+    results = filtered_query.all()
+
+    # Should return file3 and file6 (both have failure logs)
+    assert len(results) == 2
+    filehashes = {r.filehash for r in results}
+    assert "hash3" in filehashes
+    assert "hash6" in filehashes
+
+
+@pytest.mark.unit
+def test_apply_status_filter_completed(db_session, sample_files):
+    """Test filtering for completed files (success with no failures)."""
+    query = db_session.query(FileRecord)
+    filtered_query = apply_status_filter(query, db_session, "completed")
+    results = filtered_query.all()
+
+    # Should return file4 and file5 (success logs, no failures)
+    # file6 should NOT be included (has both success and failure)
+    assert len(results) == 2
+    filehashes = {r.filehash for r in results}
+    assert "hash4" in filehashes
+    assert "hash5" in filehashes
+    assert "hash6" not in filehashes  # mixed status should be excluded
+
+
+@pytest.mark.unit
+def test_apply_status_filter_empty_string(db_session, sample_files):
+    """Test that empty string status is treated like None (no filter)."""
+    query = db_session.query(FileRecord)
+    filtered_query = apply_status_filter(query, db_session, "")
+    results = filtered_query.all()
+
+    # Should return all files
+    assert len(results) == 6
+
+
+@pytest.mark.unit
+def test_apply_status_filter_invalid_status(db_session, sample_files):
+    """Test that invalid status values don't crash, just return unfiltered."""
+    query = db_session.query(FileRecord)
+    filtered_query = apply_status_filter(query, db_session, "invalid_status")
+    results = filtered_query.all()
+
+    # Should return all files (no filter applied for unknown status)
+    assert len(results) == 6
+
+
+@pytest.mark.unit
+def test_apply_status_filter_with_other_filters(db_session, sample_files):
+    """Test that status filter can be combined with other query filters."""
+    # Filter for completed PDFs
+    query = db_session.query(FileRecord).filter(FileRecord.mime_type == "application/pdf")
+    filtered_query = apply_status_filter(query, db_session, "completed")
+    results = filtered_query.all()
+
+    # Should return file4 and file5 (both are completed PDFs)
+    assert len(results) == 2
+    filehashes = {r.filehash for r in results}
+    assert "hash4" in filehashes
+    assert "hash5" in filehashes
+
+
+@pytest.mark.unit
+def test_apply_status_filter_empty_database(db_session):
+    """Test filtering on empty database doesn't crash."""
+    query = db_session.query(FileRecord)
+    filtered_query = apply_status_filter(query, db_session, "completed")
+    results = filtered_query.all()
+
+    # Should return empty list
+    assert len(results) == 0
+
+
+@pytest.mark.unit
+def test_apply_status_filter_preserves_query_order(db_session, sample_files):
+    """Test that status filter preserves existing query ordering."""
+    from sqlalchemy import desc
+
+    # Query with ordering
+    query = db_session.query(FileRecord).order_by(desc(FileRecord.file_size))
+    filtered_query = apply_status_filter(query, db_session, "completed")
+    results = filtered_query.all()
+
+    # Should return file5 before file4 (larger file size)
+    assert len(results) == 2
+    assert results[0].file_size > results[1].file_size
+    assert results[0].filehash == "hash5"
+    assert results[1].filehash == "hash4"