manage/backend/tests/test_media_index.py

"""Unit tests for the SQLite media index service."""

from typing import Any

import pytest

from media_library_viewer_api.services.media_index import (
    MediaIndex,
    MediaIndexBuildCancelled,
    build_media_index,
)


@pytest.fixture
def index(tmp_path):
    """Create a temporary media index for testing."""
    db_path = tmp_path / "test_index.sqlite"
    return MediaIndex(db_path)


@pytest.fixture
def populated_index(index):
    """Index with sample rows inserted."""
    rows = [
        {
            "id": "m1",
            "title": "The Matrix",
            "series": "",
            "season": "",
            "season_number": None,
            "episode": None,
            "type": "Movie",
            "year": 1999,
            "runtime_ticks": 81600000000,
            "runtime_min": 136,
            "size_bytes": 15_000_000_000,
            "bitrate_bps": 15_000_000,
            "hdr": 0,
            "video": "h264",
            "width": 1920,
            "height": 1080,
            "resolution": "1920x1080",
            "date_added": "2024-01-15",
            "date_added_ts": 1705276800,
            "path": "/media/movies/The Matrix (1999)/file.mkv",
            "library_id": "lib1",
            "library_name": "Movies",
            "size": "15.0 GB",
            "bitrate": "15.0 Mbps",
        },
        {
            "id": "m2",
            "title": "Dune Part Two",
            "series": "",
            "season": "",
            "season_number": None,
            "episode": None,
            "type": "Movie",
            "year": 2024,
            "runtime_ticks": 99600000000,
            "runtime_min": 166,
            "size_bytes": 45_000_000_000,
            "bitrate_bps": 40_000_000,
            "hdr": 1,
            "video": "hevc",
            "width": 3840,
            "height": 2160,
            "resolution": "3840x2160",
            "date_added": "2024-06-01",
            "date_added_ts": 1717200000,
            "path": "/media/movies/Dune Part Two (2024)/file.mkv",
            "library_id": "lib1",
            "library_name": "Movies",
            "size": "45.0 GB",
            "bitrate": "40.0 Mbps",
        },
        {
            "id": "e1",
            "title": "Pilot",
            "series": "Breaking Bad",
            "season": "S01",
            "season_number": 1,
            "episode": 1,
            "type": "Episode",
            "year": 2008,
            "runtime_ticks": 35000000000,
            "runtime_min": 58,
            "size_bytes": 3_000_000_000,
            "bitrate_bps": 8_000_000,
            "hdr": 0,
            "video": "h264",
            "width": 1920,
            "height": 1080,
            "resolution": "1920x1080",
            "date_added": "2024-03-01",
            "date_added_ts": 1709251200,
            "path": "/media/shows/Breaking Bad/S01E01.mkv",
            "library_id": "lib2",
            "library_name": "TV Shows",
            "size": "3.0 GB",
            "bitrate": "8.0 Mbps",
        },
    ]
    index.replace_items(rows)
    return index


class TestMediaIndexStatus:
    def test_nonexistent(self, tmp_path):
        idx = MediaIndex(tmp_path / "nonexistent.sqlite")
        status = idx.status()
        assert status.exists is False
        assert status.item_count == 0

    def test_empty_index(self, index):
        index.init_schema()
        status = index.status()
        assert status.exists is True
        assert status.item_count == 0

    def test_populated(self, populated_index):
        status = populated_index.status()
        assert status.exists is True
        assert status.item_count == 3


class TestMediaIndexQuery:
    def test_query_all(self, populated_index):
        rows, total = populated_index.query(
            library_ids=["lib1", "lib2"],
            media_types=["Movie", "Episode"],
        )
        assert total == 3
        assert len(rows) == 3

    def test_filter_by_library(self, populated_index):
        rows, total = populated_index.query(
            library_ids=["lib1"],
            media_types=["Movie", "Episode"],
        )
        assert total == 2
        # Only movies from lib1
        assert all(r["library"] == "Movies" for r in rows)

    def test_filter_by_type(self, populated_index):
        rows, total = populated_index.query(
            library_ids=["lib1", "lib2"],
            media_types=["Episode"],
        )
        assert total == 1
        assert rows[0]["title"] == "Pilot"

    def test_search(self, populated_index):
        rows, total = populated_index.query(
            library_ids=["lib1", "lib2"],
            media_types=["Movie", "Episode"],
            search="matrix",
        )
        assert total == 1
        assert rows[0]["title"] == "The Matrix"

    def test_hdr_filter(self, populated_index):
        rows, total = populated_index.query(
            library_ids=["lib1", "lib2"],
            media_types=["Movie", "Episode"],
            hdr_filter="HDR only",
        )
        assert total == 1
        assert rows[0]["title"] == "Dune Part Two"

    def test_sdr_filter(self, populated_index):
        rows, total = populated_index.query(
            library_ids=["lib1", "lib2"],
            media_types=["Movie", "Episode"],
            hdr_filter="SDR/unknown only",
        )
        assert total == 2

    def test_sort_by_size_desc(self, populated_index):
        rows, total = populated_index.query(
            library_ids=["lib1", "lib2"],
            media_types=["Movie", "Episode"],
            sort_key="size",
            sort_order="Descending",
        )
        assert rows[0]["title"] == "Dune Part Two"  # largest

    def test_sort_by_year_asc(self, populated_index):
        rows, total = populated_index.query(
            library_ids=["lib1", "lib2"],
            media_types=["Movie", "Episode"],
            sort_key="year",
            sort_order="Ascending",
        )
        assert rows[0]["title"] == "The Matrix"  # 1999

    def test_pagination(self, populated_index):
        rows, total = populated_index.query(
            library_ids=["lib1", "lib2"],
            media_types=["Movie", "Episode"],
            limit=2,
            offset=0,
        )
        assert total == 3
        assert len(rows) == 2

        rows2, total2 = populated_index.query(
            library_ids=["lib1", "lib2"],
            media_types=["Movie", "Episode"],
            limit=2,
            offset=2,
        )
        assert total2 == 3
        assert len(rows2) == 1

    def test_display_row_format(self, populated_index):
        rows, _ = populated_index.query(
            library_ids=["lib1"],
            media_types=["Movie"],
        )
        for row in rows:
            assert "title" in row
            assert "hdr" in row
            assert row["hdr"] in ("yes", "no")
            assert "library" in row
            assert "path" in row


class TestMediaIndexReplace:
    def test_replace_clears_old(self, index):
        index.replace_items([
            {"id": "x", "title": "Old", "type": "Movie", "library_id": "l1", "library_name": "L1"},
        ])
        status = index.status()
        assert status.item_count == 1

        index.replace_items([
            {"id": "y", "title": "New1", "type": "Movie", "library_id": "l1", "library_name": "L1"},
            {"id": "z", "title": "New2", "type": "Movie", "library_id": "l1", "library_name": "L1"},
        ])
        status = index.status()
        assert status.item_count == 2


class TestMediaIndexMetadata:
    def test_set_and_read_metadata(self, index):
        index.set_metadata("build_duration_seconds", "12.5")
        status = index.status()
        assert status.build_duration_seconds == 12.5


class TestMediaIndexBuildPaths:
    def test_build_media_index_patches_remote_media_root(self, tmp_path):
        class FakeClient:
            def items(self, **kwargs):
                return {
                    "Items": [
                        {
                            "Id": "m1",
                            "Name": "Movie One",
                            "Type": "Movie",
                            "Path": "/media/movies/Movie One/file.mkv",
                        }
                    ],
                    "TotalRecordCount": 1,
                }

        index = MediaIndex(tmp_path / "index.sqlite")
        count = build_media_index(
            FakeClient(),
            "user1",
            [{"Id": "lib1", "Name": "Movies"}],
            index,
            media_root="/srv/media",
            fallback_prefix="",
        )
        assert count == 1
        rows, total = index.query(library_ids=["lib1"], media_types=["Movie"])
        assert total == 1
        assert rows[0]["path"] == "/srv/media/movies/Movie One/file.mkv"

    def test_build_media_index_reports_progress(self, tmp_path):
        class FakeClient:
            def __init__(self):
                self.calls = []

            def items(self, **kwargs):
                self.calls.append(kwargs.get("start_index", 0))
                if kwargs.get("start_index", 0) == 0:
                    return {
                        "Items": [
                            {"Id": "m1", "Name": "Movie One", "Type": "Movie", "Path": "/media/a.mkv"}
                        ],
                        "TotalRecordCount": 2,
                    }
                return {
                    "Items": [
                        {"Id": "m2", "Name": "Movie Two", "Type": "Movie", "Path": "/media/b.mkv"}
                    ],
                    "TotalRecordCount": 2,
                }

        events: list[dict[str, Any]] = []
        index = MediaIndex(tmp_path / "index.sqlite")
        count = build_media_index(
            FakeClient(),
            "user1",
            [{"Id": "lib1", "Name": "Movies"}],
            index,
            page_size=1,
            progress_callback=events.append,
        )
        assert count == 2
        assert events[0]["stage"] == "starting"
        assert events[0]["progress"] is None
        assert any(event["stage"] == "building" for event in events)
        building_event = next(event for event in events if event["stage"] == "building")
        assert building_event["library"] == "Movies"
        assert building_event["library_progress"] in (0.5, 1.0)
        assert "elapsed_seconds" in building_event
        assert "eta_seconds" in building_event
        assert events[-1]["stage"] == "completed"
        assert events[-1]["progress"] == 1.0
        assert events[-1]["library_progress"] == 1.0

    def test_build_media_index_can_be_cancelled(self, tmp_path):
        class FakeClient:
            def __init__(self):
                self.calls = []

            def items(self, **kwargs):
                self.calls.append(kwargs.get("start_index", 0))
                if kwargs.get("start_index", 0) == 0:
                    return {
                        "Items": [
                            {"Id": "m1", "Name": "Movie One", "Type": "Movie", "Path": "/media/a.mkv"}
                        ],
                        "TotalRecordCount": 2,
                    }
                return {
                    "Items": [
                        {"Id": "m2", "Name": "Movie Two", "Type": "Movie", "Path": "/media/b.mkv"}
                    ],
                    "TotalRecordCount": 2,
                }

        events: list[dict[str, Any]] = []
        cancel_after_building = [False]

        def progress_callback(state: dict[str, Any]) -> None:
            events.append(state)
            if state["stage"] == "building":
                cancel_after_building[0] = True

        def should_cancel() -> bool:
            return cancel_after_building[0]

        index = MediaIndex(tmp_path / "index.sqlite")
        client = FakeClient()
        with pytest.raises(MediaIndexBuildCancelled):
            build_media_index(
                client,
                "user1",
                [{"Id": "lib1", "Name": "Movies"}],
                index,
                page_size=1,
                progress_callback=progress_callback,
                should_cancel=should_cancel,
            )
        assert any(event["stage"] == "building" for event in events)
        assert events[-1]["stage"] == "building"
        assert client.calls == [0]