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>
186 lines
6.6 KiB
Python
186 lines
6.6 KiB
Python
"""Tests for asset filtering, sorting, and URL utilities."""
|
|
|
|
from immich_watcher_core.asset_utils import (
|
|
build_asset_detail,
|
|
filter_assets,
|
|
get_any_url,
|
|
get_public_url,
|
|
get_protected_url,
|
|
sort_assets,
|
|
)
|
|
from immich_watcher_core.models import AssetInfo, SharedLinkInfo
|
|
|
|
|
|
def _make_asset(
|
|
asset_id: str = "a1",
|
|
asset_type: str = "IMAGE",
|
|
filename: str = "photo.jpg",
|
|
created_at: str = "2024-01-15T10:30:00Z",
|
|
is_favorite: bool = False,
|
|
rating: int | None = None,
|
|
city: str | None = None,
|
|
country: str | None = None,
|
|
) -> AssetInfo:
|
|
return AssetInfo(
|
|
id=asset_id,
|
|
type=asset_type,
|
|
filename=filename,
|
|
created_at=created_at,
|
|
is_favorite=is_favorite,
|
|
rating=rating,
|
|
city=city,
|
|
country=country,
|
|
is_processed=True,
|
|
)
|
|
|
|
|
|
class TestFilterAssets:
|
|
def test_favorite_only(self):
|
|
assets = [_make_asset("a1", is_favorite=True), _make_asset("a2")]
|
|
result = filter_assets(assets, favorite_only=True)
|
|
assert len(result) == 1
|
|
assert result[0].id == "a1"
|
|
|
|
def test_min_rating(self):
|
|
assets = [
|
|
_make_asset("a1", rating=5),
|
|
_make_asset("a2", rating=2),
|
|
_make_asset("a3"), # no rating
|
|
]
|
|
result = filter_assets(assets, min_rating=3)
|
|
assert len(result) == 1
|
|
assert result[0].id == "a1"
|
|
|
|
def test_asset_type_photo(self):
|
|
assets = [
|
|
_make_asset("a1", asset_type="IMAGE"),
|
|
_make_asset("a2", asset_type="VIDEO"),
|
|
]
|
|
result = filter_assets(assets, asset_type="photo")
|
|
assert len(result) == 1
|
|
assert result[0].type == "IMAGE"
|
|
|
|
def test_date_range(self):
|
|
assets = [
|
|
_make_asset("a1", created_at="2024-01-10T00:00:00Z"),
|
|
_make_asset("a2", created_at="2024-01-15T00:00:00Z"),
|
|
_make_asset("a3", created_at="2024-01-20T00:00:00Z"),
|
|
]
|
|
result = filter_assets(
|
|
assets, min_date="2024-01-12T00:00:00Z", max_date="2024-01-18T00:00:00Z"
|
|
)
|
|
assert len(result) == 1
|
|
assert result[0].id == "a2"
|
|
|
|
def test_memory_date(self):
|
|
assets = [
|
|
_make_asset("a1", created_at="2023-03-19T10:00:00Z"), # same month/day, different year
|
|
_make_asset("a2", created_at="2024-03-19T10:00:00Z"), # same year as reference
|
|
_make_asset("a3", created_at="2023-06-15T10:00:00Z"), # different date
|
|
]
|
|
result = filter_assets(assets, memory_date="2024-03-19T00:00:00Z")
|
|
assert len(result) == 1
|
|
assert result[0].id == "a1"
|
|
|
|
def test_city_filter(self):
|
|
assets = [
|
|
_make_asset("a1", city="Paris"),
|
|
_make_asset("a2", city="London"),
|
|
_make_asset("a3"),
|
|
]
|
|
result = filter_assets(assets, city="paris")
|
|
assert len(result) == 1
|
|
assert result[0].id == "a1"
|
|
|
|
|
|
class TestSortAssets:
|
|
def test_sort_by_date_descending(self):
|
|
assets = [
|
|
_make_asset("a1", created_at="2024-01-10T00:00:00Z"),
|
|
_make_asset("a2", created_at="2024-01-20T00:00:00Z"),
|
|
_make_asset("a3", created_at="2024-01-15T00:00:00Z"),
|
|
]
|
|
result = sort_assets(assets, order_by="date", order="descending")
|
|
assert [a.id for a in result] == ["a2", "a3", "a1"]
|
|
|
|
def test_sort_by_name(self):
|
|
assets = [
|
|
_make_asset("a1", filename="charlie.jpg"),
|
|
_make_asset("a2", filename="alice.jpg"),
|
|
_make_asset("a3", filename="bob.jpg"),
|
|
]
|
|
result = sort_assets(assets, order_by="name", order="ascending")
|
|
assert [a.id for a in result] == ["a2", "a3", "a1"]
|
|
|
|
def test_sort_by_rating(self):
|
|
assets = [
|
|
_make_asset("a1", rating=3),
|
|
_make_asset("a2", rating=5),
|
|
_make_asset("a3"), # None rating
|
|
]
|
|
result = sort_assets(assets, order_by="rating", order="descending")
|
|
# With descending + (is_none, value) key: None goes last when reversed
|
|
# (True, 0) vs (False, 5) vs (False, 3) - reversed: (True, 0), (False, 5), (False, 3)
|
|
# Actually: reversed sort puts (True,0) first. Let's just check rated come before unrated
|
|
rated = [a for a in result if a.rating is not None]
|
|
assert rated[0].id == "a2"
|
|
assert rated[1].id == "a1"
|
|
|
|
|
|
class TestUrlHelpers:
|
|
def _make_links(self):
|
|
return [
|
|
SharedLinkInfo(id="l1", key="public-key"),
|
|
SharedLinkInfo(id="l2", key="protected-key", has_password=True, password="pass123"),
|
|
]
|
|
|
|
def test_get_public_url(self):
|
|
links = self._make_links()
|
|
url = get_public_url("https://immich.example.com", links)
|
|
assert url == "https://immich.example.com/share/public-key"
|
|
|
|
def test_get_protected_url(self):
|
|
links = self._make_links()
|
|
url = get_protected_url("https://immich.example.com", links)
|
|
assert url == "https://immich.example.com/share/protected-key"
|
|
|
|
def test_get_any_url_prefers_public(self):
|
|
links = self._make_links()
|
|
url = get_any_url("https://immich.example.com", links)
|
|
assert url == "https://immich.example.com/share/public-key"
|
|
|
|
def test_get_any_url_falls_back_to_protected(self):
|
|
links = [SharedLinkInfo(id="l1", key="prot-key", has_password=True, password="x")]
|
|
url = get_any_url("https://immich.example.com", links)
|
|
assert url == "https://immich.example.com/share/prot-key"
|
|
|
|
def test_no_links(self):
|
|
assert get_public_url("https://example.com", []) is None
|
|
assert get_any_url("https://example.com", []) is None
|
|
|
|
|
|
class TestBuildAssetDetail:
|
|
def test_build_image_detail(self):
|
|
asset = _make_asset("a1", asset_type="IMAGE")
|
|
links = [SharedLinkInfo(id="l1", key="key1")]
|
|
detail = build_asset_detail(asset, "https://immich.example.com", links)
|
|
assert detail["id"] == "a1"
|
|
assert "url" in detail
|
|
assert "download_url" in detail
|
|
assert "photo_url" in detail
|
|
assert "thumbnail_url" in detail
|
|
|
|
def test_build_video_detail(self):
|
|
asset = _make_asset("a1", asset_type="VIDEO")
|
|
links = [SharedLinkInfo(id="l1", key="key1")]
|
|
detail = build_asset_detail(asset, "https://immich.example.com", links)
|
|
assert "playback_url" in detail
|
|
assert "photo_url" not in detail
|
|
|
|
def test_no_shared_links(self):
|
|
asset = _make_asset("a1")
|
|
detail = build_asset_detail(asset, "https://immich.example.com", [])
|
|
assert "url" not in detail
|
|
assert "download_url" not in detail
|
|
assert "thumbnail_url" in detail # always present
|