MediaShelf/tests/test_rules_and_api.py

"""Rule grammar (must reject everything outside the whitelist) and API shapes."""

import json

import pytest

from mediashelf.rules import RuleError, compile_rules, compile_sort

NOW = 1_757_000_000


def test_basic_conditions_compile_to_bound_parameters():
    sql, params = compile_rules(
        {"op": "and", "rules": [
            {"field": "size_bytes", "op": "gte", "value": 1024},
            {"field": "watch_count", "op": "eq", "value": 0}]}, NOW)
    assert "i.size_bytes >=" in sql and "i.watch_count =" in sql
    assert set(params.values()) == {1024, 0}
    assert "1024" not in sql, "value was interpolated instead of bound"


def test_nested_groups():
    sql, _ = compile_rules({"op": "or", "rules": [
        {"field": "kind", "op": "eq", "value": "movie"},
        {"op": "and", "rules": [
            {"field": "kind", "op": "eq", "value": "season"},
            {"field": "episode_count", "op": "gt", "value": 5}]}]}, NOW)
    assert sql.count("(") >= 2 and " OR " in sql and " AND " in sql


@pytest.mark.parametrize("field", [
    "title; DROP TABLE media_item",
    "1=1",
    "i.size_bytes) OR (1",
    "nonexistent_column",
    "",
])
def test_unknown_fields_are_rejected(field):
    with pytest.raises(RuleError):
        compile_rules({"field": field, "op": "eq", "value": 1}, NOW)


@pytest.mark.parametrize("op", ["exec", "'; --", "regexp", ""])
def test_unknown_operators_are_rejected(op):
    with pytest.raises(RuleError):
        compile_rules({"field": "title", "op": op, "value": "x"}, NOW)


def test_like_wildcards_in_user_input_are_escaped():
    sql, params = compile_rules(
        {"field": "title", "op": "contains", "value": "100% _real_"}, NOW)
    assert "ESCAPE" in sql
    assert list(params.values())[0] == r"%100\% \_real\_%"


def test_relative_date_operators():
    sql, params = compile_rules(
        {"field": "last_watched_at", "op": "older_than_days", "value": 730}, NOW)
    assert "IS NOT NULL" in sql
    assert list(params.values())[0] == NOW - 730 * 86400


def test_never_operator_treats_zero_and_null_alike():
    sql, _ = compile_rules({"field": "watch_count", "op": "never"}, NOW)
    assert "IS NULL" in sql and "= 0" in sql


def test_in_operator_bounds_list_length():
    with pytest.raises(RuleError):
        compile_rules({"field": "library_id", "op": "in",
                       "value": list(range(600))}, NOW)
    with pytest.raises(RuleError):
        compile_rules({"field": "library_id", "op": "in", "value": []}, NOW)


def test_deeply_nested_rules_are_rejected():
    node = {"field": "watch_count", "op": "eq", "value": 0}
    for _ in range(12):
        node = {"op": "and", "rules": [node]}
    with pytest.raises(RuleError):
        compile_rules(node, NOW)


@pytest.mark.parametrize("spec", ["size_bytes); DROP TABLE x--", "evil:desc", "1"])
def test_sort_whitelist(spec):
    with pytest.raises(RuleError):
        compile_sort(spec)


def test_sort_has_a_stable_tiebreak():
    assert compile_sort("size_bytes:desc").endswith("i.id ASC")


# ── API ──────────────────────────────────────────────────────────────────

def test_core_endpoints_respond(client):
    for path in ["/healthz", "/api/v1/stats/overview", "/api/v1/libraries",
                 "/api/v1/views", "/api/v1/duplicates", "/api/v1/scans",
                 "/api/v1/sources", "/api/v1/keeps", "/api/v1/accounts",
                 "/api/v1/stats/completion", "/api/v1/stats/added-over-time",
                 "/api/v1/stats/watch-distribution", "/"]:
        assert client.get(path).status_code == 200, path


def test_items_shape(client):
    data = client.get("/api/v1/items?page_size=5").get_json()
    assert {"total", "page", "page_size", "aggregate", "items"} <= set(data)
    assert data["total"] > 0
    item = data["items"][0]
    for key in ("id", "kind", "title", "library", "size_bytes", "reclaim_score",
                "reclaim_components", "kept", "flags", "pre_history"):
        assert key in item, key
    assert set(item["reclaim_components"]) == {
        "size", "staleness", "unpopularity", "solitude", "age", "rejection"}


def test_shows_are_excluded_from_the_grid_by_default(client):
    data = client.get("/api/v1/items?page_size=500").get_json()
    assert all(i["kind"] != "show" for i in data["items"])


def test_injection_via_rules_is_a_400_not_a_500(client):
    bad = json.dumps({"op": "and", "rules": [
        {"field": "title; DROP TABLE media_item", "op": "eq", "value": "x"}]})
    r = client.get("/api/v1/items?rules=" + bad)
    assert r.status_code == 400
    assert r.get_json()["error"] == "invalid_rule"
    assert client.get("/api/v1/items?page_size=1").get_json()["total"] > 0


def test_malformed_json_rules_is_a_400(client):
    assert client.get("/api/v1/items?rules=notjson").status_code == 400


def test_search_handles_hostile_input(client):
    for q in ['"', 'a OR b', 'NEAR(', '*', "'; --"]:
        assert client.get("/api/v1/items?q=" + q).status_code == 200, q


def test_csv_export_streams_every_row_not_just_a_page(client):
    total = client.get("/api/v1/items?page_size=1").get_json()["total"]
    body = client.get("/api/v1/export.csv?page_size=1").data.decode()
    lines = [l for l in body.splitlines() if l.strip()]
    assert len(lines) == total + 1


def test_saved_views_are_seeded_and_runnable(client):
    views = client.get("/api/v1/views").get_json()["views"]
    names = {v["name"] for v in views}
    assert {"Confident reclaim", "Uncertain reclaim", "Kept"} <= names
    for v in views:
        r = client.get("/api/v1/items?page_size=1&view_id=%d" % v["id"])
        assert r.status_code == 200, v["name"]


def test_builtin_views_cannot_be_deleted(client):
    v = next(v for v in client.get("/api/v1/views").get_json()["views"] if v["builtin"])
    assert client.delete("/api/v1/views/%d" % v["id"]).get_json()["deleted"] is False


def test_settings_never_leak_credentials(client):
    body = client.get("/api/v1/settings").get_json()
    assert body["plex_token"] == "***"
    assert body["tautulli_api_key"] == "***"
    assert body["secret_key"] == "***"
    assert "testtoken" not in json.dumps(body)
    assert "testkey" not in json.dumps(body)


def test_overview_reports_the_three_way_split(client):
    ov = client.get("/api/v1/stats/overview").get_json()
    for key in ("never_played_bytes", "kept_bytes", "available_bytes",
                "confident_bytes", "uncertain_bytes", "has_completion_data"):
        assert key in ov, key
    assert ov["confident_bytes"] + ov["uncertain_bytes"] <= ov["never_played_bytes"]


def test_bulk_keep(client, scanned):
    ids = [i["id"] for i in
           client.get("/api/v1/items?page_size=5").get_json()["items"]]
    r = client.post("/api/v1/keeps/bulk", json={"item_ids": ids, "note": "batch"})
    assert r.status_code == 200
    assert r.get_json()["created"] == len(ids)
    assert client.get("/api/v1/keeps").get_json()["kept_items"] >= len(ids)


def test_completion_flag_comes_from_events_not_the_coverage_table(client, scanned):
    """history_coverage is written at the END of an ingest. Reading the flag from
    it made the UI announce a Plex fallback mid-scan on every fresh database."""
    from mediashelf import queries

    assert queries.history_has_completion(scanned) is True

    # simulate mid-scan: events present, coverage not yet written
    scanned.execute("DELETE FROM history_coverage")
    assert queries.history_has_completion(scanned) is True, \
        "flag went false with events already ingested"
    assert queries.active_history_source(scanned) == "tautulli", \
        "source should fall back to the scan record when coverage is absent"

    ov = client.get("/api/v1/stats/overview").get_json()
    assert ov["has_completion_data"] is True
    assert ov["history_source"] == "tautulli"


def test_completion_flag_is_false_for_plex_only_history(scanned):
    from mediashelf import queries
    scanned.execute("UPDATE watch_event SET percent_complete = NULL")
    assert queries.history_has_completion(scanned) is False


def test_multi_part_flag_is_not_set_on_ordinary_seasons(client):
    """A season has one part per episode; flagging that as multi_part put a
    meaningless badge on every TV row in the grid."""
    data = client.get("/api/v1/items?page_size=500&kind=season").get_json()
    seasons = [i for i in data["items"] if i["kind"] == "season"]
    assert seasons
    bogus = [s for s in seasons
             if "multi_part" in s["flags"] and s["part_count"] <= (s["episode_count"] or 0)]
    assert not bogus, f"{len(bogus)} seasons flagged multi_part with no extra files"