diff --git a/app/modules/themoviedb/tmdbv3api/objs/list.py b/app/modules/themoviedb/tmdbv3api/objs/list.py index eb30c116e..fe4700dea 100644 --- a/app/modules/themoviedb/tmdbv3api/objs/list.py +++ b/app/modules/themoviedb/tmdbv3api/objs/list.py @@ -1,7 +1,10 @@ -from ..tmdb import TMDb +from ..exceptions import TMDbException +from ..tmdb import TMDb, TMDbJson class List(TMDb): + """提供同步与异步的 TMDb 片单查询和维护接口。""" + _urls = { "details": "/list/%s", "check_status": "/list/%s/item_status", @@ -12,7 +15,7 @@ class List(TMDb): "delete_list": "/list/%s", } - def details(self, list_id): + def details(self, list_id: int | str) -> TMDbJson: """ Get list details by id. :param list_id: int @@ -20,23 +23,24 @@ class List(TMDb): """ return self._request_obj(self._urls["details"] % list_id, key="items") - def check_item_status(self, list_id, movie_id): + def check_item_status(self, list_id: int | str, movie_id: int | str) -> TMDbJson: """ You can use this method to check if a movie has already been added to the list. :param list_id: int :param movie_id: int :return: """ - return self._request_obj(self._urls["check_status"] % list_id, params="movie_id=%s" % movie_id)["item_present"] + result = self._request_obj(self._urls["check_status"] % list_id, params="movie_id=%s" % movie_id) + return self._required_field(result, "item_present") - def create_list(self, name, description): + def create_list(self, name: str, description: str) -> TMDbJson: """ - You can use this method to check if a movie has already been added to the list. + Create a list and return its identifier. :param name: str :param description: str :return: """ - return self._request_obj( + result = self._request_obj( self._urls["create"], params="session_id=%s" % self.session_id, method="POST", @@ -45,9 +49,10 @@ class List(TMDb): "description": description, "language": self.language } - ).list_id + ) + return self._required_field(result, "list_id") - def add_movie(self, list_id, movie_id): + def add_movie(self, list_id: int | str, movie_id: int | str) -> None: """ Add a movie to a list. :param list_id: int @@ -60,7 +65,7 @@ class List(TMDb): json={"media_id": movie_id} ) - def remove_movie(self, list_id, movie_id): + def remove_movie(self, list_id: int | str, movie_id: int | str) -> None: """ Remove a movie from a list. :param list_id: int @@ -73,7 +78,7 @@ class List(TMDb): json={"media_id": movie_id} ) - def clear_list(self, list_id): + def clear_list(self, list_id: int | str) -> None: """ Clear all of the items from a list. :param list_id: int @@ -84,7 +89,7 @@ class List(TMDb): method="POST" ) - def delete_list(self, list_id): + def delete_list(self, list_id: int | str) -> None: """ Delete a list. :param list_id: int @@ -96,7 +101,7 @@ class List(TMDb): ) # 异步版本方法 - async def async_details(self, list_id): + async def async_details(self, list_id: int | str) -> TMDbJson: """ Get list details by id.(异步版本) :param list_id: int @@ -104,7 +109,7 @@ class List(TMDb): """ return await self._async_request_obj(self._urls["details"] % list_id, key="items") - async def async_check_item_status(self, list_id, movie_id): + async def async_check_item_status(self, list_id: int | str, movie_id: int | str) -> TMDbJson: """ You can use this method to check if a movie has already been added to the list.(异步版本) :param list_id: int @@ -112,11 +117,11 @@ class List(TMDb): :return: """ result = await self._async_request_obj(self._urls["check_status"] % list_id, params="movie_id=%s" % movie_id) - return result["item_present"] + return self._required_field(result, "item_present") - async def async_create_list(self, name, description): + async def async_create_list(self, name: str, description: str) -> TMDbJson: """ - You can use this method to check if a movie has already been added to the list.(异步版本) + Create a list and return its identifier.(异步版本) :param name: str :param description: str :return: @@ -131,9 +136,9 @@ class List(TMDb): "language": self.language } ) - return result.list_id + return self._required_field(result, "list_id") - async def async_add_movie(self, list_id, movie_id): + async def async_add_movie(self, list_id: int | str, movie_id: int | str) -> None: """ Add a movie to a list.(异步版本) :param list_id: int @@ -146,7 +151,7 @@ class List(TMDb): json={"media_id": movie_id} ) - async def async_remove_movie(self, list_id, movie_id): + async def async_remove_movie(self, list_id: int | str, movie_id: int | str) -> None: """ Remove a movie from a list.(异步版本) :param list_id: int @@ -159,7 +164,7 @@ class List(TMDb): json={"media_id": movie_id} ) - async def async_clear_list(self, list_id): + async def async_clear_list(self, list_id: int | str) -> None: """ Clear all of the items from a list.(异步版本) :param list_id: int @@ -170,7 +175,7 @@ class List(TMDb): method="POST" ) - async def async_delete_list(self, list_id): + async def async_delete_list(self, list_id: int | str) -> None: """ Delete a list.(异步版本) :param list_id: int @@ -180,3 +185,10 @@ class List(TMDb): params="session_id=%s" % self.session_id, method="DELETE" ) + + @staticmethod + def _required_field(response: TMDbJson, field: str) -> TMDbJson: + """读取响应必需字段,保留字段缺失时的 KeyError。""" + if not isinstance(response, dict): + raise TMDbException("TheMovieDb 返回数据格式异常:期望JSON对象") + return response[field] diff --git a/mypy.ini b/mypy.ini index dcc146a27..ef2ee8110 100644 --- a/mypy.ini +++ b/mypy.ini @@ -174,4 +174,5 @@ files = app/api/dependencies/subscription.py, app/modules/themoviedb/tmdbv3api/tmdb.py, app/modules/themoviedb/tmdbv3api/objs/auth.py, + app/modules/themoviedb/tmdbv3api/objs/list.py, scripts/architecture/async_blocking.py diff --git a/tests/test_tmdb_list_response_contract.py b/tests/test_tmdb_list_response_contract.py new file mode 100644 index 000000000..23941e266 --- /dev/null +++ b/tests/test_tmdb_list_response_contract.py @@ -0,0 +1,54 @@ +"""片单接口必须从 JSON 字典读取创建结果,并保留双入口一致性。""" + +import asyncio +from unittest.mock import AsyncMock, Mock, call + +import pytest + +from app.modules.themoviedb.tmdbv3api.exceptions import TMDbException +from app.modules.themoviedb.tmdbv3api.objs.list import List + + +@pytest.mark.parametrize("asynchronous", [False, True]) +def test_list_creation_reads_id_and_preserves_false_item_status(monkeypatch, asynchronous): + """创建片单读取 list_id;电影尚未加入时原样返回 False。""" + request = (AsyncMock if asynchronous else Mock)(side_effect=[ + {"success": True, "list_id": 735}, {"item_present": False}, + ]) + monkeypatch.setattr(List, "_async_request_obj" if asynchronous else "_request_obj", request) + client = List(language="zh-CN") + client.session_id = "session" + try: + if asynchronous: + created = asyncio.run(client.async_create_list("收藏", "说明")) + present = asyncio.run(client.async_check_item_status(735, 12)) + else: + created = client.create_list("收藏", "说明") + present = client.check_item_status(735, 12) + assert created == 735 + assert present is False + assert request.call_args_list == [ + call("/list", params="session_id=session", method="POST", json={ + "name": "收藏", "description": "说明", "language": "zh-CN", + }), + call("/list/735/item_status", params="movie_id=12"), + ] + finally: + client.close() + + +@pytest.mark.parametrize("asynchronous", [False, True]) +@pytest.mark.parametrize("payload,error", [({}, KeyError), ([], TMDbException), (None, TMDbException)]) +def test_list_status_rejects_missing_required_data(monkeypatch, asynchronous, payload, error): + """缺失状态字段仍抛 KeyError,非对象响应不能冒充有效片单状态。""" + request = (AsyncMock if asynchronous else Mock)(return_value=payload) + monkeypatch.setattr(List, "_async_request_obj" if asynchronous else "_request_obj", request) + client = List() + try: + with pytest.raises(error): + if asynchronous: + asyncio.run(client.async_check_item_status(735, 12)) + else: + client.check_item_status(735, 12) + finally: + client.close()