Finish typing support and enable mypy in CI

Author: jschwartzentruber

.pre-commit-config.yaml

@@ -49,6 +49,13 @@ repos:
       - id: check-useless-excludes
   - repo: local
     hooks:
+      - id: mypy
+        name: mypy
+        entry: tox -e mypy --
+        language: system
+        require_serial: true
+        exclude: ^tests/
+        types: [python]
       - id: pylint
         name: pylint
         entry: tox -e pylint --

pyproject.toml

@@ -22,41 +22,9 @@ known_first_party = "lithium"
 profile = "black"
 
 [tool.mypy]
-check_untyped_defs = true
-disallow_any_generics = true
-disallow_incomplete_defs = true
-disallow_subclassing_any = true
-disallow_untyped_calls = true
-disallow_untyped_decorators = true
-disallow_untyped_defs = true
-implicit_reexport = false
-namespace_packages = true
-no_implicit_optional = true
-python_version = "3.6"
-show_error_codes = true
-strict_equality = true
-warn_redundant_casts = true
-warn_return_any = true
-warn_unused_configs = true
-warn_unused_ignores = true
-
-[[tool.mypy.overrides]]
-module = [
-    "Collector.Collector",
-    "CovReporter",
-    "distro",
-    "FTB",
-    "FTB.ProgramConfiguration",
-    "FTB.Signatures",
-    "FTB.Signatures.CrashInfo",
-    "fasteners",
-    "lithium",  # Lithium *may* get types soon, as of mid-2021
-    "lithium.interestingness",
-    "lithium.interestingness.timed_run",
-    "lithium.interestingness.utils",
-    "Reporter.Reporter",
-]
+strict = true
 ignore_missing_imports = true
+show_error_codes = true
 
 [tool.pylint.format]
 max-line-length = 88
@@ -66,6 +34,7 @@ disable = [
     "C0330",
     "C0326",
     "bad-continuation",
+    "duplicate-code",
     "fixme",
     "import-error",
     "subprocess-run-check",

src/lithium/docs/examples/arithmetic/product_divides.py

@@ -7,14 +7,15 @@
 """This tests Lithium's main "minimize" algorithm."""
 
 import sys
+from typing import List
 
 
-def interesting(args: str, _temp_prefix: str) -> bool:
+def interesting(args: List[str], _temp_prefix: str) -> bool:
     """Interesting if the product of the numbers in the file divides the argument.
 
     Args:
-        args: The first parameter.
-        _temp_prefix: The second parameter.
+        args: Input arguments.
+        _temp_prefix: Temp directory prefix.
 
     Returns:
         True if successful, False otherwise.

src/lithium/interestingness/outputs.py

@@ -13,8 +13,7 @@
 import logging
 import os
 from pathlib import Path
-from typing import List
-from typing import Union
+from typing import List, Union
 
 from . import timed_run, utils
 

src/lithium/interestingness/repeat.py

@@ -32,7 +32,7 @@
 
 import argparse
 import logging
-from typing import List
+from typing import Any, List, cast
 
 from .utils import rel_or_abs_import
 
@@ -69,7 +69,7 @@ def interesting(cli_args: List[str], temp_prefix: str) -> bool:
     condition_args = args.cmd_with_flags[2:]
 
     if hasattr(condition_script, "init"):
-        condition_script.init(condition_args)
+        cast(Any, condition_script).init(condition_args)
 
     # Run the program over as many iterations as intended, with desired flags, replacing
     # REPEATNUM where necessary.
@@ -79,7 +79,9 @@ def interesting(cli_args: List[str], temp_prefix: str) -> bool:
             s.replace("REPEATNUM", str(i)) for s in condition_args
         ]
         log.info("Repeat number %d:", i)
-        if condition_script.interesting(replaced_condition_args, temp_prefix):
+        if cast(Any, condition_script).interesting(
+            replaced_condition_args, temp_prefix
+        ):
             return True
 
     return False

src/lithium/interestingness/timed_run.py

@@ -14,9 +14,7 @@
 import sys
 import time
 from pathlib import Path
-from typing import BinaryIO
-from typing import List
-from typing import Union
+from typing import BinaryIO, Callable, Dict, List, Optional, Union
 
 (CRASHED, TIMED_OUT, NORMAL, ABNORMAL, NONE) = range(5)
 
@@ -31,7 +29,7 @@
 class ArgumentParser(argparse.ArgumentParser):
     """Argument parser with `timeout` and `cmd_with_args`"""
 
-    def __init__(self, *args, **kwds) -> None:
+    def __init__(self, *args, **kwds) -> None:  # type: ignore
         super().__init__(*args, **kwds)
         self.add_argument(
             "-t",
@@ -68,9 +66,9 @@ def timed_run(
     cmd_with_args: List[str],
     timeout: int,
     log_prefix: str = "",
-    env=None,
+    env: Optional[Dict[str, str]] = None,
     inp: str = "",
-    preexec_fn=None,
+    preexec_fn: Optional[Callable[[], None]] = None,
 ) -> RunData:
     """If log_prefix is None, uses pipes instead of files for all output.
 

src/lithium/interestingness/utils.py

@@ -15,8 +15,7 @@
 import sys
 from pathlib import Path
 from types import ModuleType
-from typing import Tuple
-from typing import Union
+from typing import Tuple, Union
 
 
 def file_contains_str(
@@ -65,7 +64,7 @@ def file_contains_regex(
         if match was found, and matched string
     """
 
-    matched_str = ""
+    matched_str = b""
     found = False
     file_contents = Path(input_file).read_bytes()
     found_regex = re.search(regex, file_contents, flags=re.MULTILINE)

src/lithium/reducer.py

@@ -9,15 +9,16 @@
 import os
 import sys
 from pathlib import Path
-from typing import Dict
-from typing import List
-from typing import Optional
+from types import ModuleType
+from typing import Any, Dict, List, Optional, Type, cast
 
 import pkg_resources
 
 from .interestingness.utils import rel_or_abs_import
 from .strategies import DEFAULT as DEFAULT_STRATEGY
+from .strategies import Strategy
 from .testcases import DEFAULT as DEFAULT_TESTCASE
+from .testcases import Testcase
 from .util import LithiumError, quantity, summary_header
 
 LOG = logging.getLogger(__name__)
@@ -28,22 +29,22 @@ class Lithium:
 
     def __init__(self) -> None:
 
-        self.strategy = None
+        self.strategy: Optional[Strategy] = None
 
-        self.condition_script = None
-        self.condition_args = None
+        self.condition_script: Optional[ModuleType] = None
+        self.condition_args: Optional[List[str]] = None
 
         self.test_count = 0
         self.test_total = 0
 
-        self.temp_dir: Path = Path()
+        self.temp_dir: Optional[Path] = None
 
-        self.testcase = None
-        self.last_interesting = None
+        self.testcase: Optional[Testcase] = None
+        self.last_interesting: Optional[Testcase] = None
 
         self.temp_file_count = 1
 
-    def main(self, argv: Optional[List[str]]) -> int:
+    def main(self, argv: Optional[List[str]] = None) -> int:
         """Main entrypoint (parse args and call `run()`)
 
         Args:
@@ -69,7 +70,7 @@ def run(self) -> int:
             0 for successful reduction
         """
         if hasattr(self.condition_script, "init"):
-            self.condition_script.init(self.condition_args)
+            cast(Any, self.condition_script).init(self.condition_args)
 
         try:
             if self.temp_dir is None:
@@ -78,6 +79,8 @@ def run(self) -> int:
                     "Intermediate files will be stored in %s%s.", self.temp_dir, os.sep
                 )
 
+            assert self.strategy is not None
+            assert self.testcase is not None
             result = self.strategy.main(
                 self.testcase, self.interesting, self.testcase_temp_filename
             )
@@ -89,13 +92,13 @@ def run(self) -> int:
 
         finally:
             if hasattr(self.condition_script, "cleanup"):
-                self.condition_script.cleanup(self.condition_args)
+                cast(Any, self.condition_script).cleanup(self.condition_args)
 
             # Make sure we exit with an interesting testcase
             if self.last_interesting is not None:
                 self.last_interesting.dump()
 
-    def process_args(self, argv: Optional[List[str]]) -> None:
+    def process_args(self, argv: Optional[List[str]] = None) -> None:
         """Parse command-line args and initialize self.
 
         Args:
@@ -106,10 +109,12 @@ def process_args(self, argv: Optional[List[str]]) -> None:
         class _ArgParseTry(argparse.ArgumentParser):
             # pylint: disable=arguments-differ,no-self-argument
 
-            def exit(_, **kwds) -> None:
+            def exit(  # type: ignore[override]
+                self, status: int = 0, message: Optional[str] = None
+            ) -> None:
                 pass
 
-            def error(_, message) -> None:
+            def error(self, message: str) -> None:  # type: ignore[override]
                 pass
 
         early_parser = _ArgParseTry(add_help=False)
@@ -122,8 +127,8 @@ def error(_, message) -> None:
         grp_opt = parser.add_argument_group(description="Lithium options")
         grp_atoms = grp_opt.add_mutually_exclusive_group()
 
-        strategies: Dict[str, str] = {}
-        testcase_types: Dict[str, str] = {}
+        strategies: Dict[str, Type[Strategy]] = {}
+        testcase_types: Dict[str, Type[Testcase]] = {}
         for entry_point in pkg_resources.iter_entry_points("lithium_strategies"):
             try:
                 strategy_cls = entry_point.load()
@@ -174,10 +179,10 @@ def error(_, message) -> None:
         early_parser.add_argument(
             "--strategy", default=DEFAULT_STRATEGY, choices=strategies.keys()
         )
-        args = early_parser.parse_known_args(argv)
-        atom = args[0].atom if args else DEFAULT_TESTCASE
+        early_args = early_parser.parse_known_args(argv)
+        atom = early_args[0].atom if early_args else DEFAULT_TESTCASE
         self.strategy = strategies.get(
-            args[0].strategy if args else None, strategies[DEFAULT_STRATEGY]
+            early_args[0].strategy if early_args else None, strategies[DEFAULT_STRATEGY]
         )()
 
         grp_opt.add_argument(
@@ -192,6 +197,7 @@ def error(_, message) -> None:
             "-v", "--verbose", action="store_true", help="enable verbose debug logging"
         )
         # this has already been parsed above, it's only here for the help message
+        assert self.strategy is not None
         grp_opt.add_argument(
             "--strategy",
             default=self.strategy.name,
@@ -250,6 +256,9 @@ def testcase_temp_filename(
         if use_number:
             filename_stem = "%d-%s" % (self.temp_file_count, filename_stem)
             self.temp_file_count += 1
+        assert self.testcase is not None
+        assert self.testcase.extension is not None
+        assert self.temp_dir is not None
         return self.temp_dir / (filename_stem + self.testcase.extension)
 
     def create_temp_dir(self) -> None:
@@ -269,7 +278,7 @@ def create_temp_dir(self) -> None:
 
     # If the file is still interesting after the change, changes "parts" and returns
     # True.
-    def interesting(self, testcase_suggestion, write_it: bool = True) -> bool:
+    def interesting(self, testcase_suggestion: Testcase, write_it: bool = True) -> bool:
         """Test whether a testcase suggestion is interesting.
 
         Args:
@@ -286,9 +295,15 @@ def interesting(self, testcase_suggestion, write_it: bool = True) -> bool:
         self.test_count += 1
         self.test_total += len(testcase_suggestion)
 
+        assert self.temp_dir is not None
         temp_prefix = str(self.temp_dir / str(self.temp_file_count))
 
-        inter = self.condition_script.interesting(self.condition_args, temp_prefix)
+        assert self.condition_script is not None
+        inter = bool(
+            cast(Any, self.condition_script).interesting(
+                self.condition_args, temp_prefix
+            )
+        )
 
         # Save an extra copy of the file inside the temp directory.
         # This is useful if you're reducing an assertion and encounter a crash: