gitea-admin/gh-christianlouis-docuelevate
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

test: add tests for storage reorganization features

Browse Source 
- Add tests for get_unique_filepath_with_counter collision handling
- Add tests for immutable original storage
- Add tests for metadata augmentation with file paths
- Add tests for force_cloud_ocr parameter
- Verify existing process_document tests still pass

Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com>
This commit is contained in:
copilot-swe-agent[bot]
2026-02-11 20:13:17 +00:00
parent 73a0222e56
commit 952e27b339
3 changed files with 445 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/tasks/process_document.py
+41 -27
View File
@@ -151,31 +151,46 @@ def process_document(self, original_local_file: str, original_filename: str = No
file_uuid = str(uuid.uuid4())
new_filename = f"{file_uuid}{file_ext}"
# 2. Save immutable copy to /workdir/original
# This copy serves as the permanent, untouched reference of the ingested file
original_dir = os.path.join(settings.workdir, "original")
os.makedirs(original_dir, exist_ok=True)
# Use collision-resistant naming with -0001, -0002 suffixes
base_name = os.path.splitext(new_filename)[0]
original_file_path = get_unique_filepath_with_counter(original_dir, base_name, file_ext)
logger.info(f"[{task_id}] Saving immutable original to: {original_file_path}")
log_task_progress(
task_id,
"save_original",
"in_progress",
f"Saving original to {os.path.basename(original_file_path)}",
file_id=new_record.id,
)
shutil.copy(original_local_file, original_file_path)
log_task_progress(
task_id,
"save_original",
"success",
f"Original saved: {os.path.basename(original_file_path)}",
file_id=new_record.id,
)
# 2. Save immutable copy to /workdir/original (only for new files, not reprocessing)
# For reprocessing, the original_file_path should already exist in the database
if file_id is None: # New file - save original copy
# This copy serves as the permanent, untouched reference of the ingested file
original_dir = os.path.join(settings.workdir, "original")
os.makedirs(original_dir, exist_ok=True)
# Use collision-resistant naming with -0001, -0002 suffixes
base_name = os.path.splitext(new_filename)[0]
original_file_path = get_unique_filepath_with_counter(original_dir, base_name, file_ext)
logger.info(f"[{task_id}] Saving immutable original to: {original_file_path}")
log_task_progress(
task_id,
"save_original",
"in_progress",
f"Saving original to {os.path.basename(original_file_path)}",
file_id=new_record.id,
)
shutil.copy(original_local_file, original_file_path)
log_task_progress(
task_id,
"save_original",
"success",
f"Original saved: {os.path.basename(original_file_path)}",
file_id=new_record.id,
)
# Update the DB with original_file_path
new_record.original_file_path = original_file_path
else:
# Reprocessing - original should already exist
logger.info(f"[{task_id}] Reprocessing: original file already saved at {new_record.original_file_path}")
log_task_progress(
task_id,
"save_original",
"success",
"Reprocessing: using existing original",
file_id=new_record.id,
)
# 3. Copy to /workdir/tmp for processing
tmp_dir = os.path.join(settings.workdir, "tmp")
@@ -200,9 +215,8 @@ def process_document(self, original_local_file: str, original_filename: str = No
file_id=new_record.id,
)
# Update the DB with file paths
# Update the DB with local_filename
new_record.local_filename = new_local_path
new_record.original_file_path = original_file_path
db.commit()
# Store file_id before session closes to avoid DetachedInstanceError
tests/test_filename_utils.py
+88
View File
@@ -257,3 +257,91 @@ class TestFilenameUtilsEdgeCases:
assert '"' not in result
assert "|" not in result
assert "?" not in result
@pytest.mark.unit
class TestUniqueFilepathWithCounter:
"""Test unique filepath generation with numeric counter suffix"""
def test_get_unique_filepath_with_counter_no_collision(self, tmp_path):
"""Test that original filename is returned when no collision exists"""
from app.utils.filename_utils import get_unique_filepath_with_counter
result = get_unique_filepath_with_counter(str(tmp_path), "document")
assert result == str(tmp_path / "document.pdf")
def test_get_unique_filepath_with_counter_single_collision(self, tmp_path):
"""Test that -0001 suffix is added on first collision"""
from app.utils.filename_utils import get_unique_filepath_with_counter
# Create the base file
(tmp_path / "document.pdf").touch()
result = get_unique_filepath_with_counter(str(tmp_path), "document")
assert result == str(tmp_path / "document-0001.pdf")
def test_get_unique_filepath_with_counter_multiple_collisions(self, tmp_path):
"""Test that counter increments correctly for multiple collisions"""
from app.utils.filename_utils import get_unique_filepath_with_counter
# Create files with base name and first two counter suffixes
(tmp_path / "document.pdf").touch()
(tmp_path / "document-0001.pdf").touch()
(tmp_path / "document-0002.pdf").touch()
result = get_unique_filepath_with_counter(str(tmp_path), "document")
assert result == str(tmp_path / "document-0003.pdf")
def test_get_unique_filepath_with_counter_custom_extension(self, tmp_path):
"""Test with custom file extension"""
from app.utils.filename_utils import get_unique_filepath_with_counter
(tmp_path / "data.json").touch()
result = get_unique_filepath_with_counter(str(tmp_path), "data", extension=".json")
assert result == str(tmp_path / "data-0001.json")
def test_get_unique_filepath_with_counter_zero_padded(self, tmp_path):
"""Test that counter uses zero-padded 4-digit format"""
from app.utils.filename_utils import get_unique_filepath_with_counter
(tmp_path / "invoice.pdf").touch()
result = get_unique_filepath_with_counter(str(tmp_path), "invoice")
# Should be -0001, not -1
assert result == str(tmp_path / "invoice-0001.pdf")
assert "-1.pdf" not in result
def test_get_unique_filepath_with_counter_preserves_filename(self, tmp_path):
"""Test that complex filenames are preserved"""
from app.utils.filename_utils import get_unique_filepath_with_counter
filename = "2024-01-01_Invoice_Company-Name"
(tmp_path / f"{filename}.pdf").touch()
result = get_unique_filepath_with_counter(str(tmp_path), filename)
assert filename in result
assert result == str(tmp_path / f"{filename}-0001.pdf")
def test_get_unique_filepath_with_counter_high_count(self, tmp_path):
"""Test that function handles high counter values"""
from app.utils.filename_utils import get_unique_filepath_with_counter
# Create files up to -0099
(tmp_path / "test.pdf").touch()
for i in range(1, 100):
(tmp_path / f"test-{i:04d}.pdf").touch()
result = get_unique_filepath_with_counter(str(tmp_path), "test")
assert result == str(tmp_path / "test-0100.pdf")
def test_get_unique_filepath_with_counter_directory_creation(self, tmp_path):
"""Test with directory that already exists"""
from app.utils.filename_utils import get_unique_filepath_with_counter
# Directory already exists (tmp_path)
result = get_unique_filepath_with_counter(str(tmp_path), "newfile")
assert result == str(tmp_path / "newfile.pdf")
# File shouldn't be created, just path returned
assert not os.path.exists(result)
tests/test_storage_reorganization.py
+316
View File
@@ -0,0 +1,316 @@
"""
Tests for document storage reorganization features.
Tests the new functionality for storing immutable originals and processed copies,
collision handling, and forced Cloud OCR reprocessing.
"""
import os
import json
from unittest.mock import MagicMock, patch
import pytest
from sqlalchemy.orm import Session
from app.models import FileRecord
@pytest.mark.unit
@pytest.mark.requires_db
class TestImmutableOriginalStorage:
"""Test that original files are saved immutably to /workdir/original"""
def test_new_file_saves_original_copy(self, db_session, tmp_path):
"""Test that a new file creates an immutable original copy"""
from app.tasks.process_document import process_document
# Create test PDF
test_pdf = tmp_path / "test_input.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]
/Resources <<
/Font <<
/F1 <<
/Type /Font
/Subtype /Type1
/BaseFont /Helvetica
>>
>>
>>
/Contents 4 0 R
>>
endobj
4 0 obj
<<
/Length 44
>>
stream
BT
/F1 12 Tf
100 700 Td
(Test content) Tj
ET
endstream
endobj
xref
0 5
0000000000 65535 f
0000000009 00000 n
0000000058 00000 n
0000000115 00000 n
0000000306 00000 n
trailer
<<
/Size 5
/Root 1 0 R
>>
startxref
399
%%EOF
"""
test_pdf.write_bytes(pdf_content)
# Setup mocks
with (
patch("app.tasks.process_document.SessionLocal") as mock_session_local,
patch("app.tasks.process_document.settings") as mock_settings,
patch("app.tasks.process_document.log_task_progress"),
patch("app.tasks.process_document.extract_metadata_with_gpt") as mock_extract,
):
mock_settings.workdir = str(tmp_path)
mock_session_local.return_value.__enter__.return_value = db_session
mock_session_local.return_value.__exit__.return_value = None
mock_extract.delay = MagicMock()
# Call process_document
result = process_document(str(test_pdf), original_filename="test_input.pdf")
# Verify original directory was created
original_dir = tmp_path / "original"
assert original_dir.exists()
# Verify an original file was saved
original_files = list(original_dir.glob("*.pdf"))
assert len(original_files) > 0
# Verify database record has original_file_path
file_record = db_session.query(FileRecord).first()
assert file_record is not None
assert file_record.original_file_path is not None
assert "original" in file_record.original_file_path
def test_reprocessing_preserves_original(self, db_session, tmp_path):
"""Test that reprocessing doesn't create a duplicate original"""
from app.tasks.process_document import process_document
# Create test file and original
test_pdf = tmp_path / "test.pdf"
test_pdf.write_bytes(b"%PDF-1.4\ntest")
original_dir = tmp_path / "original"
original_dir.mkdir()
original_file = original_dir / "existing-original.pdf"
original_file.write_bytes(b"%PDF-1.4\noriginal")
# Create existing file record
file_record = FileRecord(
filehash="abc123",
original_filename="test.pdf",
local_filename=str(test_pdf),
original_file_path=str(original_file),
file_size=100,
mime_type="application/pdf"
)
db_session.add(file_record)
db_session.commit()
# Setup mocks
with (
patch("app.tasks.process_document.SessionLocal") as mock_session_local,
patch("app.tasks.process_document.settings") as mock_settings,
patch("app.tasks.process_document.log_task_progress"),
patch("app.tasks.process_document.extract_metadata_with_gpt") as mock_extract,
):
mock_settings.workdir = str(tmp_path)
mock_session_local.return_value.__enter__.return_value = db_session
mock_session_local.return_value.__exit__.return_value = None
mock_extract.delay = MagicMock()
# Reprocess with file_id
original_count = len(list(original_dir.glob("*.pdf")))
process_document(str(test_pdf), file_id=file_record.id)
# Should not create new original file
new_count = len(list(original_dir.glob("*.pdf")))
assert new_count == original_count
@pytest.mark.unit
class TestCollisionHandling:
"""Test filename collision handling with -0001 suffix format"""
def test_collision_handling_in_processed_dir(self, tmp_path):
"""Test that collision handling works in processed directory"""
from app.utils.filename_utils import get_unique_filepath_with_counter
processed_dir = tmp_path / "processed"
processed_dir.mkdir()
# Create first file
(processed_dir / "2024-01-01_Invoice.pdf").touch()
# Get unique path for same filename
result = get_unique_filepath_with_counter(str(processed_dir), "2024-01-01_Invoice")
assert "2024-01-01_Invoice-0001.pdf" in result
assert os.path.exists(str(processed_dir / "2024-01-01_Invoice.pdf"))
@pytest.mark.unit
class TestMetadataAugmentation:
"""Test that metadata JSON includes file path references"""
def test_metadata_includes_file_paths(self, tmp_path):
"""Test that persisted metadata includes original and processed paths"""
from app.tasks.embed_metadata_into_pdf import persist_metadata
metadata = {
"filename": "2024-01-01_Invoice",
"document_type": "Invoice",
"tags": ["finance", "2024"]
}
processed_file = tmp_path / "processed" / "2024-01-01_Invoice.pdf"
processed_file.parent.mkdir(parents=True)
processed_file.touch()
original_path = "/workdir/original/abc123.pdf"
processed_path = str(processed_file)
json_path = persist_metadata(
metadata,
str(processed_file),
original_file_path=original_path,
processed_file_path=processed_path
)
# Verify JSON was created
assert os.path.exists(json_path)
# Verify content
with open(json_path, 'r') as f:
saved_metadata = json.load(f)
assert "original_file_path" in saved_metadata
assert saved_metadata["original_file_path"] == original_path
assert "processed_file_path" in saved_metadata
assert saved_metadata["processed_file_path"] == processed_path
assert saved_metadata["filename"] == "2024-01-01_Invoice"
@pytest.mark.unit
@pytest.mark.requires_db
class TestForceCloudOCR:
"""Test forced Cloud OCR reprocessing functionality"""
def test_force_cloud_ocr_parameter(self, db_session, tmp_path):
"""Test that force_cloud_ocr parameter skips local text extraction"""
from app.tasks.process_document import process_document
# Create PDF with embedded text
test_pdf = tmp_path / "with_text.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]
/Resources <<
/Font <<
/F1 <<
/Type /Font
/Subtype /Type1
/BaseFont /Helvetica
>>
>>
>>
/Contents 4 0 R
>>
endobj
4 0 obj
<<
/Length 44
>>
stream
BT
/F1 12 Tf
100 700 Td
(Embedded text here) Tj
ET
endstream
endobj
xref
0 5
0000000000 65535 f
0000000009 00000 n
0000000058 00000 n
0000000115 00000 n
0000000306 00000 n
trailer
<<
/Size 5
/Root 1 0 R
>>
startxref
399
%%EOF
"""
test_pdf.write_bytes(pdf_content)
with (
patch("app.tasks.process_document.SessionLocal") as mock_session_local,
patch("app.tasks.process_document.settings") as mock_settings,
patch("app.tasks.process_document.log_task_progress"),
patch("app.tasks.process_document.process_with_azure_document_intelligence") as mock_azure,
):
mock_settings.workdir = str(tmp_path)
mock_session_local.return_value.__enter__.return_value = db_session
mock_session_local.return_value.__exit__.return_value = None
mock_azure.delay = MagicMock()
# Process with force_cloud_ocr=True
result = process_document(str(test_pdf), force_cloud_ocr=True)
# Should queue Azure OCR, not local extraction
mock_azure.delay.assert_called_once()
assert result["status"] == "Queued for forced OCR"