haos-hacs-immich-album-watcher/packages/core/tests/test_asset_utils.py

"""Tests for asset filtering, sorting, and URL utilities."""

from immich_watcher_core.asset_utils import (
    build_asset_detail,
    combine_album_assets,
    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


class TestCombineAlbumAssets:
    def test_even_distribution(self):
        """Both albums have plenty, split evenly."""
        a = [_make_asset(f"a{i}") for i in range(10)]
        b = [_make_asset(f"b{i}") for i in range(10)]
        result = combine_album_assets({"A": a, "B": b}, total_limit=6, order_by="name")
        assert len(result) == 6

    def test_smart_redistribution(self):
        """Album A has 1 photo, Album B has 20. Limit=10 should get 10 total."""
        a = [_make_asset("a1", created_at="2023-03-19T10:00:00Z")]
        b = [_make_asset(f"b{i}", created_at=f"2023-03-19T{10+i}:00:00Z") for i in range(20)]
        result = combine_album_assets({"A": a, "B": b}, total_limit=10, order_by="name")
        assert len(result) == 10
        # a1 should be in result
        ids = {r.id for r in result}
        assert "a1" in ids

    def test_redistribution_with_3_albums(self):
        """3 albums: A has 1, B has 2, C has 20. Limit=12."""
        a = [_make_asset("a1")]
        b = [_make_asset("b1"), _make_asset("b2")]
        c = [_make_asset(f"c{i}") for i in range(20)]
        result = combine_album_assets({"A": a, "B": b, "C": c}, total_limit=12, order_by="name")
        assert len(result) == 12
        # All of A and B should be included
        ids = {r.id for r in result}
        assert "a1" in ids
        assert "b1" in ids
        assert "b2" in ids
        # C fills the remaining 9
        c_count = sum(1 for r in result if r.id.startswith("c"))
        assert c_count == 9

    def test_all_albums_empty(self):
        result = combine_album_assets({"A": [], "B": []}, total_limit=10)
        assert result == []

    def test_single_album(self):
        a = [_make_asset(f"a{i}") for i in range(5)]
        result = combine_album_assets({"A": a}, total_limit=3, order_by="name")
        assert len(result) == 3

    def test_total_less_than_limit(self):
        """Both albums together have fewer than limit."""
        a = [_make_asset("a1")]
        b = [_make_asset("b1"), _make_asset("b2")]
        result = combine_album_assets({"A": a, "B": b}, total_limit=10, order_by="name")
        assert len(result) == 3