diff --git a/app/modules/themoviedb/tmdbv3api/objs/auth.py b/app/modules/themoviedb/tmdbv3api/objs/auth.py index c610860b6..608f946e8 100644 --- a/app/modules/themoviedb/tmdbv3api/objs/auth.py +++ b/app/modules/themoviedb/tmdbv3api/objs/auth.py @@ -1,7 +1,10 @@ -from ..tmdb import TMDb +from ..exceptions import TMDbException +from ..tmdb import TMDb, TMDbJson class Authentication(TMDb): + """按令牌、账号验证、会话创建顺序执行 TMDb 认证。""" + _urls = { "create_request_token": "/authentication/token/new", "validate_with_login": "/authentication/token/validate_with_login", @@ -9,24 +12,25 @@ class Authentication(TMDb): "delete_session": "/authentication/session", } - def __init__(self, username, password): + def __init__(self, username: str, password: str) -> None: + """保存凭据并完成认证,不吞掉上游业务错误。""" super().__init__() self.username = username self.password = password - self.expires_at = None + self.expires_at: TMDbJson = None self.request_token = self._create_request_token() self._authorise_request_token_with_login() self._create_session() - def _create_request_token(self): + def _create_request_token(self) -> str: """ Create a temporary request token that can be used to validate a TMDb user login. """ response = self._request_obj(self._urls["create_request_token"]) - self.expires_at = response.expires_at - return response.request_token + self.expires_at = self._get_json_key(response, "expires_at") + return self._required_string(response, "request_token") - def _create_session(self): + def _create_session(self) -> None: """ You can use this method to create a fully valid session ID once a user has validated the request token. """ @@ -35,9 +39,9 @@ class Authentication(TMDb): method="POST", json={"request_token": self.request_token} ) - self.session_id = response.session_id + self.session_id = self._required_string(response, "session_id") - def _authorise_request_token_with_login(self): + def _authorise_request_token_with_login(self) -> None: """ This method allows an application to validate a request token by entering a username and password. """ @@ -51,7 +55,7 @@ class Authentication(TMDb): } ) - def delete_session(self): + def delete_session(self) -> None: """ If you would like to delete (or "logout") from a session, call this method with a valid session ID. """ @@ -62,3 +66,11 @@ class Authentication(TMDb): json={"session_id": self.session_id} ) self.session_id = "" + + @classmethod + def _required_string(cls, response: TMDbJson, field: str) -> str: + """验证认证必需字段,避免把无效响应继续作为凭据发送。""" + value = cls._get_json_key(response, field) + if not isinstance(value, str) or not value: + raise TMDbException(f"TheMovieDb 认证响应缺少有效字段 {field!r}") + return value diff --git a/mypy.ini b/mypy.ini index cbb3b503b..dcc146a27 100644 --- a/mypy.ini +++ b/mypy.ini @@ -173,4 +173,5 @@ files = app/api/context.py, app/api/dependencies/subscription.py, app/modules/themoviedb/tmdbv3api/tmdb.py, + app/modules/themoviedb/tmdbv3api/objs/auth.py, scripts/architecture/async_blocking.py diff --git a/tests/test_tmdb_auth_response_contract.py b/tests/test_tmdb_auth_response_contract.py new file mode 100644 index 000000000..1f631f3e1 --- /dev/null +++ b/tests/test_tmdb_auth_response_contract.py @@ -0,0 +1,50 @@ +"""认证链使用 TMDb 已解析的 JSON 字典,而不是属性包装对象。""" + +from unittest.mock import Mock, call + +import pytest + +from app.modules.themoviedb.tmdbv3api.exceptions import TMDbException +from app.modules.themoviedb.tmdbv3api.objs.auth import Authentication + + +def test_authentication_reads_json_credentials_and_preserves_request_order(monkeypatch): + """真实字典中的 token 和 session 必须按顺序传递,注销后清空会话。""" + request = Mock(side_effect=[ + {"success": True, "expires_at": "2030-01-01 00:00:00 UTC", "request_token": "token"}, + {"success": True}, + {"success": True, "session_id": "session"}, + {"success": True}, + ]) + monkeypatch.setattr(Authentication, "_request_obj", request) + client = Authentication("user", "password") + try: + assert client.request_token == "token" + assert client.expires_at == "2030-01-01 00:00:00 UTC" + assert client.session_id == "session" + client.delete_session() + assert not client.has_session + assert request.call_args_list == [ + call("/authentication/token/new"), + call("/authentication/token/validate_with_login", method="POST", json={ + "username": "user", "password": "password", "request_token": "token", + }), + call("/authentication/session/new", method="POST", json={"request_token": "token"}), + call("/authentication/session", method="DELETE", json={"session_id": "session"}), + ] + finally: + client.close() + + +@pytest.mark.parametrize("value", [None, "", 42, [], {}]) +@pytest.mark.parametrize("field", ["request_token", "session_id"]) +def test_authentication_rejects_missing_or_invalid_credentials(monkeypatch, field, value): + """失败的凭据解析应终止认证,不能把空值或非字符串发给下一步。""" + responses = [{field: value}] + if field == "session_id": + responses = [{"request_token": "token"}, {"success": True}, *responses] + request = Mock(side_effect=responses) + monkeypatch.setattr(Authentication, "_request_obj", request) + with pytest.raises(TMDbException, match=field): + Authentication("user", "password") + assert request.call_count == len(responses)