Adding mypy

Author: CoMPaTech

.github/workflows/verify.yml

@@ -106,7 +106,6 @@ jobs:
           include-hidden-files: true
 
   mypy:
-    if: false  # disables the job --> "Code is not up to par for mypy, skipping"
     runs-on: ubuntu-latest
     name: Run mypy
     needs:
@@ -135,7 +134,7 @@ jobs:
     needs:
       - ruff
       - pytest
-      # - mypy
+      - mypy
     steps:
       - name: Check out committed code
         uses: actions/checkout@v4

.pre-commit-config.yaml

@@ -80,3 +80,12 @@ repos:
     rev: v0.45.0
     hooks:
       - id: markdownlint
+  - repo: local
+    hooks:
+      - id: mypy
+        name: mypy
+        entry: script/run-in-env.sh mypy
+        language: script
+        require_serial: true
+        types_or: [python, pyi]
+        files: ^(airos|tests|scripts)/.+\.(py|pyi)$

airos/airos8.py

@@ -54,7 +54,7 @@ def __init__(
         self._status_cgi_url = f"{self.base_url}/status.cgi"  # AirOS 8
         self._stakick_cgi_url = f"{self.base_url}/stakick.cgi"  # AirOS 8
         self._provmode_url = f"{self.base_url}/api/provmode"  # AirOS 8
-        self.current_csrf_token = None
+        self.current_csrf_token: str | None = None
 
         self._use_json_for_login_post = False
 
@@ -87,7 +87,7 @@ async def login(self) -> bool:
 
         login_request_headers = {**self._common_headers}
 
-        post_data = None
+        post_data: dict[str, str] | str | None = None
         if self._use_json_for_login_post:
             login_request_headers["Content-Type"] = "application/json"
             post_data = json.dumps(login_payload)
@@ -114,7 +114,7 @@ async def login(self) -> bool:
                         # If the AIROS_ cookie was parsed but isn't automatically added to the jar, add it manually
                         if (
                             morsel.key.startswith("AIROS_")
-                            and morsel.key not in self.session.cookie_jar
+                            and morsel.key not in self.session.cookie_jar  # type: ignore[operator]
                         ):
                             # `SimpleCookie`'s Morsel objects are designed to be compatible with cookie jars.
                             # We need to set the domain if it's missing, otherwise the cookie might not be sent.
@@ -152,7 +152,7 @@ async def login(self) -> bool:
                 if new_csrf_token:
                     self.current_csrf_token = new_csrf_token
                 else:
-                    return
+                    return False
 
                 # Re-check cookies in self.session.cookie_jar AFTER potential manual injection
                 airos_cookie_found = False
@@ -186,18 +186,16 @@ async def login(self) -> bool:
                     log = f"Login failed with status {response.status}. Full Response: {response.text}"
                     _LOGGER.error(log)
                     raise AirOSConnectionAuthenticationError from None
-        except (TimeoutError, aiohttp.client_exceptions.ClientError) as err:
+        except (TimeoutError, aiohttp.ClientError) as err:
             _LOGGER.exception("Error during login")
             raise AirOSDeviceConnectionError from err
         except asyncio.CancelledError:
             _LOGGER.info("Login task was cancelled")
             raise
 
-    def derived_data(
-        self, response: dict[str, Any] | None = None
-    ) -> dict[str, Any] | None:
+    def derived_data(self, response: dict[str, Any] = {}) -> dict[str, Any]:
         """Add derived data to the device response."""
-        derived = {
+        derived: dict[str, Any] = {
             "station": False,
             "access_point": False,
             "ptp": False,
@@ -302,14 +300,14 @@ async def status(self) -> AirOSData:
                         response_text,
                     )
                     raise AirOSDeviceConnectionError
-        except (TimeoutError, aiohttp.client_exceptions.ClientError) as err:
+        except (TimeoutError, aiohttp.ClientError) as err:
             _LOGGER.exception("Status API call failed: %s", err)
             raise AirOSDeviceConnectionError from err
         except asyncio.CancelledError:
             _LOGGER.info("API status retrieval task was cancelled")
             raise
 
-    async def stakick(self, mac_address: str = None) -> bool:
+    async def stakick(self, mac_address: str | None = None) -> bool:
         """Reconnect client station."""
         if not self.connected:
             _LOGGER.error("Not connected, login first")
@@ -340,7 +338,7 @@ async def stakick(self, mac_address: str = None) -> bool:
                 log = f"Unable to restart connection response status {response.status} with {response_text}"
                 _LOGGER.error(log)
                 return False
-        except (TimeoutError, aiohttp.client_exceptions.ClientError) as err:
+        except (TimeoutError, aiohttp.ClientError) as err:
             _LOGGER.exception("Error during call to reconnect remote: %s", err)
             raise AirOSDeviceConnectionError from err
         except asyncio.CancelledError:
@@ -379,7 +377,7 @@ async def provmode(self, active: bool = False) -> bool:
                 log = f"Unable to change provisioning mode response status {response.status} with {response_text}"
                 _LOGGER.error(log)
                 return False
-        except (TimeoutError, aiohttp.client_exceptions.ClientError) as err:
+        except (TimeoutError, aiohttp.ClientError) as err:
             _LOGGER.exception("Error during call to change provisioning mode: %s", err)
             raise AirOSDeviceConnectionError from err
         except asyncio.CancelledError:

airos/data.py

@@ -39,7 +39,7 @@ def is_ip_address(value: str) -> bool:
         return False
 
 
-def redact_data_smart(data: dict) -> dict:
+def redact_data_smart(data: dict[str, Any]) -> dict[str, Any]:
     """Recursively redacts sensitive keys in a dictionary."""
     sensitive_keys = {
         "hostname",
@@ -56,10 +56,7 @@ def redact_data_smart(data: dict) -> dict:
         "platform",
     }
 
-    def _redact(d: dict):
-        if not isinstance(d, dict):
-            return d
-
+    def _redact(d: dict[str, Any]) -> dict[str, Any]:
         redacted_d = {}
         for k, v in d.items():
             if k in sensitive_keys:
@@ -73,15 +70,15 @@ def _redact(d: dict):
                     isinstance(i, str) and is_ip_address(i) for i in v
                 ):
                     # Redact list of IPs to a dummy list
-                    redacted_d[k] = ["127.0.0.3"]
+                    redacted_d[k] = ["127.0.0.3"]  # type: ignore[assignment]
                 else:
                     redacted_d[k] = "REDACTED"
             elif isinstance(v, dict):
-                redacted_d[k] = _redact(v)
+                redacted_d[k] = _redact(v)  # type: ignore[assignment]
             elif isinstance(v, list):
                 redacted_d[k] = [
                     _redact(item) if isinstance(item, dict) else item for item in v
-                ]
+                ]  # type: ignore[assignment]
             else:
                 redacted_d[k] = v
         return redacted_d

airos/discovery.py

@@ -29,7 +29,7 @@ class AirOSDiscoveryProtocol(asyncio.DatagramProtocol):
 
     """
 
-    def __init__(self, callback: Callable[[dict[str, Any]], None]) -> None:
+    def __init__(self, callback: Callable[[dict[str, Any]], Any]) -> None:
         """Initialize AirOSDiscoveryProtocol.
 
         Args:
@@ -43,7 +43,7 @@ def __init__(self, callback: Callable[[dict[str, Any]], None]) -> None:
     def connection_made(self, transport: asyncio.BaseTransport) -> None:
         """Set up the UDP socket for broadcasting and reusing the address."""
         self.transport = transport  # type: ignore[assignment] # transport is DatagramTransport
-        sock: socket.socket = self.transport.get_extra_info("socket")
+        sock: socket.socket = transport.get_extra_info("socket")
         sock.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1)
         sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
         log = f"AirOS discovery listener (low-level) started on UDP port {DISCOVERY_PORT}."

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name            = "airos"
-version         = "0.2.8"
+version         = "0.2.9a0"
 license         = "MIT"
 description     = "Ubiquity airOS module(s) for Python 3."
 readme          = "README.md"
@@ -396,6 +396,12 @@ warn_return_any = true
 warn_unreachable = true
 exclude = []
 
+[[tool.mypy.overrides]]
+module = "tests.*"
+ignore_missing_imports = true # You'll likely need this for test-only dependencies
+disallow_untyped_decorators = false # The fix for your current errors
+check_untyped_defs = false 
+
 [tool.coverage.run]
 source = [ "airos" ]
 omit= [

requirements-test.txt

@@ -7,3 +7,4 @@ aioresponses
 aioresponses==0.7.8
 aiofiles==24.1.0
 radon==6.0.1
+types-aiofiles==24.1.0.20250809

script/generate_ha_fixture.py

@@ -15,10 +15,11 @@
 
 # NOTE: This assumes the airos module is correctly installed or available in the project path.
 # If not, you might need to adjust the import statement.
-from airos.airos8 import AirOS, AirOSData  # noqa: E402
+from airos.airos8 import AirOS  # noqa: E402
+from airos.data import AirOS8Data as AirOSData  # noqa: E402
 
 
-def generate_airos_fixtures():
+def generate_airos_fixtures() -> None:
     """Process all (intended) JSON files from the userdata directory to potential fixtures."""
 
     # Define the paths to the directories
@@ -44,7 +45,7 @@ def generate_airos_fixtures():
                 with open(base_fixture_path) as source:
                     source_data = json.loads(source.read())
 
-                derived_data = AirOS.derived_data(None, source_data)
+                derived_data = AirOS.derived_data(None, source_data)  # type: ignore[arg-type]
                 new_data = AirOSData.from_dict(derived_data)
 
                 with open(new_fixture_path, "w") as new:

script/mashumaro-step-debug.py

@@ -18,7 +18,7 @@
 _LOGGER = logging.getLogger(__name__)
 
 
-def main():
+def main() -> None:
     """Debug data."""
     if len(sys.argv) <= 1:
         _LOGGER.info("Use with file to check")

script/run-in-env.sh

@@ -0,0 +1,32 @@
+#!/usr/bin/env sh
+set -eu
+
+# Used in venv activate script.
+# Would be an error if undefined.
+OSTYPE="${OSTYPE-}"
+
+# Activate pyenv and virtualenv if present, then run the specified command
+
+# pyenv, pyenv-virtualenv
+if [ -s .python-version ]; then
+    PYENV_VERSION=$(head -n 1 .python-version)
+    export PYENV_VERSION
+fi
+
+# shellcheck source=/dev/null
+if [ -n "${VIRTUAL_ENV-}" ] && [ -f "${VIRTUAL_ENV}/bin/activate" ]; then
+  # shellcheck source=/dev/null
+  . "${VIRTUAL_ENV}/bin/activate"
+else
+  # other common virtualenvs
+  my_path=$(git rev-parse --show-toplevel)
+
+  for venv in venv .venv .; do
+    if [ -f "${my_path}/${venv}/bin/activate" ]; then
+      . "${my_path}/${venv}/bin/activate"
+      break
+    fi
+  done
+fi
+
+exec "$@"