"""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"