Use auth.tesla.cn token endpoint for the cn region

tesla_fleet_api/const.py

@@ -19,6 +19,12 @@
     "cn": "https://fleet-api.prd.cn.vn.cloud.tesla.cn",
 }
 
+TOKEN_URLS: dict[Region, str] = {
+    "na": "https://fleet-auth.prd.vn.cloud.tesla.com/oauth2/v3/token",
+    "eu": "https://fleet-auth.prd.vn.cloud.tesla.com/oauth2/v3/token",
+    "cn": "https://auth.tesla.cn/oauth2/v3/token",
+}
+
 
 def is_valid_region(value: str) -> TypeGuard[Region]:
     """Check if a string is a valid region and narrow its type."""

tesla_fleet_api/tesla/fleet.py

@@ -7,7 +7,7 @@
 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, TOKEN_URLS, Method, Region, Scope
 from tesla_fleet_api.exceptions import (
     InvalidRegion,
     LibraryError,
@@ -267,7 +267,7 @@ async def partner_login(
         }
 
         async with self.session.post(
-            "https://fleet-auth.prd.vn.cloud.tesla.com/oauth2/v3/token",
+            TOKEN_URLS.get(self.region or "na", TOKEN_URLS["na"]),
             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,
+    TOKEN_URLS,
+    Method,
+    Region,
+    Scope,
+    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",
+            TOKEN_URLS.get(self.region or "na", TOKEN_URLS["na"]),
             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",
+            TOKEN_URLS.get(self.region or "na", TOKEN_URLS["na"]),
             data={
                 "grant_type": "refresh_token",
                 "client_id": self.client_id,

tests/test_oauth_token_host.py

@@ -0,0 +1,55 @@
+"""The OAuth token endpoint follows the region, like the authorize URL."""
+
+from contextlib import asynccontextmanager
+from unittest import IsolatedAsyncioTestCase
+from unittest.mock import MagicMock
+
+from tesla_fleet_api.const import Region
+from tesla_fleet_api.tesla.fleet import TeslaFleetApi
+from tesla_fleet_api.tesla.oauth import TeslaFleetOAuth
+
+GLOBAL = "https://fleet-auth.prd.vn.cloud.tesla.com/oauth2/v3/token"
+CN = "https://auth.tesla.cn/oauth2/v3/token"
+
+
+def _session(urls: list[str]) -> MagicMock:
+    resp = MagicMock()
+    resp.ok = True
+
+    async def _json() -> dict[str, object]:
+        return {"access_token": "a", "refresh_token": "r", "expires_in": 60}
+
+    resp.json = _json
+
+    @asynccontextmanager
+    async def _ctx(url: str, **kwargs: object):
+        urls.append(url)
+        yield resp
+
+    session = MagicMock()
+    session.post = MagicMock(side_effect=_ctx)
+    return session
+
+
+class TestTokenHost(IsolatedAsyncioTestCase):
+    async def test_code_exchange_and_refresh(self) -> None:
+        for region, expected in (("na", GLOBAL), ("eu", GLOBAL), ("cn", CN)):
+            urls: list[str] = []
+            api = TeslaFleetOAuth(
+                _session(urls),
+                region,  # type: ignore[arg-type]
+                "id",
+                client_secret="secret",
+                redirect_uri="https://example.com/cb",
+            )
+            await api.get_refresh_token("code")
+            await api.refresh_access_token()
+            self.assertEqual(urls, [expected, expected], region)
+
+    async def test_partner_login(self) -> None:
+        for region, expected in (("na", GLOBAL), ("cn", CN)):
+            urls: list[str] = []
+            region_: Region = region  # type: ignore[assignment]
+            api = TeslaFleetApi(_session(urls), access_token="t", region=region_)
+            await api.partner_login("id", "secret")
+            self.assertEqual(urls, [expected], region)