From 20dc0ebdd43b18762279308a7e511f16a778ff01 Mon Sep 17 00:00:00 2001 From: "coderabbitai[bot]" <136622811+coderabbitai[bot]@users.noreply.github.com> Date: Wed, 4 Feb 2026 21:12:22 +0000 Subject: [PATCH] =?UTF-8?q?=F0=9F=93=9D=20Add=20docstrings=20to=20`retry-i?= =?UTF-8?q?nternational-auth`?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Docstrings generation was requested by @DasBasti. * https://github.com/DasBasti/pySmartHashtag/pull/161#issuecomment-3849699039 The following files were modified: * `pysmarthashtag/account.py` * `pysmarthashtag/api/authentication.py` * `pysmarthashtag/api/utils.py` * `pysmarthashtag/cli.py` * `pysmarthashtag/const.py` * `pysmarthashtag/tests/common.py` * `pysmarthashtag/vehicle/vehicle.py` --- pysmarthashtag/account.py | 135 +++++++++++++++++++++++---- pysmarthashtag/api/authentication.py | 133 ++++++++++++++++++++++---- pysmarthashtag/api/utils.py | 77 +++++++++++++-- pysmarthashtag/cli.py | 35 ++++--- pysmarthashtag/const.py | 60 +++++++++--- pysmarthashtag/tests/common.py | 15 ++- pysmarthashtag/vehicle/vehicle.py | 15 ++- 7 files changed, 398 insertions(+), 72 deletions(-) mode change 100755 => 100644 pysmarthashtag/cli.py diff --git a/pysmarthashtag/account.py b/pysmarthashtag/account.py index 8fb578f..d9902d6 100644 --- a/pysmarthashtag/account.py +++ b/pysmarthashtag/account.py @@ -44,7 +44,18 @@ class SmartAccount: """Vehicles associated with the account.""" def __post_init__(self, password, log_responses): - """Initialize the account.""" + """ + Set up endpoint URLs and client configuration for the Smart account. + + If `endpoint_urls` is None, assigns a default EndpointUrls instance. + If `config` is None, creates a SmartClientConfiguration using a SmartAuthentication + initialized with the instance `username`, the provided `password`, and the + resolved `endpoint_urls`, and applies the `log_responses` flag to the configuration. + + Parameters: + password (str): Password used to construct the SmartAuthentication instance. + log_responses (bool): Whether the created configuration should log server responses. + """ # Ensure endpoint_urls is set if self.endpoint_urls is None: self.endpoint_urls = EndpointUrls() @@ -56,15 +67,19 @@ def __post_init__(self, password, log_responses): ) def _is_global_auth(self) -> bool: - """Return True when using the Global app authentication mode.""" + """ + Determine whether the account uses Global app authentication. + + Returns: + bool: `true` if the account's authentication mode equals `GLOBAL_HMAC`, `false` otherwise. + """ return self.config.authentication.auth_mode == SmartAuthMode.GLOBAL_HMAC async def _ensure_ssl_context(self) -> None: - """Ensure SSL context is created asynchronously. - - This method creates the SSL context in a thread pool executor - to avoid blocking the async event loop when httpx creates - SSL connections. + """ + Ensure the configuration and its authentication both have an SSL context. + + If the configuration has no SSL context, obtain one via config.get_ssl_context() and assign it to config.ssl_context and config.authentication.ssl_context. """ if self.config.ssl_context is None: self.config.ssl_context = await self.config.get_ssl_context() @@ -79,7 +94,11 @@ async def login(self, force_refresh: bool = False) -> None: await self.config.authentication.login() async def _init_vehicles(self) -> None: - """Initialize vehicles from Smart servers.""" + """ + Initialize and populate account vehicles from the Smart API. + + Ensures an SSL context is available, requests the account's vehicle list from the configured Smart endpoint, records the UTC fetch timestamp, and adds each returned vehicle to the account via add_vehicle (passing the fetch time). Retries the request up to three times on token-refresh errors and reattempts initialization when a human-car-connection error occurs. + """ _LOGGER.debug("Getting initial vehicle list") await self._ensure_ssl_context() @@ -120,7 +139,11 @@ async def _init_vehicles(self) -> None: self.add_vehicle(vehicle, fetched_at) async def _init_vehicles_global(self) -> None: - """Initialize vehicles from Smart Global servers.""" + """ + Fetches the account's vehicle ownership list from the Smart Global endpoints and registers each vehicle with the account. + + Each discovered vehicle is added to the account's vehicle mapping along with the timestamp when the list was fetched. + """ _LOGGER.debug("Getting initial vehicle list (global)") await self._ensure_ssl_context() @@ -152,11 +175,24 @@ async def _init_vehicles_global(self) -> None: self.add_vehicle(vehicle, fetched_at) def add_vehicle(self, vehicle, fetched_at): - """Add a vehicle to the account.""" + """ + Add a vehicle to the account's vehicle mapping. + + Parameters: + vehicle (dict): Vehicle data from the API; must include the `vin` key used as the mapping key. + fetched_at (datetime): UTC timestamp when the vehicle data was fetched. + """ self.vehicles[vehicle.get("vin")] = SmartVehicle(self, vehicle, fetched_at=fetched_at) async def get_vehicles(self, force_init: bool = False) -> None: - """Get the vehicles associated with the account.""" + """ + Load and refresh the vehicles for this Smart account and populate the account's internal vehicle mapping. + + If the account is not authenticated, perform authentication first. When called with `force_init=True` or when no vehicles are known, fetch the list of vehicles. For global-auth mode the method fetches global vehicle listings and updates global details and abilities; for non-global mode it selects each vehicle and updates its information, state-of-charge, and OTA info by merging the retrieved data into each SmartVehicle instance. + + Parameters: + force_init (bool): If True, re-initialize the vehicle list even if vehicles are already present. + """ await self._ensure_ssl_context() if self.config.authentication.api_user_id is None: await self.config.authentication.login() @@ -182,7 +218,13 @@ async def get_vehicles(self, force_init: bool = False) -> None: vehicle.combine_data(vehicle_info, charging_settings=vehicle_soc, ota_info=vehicle_ota_info) async def select_active_vehicle(self, vin) -> None: - """Select the active vehicle.""" + """ + Selects the given vehicle as the active vehicle for subsequent operations. + + This updates the remote session to mark the vehicle identified by `vin` as active. When the account is configured for global authentication, this is a no-op. The method may perform internal retries on transient token or human-car-connection errors. + Parameters: + vin (str): Vehicle Identification Number of the vehicle to select. + """ if self._is_global_auth(): return _LOGGER.debug("Selecting vehicle") @@ -221,7 +263,15 @@ async def select_active_vehicle(self, vin) -> None: break async def get_vehicle_information(self, vin) -> str: - """Get information about a vehicle.""" + """ + Fetch the latest details and status for the vehicle identified by VIN, using global endpoints when the account is configured for global authentication. + + Returns: + dict: The `data` payload from the vehicle status response containing status, basic, and more fields; empty dict if no data was retrieved. + + Raises: + SmartAuthError: If vehicle information cannot be retrieved after retrying. + """ if self._is_global_auth(): return await self._get_vehicle_details_global(vin) _LOGGER.debug("Getting information for vehicle") @@ -266,7 +316,17 @@ async def get_vehicle_information(self, vin) -> str: return data async def get_vehicle_soc(self, vin) -> str: - """Get information about a vehicle.""" + """ + Retrieve the vehicle's state-of-charge (SOC) data. + + If the account is using global authentication, this returns an empty dict. On failure after retries, raises SmartAuthError. + + Returns: + dict: SOC data payload from the vehicle response, or an empty dict when using global authentication. + + Raises: + SmartAuthError: If the SOC data could not be retrieved after retrying. + """ if self._is_global_auth(): return {} _LOGGER.debug("Getting vehicle SOC") @@ -309,7 +369,22 @@ async def get_vehicle_soc(self, vin) -> str: return data async def get_vehicle_ota_info(self, vin) -> dict: - """Get information about a vehicle from OTA server.""" + """ + Retrieve OTA version information for the specified vehicle from the OTA server. + + If the account uses global authentication, no OTA request is performed and an empty dict is returned. + + Parameters: + vin (str): Vehicle Identification Number for the vehicle to query. + + Returns: + dict: A mapping with keys: + - `target_version`: OTA target version string or `None` if not present. + - `current_version`: Current vehicle OTA version string or `None` if not present. + + Raises: + SmartAuthError: When repeated authentication/token failures prevent retrieving OTA information. + """ if self._is_global_auth(): return {} _LOGGER.debug("Getting OTA information for vehicle") @@ -351,13 +426,25 @@ async def get_vehicle_ota_info(self, vin) -> dict: return data async def _update_global_vehicle_details(self) -> None: - """Fetch global vehicle details and abilities.""" + """ + Fetch global details and abilities for each known vehicle and merge them into the corresponding SmartVehicle objects stored on the account. + + This updates each vehicle in-place with any details and capability information returned by the global service; no value is returned. + """ for vin in list(self.vehicles.keys()): await self._get_vehicle_details_global(vin) await self._get_vehicle_abilities_global(vin) async def _get_vehicle_details_global(self, vin) -> dict: - """Get global vehicle details.""" + """ + Fetch global vehicle details for the given VIN and merge them into the corresponding SmartVehicle if available. + + Parameters: + vin (str): Vehicle Identification Number to fetch details for. + + Returns: + dict: Parsed vehicle details retrieved from the global endpoint, or an empty dict if no details were returned. + """ _LOGGER.debug("Getting global vehicle details") await self._ensure_ssl_context() async with SmartLoginClient(ssl_context=self.config.ssl_context) as client: @@ -389,7 +476,17 @@ async def _get_vehicle_details_global(self, vin) -> dict: return details or {} async def _get_vehicle_abilities_global(self, vin) -> dict: - """Get global vehicle abilities.""" + """ + Retrieve global ability information for a vehicle identified by VIN. + + Parameters: + vin (str): Vehicle Identification Number to query for abilities. + + Returns: + abilities (dict): Abilities data from the global API. The vehicle's `data["abilities"]` + is updated when abilities are present. Returns an empty dict if the vehicle has no + model code or if the API provides no abilities. + """ _LOGGER.debug("Getting global vehicle abilities") await self._ensure_ssl_context() vehicle = self.vehicles.get(vin) @@ -417,4 +514,4 @@ async def _get_vehicle_abilities_global(self, vin) -> dict: abilities = data.get("result") or data.get("data") or {} if abilities and vehicle: vehicle.data["abilities"] = abilities - return abilities or {} + return abilities or {} \ No newline at end of file diff --git a/pysmarthashtag/api/authentication.py b/pysmarthashtag/api/authentication.py index b7357bd..f258b7a 100644 --- a/pysmarthashtag/api/authentication.py +++ b/pysmarthashtag/api/authentication.py @@ -42,6 +42,21 @@ def __init__( ssl_context: Optional[ssl.SSLContext] = None, endpoint_urls: Optional[EndpointUrls] = None, ): + """ + Initialize the authentication manager with credentials, optional tokens, SSL context, and endpoint configuration. + + Parameters: + username (str): Account username used for authentication. + password (str): Account password used for authentication. + access_token (Optional[datetime.datetime]): Existing OAuth access token, if available. + expires_at (Optional[datetime.datetime]): Expiration time of `access_token`; used to determine when refresh/login is needed. + refresh_token (Optional[str]): Refresh token associated with `access_token`, if available. + ssl_context (Optional[ssl.SSLContext]): Optional SSL context to use for login requests to avoid creating one lazily. + endpoint_urls (Optional[EndpointUrls]): Endpoint configuration; if omitted a default is created and the authentication mode is inferred from it. + + Behavior: + Stores provided values on the instance, generates a random device identifier, initializes internal locks and API session token placeholders, and sets `auth_mode` by inferring it from `endpoint_urls`. + """ self.username: str = username self.password: str = password self.access_token: Optional[str] = access_token @@ -59,16 +74,13 @@ def __init__( _LOGGER.debug("Device ID initialized") async def get_ssl_context(self) -> ssl.SSLContext: - """Get or create SSL context asynchronously. - - This method returns a cached SSL context if available, or creates - a new one asynchronously using the shared ssl_context module. - Thread-safe using asyncio.Lock. - - Returns - ------- - ssl.SSLContext: An SSL context for secure connections. - + """ + Obtain the SSLContext used for secure connections, creating and caching it if necessary. + + This method returns the cached SSLContext when present; otherwise it asynchronously acquires a new SSLContext and stores it for subsequent calls. The operation is safe for concurrent callers. + + Returns: + ssl.SSLContext: SSL context configured for secure HTTP connections. """ if self.ssl_context is None: # Import here to avoid circular imports @@ -138,7 +150,14 @@ async def async_auth_flow(self, request: Request) -> AsyncGenerator[Request, Res raise async def login(self) -> None: - """Login to the Smart API.""" + """ + Perform authentication with the Smart API and store retrieved tokens and expiry. + + Attempts to refresh the access token when a refresh token is available; otherwise performs a full login. On success, normalizes and stores `access_token`, `refresh_token`, `api_access_token`, `api_refresh_token`, `api_user_id`, optional `id_token`, and `expires_at` (adjusted by EXPIRES_AT_OFFSET) on the instance. + + Raises: + SmartAPIError: If required token fields are missing from the login response. + """ _LOGGER.debug("Logging in to Smart API") token_data = {} if self.refresh_token: @@ -161,7 +180,11 @@ async def login(self) -> None: raise SmartAPIError("Could not login to Smart API") async def _refresh_access_token(self): - """Refresh the access token.""" + """ + Attempt to refresh the stored access (and related) tokens using the configured authentication mode. + + Tries a mode-specific refresh (global HMAC refresh when configured, otherwise the EU refresh). If the refresh succeeds, returns a dict with refreshed token data (e.g., `access_token`, `refresh_token`, `expires_at`, and optional `id_token`); if the refresh fails, returns an empty dict to indicate a full login is required. + """ if self.auth_mode == SmartAuthMode.GLOBAL_HMAC: try: return await self._refresh_access_token_global() @@ -176,13 +199,34 @@ async def _refresh_access_token(self): return {} async def _login(self): - """Login to Smart web services.""" + """ + Selects and executes the appropriate login flow for the configured authentication mode. + + Dispatches to the global HMAC login when auth_mode is SmartAuthMode.GLOBAL_HMAC; otherwise runs the EU OAuth login flow. + + Returns: + token_data (dict): Authentication tokens and related metadata such as `access_token`, `refresh_token`, `expires_at`, and optionally `id_token` and API session fields. + """ if self.auth_mode == SmartAuthMode.GLOBAL_HMAC: return await self._login_global() return await self._login_eu() async def _login_eu(self): - """Login to Smart web services (EU OAuth flow).""" + """ + Perform the EU OAuth login flow, exchange the OAuth access token for API session tokens, and return the resulting tokens and expiry. + + Returns: + dict: Mapping with the following keys: + access_token (str): OAuth access token obtained from the authorization redirect. + refresh_token (str): OAuth refresh token obtained from the authorization redirect. + api_access_token (str): API session access token exchanged from the OAuth access token. + api_refresh_token (str): API session refresh token exchanged from the OAuth access token. + api_user_id (str): User identifier returned by the API session exchange. + expires_at (datetime.datetime): UTC timestamp when the OAuth access token expires. + + Raises: + SmartAPIError: If the login context, login token, redirect location, or access/refresh tokens cannot be obtained. + """ ssl_ctx = await self.get_ssl_context() async with SmartLoginClient(ssl_context=ssl_ctx) as client: _LOGGER.info("Acquiring access token.") @@ -289,7 +333,18 @@ async def _login_eu(self): } async def _get_api_session(self, client: "SmartLoginClient", access_token: str) -> tuple[str, str, str]: - """Exchange OAuth access token for API session tokens.""" + """ + Exchange an OAuth access token for the API session tokens and the associated user id. + + Posts the given OAuth `access_token` to the API session endpoint and returns the API-level + access token, refresh token, and user identifier obtained from the response. + + Returns: + tuple[str, str, str]: (api_access_token, api_refresh_token, api_user_id) + + Raises: + SmartAPIError: If the response does not contain the expected token or user id fields. + """ data = json.dumps({"accessToken": access_token}).replace(" ", "") r_api_access = await client.post( # we do not know what type of car we have in our list so we fall back to the old API URL @@ -319,7 +374,22 @@ async def _get_api_session(self, client: "SmartLoginClient", access_token: str) return api_access_token, api_refresh_token, api_user_id async def _refresh_access_token_eu(self) -> dict: - """Refresh the EU OAuth access token.""" + """ + Attempt to refresh the EU (OAuth) access token and exchange it for API session tokens. + + If a refresh token is available, sends a refresh request to the OAuth token URL, computes a new expiry timestamp, exchanges the returned OAuth access token for API session tokens, and returns a mapping of tokens and metadata. Returns an empty dict if no refresh token is configured or if the refresh response does not yield an access token. + + Returns: + dict: A mapping with the following keys when successful: + - "access_token" (str): The refreshed OAuth access token. + - "refresh_token" (str): The refreshed OAuth refresh token (or the previous refresh token if not provided). + - "api_access_token" (str): The exchanged API access token for subsequent API calls. + - "api_refresh_token" (str): The exchanged API refresh token for API session renewal. + - "api_user_id" (str): The user identifier returned by the API session exchange. + - "id_token" (str | None): The ID token returned by the OAuth refresh response, if present. + - "expires_at" (datetime.datetime): UTC timestamp when the OAuth access token expires. + Returns an empty dict if no refresh was possible or the refresh response lacked an access token. + """ if not self.refresh_token: return {} @@ -362,7 +432,19 @@ async def _refresh_access_token_eu(self) -> dict: } async def _login_global(self) -> dict: - """Login to Smart Global app services (HMAC flow).""" + """ + Perform the global (HMAC) login flow and return obtained tokens and related metadata. + + Returns: + dict: Mapping with the following keys: + - access_token (str): OAuth access token from the global login. + - refresh_token (str|None): Refresh token from the global login, if provided. + - api_access_token (str): API session access token (same as `access_token` for global flow). + - api_refresh_token (str|None): API session refresh token (same as `refresh_token` for global flow). + - api_user_id (str): User identifier returned by the global login. + - id_token (str|None): ID token returned by the global login, if present. + - expires_at (datetime.datetime): UTC timestamp when the access token expires. + """ ssl_ctx = await self.get_ssl_context() async with SmartLoginClient(ssl_context=ssl_ctx) as client: _LOGGER.info("Acquiring access token (global app).") @@ -421,7 +503,20 @@ async def _login_global(self) -> dict: } async def _refresh_access_token_global(self) -> dict: - """Refresh the Global app access token.""" + """ + Refresh the Global app access token and return the refreshed token set. + + Returns: + dict: Mapping with refreshed token and session fields: + - `access_token` (str): OAuth access token. + - `refresh_token` (str): OAuth refresh token (may be original if not returned). + - `api_access_token` (str): API session access token (same as `access_token`). + - `api_refresh_token` (str): API session refresh token (same as `refresh_token`). + - `api_user_id` (str): API user identifier. + - `id_token` (str | None): ID token if provided by the server. + - `expires_at` (datetime.datetime): UTC timestamp when the access token expires. + Returns an empty dict if no refresh token is available or the refresh attempt fails. + """ if not self.refresh_token: return {} @@ -569,4 +664,4 @@ def get_retry_wait_time(response: httpx.Response) -> int: retry_after = next(iter([int(i) for i in response.json().get("message", "") if i.isdigit()])) except Exception: retry_after = 2 - return math.ceil(retry_after * 2) + return math.ceil(retry_after * 2) \ No newline at end of file diff --git a/pysmarthashtag/api/utils.py b/pysmarthashtag/api/utils.py index 42441c7..97a210d 100644 --- a/pysmarthashtag/api/utils.py +++ b/pysmarthashtag/api/utils.py @@ -73,11 +73,25 @@ def generate_default_header( def create_correct_timestamp() -> str: - """Create a correct timestamp for the request.""" + """ + Generate a timestamp string representing the current time in milliseconds since the Unix epoch. + + Returns: + timestamp (str): Current time in milliseconds since 1970-01-01 UTC, formatted as a decimal string. + """ return str(int(time.time() * 1000)) def _ensure_bytes(body: Optional[object]) -> Optional[bytes]: + """ + Normalize a request body to a UTF-8 bytes object when present. + + Parameters: + body (Optional[object]): The value to normalize. If `None`, no conversion is performed. + + Returns: + Optional[bytes]: `None` if input is `None`; the input unchanged if already `bytes`; otherwise the UTF-8 encoding of `str(body)`. + """ if body is None: return None if isinstance(body, bytes): @@ -86,7 +100,15 @@ def _ensure_bytes(body: Optional[object]) -> Optional[bytes]: def _global_md5_base64(body: bytes) -> str: - """Calculate MD5 hash and return the first 24 chars of base64 encoding.""" + """ + Return the first 24 characters of the base64-encoded MD5 digest of `body`. + + Parameters: + body (bytes): Input bytes to hash. + + Returns: + str: First 24 characters of the base64-encoded MD5 digest. + """ md5_hash = hashlib.md5(body).digest() return base64.b64encode(md5_hash).decode("utf-8")[:24] @@ -97,7 +119,20 @@ def _build_global_string_to_sign( headers: dict[str, str], content_md5: str = "", ) -> str: - """Build the string to sign for HMAC-SHA256 Global API requests.""" + """ + Construct the canonical string used to compute the HMAC-SHA256 signature for a Global API request. + + The resulting newline-separated string contains, in order: HTTP method, Accept header, the provided content MD5 value, Content-Type header, Date header, all `x-ca-*` headers (each as `key:value` on its own line), and the request path. This canonical string is intended to be the message passed to the signing HMAC. + + Parameters: + method (str): HTTP method (e.g., "GET", "POST"). + path (str): Request path, including query string if applicable. + headers (dict[str, str]): Request headers; values for "accept", "content-type", "date", and any `x-ca-*` headers are used. + content_md5 (str): Base64-encoded MD5 of the request body when present, or an empty string if absent. + + Returns: + str: The canonical string to sign with HMAC-SHA256. + """ string_to_sign = [ method, headers.get("accept", ""), @@ -129,7 +164,17 @@ def _generate_global_signature( headers: dict[str, str], body: Optional[bytes] = None, ) -> str: - """Generate HMAC-SHA256 signature for Global API requests.""" + """ + Create the HMAC-SHA256 signature used for Global API requests. + + If a request body is provided, its MD5 (base64, truncated to 24 chars) is computed and inserted into headers["content-md5"] before signing. The function builds the canonical string-to-sign from method, path, and headers, then returns the base64-encoded HMAC-SHA256 of that string using app_secret as the key. + + Parameters: + headers (dict[str, str]): Request headers; this dict will be mutated to include "content-md5" when a body is provided. + + Returns: + str: The base64-encoded HMAC-SHA256 signature. + """ content_md5 = "" if body is not None: content_md5 = _global_md5_base64(body) @@ -157,7 +202,27 @@ def generate_global_header( id_token: Optional[str] = None, extra_headers: Optional[dict[str, str]] = None, ) -> dict[str, str]: - """Generate signed headers for Global app requests.""" + """ + Builds HTTP headers for a Global API request and signs them with HMAC-SHA256. + + Assembles standard headers (date, content-type, host, user-agent, x-ca-timestamp, x-ca-nonce, x-ca-key, etc.), conditionally includes Authorization/x-smart-id/Xs-Auth-Token when provided, merges any extra_headers, and computes the `x-ca-signature` header using the provided `app_secret`. + + Parameters: + method (str): HTTP method (e.g., "GET", "POST") used when computing the signature. + path (str): Request path (URI) used in the signature calculation. + host (str): Host header value for the request. + app_key (str): Application key inserted as `x-ca-key`. + app_secret (str): Secret used to compute the HMAC-SHA256 signature. + body (Optional[object]): Request body; if provided it will be converted to bytes and included in the signature computation. + content_type (str): Value for `content-type` and `accept` headers. Defaults to "application/json". + access_token (Optional[str]): If provided, added as `Authorization: Bearer `. + user_id (Optional[str]): If provided, added as `x-smart-id`. + id_token (Optional[str]): If provided, added as `Xs-Auth-Token` (and `Xs-App-Ver` is set). + extra_headers (Optional[dict[str, str]]): Additional headers to merge into the final header set. + + Returns: + dict[str, str]: A dictionary of HTTP headers ready to attach to the request, including the computed `x-ca-signature`. + """ timestamp = create_correct_timestamp() nonce = str(uuid.uuid4()) http_date = formatdate(timeval=None, localtime=False, usegmt=True) @@ -196,4 +261,4 @@ def generate_global_header( ) _LOGGER.debug("Constructed global request header for %s %s", method, path) - return headers + return headers \ No newline at end of file diff --git a/pysmarthashtag/cli.py b/pysmarthashtag/cli.py old mode 100755 new mode 100644 index 41c7427..ce1798e --- a/pysmarthashtag/cli.py +++ b/pysmarthashtag/cli.py @@ -24,7 +24,14 @@ def environ_or_required(key): def main_parser() -> argparse.ArgumentParser: - """Create argument parser.""" + """ + Create and return the CLI ArgumentParser and configure module logging from SMART_LOG_LEVEL. + + The parser is configured with subcommands: `status`, `info`, `watch` (with `-i` interval), `climate` (with `--vin`, `--temp`, `--active`), and `seatheating` (with `--vin`, `--level`, `--temp`, `--active`). Default authentication/region arguments are added via the module helper and the parser's default `func` is set to the command dispatcher. + + Returns: + argparse.ArgumentParser: A fully configured ArgumentParser for the CLI. + """ logging_config = { "version": 1, @@ -167,7 +174,13 @@ async def set_seatheating(args) -> None: def _add_default_args(parser: argparse.ArgumentParser): - """Add the default arguments username, password to the parser.""" + """ + Add standard CLI options for Smart account credentials and API region. + + Adds the following arguments to the provided ArgumentParser: + - `--username` and `--password`: use values from `SMART_USERNAME` / `SMART_PASSWORD` when present; marked required if the corresponding environment variable is not set. + - `--region`: selects the Smart API region with choices `eu`, `intl`, `global`; defaults to the `SMART_REGION` environment variable or `"eu"` when not set. + """ parser.add_argument("--username", help="Smart username", **environ_or_required("SMART_USERNAME")) parser.add_argument("--password", help="Smart password", **environ_or_required("SMART_PASSWORD")) parser.add_argument( @@ -179,16 +192,14 @@ def _add_default_args(parser: argparse.ArgumentParser): def _get_endpoint_urls_from_args(args) -> EndpointUrls: - """Get EndpointUrls based on region argument. - - Args: - ---- - args: Parsed command line arguments containing the region. - + """ + Return endpoint URLs for the SmartRegion specified in args.region. + + Parameters: + args: Parsed command-line arguments with a `region` attribute ('eu', 'intl', or 'global'). + Returns: - ------- - EndpointUrls configured for the specified region. - + EndpointUrls: Endpoint URLs configured for the specified region. """ region = SmartRegion(args.region) return get_endpoint_urls_for_region(region) @@ -203,4 +214,4 @@ def main(): if __name__ == "__main__": - main() + main() \ No newline at end of file diff --git a/pysmarthashtag/const.py b/pysmarthashtag/const.py index 0b4691b..cb82b1f 100644 --- a/pysmarthashtag/const.py +++ b/pysmarthashtag/const.py @@ -146,34 +146,65 @@ def get_api_base_url_v2(self) -> str: return self.api_base_url_v2 if self.api_base_url_v2 is not None else API_BASE_URL_V2 def get_ota_server_url(self) -> str: - """Get the OTA server URL, using the default if not set.""" + """ + Return the configured OTA server URL or the module default. + + Returns: + str: The OTA server URL; the configured `ota_server_url` when present, otherwise `OTA_SERVER_URL`. + """ return self.ota_server_url if self.ota_server_url is not None else OTA_SERVER_URL def get_oauth_base_url(self) -> str: - """Get the OAuth base URL, using the default if not set.""" + """ + Return the OAuth base URL to use for token requests. + + Returns: + The configured OAuth base URL string, or `EU_OAUTH_BASE_URL` if no override is set. + """ return self.oauth_base_url if self.oauth_base_url is not None else EU_OAUTH_BASE_URL def get_oauth_api_key(self) -> str: - """Get the OAuth API key, using the default if not set.""" + """ + Return the configured OAuth API key or the default if none is configured. + + Returns: + str: The configured OAuth API key if set, otherwise the module default `EU_OAUTH_API_KEY`. + """ return self.oauth_api_key if self.oauth_api_key is not None else EU_OAUTH_API_KEY def get_oauth_token_url(self) -> str: - """Get the OAuth token URL.""" + """ + Builds the OAuth token endpoint URL for the configured OAuth base URL. + + Returns: + The full token endpoint URL (the OAuth base URL with a single trailing '/token'). + """ return f"{self.get_oauth_base_url().rstrip('/')}/token" def get_global_app_key(self) -> str: - """Get the Global app key, using the default if not set.""" + """ + Return the configured global app key or the default. + + Returns: + global_app_key (str): The configured global app key if set, otherwise the module default GLOBAL_APP_KEY. + """ return self.global_app_key if self.global_app_key is not None else GLOBAL_APP_KEY def get_global_app_secret(self) -> str: - """Get the Global app secret, using the default if not set.""" + """ + Return the configured global app secret for this endpoint. + + Returns: + str: The global app secret set on this instance, or the module default `GLOBAL_APP_SECRET` if none is configured. + """ return self.global_app_secret if self.global_app_secret is not None else GLOBAL_APP_SECRET def _is_global_api_base_url(self, api_base_url: str) -> bool: - """Determine if the given API base URL should use GLOBAL_HMAC auth. - - This avoids substring checks by comparing against known-good base URLs, - normalizing away a trailing slash if present. + """ + Determine whether `api_base_url` matches the module's global API base URL (ignoring a trailing slash). + + Returns: + True if `api_base_url` equals `GLOBAL_API_BASE_URL` after removing a trailing slash, False otherwise. """ normalized = api_base_url.rstrip("/") # Default global API base URL, plus any additional explicit variants if needed. @@ -181,8 +212,13 @@ def _is_global_api_base_url(self, api_base_url: str) -> bool: return normalized == global_base def infer_auth_mode(self) -> SmartAuthMode: - """Infer authentication mode based on endpoint URLs.""" + """ + Selects the authentication mode to use based on the configured API base URL. + + Returns: + SmartAuthMode: `SmartAuthMode.GLOBAL_HMAC` if the configured API base URL matches the global API base URL, `SmartAuthMode.EU_OAUTH` otherwise. + """ api_base_url = self.get_api_base_url() if self._is_global_api_base_url(api_base_url): return SmartAuthMode.GLOBAL_HMAC - return SmartAuthMode.EU_OAUTH + return SmartAuthMode.EU_OAUTH \ No newline at end of file diff --git a/pysmarthashtag/tests/common.py b/pysmarthashtag/tests/common.py index 95cd1ff..66b3495 100644 --- a/pysmarthashtag/tests/common.py +++ b/pysmarthashtag/tests/common.py @@ -33,7 +33,18 @@ def __init__( # # # # # # # # # # # # # # # # # # # # # # # # def add_login_routes(self) -> None: - """Add routes for login.""" + """ + Register mocked HTTP routes for the Smart API login flow and related endpoints. + + This sets up a stateful sequence of mocked responses used by tests, including: + - initial server redirect and authentication context endpoints, + - login endpoint returning a test session token and intermediate redirect, + - OAuth token endpoint returning access/refresh/id tokens, + - SMART session and vehicle-related endpoints (for both API_BASE_URL and API_BASE_URL_V2) used to fetch car lists, select a car, retrieve vehicle status and SOC, and perform telematics/remote-control actions, + - OTA app info endpoints for test VINs. + + Responses use predefined JSON fixtures loaded from the test RESPONSE_DIR. + """ # Login context self.get(SERVER_URL).respond(302, headers={"location": load_response(RESPONSE_DIR / "auth_context.url")}) @@ -105,4 +116,4 @@ def add_login_routes(self) -> None: self.get(OTA_SERVER_URL + "app/info/TestVIN0000000002").respond( 200, json=load_response(RESPONSE_DIR / "ota_response.json"), - ) + ) \ No newline at end of file diff --git a/pysmarthashtag/vehicle/vehicle.py b/pysmarthashtag/vehicle/vehicle.py index 842af7f..178afd7 100644 --- a/pysmarthashtag/vehicle/vehicle.py +++ b/pysmarthashtag/vehicle/vehicle.py @@ -74,7 +74,18 @@ def __init__( charging_settings: Optional[dict] = None, fetched_at: Optional[datetime.datetime] = None, ) -> None: - """Initialize the vehicle.""" + """ + Create a SmartVehicle instance by storing the account and merging provided vehicle data, then derive series code and select the appropriate API base URL. + + Merges vehicle_base, vehicle_state, and charging_settings into the instance data (optionally recording fetched_at), ensures a `seriesCodeVs` value is present when derivable, and chooses the API base URL according to the account authentication mode and the vehicle series code. Logs vehicle initialization. + + Parameters: + account (SmartAccount): Account that owns the vehicle and provides configuration and endpoint URLs. + vehicle_base (dict): Base vehicle information retrieved from the primary API. + vehicle_state (Optional[dict]): Optional dynamic state payload to merge into the base data. + charging_settings (Optional[dict]): Optional charging-related settings to merge into the base data. + fetched_at (Optional[datetime.datetime]): Optional timestamp indicating when the provided data was fetched. + """ self.account = account self.data = {} self.combine_data(vehicle_base, vehicle_state, charging_settings, None, fetched_at) @@ -161,4 +172,4 @@ def _parse_data(self) -> None: self.engine_state = get_element_from_dict_maybe( self.data, "vehicleStatus", "basicVehicleStatus", "engineStatus" - ) + ) \ No newline at end of file