fix(tmdb): read authentication credentials from JSON dictionaries

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

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

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)