76 lines
2.7 KiB
Python
76 lines
2.7 KiB
Python
import asyncio
|
|
import importlib
|
|
import io
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
from contextlib import asynccontextmanager
|
|
from types import SimpleNamespace
|
|
from unittest.mock import patch
|
|
|
|
from alembic.config import Config
|
|
from alembic.migration import MigrationContext
|
|
from alembic.operations import Operations
|
|
from alembic.script import ScriptDirectory
|
|
|
|
from app import gc
|
|
from app.blobs import blob_path, put_bytes
|
|
from app.config import get_settings
|
|
from app.models import Game
|
|
|
|
|
|
def test_gallery_migration_has_one_head_and_postgres_default_for_existing_games():
|
|
output = io.StringIO()
|
|
context = MigrationContext.configure(dialect_name="postgresql", opts={"as_sql": True, "output_buffer": output})
|
|
operations = Operations(context)
|
|
migration = importlib.import_module("migrations.versions.0006_game_gallery")
|
|
with patch("alembic.op.add_column", operations.add_column):
|
|
migration.upgrade()
|
|
sql = output.getvalue()
|
|
assert "gallery_hashes VARCHAR(80)[] DEFAULT '{}' NOT NULL" in sql
|
|
assert ScriptDirectory.from_config(Config("alembic.ini")).get_heads() == ["0006"]
|
|
|
|
|
|
def test_gc_retains_gallery_images_without_cover_or_version_references(tmp_path, monkeypatch):
|
|
monkeypatch.setenv("DATA_DIR", str(tmp_path))
|
|
get_settings.cache_clear()
|
|
live = "a" * 64 + ".webp"
|
|
orphan = "b" * 64 + ".webp"
|
|
for value in (live, orphan):
|
|
path = blob_path(value)
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_bytes(b"image")
|
|
|
|
async def scalars(statement):
|
|
expr = statement.column_descriptions[0]["expr"]
|
|
return SimpleNamespace(all=lambda: [[live]] if expr is Game.gallery_hashes else [])
|
|
|
|
@asynccontextmanager
|
|
async def session():
|
|
yield SimpleNamespace(scalars=scalars)
|
|
|
|
@asynccontextmanager
|
|
async def access(**_):
|
|
yield
|
|
|
|
monkeypatch.setattr(gc, "SessionLocal", session)
|
|
monkeypatch.setattr(gc, "storage_access", access)
|
|
try:
|
|
report = asyncio.run(gc.collect())
|
|
assert report["retainedBlobs"] == 1
|
|
assert report["files"] == 1
|
|
assert blob_path(live).exists()
|
|
finally:
|
|
get_settings.cache_clear()
|
|
|
|
|
|
def test_identical_parallel_gallery_images_share_a_blob(tmp_path, monkeypatch):
|
|
monkeypatch.setenv("DATA_DIR", str(tmp_path))
|
|
get_settings.cache_clear()
|
|
try:
|
|
with ThreadPoolExecutor(max_workers=8) as executor:
|
|
results = list(executor.map(lambda _: put_bytes(b"same image", ".webp"), range(32)))
|
|
assert len({value[0] for value in results}) == 1
|
|
assert blob_path(results[0][0]).read_bytes() == b"same image"
|
|
assert not list((tmp_path / "blobs/tmp").iterdir())
|
|
finally:
|
|
get_settings.cache_clear()
|