diff --git a/README.md b/README.md index 04a4bcc..b1fbc2f 100644 --- a/README.md +++ b/README.md @@ -52,6 +52,15 @@ async def main(): remove() ``` +## Teslemetry for Business API keys + +A Teslemetry for Business API key (`Authorization: Bearer sk_...`) works with the same class, with these differences: + +- Pass `vin`. A business key can only stream one consented product (a VIN or an energy site id) at a time, so a stream without `vin` raises `ValueError` (or `TeslemetryStreamBusinessKeyError` on connect, when the token is a callable). +- A business key may not call `/api/metadata`, so `find_server()` reads the product's region from `GET /api/business/products` instead. On the default or an empty `server`, the stream connects to that region host (`na.teslemetry.com` or `eu.teslemetry.com`). If the product is not shared with the business, the stream stops with `TeslemetryStreamBusinessKeyError`. +- The server ends a business stream after 5 minutes so that a revoked consent or key takes effect. The stream reconnects at once, and connection listeners see it go down and up again. +- `update_fields()` (PATCH) adds fields. `replace_fields()` (POST) is not available to a business key and returns 403. + ## Using Typed Listen Methods The library provides typed listen methods for various telemetry signals. These methods allow you to listen to specific signals and handle their data in a type-safe manner. Here is an example of using the typed listen methods: @@ -269,7 +278,7 @@ Return if connected. Get the current stream config. ### `find_server(self) -> None` -Find the server using metadata. +Find the server using metadata, or for a business key, the region of `vin` in the business product listing. ### `update_fields(fields: dict, vin: str) -> dict` Modify the Fleet Telemetry configuration. diff --git a/teslemetry_stream/__init__.py b/teslemetry_stream/__init__.py index a2419af..acfba7c 100644 --- a/teslemetry_stream/__init__.py +++ b/teslemetry_stream/__init__.py @@ -10,6 +10,7 @@ from .energysite import TeslemetryStreamEnergySite from .exception import ( TeslemetryStreamAuthenticationError, + TeslemetryStreamBusinessKeyError, TeslemetryStreamConnectionError, TeslemetryStreamEnded, TeslemetryStreamError, @@ -28,6 +29,7 @@ "SseTopic", "TeslemetryStream", "TeslemetryStreamAuthenticationError", + "TeslemetryStreamBusinessKeyError", "TeslemetryStreamConnectionError", "TeslemetryStreamEnded", "TeslemetryStreamEnergySite", diff --git a/teslemetry_stream/exception.py b/teslemetry_stream/exception.py index 9921d38..e9e86e7 100644 --- a/teslemetry_stream/exception.py +++ b/teslemetry_stream/exception.py @@ -1,3 +1,6 @@ +from __future__ import annotations + + class TeslemetryStreamError(Exception): """Teslemetry Stream Error""" @@ -29,3 +32,14 @@ class TeslemetryStreamAuthenticationError(TeslemetryStreamError): """Teslemetry Stream Authentication Error""" message = "The access token was rejected (401/403) and will not be retried." + + +class TeslemetryStreamBusinessKeyError(TeslemetryStreamError): + """Teslemetry Stream Business Key Error""" + + message = "This request is not available to a Teslemetry for Business API key." + + def __init__(self, message: str | None = None) -> None: + if message is not None: + self.message = message + super().__init__() diff --git a/teslemetry_stream/stream.py b/teslemetry_stream/stream.py index 4c05de4..9258f66 100644 --- a/teslemetry_stream/stream.py +++ b/teslemetry_stream/stream.py @@ -13,7 +13,11 @@ from .const import CreditsEvent, Key from .const import parse_created_at as _parse_created_at_dt from .energysite import TeslemetryStreamEnergySite -from .exception import TeslemetryStreamAuthenticationError, TeslemetryStreamEnded +from .exception import ( + TeslemetryStreamAuthenticationError, + TeslemetryStreamBusinessKeyError, + TeslemetryStreamEnded, +) from .vehicle import TeslemetryStreamVehicle LOGGER = logging.getLogger(__package__) @@ -23,6 +27,17 @@ # (no slash) means string-only. _LIBRARY_HEADER = f"python teslemetry-stream/{version('teslemetry-stream')}" +# A Teslemetry for Business key is a WorkOS organization API key. It may not +# call /api/metadata or open the account-wide /sse stream, so its region +# comes from the business product listing instead. +_BUSINESS_KEY_PREFIX = "sk_" +_DEFAULT_SERVER = "api.teslemetry.com" + + +def is_business_key(access_token: str | None) -> bool: + """Return True if the token is a Teslemetry for Business API key.""" + return access_token is not None and access_token.startswith(_BUSINESS_KEY_PREFIX) + class TeslemetryStream: """Teslemetry Stream Client""" @@ -33,7 +48,7 @@ def __init__( self, session: aiohttp.ClientSession, access_token: str | Callable[[], Awaitable[str | None]], - server: str = "api.teslemetry.com", + server: str = _DEFAULT_SERVER, vin: str | None = None, parse_timestamp: bool = False, manual: bool = False, @@ -44,7 +59,10 @@ def __init__( :param session: An aiohttp ClientSession. :param access_token: Access token for authentication. - :param server: Teslemetry server to connect to. + :param server: Teslemetry server to connect to. A Teslemetry for + Business key (`sk_...`) on the default or an empty server goes + to the region host of `vin` instead, found in the business + product listing. :param vin: Vehicle Identification Number. :param parse_timestamp: Whether to parse timestamps. :param manual: Whether to start listening manually. @@ -58,6 +76,10 @@ def __init__( """ if server and not server.endswith(".teslemetry.com"): raise ValueError("Server must be on the teslemetry.com domain") + if isinstance(access_token, str) and is_business_key(access_token) and not vin: + raise ValueError( + "A Teslemetry for Business API key can only stream one product - pass vin" + ) self.active: bool = False self.server = server @@ -90,6 +112,12 @@ def __init__( self.vehicles: dict[str, TeslemetryStreamVehicle] = {} self.energysites: dict[int, TeslemetryStreamEnergySite] = {} self.fields: dict[str, Any] = {} + # Known once the token has been read - it may be a callable. + self.business: bool = isinstance(access_token, str) and is_business_key( + access_token + ) + # Whether the region host for a business key is already resolved. + self._business_server_resolved = False if self.vin: self.vehicle: TeslemetryStreamVehicle = self.get_vehicle(self.vin) @@ -100,6 +128,7 @@ async def headers(self) -> dict[str, str]: access_token = await self.access_token() else: access_token = self.access_token + self.business = is_business_key(access_token) return { "Authorization": f"Bearer {access_token}", "X-Library": _LIBRARY_HEADER, @@ -152,9 +181,16 @@ async def get_config(self, vin: str | None = None) -> None: async def find_server(self) -> None: """ - Find the server using metadata. + Find the server using metadata, or for a Teslemetry for Business key, + the region host of `vin` in the business product listing. + + :raises TeslemetryStreamBusinessKeyError: If a business key has no + `vin`, or `vin` is not shared with the business. """ headers = await self.headers() + if self.business: + await self._find_business_server(headers) + return req = await self._session.get( "https://api.teslemetry.com/api/metadata", headers=headers, @@ -163,6 +199,30 @@ async def find_server(self) -> None: response = await req.json() self.server = f"{response['region'].lower()}.teslemetry.com" + async def _find_business_server(self, headers: dict[str, str]) -> None: + """ + Set the server to the region host of `vin` from the business product + listing. A business key may not call /api/metadata. + """ + if not self.vin: + raise TeslemetryStreamBusinessKeyError( + "A Teslemetry for Business API key can only stream one product - pass vin" + ) + req = await self._session.get( + f"https://{_DEFAULT_SERVER}/api/business/products", + headers=headers, + raise_for_status=True, + ) + response = await req.json() + for product in response.get("response") or []: + if str(product.get("product_id")) == str(self.vin): + self.server = f"{product['region'].lower()}.teslemetry.com" + self._business_server_resolved = True + return + raise TeslemetryStreamBusinessKeyError( + f"{self.vin} is not shared with this business" + ) + async def update_fields(self, fields: dict[str, Any], vin: str) -> dict[str, Any]: """ Update Fleet Telemetry configuration. @@ -258,11 +318,19 @@ async def connect(self) -> None: # already be connected, or a stop was requested outright. return + headers = await self.headers() + if ( + self.business + and self.server == _DEFAULT_SERVER + and not self._business_server_resolved + ): + # The default host proxies to the customer's region; going + # there directly saves a hop on every reconnect. + await self._find_business_server(headers) LOGGER.debug("Connecting to %s", self.server) url = f"https://{self.server}/sse" if self.vin: url += f"/{self.vin}" - headers = await self.headers() params = {"topics": ",".join(self.topics)} if self.topics else None response = await self._session.get( url, @@ -370,8 +438,19 @@ async def __anext__(self) -> dict[str, Any]: except TeslemetryStreamEnded: # A clean end of the response body is routine (e.g. a server # deploy), not an error - reconnecting is the expected outcome. - LOGGER.info("Stream ended by server, reconnecting") + # The server also ends every business-key stream after a fixed + # 5-minute lifetime, so reconnecting re-runs admission. + if self.business: + LOGGER.debug("Business stream lifetime ended, reconnecting") + else: + LOGGER.info("Stream ended by server, reconnecting") self._close_response() + except TeslemetryStreamBusinessKeyError as error: + # Retrying cannot change the business's product listing. + LOGGER.error("Business key error, not retrying: %s", error) + self.active = False + self._connection_failed() + raise except aiohttp.ClientError as error: if isinstance(error, aiohttp.ClientResponseError) and error.status in (401, 403): # A rejected token is a definitive answer, not a transient diff --git a/tests/test_business_key.py b/tests/test_business_key.py new file mode 100644 index 0000000..acdfe25 --- /dev/null +++ b/tests/test_business_key.py @@ -0,0 +1,306 @@ +"""Tests for Teslemetry for Business API keys (`sk_...`). + +A business key may not call /api/metadata or open the account-wide /sse +stream (403 business_route_not_allowed), so the stream must find the region +host of its one product in GET /api/business/products, fail clearly when it +cannot, and reconnect when the server ends a business stream after its +fixed 5-minute lifetime. Consumer tokens must behave exactly as before. +""" +from __future__ import annotations + +import asyncio +import contextlib +import logging +from typing import Any + +from teslemetry_stream import TeslemetryStreamBusinessKeyError +from teslemetry_stream.stream import LOGGER, TeslemetryStream, is_business_key + +VIN = "5YJ3E1EA1JF000001" +BUSINESS_KEY = "sk_live_test" +PRODUCTS = { + "response": [ + { + "product_type": "vehicle", + "product_id": "5YJ3E1EA1JF000009", + "region": "NA", + "customer": {"id": "a", "ref": None}, + "granted_at": "2026-10-01T00:00:00.000Z", + }, + { + "product_type": "vehicle", + "product_id": VIN, + "region": "EU", + "customer": {"id": "b", "ref": None}, + "granted_at": "2026-10-01T00:00:00.000Z", + }, + ] +} + + +class FakeContent: + """Async-iterable SSE body that blocks until ended cleanly.""" + + def __init__(self) -> None: + self._blocker: asyncio.Future[None] = asyncio.get_running_loop().create_future() + + def __aiter__(self) -> FakeContent: + return self + + async def __anext__(self) -> bytes: + await self._blocker + raise StopAsyncIteration + + def end(self) -> None: + if not self._blocker.done(): + self._blocker.set_result(None) + + +class FakeResponse: + def __init__(self, url: str, body: Any = None) -> None: + self.url = url + self.status = 200 + self.ok = True + self.body = body + self.content = FakeContent() + self.closed = False + + async def json(self) -> Any: + return self.body + + def close(self) -> None: + self.closed = True + + +class FakeSession: + """Answers the JSON routes and records every GET url.""" + + def __init__(self, products: Any = PRODUCTS) -> None: + self.urls: list[str] = [] + self.sse: list[FakeResponse] = [] + self.products = products + + async def get(self, url: str, **kwargs: Any) -> FakeResponse: + self.urls.append(url) + if url.endswith("/api/metadata"): + return FakeResponse(url, {"region": "NA"}) + if url.endswith("/api/business/products"): + return FakeResponse(url, self.products) + if "/api/config/" in url: + return FakeResponse(url, {"response": {"fields": {}}}) + response = FakeResponse(url) + self.sse.append(response) + return response + + +def check(label: str, ok: bool, detail: str = "") -> bool: + print(f"{label:<72} {'PASS' if ok else 'FAIL'}{' ' + detail if detail else ''}") + return ok + + +async def settle() -> None: + for _ in range(5): + await asyncio.sleep(0) + + +async def drain(task: asyncio.Task[Any]) -> BaseException | None: + with contextlib.suppress(asyncio.CancelledError): + try: + await task + except Exception as error: + return error + return None + + +async def test_detection(results: list[bool]) -> None: + results.append(check("sk_ prefix is a business key", is_business_key("sk_abc"))) + results.append(check("a consumer token is not", not is_business_key("abc.def"))) + results.append(check("None is not", not is_business_key(None))) + + +async def test_consumer_unchanged(results: list[bool]) -> None: + session = FakeSession() + stream = TeslemetryStream(session, "consumer-token", server="", vin=VIN, manual=True) # type: ignore[arg-type] + await stream.find_server() + results.append( + check("consumer token still uses /api/metadata", session.urls[-1].endswith("/api/metadata")) + ) + results.append(check("consumer server from metadata", stream.server == "na.teslemetry.com")) + + session = FakeSession() + stream = TeslemetryStream(session, "consumer-token", vin=VIN, manual=True) # type: ignore[arg-type] + await stream.connect() + results.append( + check( + "consumer token on the default host makes no listing call", + session.urls == [f"https://api.teslemetry.com/sse/{VIN}"], + f"got {session.urls}", + ) + ) + stream.close() + + +async def test_business_find_server(results: list[bool]) -> None: + session = FakeSession() + stream = TeslemetryStream(session, BUSINESS_KEY, server="", vin=VIN, manual=True) # type: ignore[arg-type] + await stream.connect() + results.append( + check( + "business key never calls /api/metadata", + not any(u.endswith("/api/metadata") for u in session.urls), + f"got {session.urls}", + ) + ) + results.append( + check("business server is the VIN's region", stream.server == "eu.teslemetry.com") + ) + results.append( + check( + "business stream opens /sse/ on the region host", + session.urls[-1] == f"https://eu.teslemetry.com/sse/{VIN}", + f"got {session.urls[-1]}", + ) + ) + stream.close() + + +async def test_business_default_host_and_lifetime(results: list[bool]) -> None: + session = FakeSession() + + async def token() -> str: + return BUSINESS_KEY + + stream = TeslemetryStream(session, token, vin=VIN, manual=True) # type: ignore[arg-type] + states: list[bool] = [] + stream.async_add_connection_listener(states.append) + with CaptureLogs() as records: + task = asyncio.create_task(stream.listen()) + await settle() + results.append( + check( + "callable business key on the default host goes to the region host", + session.urls[-1] == f"https://eu.teslemetry.com/sse/{VIN}", + f"got {session.urls}", + ) + ) + # The server ends a business stream after its 5-minute lifetime. + session.sse[0].content.end() + await settle() + + sse_urls = [u for u in session.urls if "/sse/" in u] + results.append( + check( + "a lifetime end reconnects to the same product", + sse_urls == [f"https://eu.teslemetry.com/sse/{VIN}"] * 2, + f"got {sse_urls}", + ) + ) + results.append( + check( + "the product listing is read once, not on every reconnect", + sum(u.endswith("/api/business/products") for u in session.urls) == 1, + ) + ) + results.append(check("the stream is still active", stream.active)) + results.append( + check("listeners saw down then up", states == [True, False, True], f"got {states}") + ) + results.append( + check( + "a lifetime end does not log at INFO or above", + not [r for r in records if r.levelno >= logging.INFO], + f"got {[r.getMessage() for r in records if r.levelno >= logging.INFO]}", + ) + ) + stream.close() + await drain(task) + + +async def test_business_not_shared(results: list[bool]) -> None: + session = FakeSession(products={"response": []}) + stream = TeslemetryStream(session, BUSINESS_KEY, vin=VIN, manual=True) # type: ignore[arg-type] + states: list[bool] = [] + stream.async_add_connection_listener(states.append) + task = asyncio.create_task(stream.listen()) + await settle() + error = await asyncio.wait_for(drain(task), 1) + results.append( + check( + "an unshared VIN raises TeslemetryStreamBusinessKeyError", + isinstance(error, TeslemetryStreamBusinessKeyError) and VIN in str(error), + f"got {error!r}", + ) + ) + results.append( + check("and stops instead of retrying", not stream.active and len(session.urls) == 1) + ) + results.append(check("and reports the connection down", states == [False], f"got {states}")) + + +async def test_business_without_vin(results: list[bool]) -> None: + try: + TeslemetryStream(FakeSession(), BUSINESS_KEY) # type: ignore[arg-type] + raised = False + except ValueError: + raised = True + results.append(check("a str business key without vin is rejected at construction", raised)) + + async def token() -> str: + return BUSINESS_KEY + + session = FakeSession() + stream = TeslemetryStream(session, token, server="", manual=True) # type: ignore[arg-type] + try: + await stream.find_server() + error: BaseException | None = None + except TeslemetryStreamBusinessKeyError as e: + error = e + results.append( + check( + "a callable business key without vin fails clearly in find_server", + error is not None and session.urls == [], + f"got {error!r} {session.urls}", + ) + ) + + +class CaptureLogs: + def __enter__(self) -> list[logging.LogRecord]: + self._records: list[logging.LogRecord] = [] + self._handler = _ListHandler(self._records) + self._prev_level = LOGGER.level + LOGGER.addHandler(self._handler) + LOGGER.setLevel(logging.DEBUG) + return self._records + + def __exit__(self, *exc_info: Any) -> None: + LOGGER.removeHandler(self._handler) + LOGGER.setLevel(self._prev_level) + + +class _ListHandler(logging.Handler): + def __init__(self, records: list[logging.LogRecord]) -> None: + super().__init__() + self._records = records + + def emit(self, record: logging.LogRecord) -> None: + self._records.append(record) + + +async def main() -> None: + results: list[bool] = [] + await test_detection(results) + await test_consumer_unchanged(results) + await test_business_find_server(results) + await test_business_default_host_and_lifetime(results) + await test_business_not_shared(results) + await test_business_without_vin(results) + + print("-" * 72) + print("ALL PASS" if all(results) else "FAILURES PRESENT") + if not all(results): + raise SystemExit(1) + + +if __name__ == "__main__": + asyncio.run(main())