alexei.dolgolyov
d0783d0b6a
Some checks failed
Validate / Hassfest (push) Has been cancelled
Extract HA-independent logic from the integration into packages/core/ as a standalone Python library (immich-watcher-core). This is the first phase of restructuring the project to support a standalone web app alongside the existing HAOS integration. Core library modules: - models: SharedLinkInfo, AssetInfo, AlbumData, AlbumChange dataclasses - immich_client: Async Immich API client (aiohttp, session-injected) - change_detector: Pure function for album change detection - asset_utils: Filtering, sorting, URL building utilities - telegram/client: Full Telegram Bot API (text, photo, video, media groups) - telegram/cache: File ID cache with pluggable storage backend - telegram/media: Media size checks, URL extraction, group splitting - notifications/queue: Persistent notification queue - storage: StorageBackend protocol + JSON file implementation All modules have zero Home Assistant imports. 50 unit tests passing. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
113 lines
3.7 KiB
Python
113 lines
3.7 KiB
Python
"""Tests for Telegram file cache."""
|
|
|
|
import pytest
|
|
from datetime import datetime, timezone, timedelta
|
|
from typing import Any
|
|
|
|
from immich_watcher_core.storage import StorageBackend
|
|
from immich_watcher_core.telegram.cache import TelegramFileCache
|
|
|
|
|
|
class InMemoryBackend:
|
|
"""In-memory storage backend for testing."""
|
|
|
|
def __init__(self, initial_data: dict[str, Any] | None = None):
|
|
self._data = initial_data
|
|
|
|
async def load(self) -> dict[str, Any] | None:
|
|
return self._data
|
|
|
|
async def save(self, data: dict[str, Any]) -> None:
|
|
self._data = data
|
|
|
|
async def remove(self) -> None:
|
|
self._data = None
|
|
|
|
|
|
@pytest.fixture
|
|
def backend():
|
|
return InMemoryBackend()
|
|
|
|
|
|
class TestTelegramFileCacheTTL:
|
|
@pytest.mark.asyncio
|
|
async def test_set_and_get(self, backend):
|
|
cache = TelegramFileCache(backend, ttl_seconds=3600)
|
|
await cache.async_load()
|
|
await cache.async_set("url1", "file_id_1", "photo")
|
|
result = cache.get("url1")
|
|
assert result is not None
|
|
assert result["file_id"] == "file_id_1"
|
|
assert result["type"] == "photo"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_miss(self, backend):
|
|
cache = TelegramFileCache(backend, ttl_seconds=3600)
|
|
await cache.async_load()
|
|
assert cache.get("nonexistent") is None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ttl_expiry(self):
|
|
# Pre-populate with an old entry
|
|
old_time = (datetime.now(timezone.utc) - timedelta(hours=100)).isoformat()
|
|
data = {"files": {"url1": {"file_id": "old", "type": "photo", "cached_at": old_time}}}
|
|
backend = InMemoryBackend(data)
|
|
cache = TelegramFileCache(backend, ttl_seconds=3600)
|
|
await cache.async_load()
|
|
# Old entry should be cleaned up on load
|
|
assert cache.get("url1") is None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_set_many(self, backend):
|
|
cache = TelegramFileCache(backend, ttl_seconds=3600)
|
|
await cache.async_load()
|
|
entries = [
|
|
("url1", "fid1", "photo", None),
|
|
("url2", "fid2", "video", None),
|
|
]
|
|
await cache.async_set_many(entries)
|
|
assert cache.get("url1")["file_id"] == "fid1"
|
|
assert cache.get("url2")["file_id"] == "fid2"
|
|
|
|
|
|
class TestTelegramFileCacheThumbhash:
|
|
@pytest.mark.asyncio
|
|
async def test_thumbhash_validation(self, backend):
|
|
cache = TelegramFileCache(backend, use_thumbhash=True)
|
|
await cache.async_load()
|
|
await cache.async_set("asset-1", "fid1", "photo", thumbhash="hash_v1")
|
|
|
|
# Match
|
|
result = cache.get("asset-1", thumbhash="hash_v1")
|
|
assert result is not None
|
|
assert result["file_id"] == "fid1"
|
|
|
|
# Mismatch - cache miss
|
|
result = cache.get("asset-1", thumbhash="hash_v2")
|
|
assert result is None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_thumbhash_max_entries(self):
|
|
# Create cache with many entries
|
|
files = {}
|
|
for i in range(2100):
|
|
files[f"asset-{i}"] = {
|
|
"file_id": f"fid-{i}",
|
|
"type": "photo",
|
|
"cached_at": datetime(2024, 1, 1 + i // 1440, (i // 60) % 24, i % 60, tzinfo=timezone.utc).isoformat(),
|
|
}
|
|
backend = InMemoryBackend({"files": files})
|
|
cache = TelegramFileCache(backend, use_thumbhash=True)
|
|
await cache.async_load()
|
|
# Should be trimmed to 2000
|
|
remaining = backend._data["files"]
|
|
assert len(remaining) == 2000
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_remove(self, backend):
|
|
cache = TelegramFileCache(backend, ttl_seconds=3600)
|
|
await cache.async_load()
|
|
await cache.async_set("url1", "fid1", "photo")
|
|
await cache.async_remove()
|
|
assert backend._data is None
|