296 lines
11 KiB
Python
296 lines
11 KiB
Python
"""Cache path migration tests."""
|
|
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from py.utils.cache_paths import (
|
|
CacheType,
|
|
migrate_library_cache_to_default,
|
|
resolve_cache_path_with_migration,
|
|
)
|
|
|
|
|
|
class TestResolveCachePathWithMigration:
|
|
"""Tests for resolve_cache_path_with_migration function."""
|
|
|
|
def test_returns_env_override_when_set(self, tmp_path, monkeypatch):
|
|
"""Test that env override takes precedence."""
|
|
settings_dir = tmp_path / "settings"
|
|
settings_dir.mkdir()
|
|
|
|
def fake_get_settings_dir(create: bool = True) -> str:
|
|
return str(settings_dir)
|
|
|
|
monkeypatch.setattr("py.utils.cache_paths.get_settings_dir", fake_get_settings_dir)
|
|
|
|
override_path = "/custom/path/cache.sqlite"
|
|
path = resolve_cache_path_with_migration(
|
|
CacheType.MODEL,
|
|
library_name="default",
|
|
env_override=override_path,
|
|
)
|
|
assert path == override_path
|
|
|
|
def test_returns_canonical_path_when_exists(self, tmp_path, monkeypatch):
|
|
"""Test that canonical path is returned when it exists."""
|
|
settings_dir = tmp_path / "settings"
|
|
settings_dir.mkdir()
|
|
|
|
def fake_get_settings_dir(create: bool = True) -> str:
|
|
return str(settings_dir)
|
|
|
|
monkeypatch.setattr("py.utils.cache_paths.get_settings_dir", fake_get_settings_dir)
|
|
|
|
# Create the canonical path
|
|
canonical = settings_dir / "cache" / "model" / "default.sqlite"
|
|
canonical.parent.mkdir(parents=True)
|
|
canonical.write_text("existing")
|
|
|
|
path = resolve_cache_path_with_migration(CacheType.MODEL, "default")
|
|
assert path == str(canonical)
|
|
|
|
def test_migrates_from_legacy_root_level_cache(self, tmp_path, monkeypatch):
|
|
"""Test migration from root-level legacy cache."""
|
|
settings_dir = tmp_path / "settings"
|
|
settings_dir.mkdir()
|
|
|
|
def fake_get_settings_dir(create: bool = True) -> str:
|
|
return str(settings_dir)
|
|
|
|
monkeypatch.setattr("py.utils.cache_paths.get_settings_dir", fake_get_settings_dir)
|
|
|
|
# Create legacy cache at root level
|
|
legacy_path = settings_dir / "model_cache.sqlite"
|
|
legacy_path.write_text("legacy data")
|
|
|
|
path = resolve_cache_path_with_migration(CacheType.MODEL, "default")
|
|
|
|
# Should return canonical path
|
|
canonical = settings_dir / "cache" / "model" / "default.sqlite"
|
|
assert path == str(canonical)
|
|
|
|
# File should be copied to canonical location
|
|
assert canonical.exists()
|
|
assert canonical.read_text() == "legacy data"
|
|
|
|
# Legacy file should be automatically cleaned up
|
|
assert not legacy_path.exists()
|
|
|
|
def test_migrates_from_legacy_per_library_cache(self, tmp_path, monkeypatch):
|
|
"""Test migration from per-library legacy cache."""
|
|
settings_dir = tmp_path / "settings"
|
|
settings_dir.mkdir()
|
|
|
|
def fake_get_settings_dir(create: bool = True) -> str:
|
|
return str(settings_dir)
|
|
|
|
monkeypatch.setattr("py.utils.cache_paths.get_settings_dir", fake_get_settings_dir)
|
|
|
|
# Create legacy per-library cache
|
|
legacy_dir = settings_dir / "model_cache"
|
|
legacy_dir.mkdir()
|
|
legacy_path = legacy_dir / "my_library.sqlite"
|
|
legacy_path.write_text("legacy library data")
|
|
|
|
path = resolve_cache_path_with_migration(CacheType.MODEL, "my_library")
|
|
|
|
# Should return canonical path
|
|
canonical = settings_dir / "cache" / "model" / "my_library.sqlite"
|
|
assert path == str(canonical)
|
|
assert canonical.exists()
|
|
assert canonical.read_text() == "legacy library data"
|
|
|
|
# Legacy file should be automatically cleaned up
|
|
assert not legacy_path.exists()
|
|
|
|
# Empty legacy directory should be cleaned up
|
|
assert not legacy_dir.exists()
|
|
|
|
def test_prefers_per_library_over_root_for_migration(self, tmp_path, monkeypatch):
|
|
"""Test that per-library cache is preferred over root for migration."""
|
|
settings_dir = tmp_path / "settings"
|
|
settings_dir.mkdir()
|
|
|
|
def fake_get_settings_dir(create: bool = True) -> str:
|
|
return str(settings_dir)
|
|
|
|
monkeypatch.setattr("py.utils.cache_paths.get_settings_dir", fake_get_settings_dir)
|
|
|
|
# Create both legacy caches
|
|
legacy_root = settings_dir / "model_cache.sqlite"
|
|
legacy_root.write_text("root legacy")
|
|
|
|
legacy_dir = settings_dir / "model_cache"
|
|
legacy_dir.mkdir()
|
|
legacy_lib = legacy_dir / "default.sqlite"
|
|
legacy_lib.write_text("library legacy")
|
|
|
|
path = resolve_cache_path_with_migration(CacheType.MODEL, "default")
|
|
|
|
canonical = settings_dir / "cache" / "model" / "default.sqlite"
|
|
assert path == str(canonical)
|
|
# Should migrate from per-library path (first in legacy list)
|
|
assert canonical.read_text() == "library legacy"
|
|
|
|
def test_returns_canonical_path_when_no_legacy_exists(self, tmp_path, monkeypatch):
|
|
"""Test that canonical path is returned when no legacy exists."""
|
|
settings_dir = tmp_path / "settings"
|
|
settings_dir.mkdir()
|
|
|
|
def fake_get_settings_dir(create: bool = True) -> str:
|
|
return str(settings_dir)
|
|
|
|
monkeypatch.setattr("py.utils.cache_paths.get_settings_dir", fake_get_settings_dir)
|
|
|
|
path = resolve_cache_path_with_migration(CacheType.MODEL, "new_library")
|
|
|
|
canonical = settings_dir / "cache" / "model" / "new_library.sqlite"
|
|
assert path == str(canonical)
|
|
# Directory should be created
|
|
assert canonical.parent.exists()
|
|
# But file should not exist yet
|
|
assert not canonical.exists()
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("cache_type", "subdir"),
|
|
[
|
|
(CacheType.MODEL, "model"),
|
|
(CacheType.MODEL_UPDATE, "model_update"),
|
|
(CacheType.RECIPE, "recipe"),
|
|
],
|
|
)
|
|
def test_named_cache_migrates_to_default(
|
|
tmp_path, monkeypatch, cache_type, subdir
|
|
):
|
|
settings_dir = tmp_path / "settings"
|
|
source = settings_dir / "cache" / subdir / "comfyui.sqlite"
|
|
source.parent.mkdir(parents=True)
|
|
source.write_text("cached data")
|
|
monkeypatch.setattr(
|
|
"py.utils.cache_paths.get_settings_dir",
|
|
lambda create=True: str(settings_dir),
|
|
)
|
|
|
|
migrate_library_cache_to_default(cache_type, "comfyui")
|
|
|
|
target = settings_dir / "cache" / subdir / "default.sqlite"
|
|
assert target.read_text() == "cached data"
|
|
assert not source.exists()
|
|
|
|
|
|
def test_named_cache_migration_does_not_overwrite_default(tmp_path, monkeypatch):
|
|
settings_dir = tmp_path / "settings"
|
|
model_dir = settings_dir / "cache" / "model"
|
|
model_dir.mkdir(parents=True)
|
|
source = model_dir / "comfyui.sqlite"
|
|
target = model_dir / "default.sqlite"
|
|
source.write_text("named")
|
|
target.write_text("default")
|
|
monkeypatch.setattr(
|
|
"py.utils.cache_paths.get_settings_dir",
|
|
lambda create=True: str(settings_dir),
|
|
)
|
|
|
|
migrate_library_cache_to_default(CacheType.MODEL, "comfyui")
|
|
|
|
assert target.read_text() == "default"
|
|
assert source.read_text() == "named"
|
|
|
|
|
|
class TestAutomaticCleanup:
|
|
"""Tests for automatic cleanup during migration."""
|
|
|
|
def test_automatic_cleanup_on_migration(self, tmp_path, monkeypatch):
|
|
"""Test that legacy files are automatically cleaned up after migration."""
|
|
settings_dir = tmp_path / "settings"
|
|
settings_dir.mkdir()
|
|
|
|
def fake_get_settings_dir(create: bool = True) -> str:
|
|
return str(settings_dir)
|
|
|
|
monkeypatch.setattr("py.utils.cache_paths.get_settings_dir", fake_get_settings_dir)
|
|
|
|
# Create a legacy cache file
|
|
legacy_dir = settings_dir / "model_cache"
|
|
legacy_dir.mkdir()
|
|
legacy_file = legacy_dir / "default.sqlite"
|
|
legacy_file.write_text("test data")
|
|
|
|
# Verify legacy file exists
|
|
assert legacy_file.exists()
|
|
|
|
# Trigger migration (this should auto-cleanup)
|
|
resolved_path = resolve_cache_path_with_migration(CacheType.MODEL, "default")
|
|
|
|
# Verify canonical file exists
|
|
canonical_path = settings_dir / "cache" / "model" / "default.sqlite"
|
|
assert resolved_path == str(canonical_path)
|
|
assert canonical_path.exists()
|
|
assert canonical_path.read_text() == "test data"
|
|
|
|
# Verify legacy file was cleaned up
|
|
assert not legacy_file.exists()
|
|
|
|
# Verify empty directory was cleaned up
|
|
assert not legacy_dir.exists()
|
|
|
|
def test_automatic_cleanup_with_verification(self, tmp_path, monkeypatch):
|
|
"""Test that cleanup verifies file integrity before deletion."""
|
|
settings_dir = tmp_path / "settings"
|
|
settings_dir.mkdir()
|
|
|
|
def fake_get_settings_dir(create: bool = True) -> str:
|
|
return str(settings_dir)
|
|
|
|
monkeypatch.setattr("py.utils.cache_paths.get_settings_dir", fake_get_settings_dir)
|
|
|
|
# Create legacy cache
|
|
legacy_dir = settings_dir / "recipe_cache"
|
|
legacy_dir.mkdir()
|
|
legacy_file = legacy_dir / "my_library.sqlite"
|
|
legacy_file.write_text("data")
|
|
|
|
# Trigger migration
|
|
resolved_path = resolve_cache_path_with_migration(CacheType.RECIPE, "my_library")
|
|
canonical_path = settings_dir / "cache" / "recipe" / "my_library.sqlite"
|
|
|
|
# Both should exist initially (migration successful)
|
|
assert canonical_path.exists()
|
|
assert legacy_file.exists() is False # Auto-cleanup removes it
|
|
|
|
# File content should match (integrity check)
|
|
assert canonical_path.read_text() == "data"
|
|
|
|
# Directory should be cleaned up
|
|
assert not legacy_dir.exists()
|
|
|
|
def test_automatic_cleanup_multiple_cache_types(self, tmp_path, monkeypatch):
|
|
"""Test automatic cleanup for different cache types."""
|
|
settings_dir = tmp_path / "settings"
|
|
settings_dir.mkdir()
|
|
|
|
def fake_get_settings_dir(create: bool = True) -> str:
|
|
return str(settings_dir)
|
|
|
|
monkeypatch.setattr("py.utils.cache_paths.get_settings_dir", fake_get_settings_dir)
|
|
|
|
# Test RECIPE_FTS migration
|
|
legacy_fts = settings_dir / "recipe_fts.sqlite"
|
|
legacy_fts.write_text("fts data")
|
|
resolve_cache_path_with_migration(CacheType.RECIPE_FTS)
|
|
canonical_fts = settings_dir / "cache" / "fts" / "recipe_fts.sqlite"
|
|
|
|
assert canonical_fts.exists()
|
|
assert not legacy_fts.exists()
|
|
|
|
# Test TAG_FTS migration
|
|
legacy_tag = settings_dir / "tag_fts.sqlite"
|
|
legacy_tag.write_text("tag data")
|
|
resolve_cache_path_with_migration(CacheType.TAG_FTS)
|
|
canonical_tag = settings_dir / "cache" / "fts" / "tag_fts.sqlite"
|
|
|
|
assert canonical_tag.exists()
|
|
assert not legacy_tag.exists()
|