feat(metadata)!: authoritative id resolver and keyless imdb provider

Supplied IDs resolve directly via a shared resolve_by_ids and never fall back to title search; metadata_providers gains per-kind ordering and a disable_metadata switch stops all automatic lookups while explicit IDs keep working.
BREAKING: --tmdb/--imdb/--tvdb are now mutually exclusive and require a usable provider; --enrich also overwrites original language; the imdbapi provider and imdb_api_enabled are replaced by a keyless imdb provider.
This commit is contained in:
imSp4rky
2026-08-11 15:07:24 -06:00
parent 0cec6c9686
commit ebb4e74318
22 changed files with 1035 additions and 293 deletions
+67
View File
@@ -0,0 +1,67 @@
"""The `disable_metadata` config switch stops automatic lookups but keeps supplied IDs working."""
from __future__ import annotations
from typing import Optional, Union
import pytest
from unshackle.core import providers
from unshackle.core.config import config
from unshackle.core.providers._base import ExternalIds, MetadataProvider, MetadataResult
class FakeProvider(MetadataProvider):
"""Stands in for a real provider; fails the test if it is searched."""
def __init__(self, name: str, result: Optional[MetadataResult] = None) -> None:
self.NAME = name
self._result = result
self.asked: list[Union[int, str]] = []
super().__init__()
def is_available(self) -> bool:
return True
def search(self, title: str, year: Optional[int], kind: str) -> Optional[MetadataResult]:
raise AssertionError("no provider may be searched while disable_metadata is set")
def get_by_id(self, provider_id: Union[int, str], kind: str) -> Optional[MetadataResult]:
self.asked.append(provider_id)
return self._result
def get_external_ids(self, provider_id: Union[int, str], kind: str) -> ExternalIds:
return ExternalIds()
@pytest.fixture(autouse=True)
def _disabled(monkeypatch: pytest.MonkeyPatch) -> None:
"""Turn metadata off and keep every test off the network."""
monkeypatch.setattr(config, "disable_metadata", True)
monkeypatch.setattr(config, "metadata_providers", [])
monkeypatch.setattr(providers, "fetch_external_ids", lambda *a, **kw: ExternalIds())
monkeypatch.setattr(providers, "enrich_ids", lambda result: None)
def test_search_metadata_returns_none(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(providers, "get_provider", lambda name: FakeProvider(name))
assert providers.search_metadata("Futurama", 1999, "tv") is None
def test_supplied_id_still_resolves(monkeypatch: pytest.MonkeyPatch) -> None:
tmdb = FakeProvider("tmdb", MetadataResult(title="Inception", year=2010, kind="movie"))
monkeypatch.setattr(providers, "get_provider", lambda name: {"tmdb": tmdb}.get(name))
result = providers.resolve_by_ids(27205, kind="movie")
assert result is not None
assert result.title == "Inception"
assert result.external_ids.tmdb_id == 27205
assert tmdb.asked == [27205]
def test_no_ids_returns_none(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(providers, "get_provider", lambda name: FakeProvider(name))
assert providers.resolve_by_ids(title="Futurama", year=1999, kind="tv") is None
+21 -5
View File
@@ -1,11 +1,11 @@
"""--enrich original-language enrichment."""
"""Original-language parsing for the metadata a supplied ID resolves."""
from __future__ import annotations
import pytest
from unshackle.commands.dl import parse_language
from unshackle.core.providers.imdbapi import primary_language as imdb_language
from unshackle.core.providers.imdb import primary_language as imdb_language
from unshackle.core.providers.omdb import primary_language as omdb_language
from unshackle.core.providers.tvdb import primary_language as tvdb_language
@@ -28,11 +28,27 @@ def test_parse_language(tag: str | None, expected: str | None) -> None:
def test_imdb_language_takes_the_first_spoken_language() -> None:
data = {"spokenLanguages": [{"code": "jpn", "name": "Japanese"}, {"code": "eng", "name": "English"}]}
assert imdb_language(data) == "jpn"
data = {"spokenLanguages": {"spokenLanguages": [{"id": "ja", "text": "Japanese"}, {"id": "en", "text": "English"}]}}
assert imdb_language(data) == "ja"
@pytest.mark.parametrize("data", [{}, {"spokenLanguages": []}, {"spokenLanguages": None}])
def test_imdb_language_is_alpha_2() -> None:
"""IMDb's GraphQL language ids are alpha-2."""
data = {"spokenLanguages": {"spokenLanguages": [{"id": "ko", "text": "Korean"}]}}
assert parse_language(imdb_language(data)) is not None
assert str(parse_language(imdb_language(data))) == "ko"
@pytest.mark.parametrize(
"data",
[
{},
{"spokenLanguages": None}, # search hits can come back with the whole block null
{"spokenLanguages": {}},
{"spokenLanguages": {"spokenLanguages": []}},
{"spokenLanguages": {"spokenLanguages": None}},
],
)
def test_imdb_language_missing(data: dict) -> None:
assert imdb_language(data) is None
+183
View File
@@ -0,0 +1,183 @@
"""IMDb GraphQL provider: persisted-query flow, field mapping, and search constraints."""
from __future__ import annotations
import json
from typing import Any, Optional
import pytest
from unshackle.core.providers.imdb import IMDBProvider, escape_graphql
class _Response:
def __init__(self, payload: dict) -> None:
self._payload = payload
def json(self) -> dict:
return self._payload
def _node(
imdb_id: str = "tt3581920",
text: str = "The Last of Us",
year: Optional[int] = 2023,
title_type: str = "tvSeries",
languages: Any = ({"id": "en", "text": "English"}, {"id": "id", "text": "Indonesian"}),
) -> dict:
return {
"id": imdb_id,
"titleText": {"text": text},
"originalTitleText": {"text": text},
"releaseYear": {"year": year} if year else None,
"titleType": {"id": title_type},
"spokenLanguages": {"spokenLanguages": list(languages)} if languages is not None else None,
}
class _FakeSession:
"""Records GET/POST calls and replays queued payloads."""
def __init__(self, responses: list[dict]) -> None:
self._responses = responses
self.gets: list[dict] = []
self.posts: list[dict] = []
def get(self, url: str, params: dict, headers: dict, timeout: int) -> _Response:
self.gets.append(params)
return _Response(self._responses.pop(0))
def post(self, url: str, json: dict, headers: dict, timeout: int) -> _Response: # noqa: A002
self.posts.append(json)
return _Response(self._responses.pop(0))
def _provider(monkeypatch: pytest.MonkeyPatch, responses: list[dict]) -> tuple[IMDBProvider, _FakeSession]:
p = IMDBProvider()
session = _FakeSession(responses)
monkeypatch.setattr(type(p), "session", property(lambda _self: session))
return p, session
NOT_FOUND = {"errors": [{"message": "PersistedQueryNotFound", "extensions": {"code": "PERSISTED_QUERY_NOT_FOUND"}}]}
def test_is_available_without_any_key() -> None:
assert IMDBProvider().is_available() is True
def test_cache_hit_uses_one_get(monkeypatch: pytest.MonkeyPatch) -> None:
p, session = _provider(monkeypatch, [{"data": {"title": _node()}}])
result = p.get_by_id("tt3581920", "tv")
assert result is not None and result.title == "The Last of Us"
assert len(session.gets) == 1 and session.posts == []
def test_persisted_query_miss_falls_back_to_one_post(monkeypatch: pytest.MonkeyPatch) -> None:
p, session = _provider(monkeypatch, [NOT_FOUND, {"data": {"title": _node()}}])
result = p.get_by_id("tt3581920", "tv")
assert result is not None and result.external_ids.imdb_id == "tt3581920"
assert len(session.gets) == 1 and len(session.posts) == 1
# the POST registers the query by sending its text, keyed by the same hash the GET sent
assert "query UnshackleTitle" in session.posts[0]["query"]
sent_hash = json.loads(session.gets[0]["extensions"])["persistedQuery"]["sha256Hash"]
assert session.posts[0]["extensions"]["persistedQuery"]["sha256Hash"] == sent_hash
def test_get_params_carry_no_spaces(monkeypatch: pytest.MonkeyPatch) -> None:
"""IMDb's GraphQL gateway decodes "+" literally, so spaced JSON would come back as a 400."""
p, session = _provider(monkeypatch, [{"data": {"title": _node()}}])
p.get_by_id("tt3581920", "tv")
assert " " not in session.gets[0]["extensions"]
assert " " not in session.gets[0]["variables"]
def test_get_by_id_maps_every_field(monkeypatch: pytest.MonkeyPatch) -> None:
p, _ = _provider(monkeypatch, [{"data": {"title": _node()}}])
result = p.get_by_id("tt3581920", "tv")
assert result is not None
assert (result.title, result.year, result.kind) == ("The Last of Us", 2023, "tv")
assert result.original_language == "en"
assert result.source == "imdb"
assert result.external_ids.imdb_id == "tt3581920"
@pytest.mark.parametrize("title_type", ["movie", "tvSeries", "videoGame"])
def test_caller_kind_wins_over_the_response_title_type(monkeypatch: pytest.MonkeyPatch, title_type: str) -> None:
"""kind stays the caller's, so it cannot desync from the tmdb_kind resolve_by_ids seeds."""
p, _ = _provider(monkeypatch, [{"data": {"title": _node(title_type=title_type)}}])
result = p.get_by_id("tt1375666", "tv")
assert result is not None and result.kind == "tv"
@pytest.mark.parametrize("payload", [{"data": {"title": None}}, {"errors": [{"message": "boom"}]}, {"data": {}}])
def test_get_by_id_returns_none_on_no_answer(monkeypatch: pytest.MonkeyPatch, payload: dict) -> None:
p, _ = _provider(monkeypatch, [payload])
assert p.get_by_id("tt3581920", "tv") is None
def _edges(*nodes: dict) -> dict:
return {"data": {"advancedTitleSearch": {"edges": [{"node": {"title": n}} for n in nodes]}}}
def test_search_uses_its_own_operation_and_one_get_on_a_cache_hit(monkeypatch: pytest.MonkeyPatch) -> None:
p, session = _provider(monkeypatch, [_edges(_node())])
result = p.search("The Last of Us", 2023, "tv")
assert result is not None and result.external_ids.imdb_id == "tt3581920"
assert session.gets[0]["operationName"] == "UnshackleTitleSearch"
assert session.posts == []
def test_search_query_text_carries_the_constraints(monkeypatch: pytest.MonkeyPatch) -> None:
p, session = _provider(monkeypatch, [NOT_FOUND, _edges(_node())])
p.search("The Last of Us", 2023, "tv")
query = session.posts[0]["query"]
assert 'searchTerm: "The Last of Us"' in query
assert 'start: "2023-01-01", end: "2023-12-31"' in query
assert '["tvSeries", "tvMiniSeries"]' in query or '["tvSeries","tvMiniSeries"]' in query
def test_search_omits_the_year_range_when_unknown(monkeypatch: pytest.MonkeyPatch) -> None:
p, session = _provider(monkeypatch, [NOT_FOUND, _edges(_node(title_type="movie"))])
p.search("Inception", None, "movie")
query = session.posts[0]["query"]
assert "releaseDateConstraint" not in query
assert '["movie", "tvMovie"]' in query or '["movie","tvMovie"]' in query
def test_search_picks_the_closest_title_not_the_first(monkeypatch: pytest.MonkeyPatch) -> None:
noise = _node(imdb_id="tt27478132", text="Making of: The Last of Us")
exact = _node()
p, _ = _provider(monkeypatch, [_edges(noise, exact)])
result = p.search("The Last of Us", 2023, "tv")
assert result is not None and result.external_ids.imdb_id == "tt3581920"
def test_search_rejects_a_title_that_is_not_a_close_match(monkeypatch: pytest.MonkeyPatch) -> None:
p, _ = _provider(monkeypatch, [_edges(_node(imdb_id="tt999", text="Something Else Entirely"))])
assert p.search("The Last of Us", 2023, "tv") is None
def test_search_survives_a_null_language_block(monkeypatch: pytest.MonkeyPatch) -> None:
p, _ = _provider(monkeypatch, [_edges(_node(languages=None))])
result = p.search("The Last of Us", 2023, "tv")
assert result is not None and result.original_language is None
def test_search_returns_none_when_nothing_comes_back(monkeypatch: pytest.MonkeyPatch) -> None:
p, _ = _provider(monkeypatch, [_edges()])
assert p.search("The Last of Us", 2023, "tv") is None
def test_escape_graphql_closes_the_string_injection() -> None:
assert escape_graphql('a"b\\c') == 'a\\"b\\\\c'
def test_search_term_with_a_quote_stays_inside_the_string(monkeypatch: pytest.MonkeyPatch) -> None:
p, session = _provider(monkeypatch, [NOT_FOUND, _edges(_node(text='The "Best" Show'))])
p.search('The "Best" Show', None, "tv")
assert 'searchTerm: "The \\"Best\\" Show"' in session.posts[0]["query"]
def test_get_external_ids_echoes_the_id() -> None:
assert IMDBProvider().get_external_ids("tt1375666", "movie").imdb_id == "tt1375666"
+114
View File
@@ -0,0 +1,114 @@
"""Input validation for the --tmdb / --imdb / --tvdb flags."""
from __future__ import annotations
import click
import pytest
from unshackle.commands.dl import validate_metadata_ids
from unshackle.core.api.handlers import validate_download_parameters
from unshackle.core.config import config
@pytest.fixture(autouse=True)
def _default_order(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(config, "metadata_providers", [])
def _with_keys(monkeypatch: pytest.MonkeyPatch, **keys: str) -> None:
for name in ("tmdb_api_key", "tvdb_api_key", "omdb_api_key", "simkl_client_id"):
monkeypatch.setattr(config, name, keys.get(name, ""))
# ---------- one ID at a time ----------
@pytest.mark.parametrize(
("tmdb", "imdb", "tvdb"),
[(27205, "tt1375666", None), (27205, None, 73871), (None, "tt1375666", 73871), (27205, "tt1375666", 73871)],
)
def test_two_ids_are_refused(tmdb: int | None, imdb: str | None, tvdb: int | None) -> None:
with pytest.raises(click.UsageError) as exc:
validate_metadata_ids(tmdb, imdb, tvdb)
assert "cannot be used together" in str(exc.value)
def test_no_ids_is_fine() -> None:
assert validate_metadata_ids(None, None, None) is None
def test_imdb_alone_passes_on_the_keyless_provider(monkeypatch: pytest.MonkeyPatch) -> None:
_with_keys(monkeypatch)
assert validate_metadata_ids(None, "tt1375666", None) is None
# ---------- the ID must have a provider that can resolve it ----------
def test_tmdb_without_its_key_names_the_key(monkeypatch: pytest.MonkeyPatch) -> None:
_with_keys(monkeypatch)
with pytest.raises(click.UsageError) as exc:
validate_metadata_ids(27205, None, None)
assert "--tmdb" in str(exc.value) and "tmdb_api_key" in str(exc.value)
def test_tvdb_without_its_key_names_the_key(monkeypatch: pytest.MonkeyPatch) -> None:
_with_keys(monkeypatch)
with pytest.raises(click.UsageError) as exc:
validate_metadata_ids(None, None, 73871)
assert "--tvdb" in str(exc.value) and "tvdb_api_key" in str(exc.value)
def test_tmdb_with_its_key_passes(monkeypatch: pytest.MonkeyPatch) -> None:
_with_keys(monkeypatch, tmdb_api_key="k")
assert validate_metadata_ids(27205, None, None) is None
def test_a_provider_filtered_out_of_the_order_points_at_the_config(monkeypatch: pytest.MonkeyPatch) -> None:
_with_keys(monkeypatch, tmdb_api_key="k")
monkeypatch.setattr(config, "metadata_providers", ["imdb", "omdb"])
with pytest.raises(click.UsageError) as exc:
validate_metadata_ids(27205, None, None)
assert "metadata_providers" in str(exc.value)
def test_imdb_is_unresolvable_when_the_order_drops_both_of_its_providers(monkeypatch: pytest.MonkeyPatch) -> None:
_with_keys(monkeypatch)
monkeypatch.setattr(config, "metadata_providers", ["tmdb", "tvdb"])
with pytest.raises(click.UsageError) as exc:
validate_metadata_ids(None, "tt1375666", None)
assert "metadata_providers" in str(exc.value)
assert "imdb" in str(exc.value) and "omdb" in str(exc.value)
def test_one_kind_keeping_the_provider_is_enough(monkeypatch: pytest.MonkeyPatch) -> None:
"""A per-kind order can drop tvdb for movies and keep it for tv, so this must not error."""
_with_keys(monkeypatch, tvdb_api_key="k")
monkeypatch.setattr(config, "metadata_providers", {"tv": ["tvdb"], "movie": ["tmdb"]})
assert validate_metadata_ids(None, None, 73871) is None
# ---------- the REST path rejects the same combinations ----------
@pytest.mark.parametrize(
"payload",
[
{"tmdb_id": 27205, "imdb_id": "tt1375666"},
{"tmdb_id": 27205, "tvdb_id": 73871},
{"imdb_id": "tt1375666", "tvdb_id": 73871},
],
)
def test_api_refuses_two_ids(payload: dict) -> None:
err = validate_download_parameters(payload)
assert err and "multiple external IDs" in err
@pytest.mark.parametrize("payload", [{"tmdb_id": 27205}, {"imdb_id": "tt1375666"}, {"tvdb_id": 73871}, {}])
def test_api_accepts_one_id(payload: dict) -> None:
assert validate_download_parameters(payload) is None
def test_api_still_reports_a_malformed_id_first() -> None:
err = validate_download_parameters({"tmdb_id": "12345", "imdb_id": "tt1"})
assert err and "positive integer" in err
+131
View File
@@ -0,0 +1,131 @@
"""Precedence of user-supplied external IDs in providers.resolve_by_ids."""
from __future__ import annotations
from typing import Optional, Union
import pytest
from unshackle.core import providers
from unshackle.core.config import config
from unshackle.core.providers._base import ExternalIds, MetadataProvider, MetadataResult
class FakeProvider(MetadataProvider):
"""Stands in for a real provider; records the ID it was asked for."""
def __init__(self, name: str, result: Optional[MetadataResult] = None, raises: bool = False) -> None:
self.NAME = name
self._result = result
self._raises = raises
self.asked: list[Union[int, str]] = []
super().__init__()
def is_available(self) -> bool:
return True
def search(self, title: str, year: Optional[int], kind: str) -> Optional[MetadataResult]:
raise AssertionError("resolve_by_ids must not search when an ID was supplied")
def get_by_id(self, provider_id: Union[int, str], kind: str) -> Optional[MetadataResult]:
self.asked.append(provider_id)
if self._raises:
raise ValueError("provider is down")
return self._result
def get_external_ids(self, provider_id: Union[int, str], kind: str) -> ExternalIds:
return ExternalIds()
@pytest.fixture(autouse=True)
def _isolate(monkeypatch: pytest.MonkeyPatch) -> None:
"""Keep every test off the network and on the default provider order."""
monkeypatch.setattr(config, "metadata_providers", [])
monkeypatch.setattr(providers, "fetch_external_ids", lambda *a, **kw: ExternalIds())
monkeypatch.setattr(providers, "enrich_ids", lambda result: None)
monkeypatch.setattr(
providers,
"search_metadata",
lambda *a, **kw: pytest.fail("search_metadata must not run when an ID was supplied"),
)
def _patch_providers(monkeypatch: pytest.MonkeyPatch, available: dict[str, FakeProvider]) -> None:
monkeypatch.setattr(providers, "get_provider", lambda name: available.get(name))
def test_supplied_ids_survive_a_providers_own_answer(monkeypatch: pytest.MonkeyPatch) -> None:
# tmdb answers with a different imdb id; the one the user gave must win
tmdb = FakeProvider(
"tmdb",
MetadataResult(title="Inception", year=2010, kind="movie", external_ids=ExternalIds(imdb_id="tt9999999")),
)
_patch_providers(monkeypatch, {"tmdb": tmdb})
result = providers.resolve_by_ids(27205, "tt1375666", kind="movie")
assert result is not None
assert result.external_ids.tmdb_id == 27205
assert result.external_ids.tmdb_kind == "movie"
assert result.external_ids.imdb_id == "tt1375666"
def test_imdb_falls_through_to_omdb(monkeypatch: pytest.MonkeyPatch) -> None:
imdb = FakeProvider("imdb", raises=True)
omdb = FakeProvider("omdb", MetadataResult(title="Inception", year=2010, kind="movie"))
_patch_providers(monkeypatch, {"imdb": imdb, "omdb": omdb})
result = providers.resolve_by_ids(imdb_id="tt1375666", kind="movie")
assert result is not None
assert result.title == "Inception"
assert imdb.asked == ["tt1375666"]
assert omdb.asked == ["tt1375666"]
def test_all_providers_down_synthesizes_a_result(monkeypatch: pytest.MonkeyPatch) -> None:
_patch_providers(monkeypatch, {})
result = providers.resolve_by_ids(tvdb_id=73871, title="Futurama", year=1999, kind="tv")
assert result is not None
assert result.title == "Futurama"
assert result.year == 1999
assert result.external_ids.tvdb_id == 73871
def test_no_ids_searches(monkeypatch: pytest.MonkeyPatch) -> None:
calls: list[tuple] = []
monkeypatch.setattr(providers, "search_metadata", lambda *a, **kw: calls.append(a) or None)
_patch_providers(monkeypatch, {})
providers.resolve_by_ids(title="Futurama", year=1999, kind="tv")
assert calls and calls[0][:3] == ("Futurama", 1999, "tv")
def test_no_ids_and_no_title_returns_none(monkeypatch: pytest.MonkeyPatch) -> None:
_patch_providers(monkeypatch, {})
assert providers.resolve_by_ids(kind="tv") is None
def test_per_kind_order_puts_tvdb_first_for_tv(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(config, "metadata_providers", {"tv": ["tvdb", "tmdb", "imdb", "omdb"]})
tvdb = FakeProvider("tvdb", MetadataResult(title="Futurama", year=1999, kind="tv"))
tmdb = FakeProvider("tmdb", MetadataResult(title="Wrong Show", year=2020, kind="tv"))
_patch_providers(monkeypatch, {"tvdb": tvdb, "tmdb": tmdb})
result = providers.resolve_by_ids(615, tvdb_id=73871, kind="tv")
assert result is not None
assert result.title == "Futurama"
assert tmdb.asked == []
# movies keep the default order, so tmdb is consulted before tvdb there
assert [cls.NAME for cls in providers.provider_order("movie")] == list(providers.DEFAULT_ORDER)
def test_flat_list_config_applies_to_both_kinds(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(config, "metadata_providers", ["tvdb", "tmdb"])
assert [cls.NAME for cls in providers.provider_order("tv")] == ["tvdb", "tmdb"]
assert [cls.NAME for cls in providers.provider_order("movie")] == ["tvdb", "tmdb"]
assert [cls.NAME for cls in providers.provider_order()] == ["tvdb", "tmdb"]