aboutsummaryrefslogtreecommitdiffstats
path: root/packages/meshbay-node/tests/test_tmdb.py
blob: 4500feb9b5fe535e000fb0ad295ed42a0a568ef8 (plain) (blame)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
"""Tests for tmdb.py against a mocked httpx transport — no live network in CI."""

import httpx
import pytest

from meshbay_node.tmdb import TmdbClient


class FakeRoster:
    def __init__(self, enabled: bool = True, token: str | None = "fake-token",
                 language: str | None = None):
        self._enabled = enabled
        self._token = token
        self._language = language

    async def tmdb_config(self):
        return self._enabled, self._token, self._language


def _handler(response_map):
    def handle(request: httpx.Request) -> httpx.Response:
        path = request.url.path
        for prefix, body in response_map.items():
            if path.endswith(prefix):
                return httpx.Response(200, json=body)
        return httpx.Response(404, json={"results": []})
    return handle


@pytest.mark.asyncio
async def test_search_movie_returns_top_result_and_confidence():
    body = {"results": [{"id": 42, "title": "The Great Adventure", "release_date": "2015-01-01"}]}
    client = TmdbClient(
        roster=FakeRoster(),
        transport=httpx.MockTransport(_handler({"search/movie": body})),
    )
    result, ratio = await client.search_movie("The Great Adventure", 2015)

    assert result is not None
    assert result["id"] == 42
    assert ratio > 0.9
    await client.close()


@pytest.mark.asyncio
async def test_search_tv_returns_top_result():
    body = {"results": [{"id": 7, "name": "Some Show"}]}
    client = TmdbClient(
        roster=FakeRoster(),
        transport=httpx.MockTransport(_handler({"search/tv": body})),
    )
    result, ratio = await client.search_tv("Some Show")

    assert result is not None and result["id"] == 7
    assert ratio > 0.9
    await client.close()


@pytest.mark.asyncio
async def test_configured_language_is_sent_to_tmdb():
    captured = {}

    def handle(request: httpx.Request) -> httpx.Response:
        captured["language"] = request.url.params.get("language")
        return httpx.Response(200, json={"results": []})

    client = TmdbClient(
        roster=FakeRoster(language="fr-FR"),
        transport=httpx.MockTransport(handle),
    )
    await client.search_movie("Anything")

    assert captured["language"] == "fr-FR"
    await client.close()


@pytest.mark.asyncio
async def test_no_language_configured_omits_the_param():
    captured = {}

    def handle(request: httpx.Request) -> httpx.Response:
        captured["has_language"] = "language" in request.url.params
        return httpx.Response(200, json={"results": []})

    client = TmdbClient(
        roster=FakeRoster(language=None),
        transport=httpx.MockTransport(handle),
    )
    await client.search_movie("Anything")

    assert captured["has_language"] is False
    await client.close()


@pytest.mark.asyncio
async def test_no_results_returns_none_and_zero_confidence():
    client = TmdbClient(
        roster=FakeRoster(),
        transport=httpx.MockTransport(_handler({"search/movie": {"results": []}})),
    )
    result, ratio = await client.search_movie("Nonexistent Obscure Title")

    assert result is None
    assert ratio == 0.0
    await client.close()


@pytest.mark.asyncio
async def test_disabled_via_roster_setting_makes_no_request():
    calls = []

    def handle(request: httpx.Request) -> httpx.Response:
        calls.append(request)
        return httpx.Response(200, json={"results": []})

    client = TmdbClient(
        roster=FakeRoster(enabled=False),
        transport=httpx.MockTransport(handle),
    )
    result, ratio = await client.search_movie("Anything")

    assert result is None
    assert calls == []   # confirms the disabled check short-circuits before any request
    await client.close()


@pytest.mark.asyncio
async def test_no_token_resolvable_makes_no_request(monkeypatch):
    monkeypatch.delenv("MESHBAY_TMDB_DEFAULT_TOKEN", raising=False)
    calls = []

    def handle(request: httpx.Request) -> httpx.Response:
        calls.append(request)
        return httpx.Response(200, json={"results": []})

    client = TmdbClient(
        roster=FakeRoster(enabled=True, token=None),
        transport=httpx.MockTransport(handle),
    )
    result, ratio = await client.search_movie("Anything")

    assert result is None
    assert calls == []
    await client.close()


@pytest.mark.asyncio
async def test_http_error_returns_none_gracefully():
    def handle(request: httpx.Request) -> httpx.Response:
        return httpx.Response(500, json={"status_message": "server error"})

    client = TmdbClient(
        roster=FakeRoster(),
        transport=httpx.MockTransport(handle),
    )
    result, ratio = await client.search_movie("Anything")

    assert result is None
    assert ratio == 0.0
    await client.close()


@pytest.mark.asyncio
async def test_poster_url_builds_full_url():
    assert TmdbClient.poster_url("/abc123.jpg") == "https://image.tmdb.org/t/p/w500/abc123.jpg"
    assert TmdbClient.poster_url(None) is None