From f54b7b2dbfd36ff0e9e9198b0be34043c860b0f5 Mon Sep 17 00:00:00 2001 From: Andre Basche Date: Sat, 15 Apr 2023 15:55:22 +0200 Subject: [PATCH] Add mypy checks --- .github/workflows/python-check.yml | 6 ++- pyhon/appliance.py | 26 +++++++----- pyhon/connection/auth.py | 64 +++++++++++++++--------------- pyhon/connection/handler/base.py | 10 ++--- pyhon/connection/handler/hon.py | 24 +++++------ pyhon/hon.py | 6 +-- 6 files changed, 73 insertions(+), 63 deletions(-) diff --git a/.github/workflows/python-check.yml b/.github/workflows/python-check.yml index d0d9b7a..6731bd8 100644 --- a/.github/workflows/python-check.yml +++ b/.github/workflows/python-check.yml @@ -25,12 +25,16 @@ jobs: run: | python -m pip install --upgrade pip python -m pip install -r requirements.txt - python -m pip install flake8 pylint black + python -m pip install flake8 pylint black mypy - name: Lint with flake8 run: | # stop the build if there are Python syntax errors or undefined names flake8 . --count --select=E9,F63,F7,F82 --show-source --statistics flake8 . --count --exit-zero --max-complexity=10 --max-line-length=88 --statistics + - name: Type check with mypy + run: | + # stop the build if there are Python syntax errors or undefined names + mypy pyhon/ # - name: Analysing the code with pylint # run: | # pylint --max-line-length 88 $(git ls-files '*.py') diff --git a/pyhon/appliance.py b/pyhon/appliance.py index 693e9c7..16d2921 100644 --- a/pyhon/appliance.py +++ b/pyhon/appliance.py @@ -1,23 +1,29 @@ import importlib from contextlib import suppress -from typing import Optional, Dict +from typing import Optional, Dict, Any +from typing import TYPE_CHECKING from pyhon import helper from pyhon.commands import HonCommand from pyhon.parameter import HonParameterFixed +if TYPE_CHECKING: + from pyhon import HonAPI + class HonAppliance: - def __init__(self, api, info: Dict, zone: int = 0) -> None: + def __init__( + self, api: Optional["HonAPI"], info: Dict[str, Any], zone: int = 0 + ) -> None: if attributes := info.get("attributes"): info["attributes"] = {v["parName"]: v["parValue"] for v in attributes} - self._info = info - self._api = api - self._appliance_model = {} + self._info: Dict = info + self._api: Optional[HonAPI] = api + self._appliance_model: Dict = {} - self._commands = {} - self._statistics = {} - self._attributes = {} + self._commands: Dict = {} + self._statistics: Dict = {} + self._attributes: Dict = {} self._zone = zone try: @@ -58,11 +64,11 @@ class HonAppliance: @property def appliance_model_id(self) -> str: - return self._info.get("applianceModelId") + return self._info.get("applianceModelId", "") @property def appliance_type(self) -> str: - return self._info.get("applianceTypeName") + return self._info.get("applianceTypeName", "") @property def mac_address(self) -> str: diff --git a/pyhon/connection/auth.py b/pyhon/connection/auth.py index 9c5802a..a48c054 100644 --- a/pyhon/connection/auth.py +++ b/pyhon/connection/auth.py @@ -3,10 +3,11 @@ import logging import re import secrets import urllib +from contextlib import suppress from dataclasses import dataclass from datetime import datetime, timedelta from pprint import pformat -from typing import Dict, Optional +from typing import Dict, Optional, List from urllib import parse from urllib.parse import quote @@ -82,14 +83,15 @@ class HonAuth: if fail: raise exceptions.HonAuthenticationError("Can't login") - def _generate_nonce(self) -> str: + @staticmethod + def _generate_nonce() -> str: nonce = secrets.token_hex(16) return f"{nonce[:8]}-{nonce[8:12]}-{nonce[12:16]}-{nonce[16:20]}-{nonce[20:]}" - async def _load_login(self): + async def _load_login(self) -> bool: login_url = await self._introduce() login_url = await self._handle_redirects(login_url) - await self._login_url(login_url) + return await self._login_url(login_url) async def _introduce(self) -> str: redirect_uri = urllib.parse.quote(f"{const.APP}://mobilesdk/detect/oauth/done") @@ -101,8 +103,8 @@ class HonAuth: "scope": "api openid refresh_token web", "nonce": self._generate_nonce(), } - params = "&".join([f"{k}={v}" for k, v in params.items()]) - url = f"{const.AUTH_API}/services/oauth2/authorize/expid_Login?{params}" + params_encode = "&".join([f"{k}={v}" for k, v in params.items()]) + url = f"{const.AUTH_API}/services/oauth2/authorize/expid_Login?{params_encode}" async with self._request.get(url) as response: text = await response.text() self._expires = datetime.utcnow() @@ -115,7 +117,7 @@ class HonAuth: async def _manual_redirect(self, url: str) -> str: async with self._request.get(url, allow_redirects=False) as response: - if not (new_location := response.headers.get("Location")): + if not (new_location := response.headers.get("Location", "")): await self._error_logger(response) return new_location @@ -138,11 +140,11 @@ class HonAuth: ) return True await self._error_logger(response) + return False - async def _login(self): - start_url = parse.unquote(self._login_data.url.split("startURL=")[-1]).split( - "%3D" - )[0] + async def _login(self) -> str: + start_url = self._login_data.url.rsplit("startURL=", maxsplit=1)[-1] + start_url = parse.unquote(start_url).split("%3D")[0] action = { "id": "79;a", "descriptor": "apex://LightningLoginCustomController/ACTION$login", @@ -175,19 +177,13 @@ class HonAuth: params=params, ) as response: if response.status == 200: - try: - data = await response.json() - return data["events"][0]["attributes"]["values"]["url"] - except json.JSONDecodeError: - pass - except KeyError: - _LOGGER.error( - "Can't get login url - %s", pformat(await response.json()) - ) + with suppress(json.JSONDecodeError, KeyError): + result = await response.json() + return result["events"][0]["attributes"]["values"]["url"] await self._error_logger(response) return "" - def _parse_token_data(self, text): + def _parse_token_data(self, text: str) -> None: if access_token := re.findall("access_token=(.*?)&", text): self._access_token = access_token[0] if refresh_token := re.findall("refresh_token=(.*?)&", text): @@ -195,22 +191,26 @@ class HonAuth: if id_token := re.findall("id_token=(.*?)&", text): self._id_token = id_token[0] - async def _get_token(self, url): + async def _get_token(self, url: str) -> bool: async with self._request.get(url) as response: if response.status != 200: await self._error_logger(response) return False - url = re.findall("href\\s*=\\s*[\"'](.+?)[\"']", await response.text()) - if not url: + url_search = re.findall( + "href\\s*=\\s*[\"'](.+?)[\"']", await response.text() + ) + if not url_search: await self._error_logger(response) return False - if "ProgressiveLogin" in url[0]: - async with self._request.get(url[0]) as response: + if "ProgressiveLogin" in url_search[0]: + async with self._request.get(url_search[0]) as response: if response.status != 200: await self._error_logger(response) return False - url = re.findall("href\\s*=\\s*[\"'](.*?)[\"']", await response.text()) - url = "/".join(const.AUTH_API.split("/")[:-1]) + url[0] + url_search = re.findall( + "href\\s*=\\s*[\"'](.*?)[\"']", await response.text() + ) + url = "/".join(const.AUTH_API.split("/")[:-1]) + url_search[0] async with self._request.get(url) as response: if response.status != 200: await self._error_logger(response) @@ -218,7 +218,7 @@ class HonAuth: self._parse_token_data(await response.text()) return True - async def _api_auth(self): + async def _api_auth(self) -> bool: post_headers = {"id-token": self._id_token} data = self._device.get() async with self._request.post( @@ -232,7 +232,7 @@ class HonAuth: self._cognito_token = json_data["cognitoUser"]["Token"] return True - async def authenticate(self): + async def authenticate(self) -> None: self.clear() try: if not await self._load_login(): @@ -246,7 +246,7 @@ class HonAuth: except exceptions.HonNoAuthenticationNeeded: return - async def refresh(self): + async def refresh(self) -> bool: params = { "client_id": const.CLIENT_ID, "refresh_token": self._refresh_token, @@ -264,7 +264,7 @@ class HonAuth: self._access_token = data["access_token"] return await self._api_auth() - def clear(self): + def clear(self) -> None: self._session.cookie_jar.clear_domain(const.AUTH_API.split("/")[-2]) self._request.called_urls = [] self._cognito_token = "" diff --git a/pyhon/connection/handler/base.py b/pyhon/connection/handler/base.py index 7542df6..ad77ebe 100644 --- a/pyhon/connection/handler/base.py +++ b/pyhon/connection/handler/base.py @@ -1,7 +1,7 @@ import logging from collections.abc import AsyncIterator from contextlib import asynccontextmanager -from typing import Optional, Callable, Dict +from typing import Optional, Callable, Dict, Any import aiohttp from typing_extensions import Self @@ -37,18 +37,18 @@ class ConnectionHandler: raise NotImplementedError @asynccontextmanager - async def get(self, *args, **kwargs) -> AsyncIterator[Callable]: + async def get(self, *args, **kwargs) -> AsyncIterator[aiohttp.ClientResponse]: if self._session is None: raise exceptions.NoSessionException() - response: Callable + response: aiohttp.ClientResponse async with self._intercept(self._session.get, *args, **kwargs) as response: yield response @asynccontextmanager - async def post(self, *args, **kwargs) -> AsyncIterator[Callable]: + async def post(self, *args, **kwargs) -> AsyncIterator[aiohttp.ClientResponse]: if self._session is None: raise exceptions.NoSessionException() - response: Callable + response: aiohttp.ClientResponse async with self._intercept(self._session.post, *args, **kwargs) as response: yield response diff --git a/pyhon/connection/handler/hon.py b/pyhon/connection/handler/hon.py index 7da82a8..f50c3a9 100644 --- a/pyhon/connection/handler/hon.py +++ b/pyhon/connection/handler/hon.py @@ -10,7 +10,7 @@ from typing_extensions import Self from pyhon.connection.auth import HonAuth from pyhon.connection.device import HonDevice from pyhon.connection.handler.base import ConnectionHandler -from pyhon.exceptions import HonAuthenticationError +from pyhon.exceptions import HonAuthenticationError, NoAuthenticationException _LOGGER = logging.getLogger(__name__) @@ -30,7 +30,9 @@ class HonConnectionHandler(ConnectionHandler): self._auth: Optional[HonAuth] = None @property - def auth(self) -> Optional[HonAuth]: + def auth(self) -> HonAuth: + if self._auth is None: + raise NoAuthenticationException() return self._auth @property @@ -39,16 +41,14 @@ class HonConnectionHandler(ConnectionHandler): async def create(self) -> Self: await super().create() - self._auth: HonAuth = HonAuth( - self._session, self._email, self._password, self._device - ) + self._auth = HonAuth(self._session, self._email, self._password, self._device) return self async def _check_headers(self, headers: Dict) -> Dict: - if not (self._auth.cognito_token and self._auth.id_token): - await self._auth.authenticate() - headers["cognito-token"] = self._auth.cognito_token - headers["id-token"] = self._auth.id_token + if not (self.auth.cognito_token and self.auth.id_token): + await self.auth.authenticate() + headers["cognito-token"] = self.auth.cognito_token + headers["id-token"] = self.auth.id_token return self._HEADERS | headers @asynccontextmanager @@ -58,16 +58,16 @@ class HonConnectionHandler(ConnectionHandler): kwargs["headers"] = await self._check_headers(kwargs.get("headers", {})) async with method(*args, **kwargs) as response: if ( - self._auth.token_expires_soon or response.status in [401, 403] + self.auth.token_expires_soon or response.status in [401, 403] ) and loop == 0: _LOGGER.info("Try refreshing token...") - await self._auth.refresh() + await self.auth.refresh() async with self._intercept( method, *args, loop=loop + 1, **kwargs ) as result: yield result elif ( - self._auth.token_is_expired or response.status in [401, 403] + self.auth.token_is_expired or response.status in [401, 403] ) and loop == 1: _LOGGER.warning( "%s - Error %s - %s", diff --git a/pyhon/hon.py b/pyhon/hon.py index 327f85e..9be1d78 100644 --- a/pyhon/hon.py +++ b/pyhon/hon.py @@ -1,5 +1,5 @@ import asyncio -from typing import List, Optional, Dict +from typing import List, Optional, Dict, Any from typing_extensions import Self from aiohttp import ClientSession @@ -39,8 +39,8 @@ class Hon: def appliances(self) -> List[HonAppliance]: return self._appliances - async def _create_appliance(self, appliance: Dict, zone=0) -> None: - appliance = HonAppliance(self._api, appliance, zone=zone) + async def _create_appliance(self, appliance_data: Dict[str, Any], zone=0) -> None: + appliance = HonAppliance(self._api, appliance_data, zone=zone) if appliance.mac_address is None: return await asyncio.gather(