Daniel Bedeleanu
ea49cd6e4a
- Create backend/services/image_storage.py with 4 core functions:
- sanitize_filename(): remove unsafe chars, limit to 255 chars, convert to lowercase
- get_unique_filename(): handle collisions with UUID suffix (format: {name}_{uuid8}_{variant}.jpg)
- ensure_image_directories(): create /images/ root and category subdirs on startup
- save_image(): save bytes to /images/{category}/{filename}, returns relative path
- Create comprehensive test suite (22 tests) covering all functionality
- Integrate ensure_image_directories() into FastAPI startup event
- Directory structure: /images/{category}/{filename}
- Collision handling: auto-suffix with UUID if filename exists
- All tests passing, pathlib.Path for safe operations
274 lines
12 KiB
Python
274 lines
12 KiB
Python
"""Tests for image storage utilities."""
|
|
import pytest
|
|
import tempfile
|
|
from pathlib import Path
|
|
from unittest.mock import patch, MagicMock
|
|
|
|
# Import will work after we create the module
|
|
from backend.services.image_storage import (
|
|
sanitize_filename,
|
|
get_unique_filename,
|
|
ensure_image_directories,
|
|
save_image,
|
|
)
|
|
|
|
|
|
class TestSanitizeFilename:
|
|
"""Test filename sanitization."""
|
|
|
|
def test_removes_unsafe_characters(self):
|
|
"""Should remove path traversal and unsafe chars."""
|
|
# Test path traversal attempts - ".." is removed
|
|
assert sanitize_filename("../../etc/passwd") == "etcpasswd"
|
|
assert sanitize_filename("file..txt") == "filetxt" # ".." removed (path traversal)
|
|
assert sanitize_filename("file/with/slashes.jpg") == "filewithslashes.jpg"
|
|
assert sanitize_filename("file\\with\\backslashes.jpg") == "filewithbackslashes.jpg"
|
|
|
|
def test_converts_to_lowercase(self):
|
|
"""Should convert to lowercase."""
|
|
assert sanitize_filename("MyFile.JPG") == "myfile.jpg"
|
|
assert sanitize_filename("UPPERCASE.PNG") == "uppercase.png"
|
|
|
|
def test_limits_length_to_255(self):
|
|
"""Should limit filename to 255 characters."""
|
|
long_name = "a" * 300 + ".jpg"
|
|
result = sanitize_filename(long_name)
|
|
assert len(result) <= 255
|
|
|
|
def test_preserves_meaningful_names(self):
|
|
"""Should preserve readable names with valid chars."""
|
|
assert sanitize_filename("SFP-LR_original.jpg") == "sfp-lr_original.jpg"
|
|
assert sanitize_filename("networking-equipment-2024.jpg") == "networking-equipment-2024.jpg"
|
|
assert sanitize_filename("item_123_v2.png") == "item_123_v2.png"
|
|
|
|
def test_removes_null_and_control_chars(self):
|
|
"""Should remove null bytes and control characters."""
|
|
assert sanitize_filename("file\x00null.jpg") == "filenull.jpg"
|
|
assert sanitize_filename("file\n\r\t.jpg") == "file.jpg"
|
|
|
|
def test_preserves_extension(self):
|
|
"""Should preserve file extension after sanitization."""
|
|
assert sanitize_filename("My_File.JPG").endswith(".jpg")
|
|
assert sanitize_filename("test.PNG").endswith(".png")
|
|
assert sanitize_filename("doc.PDF").endswith(".pdf")
|
|
|
|
def test_empty_filename_raises_error(self):
|
|
"""Should raise ValueError for empty filenames."""
|
|
with pytest.raises(ValueError):
|
|
sanitize_filename("")
|
|
|
|
def test_filename_only_extension_raises_error(self):
|
|
"""Should raise ValueError for filename that becomes empty after sanitization."""
|
|
# ".jpg" becomes ".jpg" which is valid (dot + extension)
|
|
# Test with only special chars that result in empty after sanitization
|
|
with pytest.raises(ValueError):
|
|
sanitize_filename("...")
|
|
|
|
|
|
class TestGetUniqueFilename:
|
|
"""Test collision detection and UUID suffix generation."""
|
|
|
|
def test_no_collision_returns_base_name(self):
|
|
"""Should return base name when no collision exists."""
|
|
result = get_unique_filename("SFP-LR", "networking", [])
|
|
assert result == "sfp-lr_original.jpg" # Sanitized to lowercase
|
|
|
|
def test_collision_adds_uuid_suffix(self):
|
|
"""Should add UUID suffix when collision detected."""
|
|
existing = ["sfp-lr_original.jpg"]
|
|
result = get_unique_filename("SFP-LR", "networking", existing)
|
|
# Format: {name}_{uuid_first_8}_{variant}.{ext}
|
|
assert result.startswith("sfp-lr_")
|
|
assert "_original.jpg" in result
|
|
assert len(result.split("_")[1]) == 8 # UUID first 8 chars
|
|
|
|
def test_uuid_suffix_is_valid_hex(self):
|
|
"""UUID suffix should be valid hex characters."""
|
|
existing = ["sfp-lr_original.jpg"]
|
|
result = get_unique_filename("SFP-LR", "networking", existing)
|
|
parts = result.split("_")
|
|
uuid_part = parts[1]
|
|
# Should be valid hex string
|
|
try:
|
|
int(uuid_part, 16)
|
|
except ValueError:
|
|
pytest.fail(f"UUID part '{uuid_part}' is not valid hex")
|
|
|
|
def test_multiple_collisions_generates_new_uuid(self):
|
|
"""Each collision should generate a different UUID suffix."""
|
|
existing = [
|
|
"sfp-lr_original.jpg",
|
|
"sfp-lr_a1b2c3_original.jpg",
|
|
"sfp-lr_d4e5f6_original.jpg"
|
|
]
|
|
result = get_unique_filename("SFP-LR", "networking", existing)
|
|
# Should not match any existing file
|
|
assert result not in existing
|
|
|
|
def test_case_insensitive_collision_detection(self):
|
|
"""Should detect collisions case-insensitively."""
|
|
existing = ["SFP-LR_ORIGINAL.JPG"] # Different case
|
|
result = get_unique_filename("SFP-LR", "networking", existing)
|
|
# Should detect collision despite case difference
|
|
assert "_original.jpg" in result
|
|
assert result != "SFP-LR_original.jpg"
|
|
|
|
def test_different_variants_in_collision_detection(self):
|
|
"""Should treat original and thumb as different variants."""
|
|
# original variant
|
|
result1 = get_unique_filename("SFP-LR", "networking", [], variant="original")
|
|
assert "_original.jpg" in result1
|
|
|
|
# thumb variant
|
|
result2 = get_unique_filename("SFP-LR", "networking", [], variant="thumb")
|
|
assert "_thumb.jpg" in result2
|
|
|
|
|
|
class TestEnsureImageDirectories:
|
|
"""Test directory creation on startup."""
|
|
|
|
def test_creates_images_root_directory(self):
|
|
"""Should create /images/ directory if it doesn't exist."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
images_dir = Path(tmpdir) / "images"
|
|
assert not images_dir.exists()
|
|
|
|
with patch("backend.services.image_storage.IMAGES_ROOT", images_dir):
|
|
ensure_image_directories()
|
|
assert images_dir.exists()
|
|
assert images_dir.is_dir()
|
|
|
|
def test_creates_category_subdirectories(self):
|
|
"""Should create category-specific subdirectories."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
images_dir = Path(tmpdir) / "images"
|
|
categories = ["networking", "electronics", "mechanical"]
|
|
|
|
with patch("backend.services.image_storage.IMAGES_ROOT", images_dir), \
|
|
patch("backend.services.image_storage.get_categories") as mock_get_cats:
|
|
mock_get_cats.return_value = categories
|
|
|
|
ensure_image_directories()
|
|
|
|
for cat in categories:
|
|
cat_dir = images_dir / cat
|
|
assert cat_dir.exists(), f"Category directory {cat_dir} not created"
|
|
assert cat_dir.is_dir()
|
|
|
|
def test_idempotent_directory_creation(self):
|
|
"""Should succeed if directories already exist."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
images_dir = Path(tmpdir) / "images"
|
|
images_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
with patch("backend.services.image_storage.IMAGES_ROOT", images_dir):
|
|
# Should not raise error
|
|
ensure_image_directories()
|
|
assert images_dir.exists()
|
|
|
|
|
|
class TestSaveImage:
|
|
"""Test image file saving."""
|
|
|
|
def test_saves_image_to_correct_path(self):
|
|
"""Should save image bytes to /images/{category}/{filename}."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
images_dir = Path(tmpdir) / "images"
|
|
images_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
image_bytes = b"fake image data"
|
|
category = "networking"
|
|
filename_base = "SFP-LR"
|
|
|
|
with patch("backend.services.image_storage.IMAGES_ROOT", images_dir):
|
|
result_path = save_image(image_bytes, category, filename_base, "original")
|
|
|
|
# Check file was created
|
|
expected_file = images_dir / category / "sfp-lr_original.jpg"
|
|
assert expected_file.exists()
|
|
assert expected_file.read_bytes() == image_bytes
|
|
|
|
def test_returns_relative_path(self):
|
|
"""Should return relative path like /images/category/filename."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
images_dir = Path(tmpdir) / "images"
|
|
images_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
image_bytes = b"test"
|
|
with patch("backend.services.image_storage.IMAGES_ROOT", images_dir):
|
|
result = save_image(image_bytes, "networking", "test-item", "original")
|
|
|
|
assert result.startswith("/images/networking/")
|
|
assert result.endswith("_original.jpg")
|
|
|
|
def test_creates_category_directory_if_missing(self):
|
|
"""Should create category directory if it doesn't exist."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
images_dir = Path(tmpdir) / "images"
|
|
images_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
image_bytes = b"test"
|
|
category = "new_category"
|
|
|
|
with patch("backend.services.image_storage.IMAGES_ROOT", images_dir):
|
|
result = save_image(image_bytes, category, "test", "original")
|
|
|
|
cat_dir = images_dir / category
|
|
assert cat_dir.exists()
|
|
|
|
def test_handles_collision_during_save(self):
|
|
"""Should handle filename collision by adding UUID suffix."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
images_dir = Path(tmpdir) / "images"
|
|
images_dir.mkdir(parents=True, exist_ok=True)
|
|
cat_dir = images_dir / "networking"
|
|
cat_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
# Create first file
|
|
(cat_dir / "sfp-lr_original.jpg").write_bytes(b"first")
|
|
|
|
# Try to save with same name
|
|
image_bytes = b"second"
|
|
with patch("backend.services.image_storage.IMAGES_ROOT", images_dir):
|
|
result = save_image(image_bytes, "networking", "SFP-LR", "original")
|
|
|
|
# Should have created a different file
|
|
assert result != "/images/networking/sfp-lr_original.jpg"
|
|
# New file should exist with UUID suffix
|
|
assert "_original.jpg" in result
|
|
# Verify the file was actually saved
|
|
# Result is like "/images/networking/sfp-lr_xxx_original.jpg"
|
|
# So relative path from tmpdir is just "images/networking/..."
|
|
result_file = Path(tmpdir) / result.lstrip("/")
|
|
assert result_file.exists()
|
|
assert result_file.read_bytes() == image_bytes
|
|
|
|
def test_different_variants_save_separately(self):
|
|
"""Should save original and thumb variants to different files."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
images_dir = Path(tmpdir) / "images"
|
|
images_dir.mkdir(parents=True, exist_ok=True)
|
|
cat_dir = images_dir / "networking"
|
|
cat_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
original_bytes = b"original image data"
|
|
thumb_bytes = b"thumbnail data"
|
|
|
|
with patch("backend.services.image_storage.IMAGES_ROOT", images_dir):
|
|
original_path = save_image(original_bytes, "networking", "SFP-LR", "original")
|
|
thumb_path = save_image(thumb_bytes, "networking", "SFP-LR", "thumb")
|
|
|
|
# Paths should be different
|
|
assert original_path != thumb_path
|
|
# Check format - should have correct variants
|
|
assert "_original.jpg" in original_path
|
|
assert "_thumb.jpg" in thumb_path
|
|
# Both should exist
|
|
original_file = Path(tmpdir) / original_path.lstrip("/")
|
|
thumb_file = Path(tmpdir) / thumb_path.lstrip("/")
|
|
assert original_file.exists()
|
|
assert thumb_file.exists()
|
|
# Content should be different
|
|
assert original_file.read_bytes() == original_bytes
|
|
assert thumb_file.read_bytes() == thumb_bytes
|