fix(oauth): send the OAuth token request to the region's auth host

tesla_fleet_api/const.py

@@ -25,6 +25,18 @@ def is_valid_region(value: str) -> TypeGuard[Region]:
     return value in SERVERS
 
 
+def auth_server(region: Region | None) -> str:
+    """The OAuth token endpoint for ``region``.
+
+    China is served by its own auth host; every other region - and an unset
+    region - shares the global one. The authorize endpoint follows the same
+    split (``auth.tesla.cn`` for China, ``auth.tesla.com`` elsewhere).
+    """
+    if region == "cn":
+        return "https://auth.tesla.cn/oauth2/v3/token"
+    return "https://fleet-auth.prd.vn.cloud.tesla.com/oauth2/v3/token"
+
+
 class IntEnum(int, Enum):
     """Integer Enum."""
 

tesla_fleet_api/tesla/fleet.py

@@ -7,7 +7,14 @@
 import aiohttp
 
 from tesla_fleet_api import __version__ as VERSION
-from tesla_fleet_api.const import LOGGER, SERVERS, Method, Region, Scope
+from tesla_fleet_api.const import (
+    LOGGER,
+    SERVERS,
+    Method,
+    Region,
+    Scope,
+    auth_server,
+)
 from tesla_fleet_api.exceptions import (
     InvalidRegion,
     LibraryError,
@@ -267,7 +274,7 @@ async def partner_login(
         }
 
         async with self.session.post(
-            "https://fleet-auth.prd.vn.cloud.tesla.com/oauth2/v3/token",
+            auth_server(self.region),
             headers={"Content-Type": "application/x-www-form-urlencoded"},
             data=data,
         ) as resp:

tesla_fleet_api/tesla/oauth.py

@@ -4,7 +4,14 @@
 
 import aiohttp
 
-from tesla_fleet_api.const import SERVERS, Method, Region, Scope, is_valid_region
+from tesla_fleet_api.const import (
+    SERVERS,
+    Method,
+    Region,
+    Scope,
+    auth_server,
+    is_valid_region,
+)
 from tesla_fleet_api.tesla import TeslaFleetApi
 
 
@@ -68,7 +75,7 @@ async def get_refresh_token(self, code: str) -> None:
             self.server = SERVERS.get(self.region)
 
         async with self.session.post(
-            "https://fleet-auth.prd.vn.cloud.tesla.com/oauth2/v3/token",
+            auth_server(self.region),
             data={
                 "grant_type": "authorization_code",
                 "client_id": self.client_id,
@@ -106,7 +113,7 @@ async def refresh_access_token(self) -> dict[str, Any]:
         if not self.refresh_token:
             raise ValueError("Refresh token is missing")
         async with self.session.post(
-            "https://fleet-auth.prd.vn.cloud.tesla.com/oauth2/v3/token",
+            auth_server(self.region),
             data={
                 "grant_type": "refresh_token",
                 "client_id": self.client_id,

tests/test_fleet_auth_refresh.py

@@ -3,15 +3,20 @@
 These tests assert what ``_request`` does today; they do not assume any
 auth-refresh / retry feature exists. If that feature is added later, extend
 these tests to cover the retry path rather than deleting them.
+
+``OAuthTokenEndpointTests`` covers the OAuth *token* endpoint, which has to
+follow the client's region: China is served by its own auth host
+(``auth.tesla.cn``), every other region by the shared global one.
 """
 
 from contextlib import asynccontextmanager
 from unittest import IsolatedAsyncioTestCase
 from unittest.mock import AsyncMock, MagicMock
 
-from tesla_fleet_api.const import Method
+from tesla_fleet_api.const import Method, Scope
 from tesla_fleet_api.exceptions import MissingToken, NotFound, ResponseError
 from tesla_fleet_api.tesla.fleet import TeslaFleetApi
+from tesla_fleet_api.tesla.oauth import TeslaFleetOAuth
 
 
 class _RequestTestApi(TeslaFleetApi):
@@ -114,3 +119,108 @@ async def test_error_status_routes_through_raise_for_status(self) -> None:
         api = _make_api(response=resp)
         with self.assertRaises(NotFound):
             await api.request(Method.GET, "api/1/products")
+
+
+def _fake_token_response() -> MagicMock:
+    """A 200 response carrying an OAuth token payload."""
+    resp = MagicMock()
+    resp.ok = True
+    resp.status = 200
+    resp.json = AsyncMock(
+        return_value={
+            "access_token": "access-token",
+            "refresh_token": "refresh-token",
+            "expires_in": 3600,
+        }
+    )
+    return resp
+
+
+def _make_post_session() -> MagicMock:
+    """A session whose ``post`` returns an async context manager yielding a token response."""
+    session = MagicMock()
+
+    @asynccontextmanager
+    async def _ctx(*args, **kwargs):
+        yield _fake_token_response()
+
+    session.post = MagicMock(side_effect=lambda *a, **k: _ctx(*a, **k))
+    return session
+
+
+class OAuthTokenEndpointTests(IsolatedAsyncioTestCase):
+    """The OAuth token endpoint must follow the client's region.
+
+    China is served by its own auth host (``auth.tesla.cn``); every other
+    region shares the global one (``fleet-auth.prd.vn.cloud.tesla.com``).
+    The *authorize* URL already switched host for China - the token URL did
+    not, so a China client redeemed its authorization code against the
+    global host and got its refresh token from the wrong region.
+    """
+
+    GLOBAL_TOKEN_URL = "https://fleet-auth.prd.vn.cloud.tesla.com/oauth2/v3/token"
+    CN_TOKEN_URL = "https://auth.tesla.cn/oauth2/v3/token"
+
+    def _make_oauth(self, region: str) -> tuple[TeslaFleetOAuth, MagicMock]:
+        session = _make_post_session()
+        api = TeslaFleetOAuth(
+            session=session,
+            region=region,
+            client_id="client-id",
+            client_secret="client-secret",
+            redirect_uri="https://example.com/callback",
+            refresh_token="refresh-token",
+        )
+        return api, session
+
+    async def test_cn_exchanges_the_authorization_code_on_the_cn_host(self) -> None:
+        api, session = self._make_oauth("cn")
+
+        await api.get_refresh_token("cn_abc123")
+
+        self.assertEqual(session.post.call_args.args[0], self.CN_TOKEN_URL)
+
+    async def test_cn_refreshes_the_access_token_on_the_cn_host(self) -> None:
+        api, session = self._make_oauth("cn")
+
+        await api.refresh_access_token()
+
+        self.assertEqual(session.post.call_args.args[0], self.CN_TOKEN_URL)
+
+    async def test_na_exchanges_the_authorization_code_on_the_global_host(self) -> None:
+        api, session = self._make_oauth("na")
+
+        await api.get_refresh_token("na_abc123")
+
+        self.assertEqual(session.post.call_args.args[0], self.GLOBAL_TOKEN_URL)
+
+    async def test_na_refreshes_the_access_token_on_the_global_host(self) -> None:
+        api, session = self._make_oauth("na")
+
+        await api.refresh_access_token()
+
+        self.assertEqual(session.post.call_args.args[0], self.GLOBAL_TOKEN_URL)
+
+    async def test_cn_partner_login_uses_the_cn_host(self) -> None:
+        session = _make_post_session()
+        api = TeslaFleetApi(session=session, access_token="access-token", region="cn")
+
+        await api.partner_login("client-id", "client-secret")
+
+        self.assertEqual(session.post.call_args.args[0], self.CN_TOKEN_URL)
+
+    async def test_na_partner_login_uses_the_global_host(self) -> None:
+        session = _make_post_session()
+        api = TeslaFleetApi(session=session, access_token="access-token", region="na")
+
+        await api.partner_login("client-id", "client-secret")
+
+        self.assertEqual(session.post.call_args.args[0], self.GLOBAL_TOKEN_URL)
+
+    async def test_cn_authorize_url_already_uses_the_cn_host(self) -> None:
+        """Characterization: the authorize URL switched host for China, the token URL did not."""
+        api, _ = self._make_oauth("cn")
+
+        url = api.get_login_url([Scope.OPENID])
+
+        self.assertTrue(url.startswith("https://auth.tesla.cn/oauth2/v3/authorize"))