Fix pylint/mypy stuff

Author: ianmcorvidae

meshtastic/tests/firmware_harness.py

@@ -89,8 +89,13 @@ def start(self, binary: str) -> None:
         self.workdir = tempfile.mkdtemp(prefix=f"mtd_node{self.node_id}_")
         vfs_dir = os.path.join(self.workdir, "vfs")
         os.mkdir(vfs_dir)
-        log_stdout = open(os.path.join(self.workdir, "meshtasticd.log"), "wb", buffering=0)
-        log_stderr = open(os.path.join(self.workdir, "meshtasticd.err"), "wb", buffering=0)
+        # Files are closed in _kill(); keep them open for the process lifetime.
+        log_stdout = open(  # pylint: disable=consider-using-with
+            os.path.join(self.workdir, "meshtasticd.log"), "wb", buffering=0
+        )
+        log_stderr = open(  # pylint: disable=consider-using-with
+            os.path.join(self.workdir, "meshtasticd.err"), "wb", buffering=0
+        )
         self._log_files = [log_stdout, log_stderr]
         self.process = subprocess.Popen(  # pylint: disable=consider-using-with
             [
@@ -244,9 +249,11 @@ def stop(self) -> None:
         self._started = False
 
     def get_node(self, idx: int) -> SimNode:
+        """Return the SimNode at the given index."""
         return self.nodes[idx]
 
     def get_iface(self, idx: int) -> TCPInterface:
+        """Return the TCPInterface for the node at the given index."""
         iface = self.nodes[idx].iface
         assert iface is not None, f"node {idx} has no interface"
         return iface

meshtastic/tests/test_smokemesh.py

@@ -6,7 +6,7 @@
 according to this topology.
 """
 import time
-from typing import List
+from typing import Any, Callable, List, Optional
 
 import pytest
 from pubsub import pub  # type: ignore[import-untyped]
@@ -19,11 +19,14 @@ class _PacketCollector:
 
     def __init__(self):
         self.packets: List[dict] = []
+        self._handler: Optional[Callable[..., Any]] = None
 
     def on_receive(self, packet, interface):  # pylint: disable=unused-argument
+        """Store a received packet."""
         self.packets.append(packet)
 
     def wait_for(self, count: int, timeout: float = RECEIVE_TIMEOUT) -> bool:
+        """Wait until *count* packets have been collected or *timeout* expires."""
         deadline = time.monotonic() + timeout
         while time.monotonic() < deadline:
             if len(self.packets) >= count:
@@ -33,6 +36,7 @@ def wait_for(self, count: int, timeout: float = RECEIVE_TIMEOUT) -> bool:
 
     @property
     def texts(self) -> List[str]:
+        """Return text payloads from TEXT_MESSAGE_APP packets."""
         return [
             p.get("decoded", {}).get("text", "")
             for p in self.packets
@@ -41,12 +45,14 @@ def texts(self) -> List[str]:
 
     @property
     def traceroutes(self) -> List[dict]:
+        """Return TRACEROUTE_APP packets."""
         return [
             p for p in self.packets
             if p.get("decoded", {}).get("portnum") == "TRACEROUTE_APP"
         ]
 
     def reset(self):
+        """Clear all collected packets."""
         self.packets.clear()