Add mypy and more flake checks

.pre-commit-config.yaml

@@ -14,9 +14,23 @@ repos:
     hooks:
       - id: flake8
         entry: pflake8
-        additional_dependencies: ['pyproject-flake8==0.0.1a2']
+        additional_dependencies: 
+          - pyproject-flake8==0.0.1a2
+          - flake8-bugbear==22.1.11
+          - flake8-comprehensions==3.8.0
+          - flake8_2020==1.6.1
+          - mccabe==0.6.1
+          - pycodestyle==2.8.0
+          - pyflakes==2.4.0
 
   - repo: https://github.com/PyCQA/isort
     rev: 5.10.1
     hooks:
       - id: isort
+
+  - repo: https://github.com/pre-commit/mirrors-mypy
+    rev: v0.931
+    hooks:
+      - id: mypy
+        additional_dependencies:
+          - zigpy==0.43.0

setup.cfg

@@ -39,3 +39,23 @@ testing =
 
 [coverage:run]
 source = zigpy_znp
+
+[flake8]
+max-line-length = 88
+
+[mypy]
+ignore_missing_imports = True
+install_types = True
+non_interactive = True
+check_untyped_defs = True
+show_error_codes = True
+show_error_context = True
+disable_error_code =
+   attr-defined,
+   arg-type,
+   type-var,
+   var-annotated,
+   assignment,
+   call-overload,
+   name-defined,
+   union-attr

tests/conftest.py

@@ -14,7 +14,7 @@
     # Python 3.8 already has this
     from unittest.mock import AsyncMock as CoroutineMock  # noqa: F401
 except ImportError:
-    from asynctest import CoroutineMock  # noqa: F401
+    from asynctest import CoroutineMock  # type:ignore[no-redef]  # noqa: F401
 
 import zigpy.endpoint
 import zigpy.zdo.types as zdo_t
@@ -69,8 +69,9 @@ def write(self, data):
         assert self._is_connected
         self.protocol.data_received(data)
 
-    def close(self, *, error=ValueError("Connection was closed")):
+    def close(self, *, error=ValueError("Connection was closed")):  # noqa: B008
         LOGGER.debug("Closing %s", self)
+
         if not self._is_connected:
             return
 

tests/test_commands.py

@@ -137,7 +137,7 @@ def test_commands_schema():
 
                     commands_by_id[cmd.Req.header].append(cmd.Req)
                 else:
-                    assert False, "Command is empty"
+                    assert False, "Command is empty"  # noqa: B011
             elif cmd.type == t.CommandType.SRSP:
                 # The one command like this is RPCError
                 assert cmd is c.RPCError.CommandNotRecognized
@@ -153,7 +153,7 @@ def test_commands_schema():
 
                 commands_by_id[cmd.Rsp.header].append(cmd.Rsp)
             else:
-                assert False, "Command has unknown type"
+                assert False, "Command has unknown type"  # noqa: B011
 
     duplicate_commands = {
         cmd: commands for cmd, commands in commands_by_id.items() if len(commands) > 1

tests/test_types_cstruct.py

@@ -1,7 +1,12 @@
+import typing
+
 import pytest
 
 import zigpy_znp.types as t
 
+if typing.TYPE_CHECKING:
+    import typing_extensions
+
 
 def test_struct_fields():
     class TestStruct(t.CStruct):
@@ -266,7 +271,7 @@ class TestStruct(t.CStruct):
 
 
 def test_old_nib_deserialize():
-    PaddingByte = t.uint8_t
+    PaddingByte: typing_extensions.TypeAlias = t.uint8_t
 
     class NwkState16(t.enum_uint16):
         NWK_INIT = 0
@@ -330,11 +335,11 @@ class OldNIB(t.CStruct):
         nwkConcentratorDiscoveryTime: t.uint8_t
         nwkConcentratorRadius: t.uint8_t
         nwkAllFresh: t.uint8_t
-        PaddingByte3: PaddingByte
+        PaddingByte3: PaddingByte  # type:ignore[valid-type]
         nwkManagerAddr: t.NWK
         nwkTotalTransmissions: t.uint16_t
         nwkUpdateId: t.uint8_t
-        PaddingByte4: PaddingByte
+        PaddingByte4: PaddingByte  # type:ignore[valid-type]
 
     nib = t.NIB(
         SequenceNum=54,

tests/test_uart.py

@@ -218,7 +218,7 @@ def test_uart_frame_received_error(connected_uart, mocker):
     uart.data_received(test_frame_bytes * 3)
 
     # We should have received all three frames
-    znp.frame_received.call_count == 3
+    assert znp.frame_received.call_count == 3
 
 
 async def test_connection_lost(dummy_serial_conn, mocker, event_loop):

zigpy_znp/api.py

@@ -2,6 +2,7 @@
 
 import os
 import time
+import typing
 import asyncio
 import logging
 import itertools
@@ -30,6 +31,9 @@
 from zigpy_znp.exceptions import CommandNotRecognized, InvalidCommandResponse
 from zigpy_znp.types.nvids import ExNvIds, OsalNvIds
 
+if typing.TYPE_CHECKING:
+    import typing_extensions
+
 LOGGER = logging.getLogger(__name__)
 
 
@@ -50,8 +54,8 @@ def __init__(self, config: conf.ConfigType):
         self._listeners = defaultdict(list)
         self._sync_request_lock = asyncio.Lock()
 
-        self.capabilities = None
-        self.version = None
+        self.capabilities = None  # type: int
+        self.version = None  # type: float
 
         self.nvram = NVRAMHelper(self)
         self.network_info: zigpy.state.NetworkInformation = None
@@ -542,7 +546,7 @@ async def ping_task():
 
                 try:
                     async with async_timeout.timeout(CONNECT_PING_TIMEOUT):
-                        result = await ping_task
+                        result = await ping_task  # type:ignore[misc]
                 except asyncio.TimeoutError:
                     ping_task.cancel()
 
@@ -609,7 +613,7 @@ def close(self) -> None:
 
         self._app = None
 
-        for header, listeners in self._listeners.items():
+        for _header, listeners in self._listeners.items():
             for listener in listeners:
                 listener.cancel()
 
@@ -659,7 +663,7 @@ def remove_listener(self, listener: BaseResponseListener) -> None:
             counts[OneShotResponseListener],
         )
 
-    def frame_received(self, frame: GeneralFrame) -> bool:
+    def frame_received(self, frame: GeneralFrame) -> bool | None:
         """
         Called when a frame has been received. Returns whether or not the frame was
         handled by any listener.
@@ -669,7 +673,7 @@ def frame_received(self, frame: GeneralFrame) -> bool:
 
         if frame.header not in c.COMMANDS_BY_ID:
             LOGGER.error("Received an unknown frame: %s", frame)
-            return
+            return None
 
         command_cls = c.COMMANDS_BY_ID[frame.header]
 
@@ -680,7 +684,7 @@ def frame_received(self, frame: GeneralFrame) -> bool:
             # https://github.com/home-assistant/core/issues/50005
             if command_cls == c.ZDO.ParentAnnceRsp.Callback:
                 LOGGER.warning("Failed to parse broken %s as %s", frame, command_cls)
-                return
+                return None
 
             raise
 
@@ -760,7 +764,21 @@ def callback_for_response(
 
         return self.callback_for_responses([response], callback)
 
-    def wait_for_responses(self, responses, *, context=False) -> asyncio.Future:
+    @typing.overload
+    def wait_for_responses(
+        self, responses, *, context: typing_extensions.Literal[False] = ...
+    ) -> asyncio.Future:
+        ...
+
+    @typing.overload
+    def wait_for_responses(
+        self, responses, *, context: typing_extensions.Literal[True]
+    ) -> tuple[asyncio.Future, OneShotResponseListener]:
+        ...
+
+    def wait_for_responses(
+        self, responses, *, context: bool = False
+    ) -> asyncio.Future | tuple[asyncio.Future, OneShotResponseListener]:
         """
         Creates a one-shot listener that matches any *one* of the given responses.
         """
@@ -787,7 +805,9 @@ def wait_for_response(self, response: t.CommandBase) -> asyncio.Future:
 
         return self.wait_for_responses([response])
 
-    async def request(self, request: t.CommandBase, **response_params) -> t.CommandBase:
+    async def request(
+        self, request: t.CommandBase, **response_params
+    ) -> t.CommandBase | None:
         """
         Sends a SREQ/AREQ request and returns its SRSP (only for SREQ), failing if any
         of the SRSP's parameters don't match `response_params`.
@@ -827,7 +847,7 @@ async def request(self, request: t.CommandBase, **response_params) -> t.CommandB
             if not request.Rsp:
                 LOGGER.debug("Request has no response, not waiting for one.")
                 self._uart.send(frame)
-                return
+                return None
 
             # We need to create the response listener before we send the request
             response_future = self.wait_for_responses(

zigpy_znp/tools/common.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import sys
+import typing
 import logging
 import argparse
 
@@ -116,7 +117,7 @@ def validate_backup_json(backup: t.JSONType) -> None:
 
 
 class CustomArgumentParser(argparse.ArgumentParser):
-    def parse_args(self, args: list[str] = None, namespace=None):
+    def parse_args(self, args: typing.Sequence[str] | None = None, namespace=None):
         args = super().parse_args(args, namespace)
 
         # Since we're running as a CLI tool, install our own log level and color logger