Run mypy in a stricter configuration and address missing types

CHANGELOG.md

@@ -11,6 +11,7 @@ Other changes
 - Fixed a bug where notifications without a payload were not recognized as such
 - Invalid octal sequences produced by GDB are left unchanged instead of causing a `UnicodeDecodeError` (#64)
 - Fix a crash on Windows by waiting for the GDB process to exit in `GdbController.exit`
+- Added type annotations to the whole public API
 - Updated the examples in `README.md` to use the current API and show the results printed by this version of pygdbmi (#69)
 
 Internal changes
@@ -21,6 +22,7 @@ Internal changes
 - Added `nox -s format` to re-format the source code using the correct options
 - Reformatted all imports with `isort`, and use it as part of `nox -s lint` and `nox -s format`
 - Converted tests to use pytest's test structure rather than the unittest-based one
+- Added mypy configuration to detect more problems and to force all code to be annotated
 - Excluded some common backup and cache files from `MANIFEST.in` to prevent unwanted files to be included which causes `check-manifest` to fail
 
 ## 0.10.0.2

MANIFEST.in

@@ -8,11 +8,12 @@ exclude .flake8
 exclude noxfile.py
 exclude mkdocs.yml
 exclude __pycache__
+exclude mypy.ini
 exclude *.in
 exclude *.txt
 exclude *.pyc
 exclude *.sw[po]
 exclude *~
 
 prune tests
-prune docs
+prune docs

example.py

@@ -25,14 +25,13 @@
     MAKE_CMD = "make"
 
 
-def main(verbose=True):
+def main() -> None:
     """Build and debug an application programatically
 
     For a list of GDB MI commands, see https://www.sourceware.org/gdb/onlinedocs/gdb/GDB_002fMI.html
     """
 
     # Build C program
-    find_executable(MAKE_CMD)
     if not find_executable(MAKE_CMD):
         print(
             'Could not find executable "%s". Ensure it is installed and on your $PATH.'
@@ -42,7 +41,7 @@ def main(verbose=True):
     subprocess.check_output([MAKE_CMD, "-C", SAMPLE_C_CODE_DIR, "--quiet"])
 
     # Initialize object that manages gdb subprocess
-    gdbmi = GdbController(verbose=verbose)
+    gdbmi = GdbController()
 
     # Send gdb commands. Gdb machine interface commands are easier to script around,
     # hence the name "machine interface".

mypy.ini

@@ -0,0 +1,6 @@
+[mypy]
+python_version = 3.7
+show_error_codes = True
+warn_redundant_casts = True
+strict_equality = True
+disallow_untyped_defs = True

noxfile.py

@@ -2,7 +2,8 @@
 import shutil
 from pathlib import Path
 
-import nox  # type: ignore
+import nox
+from nox.sessions import Session
 
 
 nox.options.sessions = ["tests", "lint", "docs"]
@@ -13,7 +14,7 @@
 # If these are modified, also modify .github/workflows/tests.yml and the list of supported versions
 # in setup.py.
 @nox.session(python=["3.7", "3.10"])
-def tests(session):
+def tests(session: Session) -> None:
     session.install(".", "pytest")
     session.run("pytest", *session.posargs)
 
@@ -33,7 +34,7 @@ def tests(session):
 
 # `format` is a builtin so the function is named differently.
 @nox.session(name="format")
-def format_(session):
+def format_(session: Session) -> None:
     """Re-format all Python source files or, if positionals are passed, only the
     specified files."""
     files = LINTED_PATHS
@@ -47,7 +48,7 @@ def format_(session):
 
 
 @nox.session()
-def lint(session):
+def lint(session: Session) -> None:
     session.install(
         # Packages needed as they are used directly.
         "black",
@@ -56,6 +57,7 @@ def lint(session):
         "isort",
         "mypy",
         # Packages needed to provide types for mypy.
+        "nox",
         "pytest",
     )
     session.run("isort", "--check-only", *ISORT_OPTIONS, *LINTED_PATHS)
@@ -66,30 +68,30 @@ def lint(session):
     session.run("python", "setup.py", "check", "--metadata", "--strict")
 
 
-def install_mkdoc_dependencies(session):
+def install_mkdoc_dependencies(session: Session) -> None:
     session.install("-r", "mkdoc_requirements.txt")
 
 
 @nox.session
-def docs(session):
+def docs(session: Session) -> None:
     install_mkdoc_dependencies(session)
     session.run("mkdocs", "build")
 
 
 @nox.session
-def serve_docs(session):
+def serve_docs(session: Session) -> None:
     install_mkdoc_dependencies(session)
     session.run("mkdocs", "serve")
 
 
 @nox.session
-def publish_docs(session):
+def publish_docs(session: Session) -> None:
     install_mkdoc_dependencies(session)
     session.run("mkdocs", "gh-deploy")
 
 
 @nox.session(python="3.7")
-def build(session):
+def build(session: Session) -> None:
     session.install("setuptools", "wheel", "twine")
     shutil.rmtree("dist", ignore_errors=True)
     shutil.rmtree("build", ignore_errors=True)
@@ -98,7 +100,7 @@ def build(session):
 
 
 @nox.session(python="3.7")
-def publish(session):
+def publish(session: Session) -> None:
     build(session)
     print("REMINDER: Has the changelog been updated?")
     session.run("python", "-m", "twine", "upload", "dist/*")

pygdbmi/IoManager.py

@@ -2,13 +2,12 @@
 which manages I/O for file objects connected to an existing gdb process
 or pty.
 """
-import io
 import logging
 import os
 import select
 import time
 from pprint import pformat
-from typing import Any, Dict, List, Optional, Tuple, Union
+from typing import IO, Any, Dict, List, Optional, Tuple, Union
 
 from pygdbmi import gdbmiparser
 from pygdbmi.constants import (
@@ -36,11 +35,11 @@
 class IoManager:
     def __init__(
         self,
-        stdin: io.BufferedWriter,
-        stdout: io.BufferedReader,
-        stderr: Optional[io.BufferedReader],
-        time_to_check_for_additional_output_sec=DEFAULT_TIME_TO_CHECK_FOR_ADDITIONAL_OUTPUT_SEC,
-    ):
+        stdin: IO[bytes],
+        stdout: IO[bytes],
+        stderr: Optional[IO[bytes]],
+        time_to_check_for_additional_output_sec: float = DEFAULT_TIME_TO_CHECK_FOR_ADDITIONAL_OUTPUT_SEC,
+    ) -> None:
         """
         Manage I/O for file objects created before calling this class
         This can be useful if the gdb process is managed elsewhere, or if a
@@ -72,8 +71,10 @@ def __init__(
             _make_non_blocking(self.stderr)
 
     def get_gdb_response(
-        self, timeout_sec: float = DEFAULT_GDB_TIMEOUT_SEC, raise_error_on_timeout=True
-    ):
+        self,
+        timeout_sec: float = DEFAULT_GDB_TIMEOUT_SEC,
+        raise_error_on_timeout: bool = True,
+    ) -> List[Dict]:
         """Get response from GDB, and block while doing so. If GDB does not have any response ready to be read
         by timeout_sec, an exception is raised.
 
@@ -107,7 +108,7 @@ def get_gdb_response(
         else:
             return retval
 
-    def _get_responses_windows(self, timeout_sec):
+    def _get_responses_windows(self, timeout_sec: float) -> List[Dict]:
         """Get responses on windows. Assume no support for select and use a while loop."""
         timeout_time_sec = time.time() + timeout_sec
         responses = []
@@ -120,12 +121,13 @@ def _get_responses_windows(self, timeout_sec):
             except IOError:
                 pass
 
-            try:
-                self.stderr.flush()
-                raw_output = self.stderr.readline().replace(b"\r", b"\n")
-                responses_list += self._get_responses_list(raw_output, "stderr")
-            except IOError:
-                pass
+            if self.stderr is not None:
+                try:
+                    self.stderr.flush()
+                    raw_output = self.stderr.readline().replace(b"\r", b"\n")
+                    responses_list += self._get_responses_list(raw_output, "stderr")
+                except IOError:
+                    pass
 
             responses += responses_list
             if timeout_sec == 0:
@@ -140,7 +142,7 @@ def _get_responses_windows(self, timeout_sec):
 
         return responses
 
-    def _get_responses_unix(self, timeout_sec):
+    def _get_responses_unix(self, timeout_sec: float) -> List[Dict]:
         """Get responses on unix-like system. Use select to wait for output."""
         timeout_time_sec = time.time() + timeout_sec
         responses = []
@@ -158,6 +160,7 @@ def _get_responses_unix(self, timeout_sec):
                     stream = "stdout"
 
                 elif fileno == self.stderr_fileno:
+                    assert self.stderr is not None
                     self.stderr.flush()
                     raw_output = self.stderr.read()
                     stream = "stderr"
@@ -222,10 +225,10 @@ def _get_responses_list(
     def write(
         self,
         mi_cmd_to_write: Union[str, List[str]],
-        timeout_sec=DEFAULT_GDB_TIMEOUT_SEC,
+        timeout_sec: float = DEFAULT_GDB_TIMEOUT_SEC,
         raise_error_on_timeout: bool = True,
         read_response: bool = True,
-    ):
+    ) -> List[Dict]:
         """Write to gdb process. Block while parsing responses from gdb for a maximum of timeout_sec.
 
         Args:
@@ -324,7 +327,7 @@ def _buffer_incomplete_responses(
     return (raw_output, buf)
 
 
-def _make_non_blocking(file_obj: io.IOBase):
+def _make_non_blocking(file_obj: IO) -> None:
     """make file object non-blocking
     Windows doesn't have the fcntl module, but someone on
     stack overflow supplied this code as an answer, and it works

pygdbmi/StringStream.py

@@ -1,3 +1,5 @@
+from typing import List
+
 from pygdbmi.gdbescapes import advance_past_string_with_gdb_escapes
 
 
@@ -13,12 +15,12 @@ class StringStream:
     to the project.
     """
 
-    def __init__(self, raw_text, debug=False):
+    def __init__(self, raw_text: str, debug: bool = False) -> None:
         self.raw_text = raw_text
         self.index = 0
         self.len = len(raw_text)
 
-    def read(self, count):
+    def read(self, count: int) -> str:
         """Read count characters starting at self.index,
         and return those characters as a string
         """
@@ -31,11 +33,11 @@ def read(self, count):
 
         return buf
 
-    def seek(self, offset):
+    def seek(self, offset: int) -> None:
         """Advance the index of this StringStream by offset characters"""
         self.index = self.index + offset
 
-    def advance_past_chars(self, chars):
+    def advance_past_chars(self, chars: List[str]) -> str:
         """Advance the index past specific chars
         Args chars (list): list of characters to advance past
 

pygdbmi/gdbcontroller.py

@@ -6,7 +6,7 @@
 import logging
 import subprocess
 from distutils.spawn import find_executable
-from typing import List, Optional, Union
+from typing import Dict, List, Optional, Union
 
 from pygdbmi.constants import (
     DEFAULT_GDB_TIMEOUT_SEC,
@@ -26,8 +26,8 @@ class GdbController:
     def __init__(
         self,
         command: Optional[List[str]] = None,
-        time_to_check_for_additional_output_sec=DEFAULT_TIME_TO_CHECK_FOR_ADDITIONAL_OUTPUT_SEC,
-    ):
+        time_to_check_for_additional_output_sec: float = DEFAULT_TIME_TO_CHECK_FOR_ADDITIONAL_OUTPUT_SEC,
+    ) -> None:
         """
         Run gdb as a subprocess. Send commands and receive structured output.
         Create new object, along with a gdb subprocess
@@ -48,11 +48,11 @@ def __init__(
                 + "See https://sourceware.org/gdb/onlinedocs/gdb/Mode-Options.html#Mode-Options."
             )
         self.abs_gdb_path = None  # abs path to gdb executable
-        self.command = command  # type: List[str]
+        self.command: List[str] = command
         self.time_to_check_for_additional_output_sec = (
             time_to_check_for_additional_output_sec
         )
-        self.gdb_process = None
+        self.gdb_process: Optional[subprocess.Popen] = None
         self._allow_overwrite_timeout_times = (
             self.time_to_check_for_additional_output_sec > 0
         )
@@ -72,7 +72,7 @@ def __init__(
 
         self.spawn_new_gdb_subprocess()
 
-    def spawn_new_gdb_subprocess(self):
+    def spawn_new_gdb_subprocess(self) -> int:
         """Spawn a new gdb subprocess with the arguments supplied to the object
         during initialization. If gdb subprocess already exists, terminate it before
         spanwing a new one.
@@ -96,6 +96,8 @@ def spawn_new_gdb_subprocess(self):
             bufsize=0,
         )
 
+        assert self.gdb_process.stdin is not None
+        assert self.gdb_process.stdout is not None
         self.io_manager = IoManager(
             self.gdb_process.stdin,
             self.gdb_process.stdout,
@@ -105,18 +107,20 @@ def spawn_new_gdb_subprocess(self):
         return self.gdb_process.pid
 
     def get_gdb_response(
-        self, timeout_sec: float = DEFAULT_GDB_TIMEOUT_SEC, raise_error_on_timeout=True
-    ):
+        self,
+        timeout_sec: float = DEFAULT_GDB_TIMEOUT_SEC,
+        raise_error_on_timeout: bool = True,
+    ) -> List[Dict]:
         """Get gdb response. See IoManager.get_gdb_response() for details"""
         return self.io_manager.get_gdb_response(timeout_sec, raise_error_on_timeout)
 
     def write(
         self,
         mi_cmd_to_write: Union[str, List[str]],
-        timeout_sec=DEFAULT_GDB_TIMEOUT_SEC,
+        timeout_sec: float = DEFAULT_GDB_TIMEOUT_SEC,
         raise_error_on_timeout: bool = True,
         read_response: bool = True,
-    ):
+    ) -> List[Dict]:
         """Write command to gdb. See IoManager.write() for details"""
         return self.io_manager.write(
             mi_cmd_to_write, timeout_sec, raise_error_on_timeout, read_response