Add mypy & typeguard tests + fix dependencies + add some typing

.github/workflows/tests.yml

@@ -15,17 +15,17 @@ jobs:
         include:
           - { python-version: "3.11", os: ubuntu-latest, session: "pre-commit" }
           - { python-version: "3.11", os: ubuntu-latest, session: "safety" }
-          # - { python-version: "3.11", os: ubuntu-latest, session: "mypy" }
-          # - { python-version: "3.10", os: ubuntu-latest, session: "mypy" }
-          # - { python-version: "3.9", os: ubuntu-latest, session: "mypy" }
-          # - { python-version: "3.8", os: ubuntu-latest, session: "mypy" }
+          - { python-version: "3.11", os: ubuntu-latest, session: "mypy" }
+          - { python-version: "3.10", os: ubuntu-latest, session: "mypy" }
+          - { python-version: "3.9", os: ubuntu-latest, session: "mypy" }
+          - { python-version: "3.8", os: ubuntu-latest, session: "mypy" }
           - { python-version: "3.11", os: ubuntu-latest, session: "tests" }
           - { python-version: "3.10", os: ubuntu-latest, session: "tests" }
           - { python-version: "3.9", os: ubuntu-latest, session: "tests" }
           - { python-version: "3.8", os: ubuntu-latest, session: "tests" }
           - { python-version: "3.11", os: windows-latest, session: "tests" }
           - { python-version: "3.11", os: macos-latest, session: "tests" }
-          # - { python-version: "3.11", os: ubuntu-latest, session: "typeguard" }
+          - { python-version: "3.11", os: ubuntu-latest, session: "typeguard" }
           - { python-version: "3.11", os: ubuntu-latest, session: "xdoctest" }
           - { python-version: "3.11", os: ubuntu-latest, session: "docs-build" }
 

noxfile.py

@@ -25,9 +25,9 @@
 nox.options.sessions = (
     "pre-commit",
     "safety",
-    # "mypy",
+    "mypy",
     "tests",
-    # "typeguard",
+    "typeguard",
     "xdoctest",
     "docs-build",
 )
@@ -172,7 +172,7 @@ def docs_build(session: Session) -> None:
     """Build the documentation."""
     args = session.posargs or ["docs", "docs/_build"]
     session.install(".")
-    session.install("sphinx", "sphinx-click", "sphinx-rtd-theme")
+    session.install("sphinx", "sphinx-click")
 
     build_dir = Path("docs", "_build")
     if build_dir.exists():
@@ -186,7 +186,7 @@ def docs(session: Session) -> None:
     """Build and serve the documentation with live reloading on file changes."""
     args = session.posargs or ["--open-browser", "docs", "docs/_build"]
     session.install(".")
-    session.install("sphinx", "sphinx-autobuild", "sphinx-click", "sphinx-rtd-theme")
+    session.install("sphinx", "sphinx-autobuild", "sphinx-click")
 
     build_dir = Path("docs", "_build")
     if build_dir.exists():

pyproject.toml

@@ -29,7 +29,7 @@ Changelog = "https://github.com/hacf-fr/freebox-api/releases"
 
 [tool.poetry.dependencies]
 python = "^3.8.18"
-urllib3 = "^1.26.17"
+urllib3 = "^1.26.18"
 aiohttp = ">=3,<4"
 importlib-metadata = {version = ">=3.3,<5.0", python = "<3.12"}
 
@@ -39,7 +39,7 @@ coverage = {extras = ["toml"], version = "^7.2"}
 safety = "^2.3.5"
 mypy = "^1.6"
 typeguard = "^4.1.5"
-xdoctest = {extras = ["colors"], version = "^1.6.1"}
+xdoctest = {extras = ["colors"], version = "^1.1.1"}
 sphinx = "^4.2.0"
 sphinx-autobuild = "^2021.3.14"
 pre-commit = "^3.5.0"
@@ -75,6 +75,12 @@ pretty = true
 show_column_numbers = true
 show_error_codes = true
 show_error_context = true
+# TODO: Work on removing that
+allow_untyped_defs = true
+allow_untyped_calls = true
+exclude = [
+    '^tests/example\.py$',
+]
 
 [build-system]
 requires = ["poetry-core>=1.0.0"]

src/freebox_api/__init__.py

@@ -6,13 +6,13 @@
 # importlib.metadata available from Python 3.8 use importlib_metadata for
 # earlier versions.
 try:
-    from importlib.metadata import version, PackageNotFoundError  # type: ignore
+    from importlib.metadata import version, PackageNotFoundError
 except ImportError:  # pragma: no cover
     from importlib_metadata import version, PackageNotFoundError  # type: ignore
 
 
 try:
-    __version__ = version(__name__)
+    __version__: str = version(__name__)
 except PackageNotFoundError:  # pragma: no cover
     __version__ = "unknown"
 

src/freebox_api/access.py

@@ -1,8 +1,13 @@
 import hmac
 import json
 import logging
+from typing import Any
+from typing import Dict
+from typing import Optional
 from urllib.parse import urljoin
 
+from aiohttp import ClientSession
+
 from freebox_api.exceptions import AuthorizationError
 from freebox_api.exceptions import HttpRequestError
 from freebox_api.exceptions import InsufficientPermissionsError
@@ -11,30 +16,39 @@
 
 
 class Access:
-    def __init__(self, session, base_url, app_token, app_id, http_timeout):
+    def __init__(
+        self,
+        session: ClientSession,
+        base_url: str,
+        app_token: str,
+        app_id: str,
+        http_timeout: int,
+    ):
         self.session = session
         self.base_url = base_url
         self.app_token = app_token
         self.app_id = app_id
         self.timeout = http_timeout
-        self.session_token = None
-        self.session_permissions = None
+        self.session_token: Optional[str] = None
+        self.session_permissions: Optional[Dict[str, bool]] = None
 
     async def _get_challenge(self, base_url, timeout=10):
         """
         Return challenge from freebox API
         """
         url = urljoin(base_url, "login")
-        r = await self.session.get(url, timeout=timeout)
-        resp = await r.json()
+        resp = await self.session.get(url, timeout=timeout)
+        resp_data = await resp.json()
 
         # raise exception if resp.success != True
-        if not resp.get("success"):
+        if not resp_data.get("success"):
             raise AuthorizationError(
-                "Getting challenge failed (APIResponse: {})".format(json.dumps(resp))
+                "Getting challenge failed (APIResponse: {})".format(
+                    json.dumps(resp_data)
+                )
             )
 
-        return resp["result"]["challenge"]
+        return resp_data["result"]["challenge"]
 
     async def _get_session_token(self, base_url, app_token, app_id, timeout=10):
         """
@@ -50,17 +64,19 @@
 
         url = urljoin(base_url, "login/session/")
         data = json.dumps({"app_id": app_id, "password": password})
-        r = await self.session.post(url, data=data, timeout=timeout)
-        resp = await r.json()
+        resp = await self.session.post(url, data=data, timeout=timeout)
+        resp_data = await resp.json()
 
         # raise exception if resp.success != True
-        if not resp.get("success"):
+        if not resp_data.get("success"):
             raise AuthorizationError(
-                "Starting session failed (APIResponse: {})".format(json.dumps(resp))
+                "Starting session failed (APIResponse: {})".format(
+                    json.dumps(resp_data)
+                )
             )
 
-        session_token = resp.get("result").get("session_token")
-        session_permissions = resp.get("result").get("permissions")
+        session_token = resp_data["result"].get("session_token")
+        session_permissions = resp_data["result"].get("permissions")
 
         return (session_token, session_permissions)
 
@@ -75,7 +91,7 @@
         self.session_token = session_token
         self.session_permissions = session_permissions
 
-    def _get_headers(self):
+    def _get_headers(self) -> Dict[str, Optional[str]]:
         return {"X-Fbx-App-Auth": self.session_token}
 
     async def _perform_request(self, verb, end_url, **kwargs):
@@ -91,58 +107,64 @@
             "headers": self._get_headers(),
             "timeout": self.timeout,
         }
-        r = await verb(url, **request_params)
+        resp = await verb(url, **request_params)
 
         # Return response if content is not json
-        if r.content_type != "application/json":
-            return r
-        else:
-            resp = await r.json()
+        if resp.content_type != "application/json":
+            return resp
 
-            if resp.get("error_code") in ["auth_required", "invalid_session"]:
-                logger.debug("Invalid session")
-                await self._refresh_session_token()
-                request_params["headers"] = self._get_headers()
-                r = await verb(url, **request_params)
-                resp = await r.json()
+        resp_data = await resp.json()
+        if resp_data.get("error_code") in ["auth_required", "invalid_session"]:
+            logger.debug("Invalid session")
+            await self._refresh_session_token()
+            request_params["headers"] = self._get_headers()
+            resp = await verb(url, **request_params)
+            resp_data = await resp.json()
 
-            if not resp["success"]:
-                err_msg = "Request failed (APIResponse: {})".format(json.dumps(resp))
-                if resp.get("error_code") == "insufficient_rights":
-                    raise InsufficientPermissionsError(err_msg)
-                else:
-                    raise HttpRequestError(err_msg)
+        if not resp_data["success"]:
+            err_msg = "Request failed (APIResponse: {})".format(json.dumps(resp_data))
+            if resp_data.get("error_code") == "insufficient_rights":
+                raise InsufficientPermissionsError(err_msg)
+            raise HttpRequestError(err_msg)
 
-            return resp["result"] if "result" in resp else None
+        return resp_data.get("result")
 
-    async def get(self, end_url):
+    async def get(
+        self, end_url: str
+    ) -> Any:  # Union[Dict[str, Any], List[Dict[str, Any]]]:
         """
         Send get request and return results
         """
         return await self._perform_request(self.session.get, end_url)
 
-    async def post(self, end_url, payload=None):
+    async def post(
+        self, end_url: str, payload: Optional[Dict[str, Any]] = None
+    ) -> Dict[str, Any]:
         """
         Send post request and return results
         """
-        data = json.dumps(payload) if payload is not None else None
-        return await self._perform_request(self.session.post, end_url, data=data)
+        data = json.dumps(payload) if payload else None
+        return await self._perform_request(self.session.post, end_url, data=data)  # type: ignore
 
-    async def put(self, end_url, payload=None):
+    async def put(
+        self, end_url: str, payload: Optional[Dict[str, Any]] = None
+    ) -> Dict[str, Any]:
         """
         Send post request and return results
         """
-        data = json.dumps(payload) if payload is not None else None
-        return await self._perform_request(self.session.put, end_url, data=data)
+        data = json.dumps(payload) if payload else None
+        return await self._perform_request(self.session.put, end_url, data=data)  # type: ignore
 
-    async def delete(self, end_url, payload=None):
+    async def delete(
+        self, end_url: str, payload: Optional[Dict[str, Any]] = None
+    ) -> Optional[Dict[str, bool]]:
         """
         Send delete request and return results
         """
-        data = json.dumps(payload) if payload is not None else None
-        return await self._perform_request(self.session.delete, end_url, data=data)
+        data = json.dumps(payload) if payload else None
+        return await self._perform_request(self.session.delete, end_url, data=data)  # type: ignore
 
-    async def get_permissions(self):
+    async def get_permissions(self) -> Optional[Dict[str, bool]]:
         """
         Returns the permissions for this session/app.
         """