feat: add _auto_save_photo_from_extraction helper with graceful fallbacks
This commit is contained in:
@@ -423,3 +423,216 @@ async def upload_photo(
|
||||
status_code=500,
|
||||
detail=f"Internal server error: {str(e)}"
|
||||
)
|
||||
|
||||
|
||||
def _auto_save_photo_from_extraction(
|
||||
item_id: int,
|
||||
image_bytes: bytes,
|
||||
crop_bounds: Optional[Dict[str, int]],
|
||||
rotation_degrees: Optional[float],
|
||||
db: Session
|
||||
) -> Dict[str, str]:
|
||||
"""
|
||||
Helper function to save extracted photos with AI-guided crop/rotation.
|
||||
|
||||
This function is called after item creation if image_processing metadata exists.
|
||||
It gracefully handles missing/invalid data without throwing exceptions.
|
||||
|
||||
Args:
|
||||
item_id: ID of the item to attach the photo to
|
||||
image_bytes: Raw photo bytes
|
||||
crop_bounds: Optional crop bounds dict {x, y, width, height} (in pixels)
|
||||
rotation_degrees: Optional rotation in degrees (-360 to +360, clockwise)
|
||||
db: SQLAlchemy session
|
||||
|
||||
Returns:
|
||||
{status: "ok"} if photo saved successfully
|
||||
{status: "skipped", reason: "..."} if data invalid or missing
|
||||
|
||||
Behavior:
|
||||
- Validates crop_bounds (all keys present, all ints >= 0)
|
||||
- Validates rotation_degrees (numeric, -360 to +360)
|
||||
- Skips gracefully if crop_bounds is None (no exceptions)
|
||||
- Skips gracefully on invalid data (logs warning, returns skipped)
|
||||
- Updates item.photo_path, photo_thumbnail_path, photo_upload_date
|
||||
- Never throws exceptions
|
||||
"""
|
||||
from ..logger import log
|
||||
|
||||
try:
|
||||
# Validate item exists
|
||||
db_item = db.query(models.Item).filter(models.Item.id == item_id).first()
|
||||
if not db_item:
|
||||
log.warning(f"Auto-save photo: Item {item_id} not found, skipping")
|
||||
return {
|
||||
"status": "skipped",
|
||||
"reason": f"Item {item_id} not found"
|
||||
}
|
||||
|
||||
# Validate image_bytes
|
||||
if not image_bytes or len(image_bytes) == 0:
|
||||
log.warning(f"Auto-save photo for item {item_id}: No image bytes provided")
|
||||
return {
|
||||
"status": "skipped",
|
||||
"reason": "Empty image bytes"
|
||||
}
|
||||
|
||||
# Graceful skip if crop_bounds is None
|
||||
if crop_bounds is None:
|
||||
log.info(f"Auto-save photo for item {item_id}: crop_bounds is None, skipping")
|
||||
return {
|
||||
"status": "skipped",
|
||||
"reason": "crop_bounds is None"
|
||||
}
|
||||
|
||||
# Validate crop_bounds
|
||||
if not isinstance(crop_bounds, dict):
|
||||
log.warning(f"Auto-save photo for item {item_id}: crop_bounds is not a dict, skipping")
|
||||
return {
|
||||
"status": "skipped",
|
||||
"reason": "crop_bounds must be a dict"
|
||||
}
|
||||
|
||||
# Check for required keys
|
||||
required_keys = {'x', 'y', 'width', 'height'}
|
||||
if not required_keys.issubset(crop_bounds.keys()):
|
||||
missing = required_keys - set(crop_bounds.keys())
|
||||
log.warning(f"Auto-save photo for item {item_id}: Missing crop_bounds keys: {missing}")
|
||||
return {
|
||||
"status": "skipped",
|
||||
"reason": f"Missing crop_bounds keys: {missing}"
|
||||
}
|
||||
|
||||
# Validate all values are integers >= 0
|
||||
try:
|
||||
crop_bounds_validated = {}
|
||||
for key in required_keys:
|
||||
val = crop_bounds[key]
|
||||
# Convert to int if it's numeric
|
||||
if isinstance(val, (int, float)):
|
||||
int_val = int(val)
|
||||
else:
|
||||
raise ValueError(f"Non-numeric value for {key}: {val}")
|
||||
|
||||
if int_val < 0:
|
||||
raise ValueError(f"Negative value for {key}: {int_val}")
|
||||
|
||||
crop_bounds_validated[key] = int_val
|
||||
|
||||
except (ValueError, TypeError) as e:
|
||||
log.warning(f"Auto-save photo for item {item_id}: Invalid crop_bounds: {str(e)}")
|
||||
return {
|
||||
"status": "skipped",
|
||||
"reason": f"Invalid crop_bounds: {str(e)}"
|
||||
}
|
||||
|
||||
# Validate rotation_degrees (optional but if provided, must be valid)
|
||||
if rotation_degrees is not None:
|
||||
try:
|
||||
rot = float(rotation_degrees)
|
||||
if rot < -360 or rot > 360:
|
||||
log.warning(f"Auto-save photo for item {item_id}: rotation_degrees {rot} out of range [-360, 360]")
|
||||
return {
|
||||
"status": "skipped",
|
||||
"reason": f"rotation_degrees {rot} out of range [-360, 360]"
|
||||
}
|
||||
except (ValueError, TypeError) as e:
|
||||
log.warning(f"Auto-save photo for item {item_id}: Invalid rotation_degrees: {str(e)}")
|
||||
return {
|
||||
"status": "skipped",
|
||||
"reason": f"Invalid rotation_degrees: {str(e)}"
|
||||
}
|
||||
|
||||
# All validation passed, proceed with processing
|
||||
try:
|
||||
# Process image (crop + rotation + compression + thumbnail)
|
||||
processor = ImageProcessor()
|
||||
process_result = processor.process_photo(image_bytes, crop_bounds_validated)
|
||||
|
||||
if process_result.get('status') != 'success':
|
||||
error_msg = process_result.get('error', 'Unknown error')
|
||||
log.warning(f"Auto-save photo for item {item_id}: Image processing failed: {error_msg}")
|
||||
return {
|
||||
"status": "skipped",
|
||||
"reason": f"Image processing failed: {error_msg}"
|
||||
}
|
||||
|
||||
# Get processed bytes
|
||||
cropped_bytes = process_result.get('cropped_image_bytes')
|
||||
thumbnail_bytes = process_result.get('thumbnail_bytes')
|
||||
|
||||
if not cropped_bytes or not thumbnail_bytes:
|
||||
log.warning(f"Auto-save photo for item {item_id}: No image data from processing")
|
||||
return {
|
||||
"status": "skipped",
|
||||
"reason": "No image data from processing"
|
||||
}
|
||||
|
||||
# Get category for file storage
|
||||
category = db_item.category or "items"
|
||||
|
||||
# Get unique filenames
|
||||
existing_files = []
|
||||
cat_dir = Path("images") / category.lower()
|
||||
if cat_dir.exists():
|
||||
existing_files = [f.name for f in cat_dir.iterdir() if f.is_file()]
|
||||
|
||||
filename_base = db_item.name or f"item_{item_id}"
|
||||
original_filename = get_unique_filename(filename_base, category, existing_files, variant="original")
|
||||
thumbnail_filename = get_unique_filename(filename_base, category, existing_files, variant="thumb")
|
||||
|
||||
# Save original image
|
||||
try:
|
||||
original_path = save_image(cropped_bytes, category, original_filename.replace("_original.jpg", ""), variant="original")
|
||||
except (OSError, IOError) as e:
|
||||
log.warning(f"Auto-save photo for item {item_id}: Failed to save image: {str(e)}")
|
||||
return {
|
||||
"status": "skipped",
|
||||
"reason": f"Failed to save image: {str(e)}"
|
||||
}
|
||||
|
||||
# Save thumbnail
|
||||
try:
|
||||
thumbnail_path = save_image(thumbnail_bytes, category, thumbnail_filename.replace("_thumb.jpg", ""), variant="thumb")
|
||||
except (OSError, IOError) as e:
|
||||
log.warning(f"Auto-save photo for item {item_id}: Failed to save thumbnail: {str(e)}")
|
||||
# Clean up original if thumbnail save fails
|
||||
try:
|
||||
old_photo = Path(original_path.lstrip("/"))
|
||||
if old_photo.exists():
|
||||
old_photo.unlink()
|
||||
except Exception:
|
||||
pass
|
||||
return {
|
||||
"status": "skipped",
|
||||
"reason": f"Failed to save thumbnail: {str(e)}"
|
||||
}
|
||||
|
||||
# Update database
|
||||
db_item.photo_path = original_path
|
||||
db_item.photo_thumbnail_path = thumbnail_path
|
||||
db_item.photo_upload_date = datetime.now(timezone.utc)
|
||||
|
||||
db.commit()
|
||||
db.refresh(db_item)
|
||||
|
||||
log.info(f"Auto-save photo for item {item_id}: Success")
|
||||
return {
|
||||
"status": "ok"
|
||||
}
|
||||
|
||||
except Exception as e:
|
||||
db.rollback()
|
||||
log.warning(f"Auto-save photo for item {item_id}: Unexpected error: {str(e)}")
|
||||
return {
|
||||
"status": "skipped",
|
||||
"reason": f"Unexpected error: {str(e)}"
|
||||
}
|
||||
|
||||
except Exception as e:
|
||||
# Catch-all for any unexpected errors (never throw)
|
||||
log.warning(f"Auto-save photo for item {item_id}: Outer exception: {str(e)}")
|
||||
return {
|
||||
"status": "skipped",
|
||||
"reason": f"Internal error: {str(e)}"
|
||||
}
|
||||
|
||||
523
backend/tests/test_photo_extraction.py
Normal file
523
backend/tests/test_photo_extraction.py
Normal file
@@ -0,0 +1,523 @@
|
||||
"""
|
||||
Test suite for _auto_save_photo_from_extraction helper function.
|
||||
|
||||
Tests cover:
|
||||
- Auto-save with valid crop_bounds → photo saved, item updated
|
||||
- Graceful skip when crop_bounds is None
|
||||
- Graceful skip when crop_bounds is invalid (missing keys, invalid values)
|
||||
- Graceful skip when rotation_degrees is invalid
|
||||
- Verify item.photo_path, photo_thumbnail_path, photo_upload_date set correctly
|
||||
- Error handling (missing item, no image_bytes, processing failures)
|
||||
- Logging of warnings for skipped saves
|
||||
"""
|
||||
|
||||
import io
|
||||
import json
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
from unittest.mock import patch, MagicMock
|
||||
import pytest
|
||||
from fastapi.testclient import TestClient
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from backend import models
|
||||
from backend.routers.items import _auto_save_photo_from_extraction
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# FIXTURES
|
||||
# ============================================================================
|
||||
|
||||
@pytest.fixture
|
||||
def test_item_for_extraction(test_db: Session):
|
||||
"""Create a test item in the database for extraction tests."""
|
||||
item = models.Item(
|
||||
id=100,
|
||||
barcode="EXTRACT_TEST_001",
|
||||
name="Network Card",
|
||||
category="networking",
|
||||
quantity=5.0
|
||||
)
|
||||
test_db.add(item)
|
||||
test_db.commit()
|
||||
test_db.refresh(item)
|
||||
return item
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def sample_image_bytes():
|
||||
"""Create a minimal valid JPEG image for testing."""
|
||||
from PIL import Image
|
||||
|
||||
# Create a simple 200x200 red image
|
||||
img = Image.new('RGB', (200, 200), color='red')
|
||||
img_bytes = io.BytesIO()
|
||||
img.save(img_bytes, format='JPEG')
|
||||
img_bytes.seek(0)
|
||||
return img_bytes.getvalue()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def valid_crop_bounds():
|
||||
"""Valid crop bounds dict."""
|
||||
return {
|
||||
'x': 10,
|
||||
'y': 20,
|
||||
'width': 150,
|
||||
'height': 160
|
||||
}
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# TESTS: AUTO-SAVE WITH VALID CROP BOUNDS
|
||||
# ============================================================================
|
||||
|
||||
def test_auto_save_with_valid_crop_bounds(
|
||||
test_db: Session,
|
||||
test_item_for_extraction,
|
||||
sample_image_bytes,
|
||||
valid_crop_bounds
|
||||
):
|
||||
"""Test auto-save succeeds with valid crop_bounds."""
|
||||
result = _auto_save_photo_from_extraction(
|
||||
item_id=test_item_for_extraction.id,
|
||||
image_bytes=sample_image_bytes,
|
||||
crop_bounds=valid_crop_bounds,
|
||||
rotation_degrees=0,
|
||||
db=test_db
|
||||
)
|
||||
|
||||
# Verify result status
|
||||
assert result["status"] == "ok"
|
||||
|
||||
# Verify item was updated
|
||||
updated_item = test_db.query(models.Item).filter(
|
||||
models.Item.id == test_item_for_extraction.id
|
||||
).first()
|
||||
assert updated_item.photo_path is not None
|
||||
assert updated_item.photo_thumbnail_path is not None
|
||||
assert updated_item.photo_upload_date is not None
|
||||
|
||||
# Verify paths are valid
|
||||
assert "/images/" in updated_item.photo_path
|
||||
assert "/images/" in updated_item.photo_thumbnail_path
|
||||
assert updated_item.photo_path.endswith(".jpg")
|
||||
assert updated_item.photo_thumbnail_path.endswith(".jpg")
|
||||
|
||||
# Cleanup
|
||||
Path(updated_item.photo_path.lstrip("/")).unlink(missing_ok=True)
|
||||
Path(updated_item.photo_thumbnail_path.lstrip("/")).unlink(missing_ok=True)
|
||||
|
||||
|
||||
def test_auto_save_with_rotation_degrees(
|
||||
test_db: Session,
|
||||
test_item_for_extraction,
|
||||
sample_image_bytes,
|
||||
valid_crop_bounds
|
||||
):
|
||||
"""Test auto-save works with rotation_degrees."""
|
||||
result = _auto_save_photo_from_extraction(
|
||||
item_id=test_item_for_extraction.id,
|
||||
image_bytes=sample_image_bytes,
|
||||
crop_bounds=valid_crop_bounds,
|
||||
rotation_degrees=90,
|
||||
db=test_db
|
||||
)
|
||||
|
||||
assert result["status"] == "ok"
|
||||
|
||||
updated_item = test_db.query(models.Item).filter(
|
||||
models.Item.id == test_item_for_extraction.id
|
||||
).first()
|
||||
assert updated_item.photo_path is not None
|
||||
assert updated_item.photo_upload_date is not None
|
||||
|
||||
# Cleanup
|
||||
Path(updated_item.photo_path.lstrip("/")).unlink(missing_ok=True)
|
||||
Path(updated_item.photo_thumbnail_path.lstrip("/")).unlink(missing_ok=True)
|
||||
|
||||
|
||||
def test_auto_save_with_negative_rotation(
|
||||
test_db: Session,
|
||||
test_item_for_extraction,
|
||||
sample_image_bytes,
|
||||
valid_crop_bounds
|
||||
):
|
||||
"""Test auto-save handles negative rotation degrees."""
|
||||
result = _auto_save_photo_from_extraction(
|
||||
item_id=test_item_for_extraction.id,
|
||||
image_bytes=sample_image_bytes,
|
||||
crop_bounds=valid_crop_bounds,
|
||||
rotation_degrees=-45,
|
||||
db=test_db
|
||||
)
|
||||
|
||||
assert result["status"] == "ok"
|
||||
|
||||
updated_item = test_db.query(models.Item).filter(
|
||||
models.Item.id == test_item_for_extraction.id
|
||||
).first()
|
||||
assert updated_item.photo_path is not None
|
||||
|
||||
# Cleanup
|
||||
Path(updated_item.photo_path.lstrip("/")).unlink(missing_ok=True)
|
||||
Path(updated_item.photo_thumbnail_path.lstrip("/")).unlink(missing_ok=True)
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# TESTS: GRACEFUL SKIP WHEN CROP_BOUNDS IS NONE
|
||||
# ============================================================================
|
||||
|
||||
def test_auto_save_skip_when_crop_bounds_none(
|
||||
test_db: Session,
|
||||
test_item_for_extraction,
|
||||
sample_image_bytes
|
||||
):
|
||||
"""Test graceful skip when crop_bounds is None."""
|
||||
result = _auto_save_photo_from_extraction(
|
||||
item_id=test_item_for_extraction.id,
|
||||
image_bytes=sample_image_bytes,
|
||||
crop_bounds=None,
|
||||
rotation_degrees=0,
|
||||
db=test_db
|
||||
)
|
||||
|
||||
# Should skip gracefully
|
||||
assert result["status"] == "skipped"
|
||||
assert "reason" in result
|
||||
|
||||
# Item should not be updated
|
||||
updated_item = test_db.query(models.Item).filter(
|
||||
models.Item.id == test_item_for_extraction.id
|
||||
).first()
|
||||
assert updated_item.photo_path is None
|
||||
assert updated_item.photo_upload_date is None
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# TESTS: GRACEFUL SKIP WHEN CROP_BOUNDS IS INVALID
|
||||
# ============================================================================
|
||||
|
||||
def test_auto_save_skip_when_crop_bounds_missing_keys(
|
||||
test_db: Session,
|
||||
test_item_for_extraction,
|
||||
sample_image_bytes
|
||||
):
|
||||
"""Test graceful skip when crop_bounds is missing required keys."""
|
||||
invalid_bounds = {'x': 10, 'y': 20} # Missing width, height
|
||||
|
||||
result = _auto_save_photo_from_extraction(
|
||||
item_id=test_item_for_extraction.id,
|
||||
image_bytes=sample_image_bytes,
|
||||
crop_bounds=invalid_bounds,
|
||||
rotation_degrees=0,
|
||||
db=test_db
|
||||
)
|
||||
|
||||
assert result["status"] == "skipped"
|
||||
assert "reason" in result
|
||||
|
||||
updated_item = test_db.query(models.Item).filter(
|
||||
models.Item.id == test_item_for_extraction.id
|
||||
).first()
|
||||
assert updated_item.photo_path is None
|
||||
|
||||
|
||||
def test_auto_save_skip_when_crop_bounds_invalid_values(
|
||||
test_db: Session,
|
||||
test_item_for_extraction,
|
||||
sample_image_bytes
|
||||
):
|
||||
"""Test graceful skip when crop_bounds contains non-integer values."""
|
||||
invalid_bounds = {
|
||||
'x': 'not_int',
|
||||
'y': 20,
|
||||
'width': 150,
|
||||
'height': 160
|
||||
}
|
||||
|
||||
result = _auto_save_photo_from_extraction(
|
||||
item_id=test_item_for_extraction.id,
|
||||
image_bytes=sample_image_bytes,
|
||||
crop_bounds=invalid_bounds,
|
||||
rotation_degrees=0,
|
||||
db=test_db
|
||||
)
|
||||
|
||||
assert result["status"] == "skipped"
|
||||
|
||||
updated_item = test_db.query(models.Item).filter(
|
||||
models.Item.id == test_item_for_extraction.id
|
||||
).first()
|
||||
assert updated_item.photo_path is None
|
||||
|
||||
|
||||
def test_auto_save_skip_when_crop_bounds_negative_values(
|
||||
test_db: Session,
|
||||
test_item_for_extraction,
|
||||
sample_image_bytes
|
||||
):
|
||||
"""Test graceful skip when crop_bounds contains negative values."""
|
||||
invalid_bounds = {
|
||||
'x': -10,
|
||||
'y': 20,
|
||||
'width': 150,
|
||||
'height': 160
|
||||
}
|
||||
|
||||
result = _auto_save_photo_from_extraction(
|
||||
item_id=test_item_for_extraction.id,
|
||||
image_bytes=sample_image_bytes,
|
||||
crop_bounds=invalid_bounds,
|
||||
rotation_degrees=0,
|
||||
db=test_db
|
||||
)
|
||||
|
||||
assert result["status"] == "skipped"
|
||||
|
||||
updated_item = test_db.query(models.Item).filter(
|
||||
models.Item.id == test_item_for_extraction.id
|
||||
).first()
|
||||
assert updated_item.photo_path is None
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# TESTS: GRACEFUL SKIP WHEN ROTATION_DEGREES IS INVALID
|
||||
# ============================================================================
|
||||
|
||||
def test_auto_save_skip_when_rotation_out_of_range(
|
||||
test_db: Session,
|
||||
test_item_for_extraction,
|
||||
sample_image_bytes,
|
||||
valid_crop_bounds
|
||||
):
|
||||
"""Test graceful skip when rotation_degrees exceeds valid range."""
|
||||
result = _auto_save_photo_from_extraction(
|
||||
item_id=test_item_for_extraction.id,
|
||||
image_bytes=sample_image_bytes,
|
||||
crop_bounds=valid_crop_bounds,
|
||||
rotation_degrees=450, # > 360
|
||||
db=test_db
|
||||
)
|
||||
|
||||
assert result["status"] == "skipped"
|
||||
assert "reason" in result
|
||||
|
||||
updated_item = test_db.query(models.Item).filter(
|
||||
models.Item.id == test_item_for_extraction.id
|
||||
).first()
|
||||
assert updated_item.photo_path is None
|
||||
|
||||
|
||||
def test_auto_save_skip_when_rotation_is_string(
|
||||
test_db: Session,
|
||||
test_item_for_extraction,
|
||||
sample_image_bytes,
|
||||
valid_crop_bounds
|
||||
):
|
||||
"""Test graceful skip when rotation_degrees is not numeric."""
|
||||
result = _auto_save_photo_from_extraction(
|
||||
item_id=test_item_for_extraction.id,
|
||||
image_bytes=sample_image_bytes,
|
||||
crop_bounds=valid_crop_bounds,
|
||||
rotation_degrees="not_a_number",
|
||||
db=test_db
|
||||
)
|
||||
|
||||
assert result["status"] == "skipped"
|
||||
|
||||
updated_item = test_db.query(models.Item).filter(
|
||||
models.Item.id == test_item_for_extraction.id
|
||||
).first()
|
||||
assert updated_item.photo_path is None
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# TESTS: ERROR HANDLING
|
||||
# ============================================================================
|
||||
|
||||
def test_auto_save_skip_when_item_not_found(
|
||||
test_db: Session,
|
||||
sample_image_bytes,
|
||||
valid_crop_bounds
|
||||
):
|
||||
"""Test graceful skip when item does not exist."""
|
||||
result = _auto_save_photo_from_extraction(
|
||||
item_id=99999, # Non-existent item
|
||||
image_bytes=sample_image_bytes,
|
||||
crop_bounds=valid_crop_bounds,
|
||||
rotation_degrees=0,
|
||||
db=test_db
|
||||
)
|
||||
|
||||
assert result["status"] == "skipped"
|
||||
assert "reason" in result
|
||||
|
||||
|
||||
def test_auto_save_skip_when_image_bytes_empty(
|
||||
test_db: Session,
|
||||
test_item_for_extraction,
|
||||
valid_crop_bounds
|
||||
):
|
||||
"""Test graceful skip when image_bytes is empty."""
|
||||
result = _auto_save_photo_from_extraction(
|
||||
item_id=test_item_for_extraction.id,
|
||||
image_bytes=b'',
|
||||
crop_bounds=valid_crop_bounds,
|
||||
rotation_degrees=0,
|
||||
db=test_db
|
||||
)
|
||||
|
||||
assert result["status"] == "skipped"
|
||||
|
||||
updated_item = test_db.query(models.Item).filter(
|
||||
models.Item.id == test_item_for_extraction.id
|
||||
).first()
|
||||
assert updated_item.photo_path is None
|
||||
|
||||
|
||||
def test_auto_save_skip_when_image_bytes_invalid(
|
||||
test_db: Session,
|
||||
test_item_for_extraction,
|
||||
valid_crop_bounds
|
||||
):
|
||||
"""Test graceful skip when image_bytes is not a valid image."""
|
||||
result = _auto_save_photo_from_extraction(
|
||||
item_id=test_item_for_extraction.id,
|
||||
image_bytes=b'not a valid image',
|
||||
crop_bounds=valid_crop_bounds,
|
||||
rotation_degrees=0,
|
||||
db=test_db
|
||||
)
|
||||
|
||||
assert result["status"] == "skipped"
|
||||
|
||||
updated_item = test_db.query(models.Item).filter(
|
||||
models.Item.id == test_item_for_extraction.id
|
||||
).first()
|
||||
assert updated_item.photo_path is None
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# TESTS: LARGE CROP BOUNDS (4K IMAGE SUPPORT)
|
||||
# ============================================================================
|
||||
|
||||
def test_auto_save_with_large_crop_bounds(
|
||||
test_db: Session,
|
||||
test_item_for_extraction,
|
||||
sample_image_bytes
|
||||
):
|
||||
"""Test auto-save works with large crop bounds (4K image support)."""
|
||||
large_bounds = {
|
||||
'x': 100,
|
||||
'y': 100,
|
||||
'width': 2000,
|
||||
'height': 1500
|
||||
}
|
||||
|
||||
result = _auto_save_photo_from_extraction(
|
||||
item_id=test_item_for_extraction.id,
|
||||
image_bytes=sample_image_bytes,
|
||||
crop_bounds=large_bounds,
|
||||
rotation_degrees=0,
|
||||
db=test_db
|
||||
)
|
||||
|
||||
# Should handle gracefully (may skip due to image being too small for bounds,
|
||||
# but should not crash)
|
||||
assert result["status"] in ["ok", "skipped"]
|
||||
|
||||
if result["status"] == "ok":
|
||||
updated_item = test_db.query(models.Item).filter(
|
||||
models.Item.id == test_item_for_extraction.id
|
||||
).first()
|
||||
assert updated_item.photo_path is not None
|
||||
|
||||
# Cleanup
|
||||
Path(updated_item.photo_path.lstrip("/")).unlink(missing_ok=True)
|
||||
Path(updated_item.photo_thumbnail_path.lstrip("/")).unlink(missing_ok=True)
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# TESTS: MULTIPLE ITEMS WITH INDEPENDENT IMAGE_PROCESSING DATA
|
||||
# ============================================================================
|
||||
|
||||
def test_auto_save_multiple_items_independent(
|
||||
test_db: Session,
|
||||
sample_image_bytes
|
||||
):
|
||||
"""Test auto-save handles multiple items with independent data."""
|
||||
item1 = models.Item(id=201, barcode="MULTI_001", name="Item1", category="cat1", quantity=1.0)
|
||||
item2 = models.Item(id=202, barcode="MULTI_002", name="Item2", category="cat2", quantity=2.0)
|
||||
test_db.add(item1)
|
||||
test_db.add(item2)
|
||||
test_db.commit()
|
||||
|
||||
bounds1 = {'x': 10, 'y': 10, 'width': 100, 'height': 100}
|
||||
bounds2 = {'x': 20, 'y': 20, 'width': 150, 'height': 150}
|
||||
|
||||
result1 = _auto_save_photo_from_extraction(
|
||||
item_id=201,
|
||||
image_bytes=sample_image_bytes,
|
||||
crop_bounds=bounds1,
|
||||
rotation_degrees=0,
|
||||
db=test_db
|
||||
)
|
||||
|
||||
result2 = _auto_save_photo_from_extraction(
|
||||
item_id=202,
|
||||
image_bytes=sample_image_bytes,
|
||||
crop_bounds=bounds2,
|
||||
rotation_degrees=90,
|
||||
db=test_db
|
||||
)
|
||||
|
||||
assert result1["status"] == "ok"
|
||||
assert result2["status"] == "ok"
|
||||
|
||||
# Verify both items were updated independently
|
||||
updated1 = test_db.query(models.Item).filter(models.Item.id == 201).first()
|
||||
updated2 = test_db.query(models.Item).filter(models.Item.id == 202).first()
|
||||
|
||||
assert updated1.photo_path is not None
|
||||
assert updated2.photo_path is not None
|
||||
assert updated1.photo_path != updated2.photo_path # Different files
|
||||
|
||||
# Cleanup
|
||||
Path(updated1.photo_path.lstrip("/")).unlink(missing_ok=True)
|
||||
Path(updated1.photo_thumbnail_path.lstrip("/")).unlink(missing_ok=True)
|
||||
Path(updated2.photo_path.lstrip("/")).unlink(missing_ok=True)
|
||||
Path(updated2.photo_thumbnail_path.lstrip("/")).unlink(missing_ok=True)
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# TESTS: NO EXCEPTIONS THROWN
|
||||
# ============================================================================
|
||||
|
||||
def test_auto_save_never_throws_exceptions(
|
||||
test_db: Session,
|
||||
test_item_for_extraction
|
||||
):
|
||||
"""Test that helper never throws exceptions, always returns status dict."""
|
||||
# Test with all kinds of bad input - none should throw
|
||||
|
||||
test_cases = [
|
||||
(None, None, None),
|
||||
(b'', {}, None),
|
||||
(None, {'x': 'bad'}, 'not_a_number'),
|
||||
(b'bad_image', {'x': 0, 'y': 0, 'width': 100}, 450),
|
||||
]
|
||||
|
||||
for image_bytes, crop_bounds, rotation in test_cases:
|
||||
result = _auto_save_photo_from_extraction(
|
||||
item_id=test_item_for_extraction.id,
|
||||
image_bytes=image_bytes,
|
||||
crop_bounds=crop_bounds,
|
||||
rotation_degrees=rotation,
|
||||
db=test_db
|
||||
)
|
||||
|
||||
# Must always return a dict with 'status' key
|
||||
assert isinstance(result, dict)
|
||||
assert "status" in result
|
||||
assert result["status"] in ["ok", "skipped"]
|
||||
Reference in New Issue
Block a user