From 07d480832f7d4d44f993170aadf6ce3cbf0ca75d Mon Sep 17 00:00:00 2001 From: Brett Adams Date: Tue, 20 Oct 2026 13:00:39 +1000 Subject: [PATCH] Use the region's token host for OAuth code exchange, refresh and partner login Tesla's China docs (https://developer.tesla.cn/docs/fleet-api/authentication/third-party-tokens and partner-tokens) POST to https://auth.tesla.cn/oauth2/v3/token. Fixes #187 --- tesla_fleet_api/const.py | 6 ++++ tesla_fleet_api/tesla/fleet.py | 4 +-- tesla_fleet_api/tesla/oauth.py | 13 ++++++-- tests/test_oauth_token_host.py | 55 ++++++++++++++++++++++++++++++++++ 4 files changed, 73 insertions(+), 5 deletions(-) create mode 100644 tests/test_oauth_token_host.py diff --git a/tesla_fleet_api/const.py b/tesla_fleet_api/const.py index 09b3b2c..d344b93 100644 --- a/tesla_fleet_api/const.py +++ b/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.""" diff --git a/tesla_fleet_api/tesla/fleet.py b/tesla_fleet_api/tesla/fleet.py index 6ea8be7..41f49ee 100644 --- a/tesla_fleet_api/tesla/fleet.py +++ b/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: diff --git a/tesla_fleet_api/tesla/oauth.py b/tesla_fleet_api/tesla/oauth.py index 48d3c69..abeba7f 100644 --- a/tesla_fleet_api/tesla/oauth.py +++ b/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, diff --git a/tests/test_oauth_token_host.py b/tests/test_oauth_token_host.py new file mode 100644 index 0000000..46087fb --- /dev/null +++ b/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)