agatha
61d923d5be
Short IDs become the canonical identifier in URLs (/i/:short_id), MinIO/R2 storage keys, and all API responses. Hash-based deduplication is preserved. Includes two-phase Alembic migration (003 adds nullable column, 004 enforces NOT NULL) with a backfill script to copy storage objects and populate short_id for existing images. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
111 lines
3.5 KiB
Python
111 lines
3.5 KiB
Python
"""Unit tests for migrate_to_short_ids script logic."""
|
|
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_image_null_short_id():
|
|
img = MagicMock()
|
|
img.id = "img-uuid-1"
|
|
img.short_id = None
|
|
img.storage_key = "oldhashkey1234567890"
|
|
img.thumbnail_key = "oldhashkey1234567890-thumb"
|
|
img.mime_type = "image/jpeg"
|
|
return img
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_image_with_short_id():
|
|
img = MagicMock()
|
|
img.id = "img-uuid-2"
|
|
img.short_id = "AbCd1234"
|
|
img.storage_key = "AbCd1234"
|
|
img.thumbnail_key = "AbCd1234-thumb"
|
|
img.mime_type = "image/jpeg"
|
|
return img
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_migrate_processes_image_without_short_id(mock_image_null_short_id):
|
|
"""Images with short_id IS NULL are processed: storage copied, DB updated, old keys deleted."""
|
|
from scripts.migrate_to_short_ids import migrate_image
|
|
|
|
storage = MagicMock()
|
|
storage.get = AsyncMock(return_value=b"imagedata")
|
|
storage.put = AsyncMock()
|
|
storage.delete = AsyncMock()
|
|
|
|
session = MagicMock()
|
|
session.execute = AsyncMock()
|
|
session.flush = AsyncMock()
|
|
|
|
old_key = mock_image_null_short_id.storage_key
|
|
new_short_id = "NewSh123"
|
|
with patch("scripts.migrate_to_short_ids.generate_short_id", return_value=new_short_id):
|
|
result = await migrate_image(mock_image_null_short_id, storage, session)
|
|
|
|
assert result is True
|
|
storage.put.assert_any_call(new_short_id, b"imagedata", "image/jpeg")
|
|
storage.delete.assert_any_call(old_key)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_migrate_skips_image_with_short_id(mock_image_with_short_id):
|
|
"""Images that already have a short_id are skipped."""
|
|
from scripts.migrate_to_short_ids import migrate_image
|
|
|
|
storage = MagicMock()
|
|
session = MagicMock()
|
|
|
|
result = await migrate_image(mock_image_with_short_id, storage, session)
|
|
|
|
assert result is False
|
|
storage.get.assert_not_called() if hasattr(storage.get, "assert_not_called") else None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_migrate_continues_on_storage_error(mock_image_null_short_id):
|
|
"""If storage copy fails, error is logged and migrate_image returns False without aborting."""
|
|
from scripts.migrate_to_short_ids import migrate_image
|
|
|
|
storage = MagicMock()
|
|
storage.get = AsyncMock(side_effect=Exception("storage read error"))
|
|
storage.put = AsyncMock()
|
|
storage.delete = AsyncMock()
|
|
|
|
session = MagicMock()
|
|
session.execute = AsyncMock()
|
|
session.flush = AsyncMock()
|
|
|
|
with patch("scripts.migrate_to_short_ids.generate_short_id", return_value="ErrSh123"):
|
|
result = await migrate_image(mock_image_null_short_id, storage, session)
|
|
|
|
assert result is False
|
|
storage.put.assert_not_called()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_migrate_summary_counts(mock_image_null_short_id, mock_image_with_short_id):
|
|
"""run_migration reports correct migrated and skipped counts."""
|
|
from scripts.migrate_to_short_ids import run_migration
|
|
|
|
storage = MagicMock()
|
|
storage.get = AsyncMock(return_value=b"data")
|
|
storage.put = AsyncMock()
|
|
storage.delete = AsyncMock()
|
|
|
|
session = MagicMock()
|
|
session.execute = AsyncMock()
|
|
session.flush = AsyncMock()
|
|
|
|
images = [mock_image_null_short_id, mock_image_with_short_id]
|
|
|
|
with patch("scripts.migrate_to_short_ids.generate_short_id", return_value="NewSh999"):
|
|
migrated, skipped, failed = await run_migration(images, storage, session)
|
|
|
|
assert migrated == 1
|
|
assert skipped == 1
|
|
assert failed == 0
|