MediaShelf/tests/test_ingest.py

"""Ingest correctness. The headline property is idempotency: a scanner that
double-counts produces a report that looks plausible and is wrong, which is worse
than one that crashes."""

import mockserver


def snapshot(db):
    return {
        "items": db.scalar("SELECT COUNT(*) FROM media_item"),
        "unit_bytes": db.scalar(
            "SELECT COALESCE(SUM(size_bytes),0) FROM media_item "
            "WHERE kind IN ('movie','season')"),
        "parts": db.scalar("SELECT COUNT(*) FROM media_part"),
        "part_bytes": db.scalar("SELECT COALESCE(SUM(size_bytes),0) FROM media_part"),
        "episodes": db.scalar("SELECT COUNT(*) FROM episode"),
        "events": db.scalar("SELECT COUNT(*) FROM watch_event"),
        "watch_sum": db.scalar(
            "SELECT COALESCE(SUM(watch_count),0) FROM media_item WHERE kind='movie'"),
        "missing": db.scalar("SELECT COUNT(*) FROM media_item WHERE status='missing'"),
    }


def test_scan_succeeds_and_totals_match_source(scanned):
    expected = mockserver.stats()
    assert scanned.scalar(
        "SELECT COALESCE(SUM(size_bytes),0) FROM media_item WHERE kind IN ('movie','season')"
    ) == expected["total_bytes"]
    assert scanned.scalar("SELECT COUNT(*) FROM episode") == expected["episodes"]
    assert scanned.scalar("SELECT COUNT(*) FROM watch_event") == expected["history"]


def test_scan_is_idempotent(scanned, rescan):
    first = snapshot(scanned)
    rescan("full")
    second = snapshot(scanned)
    rescan("full")
    third = snapshot(scanned)
    assert first == second == third, "a repeated scan changed the data"


def test_incremental_adds_no_duplicate_events(scanned, rescan):
    before = scanned.scalar("SELECT COUNT(*) FROM watch_event")
    r = rescan("incremental")
    assert r.status == "succeeded"
    assert scanned.scalar("SELECT COUNT(*) FROM watch_event") == before


def test_seasons_are_the_unit_for_tv(scanned):
    seasons = scanned.query(
        "SELECT * FROM media_item WHERE kind='season' AND episode_count > 0")
    assert seasons, "no seasons were built"
    for s in seasons:
        rolled = scanned.scalar(
            "SELECT COALESCE(SUM(size_bytes),0) FROM episode "
            "WHERE season_item_id=? AND status='present'", (s["id"],))
        assert s["size_bytes"] == rolled
        assert s["parent_id"] is not None, "season is not linked to its show"


def test_multi_part_items_sum_all_parts(scanned):
    """A movie held as two files must report the total, not the first part."""
    rows = scanned.query(
        "SELECT id, size_bytes, part_count FROM media_item "
        "WHERE kind='movie' AND part_count > 1")
    assert rows, "fixture has no multi-part movies"
    for r in rows:
        total = scanned.scalar(
            "SELECT SUM(size_bytes) FROM media_part WHERE media_item_id=?", (r["id"],))
        assert r["size_bytes"] == total


def test_dispositions_are_classified(scanned, cfg):
    got = {r[0]: r[1] for r in scanned.query(
        "SELECT disposition, COUNT(*) FROM watch_event GROUP BY 1")}
    assert set(got) <= {"completed", "partial", "abandoned"}
    assert got.get("abandoned", 0) > 0, "fixture should produce abandoned plays"
    bad = scanned.scalar(
        "SELECT COUNT(*) FROM watch_event WHERE disposition='completed' "
        "AND percent_complete IS NOT NULL AND percent_complete < ?",
        (cfg.completion_threshold,))
    assert bad == 0


def test_pre_history_flag_tracks_coverage(scanned):
    cov = scanned.one("SELECT * FROM history_coverage ORDER BY event_count DESC LIMIT 1")
    assert cov and cov["earliest_event_at"]
    wrong = scanned.scalar(
        "SELECT COUNT(*) FROM media_item WHERE pre_history=1 AND added_at >= ?",
        (cov["earliest_event_at"],))
    assert wrong == 0
    assert scanned.scalar("SELECT COUNT(*) FROM media_item WHERE pre_history=1") > 0


def test_missing_items_are_flagged_not_deleted(scanned, rescan, monkeypatch):
    victim = scanned.one("SELECT * FROM media_item WHERE kind='movie' LIMIT 1")
    removed = [m for m in mockserver.MOVIES
               if m["ratingKey"] == victim["provider_item_id"]]
    assert removed
    monkeypatch.setattr(mockserver, "MOVIES",
                        [m for m in mockserver.MOVIES
                         if m["ratingKey"] != victim["provider_item_id"]])
    rescan("full")
    row = scanned.one("SELECT * FROM media_item WHERE id=?", (victim["id"],))
    assert row is not None, "a vanished item was deleted rather than flagged"
    assert row["status"] == "missing"


def test_refuses_history_from_a_different_plex_server(db, cfg, monkeypatch):
    """Joining another server's history would produce confident nonsense (§4.11)."""
    from mediashelf import ingest, providers
    from mediashelf.providers.base import ProviderError, ServerInfo

    media = providers.build_media_provider(cfg)
    history, _ = providers.build_history_provider(cfg, media)
    monkeypatch.setattr(history, "server_info",
                        lambda: ServerInfo(kind="tautulli", name="Other",
                                           server_id="TOTALLY-DIFFERENT"))
    result = ingest.Ingest(db, cfg, media, history).run("full", "manual")
    assert result.status == "failed"
    assert "different plex server" in (result.error or "").lower()