Files
tfm_ainventory/backend/tests/test_photo_extraction.py

673 lines
20 KiB
Python

"""
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"]
# ============================================================================
# INTEGRATION TESTS: FULL FLOW (create_item endpoint with auto-save)
# ============================================================================
def test_create_item_with_image_processing_integration(
admin_client,
sample_image_bytes
):
"""Integration test: create item with extracted image → photo auto-saved with crop/rotation."""
import base64
# Encode image as base64 for API payload
image_base64 = base64.b64encode(sample_image_bytes).decode()
item_data = {
"name": "NVMe Storage Drive",
"category": "Storage",
"type": "NVMe",
"quantity": 1,
"barcode": "NVM-2024-001",
"part_number": "P66093-002",
"extracted_image_bytes": image_base64,
"image_processing": {
"crop_bounds": {"x": 45, "y": 80, "width": 350, "height": 220},
"rotation_degrees": 12,
"confidence": 0.94
}
}
response = admin_client.post("/items/", json=item_data)
# Verify item was created
assert response.status_code == 201
data = response.json()
assert data["id"] is not None
assert data["name"] == "NVMe Storage Drive"
assert data["barcode"] == "NVM-2024-001"
# Verify photo was auto-saved
assert data["photo_path"] is not None
assert data["photo_thumbnail_path"] is not None
assert data["photo_upload_date"] is not None
# Verify photo paths are valid
assert "/images/" in data["photo_path"]
assert "/images/" in data["photo_thumbnail_path"]
assert data["photo_path"].endswith(".jpg")
assert data["photo_thumbnail_path"].endswith(".jpg")
# Cleanup
Path(data["photo_path"].lstrip("/")).unlink(missing_ok=True)
Path(data["photo_thumbnail_path"].lstrip("/")).unlink(missing_ok=True)
def test_create_item_with_invalid_image_processing(
admin_client,
sample_image_bytes
):
"""Integration test: Item created even if image_processing is invalid, photo skipped gracefully."""
import base64
image_base64 = base64.b64encode(sample_image_bytes).decode()
item_data = {
"name": "Test Item Invalid",
"category": "Storage",
"type": "SSD",
"quantity": 1,
"barcode": "TEST-INVALID-001",
"extracted_image_bytes": image_base64,
"image_processing": {
# Missing crop_bounds or invalid values
"rotation_degrees": 999, # Invalid (out of range)
"confidence": 1.5 # Invalid (>1.0)
}
}
response = admin_client.post("/items/", json=item_data)
# Item should still be created successfully
assert response.status_code == 201
data = response.json()
assert data["id"] is not None
assert data["name"] == "Test Item Invalid"
# Photo should not be saved (invalid image_processing)
assert data["photo_path"] is None
assert data["photo_thumbnail_path"] is None
assert data["photo_upload_date"] is None
def test_create_item_without_image_processing(
admin_client
):
"""Integration test: Backward compatibility - old clients without image_processing work."""
item_data = {
"name": "Old Style Item",
"category": "Storage",
"type": "SSD",
"quantity": 1,
"barcode": "OLD-STYLE-001"
}
response = admin_client.post("/items/", json=item_data)
# Item should be created
assert response.status_code == 201
data = response.json()
assert data["id"] is not None
assert data["name"] == "Old Style Item"
# No photo expected (no extracted_image_bytes provided)
assert data["photo_path"] is None
assert data["photo_thumbnail_path"] is None
assert data["photo_upload_date"] is None
def test_create_item_with_image_bytes_but_no_processing(
admin_client,
sample_image_bytes
):
"""Integration test: Image bytes without image_processing → item created, photo not saved."""
import base64
image_base64 = base64.b64encode(sample_image_bytes).decode()
item_data = {
"name": "Image Bytes Only",
"category": "Storage",
"type": "SATA",
"quantity": 2,
"barcode": "BYTES-ONLY-001",
"extracted_image_bytes": image_base64
# No image_processing field
}
response = admin_client.post("/items/", json=item_data)
# Item should be created
assert response.status_code == 201
data = response.json()
assert data["id"] is not None
# Photo not saved (no image_processing means no crop info)
assert data["photo_path"] is None
assert data["photo_thumbnail_path"] is None
assert data["photo_upload_date"] is None