topotato: begin mypy --check-untyped-defs sweep

mypy doesn't check functions that have no type annotations at all by
default.  Run with `--check-untyped-defs` and begin cleaning that up.

Signed-off-by: David Lamparter <[email protected]>

topotato/control.py

@@ -15,9 +15,11 @@
 import logging
 import typing
 from typing import (
+    cast,
     ClassVar,
     Dict,
     List,
+    Mapping,
     Optional,
     Protocol,
     Type,
@@ -63,7 +65,7 @@ def __init_subclass__(cls, /, name=None, **kwargs):
 
     @classmethod
     def dispatch(
-        cls, control: "Control", name: str, items: Dict[str, str]
+        cls, control: "Control", name: str, items: Mapping[str, str]
     ) -> "ControlSection":
         _name = name.split(":", 1)[0]
         if _name not in cls.sections:
@@ -74,12 +76,12 @@ def dispatch(
 
     @classmethod
     def make(
-        cls, control: "Control", name: str, items: Dict[str, str]
+        cls, control: "Control", name: str, items: Mapping[str, str]
     ) -> "ControlSection":
         return cls(control, name, items)
 
     # pylint: disable=too-many-branches
-    def __init__(self, control: "Control", name: str, items: Dict[str, str]):
+    def __init__(self, control: "Control", name: str, items: Mapping[str, str]):
         self.control = control
 
         def try_type(typ, raw_value):
@@ -171,7 +173,7 @@ def __init__(self, session):
         self.session = session
         self.config = configparser.ConfigParser()
         self.configfiles = []
-        self.typed_sections = {}
+        self.typed_sections = cast(_ControlSectionTypeDict, {})
         self.targets = {}
 
     def load_configs(self, configs):
@@ -246,10 +248,10 @@ def __init_subclass__(cls, /, name=None, **kwargs):
 
     @classmethod
     def make(
-        cls, control: "Control", name: str, items: Dict[str, str]
+        cls, control: "Control", name: str, items: Mapping[str, str]
     ) -> "ControlSection":
         return cls._types[items["type"]](control, name, items)
 
-    def __init__(self, control: "Control", name: str, items: Dict[str, str]):
+    def __init__(self, control: "Control", name: str, items: Mapping[str, str]):
         super().__init__(control, name, items)
         control.targets[self.name] = self

topotato/generatorwrap.py

@@ -14,8 +14,8 @@
 from typing import (
     Any,
     Callable,
+    Generator,
     Generic,
-    Iterable,
     List,
     Optional,
     TypeVar,
@@ -33,15 +33,15 @@ class GeneratorDeletedUnused(Exception):
     are normally printed but ignored there.
     """
 
-    origin: Iterable[str]
+    origin: List[str]
     """
     Formatted string traceback for the generator's call location.
 
     Not saving FrameInfo/related here since the frame continues executing
     after the call site and these objects would change due to that.
     """
 
-    def __init__(self, origin: Iterable[str]):
+    def __init__(self, origin: List[str]):
         self.origin = origin
 
 
@@ -124,7 +124,7 @@ def __exit__(
         raise GeneratorsUnused(self._errs)
 
 
-TG = TypeVar("TG")
+TG = TypeVar("TG", bound=Generator)
 
 
 class GeneratorWrapper(Generic[TG]):
@@ -158,12 +158,12 @@ def my_generator(foo):
     """
     Has this generator actually been iterated over?  (Not necessarily to completion.)
     """
-    _loc: Iterable[str]
+    _loc: List[str]
     """
     Location of call to pass to :py:class:`GeneratorDeletedUnused`, see there.
     """
 
-    def __init__(self, wraps: TG, loc: Iterable[str]):
+    def __init__(self, wraps: TG, loc: List[str]):
         self._wraps = wraps
         self._loc = loc
         self._run = False
@@ -194,7 +194,12 @@ def apply(cls, function: Callable[..., TG]) -> Callable[..., TG]:
 
         @functools.wraps(function)
         def wrapper(*args, **kwargs):
-            loc = traceback.format_stack(inspect.currentframe().f_back)
+            f = inspect.currentframe()
+            if f is None:
+                loc = ["???"]
+            else:
+                loc = traceback.format_stack(f.f_back)
+            del f
             return cls(function(*args, **kwargs), loc)
 
         return wrapper

topotato/pcapng.py

@@ -430,4 +430,4 @@ def flush(self):
 
     def close(self):
         self.fd.close()
-        self.fd = None
+        self.fd = None  # type: ignore

topotato/scapy.py

@@ -7,15 +7,21 @@
 
 import logging
 
+import typing
 from typing import (
+    cast,
     Any,
     Optional,
+    Self,
 )
 
 import pytest
 
 from .assertions import TopotatoModifier
 
+if typing.TYPE_CHECKING:
+    from .nswrap import LinuxNamespace
+
 _logger = logging.getLogger(__name__)
 
 try:
@@ -45,8 +51,11 @@ class ScapySend(TopotatoModifier):
     @classmethod
     def from_parent(cls, parent, name, rtr, iface, pkt, *, repeat=None, interval=None):
         path = "/".join([l.__name__ for l in pkt.layers()])
-        self = super().from_parent(
-            parent, name="%s:%s/scapy[%s/%s]" % (name, rtr.name, iface, path)
+        self = cast(
+            Self,
+            super().from_parent(
+                parent, name="%s:%s/scapy[%s/%s]" % (name, rtr.name, iface, path)
+            ),
         )
         self._rtr = rtr
         self._iface = iface
@@ -64,16 +73,16 @@ def from_parent(cls, parent, name, rtr, iface, pkt, *, repeat=None, interval=Non
 
     def __call__(self):
         if scapy_exc:
-            pytest.skip(scapy_exc)
+            pytest.skip(str(scapy_exc))
 
-        router = self.instance.routers[self._rtr.name]
+        router = cast("LinuxNamespace", self.instance.routers[self._rtr.name])
         with router:
             sock = NetnsL2Socket(iface=self._iface, promisc=False)
             sock.send(self._pkt)
 
         if self._repeat:
             for _ in range(1, self._repeat):
-                self.timeline.sleep(self._interval)
+                self.timeline.sleep(self._interval or 0.0)
                 with router:
                     sock.send(self._pkt)
 

topotato/timeline.py

@@ -33,7 +33,7 @@
 
 @dataclass
 class TimingParams:
-    delay: Optional[float]
+    delay: float
     maxwait: Optional[float]
     full_history: bool = False
 

topotato/utils.py

@@ -21,9 +21,16 @@
 import inspect
 
 from typing import (
+    cast,
+    Any,
     Callable,
+    Dict,
     List,
+    Optional,
     Union,
+    TextIO,
+    Tuple,
+    Type,
     TypeVar,
 )
 
@@ -81,7 +88,13 @@ def deindent(text: str, trim=False) -> str:
     return "\n".join(do_trim(line)[len(common_prefix) :] for line in lines)
 
 
-def get_textdiff(text1: str, text2: str, title1="", title2="", **opts) -> str:
+def get_textdiff(
+    text1: Union[str, List[str]],
+    text2: Union[str, List[str]],
+    title1="",
+    title2="",
+    **opts,
+) -> str:
     """
     Diff formatting wrapper (just cleans up line endings)
 
@@ -190,7 +203,7 @@ class JSONCompareListKeyedDict(JSONCompareDirective):
     :param keying: dict keys to look up/match up.
     """
 
-    keying: List[Union[int, str]]
+    keying: Tuple[Union[int, str]]
 
     def __init__(self, *keying):
         super().__init__()
@@ -218,22 +231,22 @@ def _json_diff(d1, d2):
     """
     Returns a string with the difference between JSON data.
     """
-    json_format_opts = {
-        "indent": 4,
-        "sort_keys": True,
-    }
-    dstr1 = json.dumps(d1, **json_format_opts)
-    dstr2 = json.dumps(d2, **json_format_opts)
 
-    dstr1 = ("\n".join(dstr1.rstrip().splitlines()) + "\n").splitlines(1)
-    dstr2 = ("\n".join(dstr2.rstrip().splitlines()) + "\n").splitlines(1)
+    def json_dump_fmt(data):
+        return json.dumps(data, indent=4, sort_keys=True)
+
+    dstr1 = json_dump_fmt(d1)
+    dstr2 = json_dump_fmt(d2)
+
+    dstr1 = ("\n".join(dstr1.rstrip().splitlines()) + "\n").splitlines(True)
+    dstr2 = ("\n".join(dstr2.rstrip().splitlines()) + "\n").splitlines(True)
     return get_textdiff(
         dstr2, dstr1, title1="Expected value", title2="Current value", n=0
     )
 
 
 # pylint: disable=too-many-locals,too-many-branches
-def _json_list_cmp(list1, list2, parent, result):
+def _json_list_cmp(list1, list2, parent, result: json_cmp_result) -> None:
     "Handles list type entries."
     if isinstance(list1, JSONCompareIgnoreContent) or isinstance(
         list2, JSONCompareIgnoreContent
@@ -250,12 +263,13 @@ def _json_list_cmp(list1, list2, parent, result):
         )
         return
 
-    flags = [{}, {}]
+    flags: List[Dict[Type[JSONCompareDirective], JSONCompareDirective]] = [{}, {}]
     # don't modify input list with l.pop(0) below...
     list1 = list1[:]
     list2 = list2[:]
+    process: List[Tuple[int, List[Any]]] = [(0, list1), (1, list2)]
 
-    for i, l in [(0, list1), (1, list2)]:
+    for i, l in process:
         while l and isinstance(l[0], JSONCompareDirective):
             item = l.pop(0)
             flags[i][type(item)] = item
@@ -276,12 +290,12 @@ def _json_list_cmp(list1, list2, parent, result):
 
     # List all unmatched items errors
     if JSONCompareListKeyedDict in flags[1]:
-        keys = flags[1][JSONCompareListKeyedDict].keying
-        for expected in list2:
+        keys = cast(JSONCompareListKeyedDict, flags[1][JSONCompareListKeyedDict]).keying
+        for expected in cast(List[Dict[Any, Any]], list2):
             assert isinstance(expected, dict)
 
             keymatch = []
-            for value in list1:
+            for value in cast(List[Dict[Any, Any]], list1):
                 if not isinstance(value, dict):
                     continue
                 for key in keys:
@@ -563,6 +577,7 @@ class LockedFile:
     """
     File name relative to _dir_fd.
     """
+    _fd: Optional[TextIO]
 
     def __init__(self, filename: _PathLike, dir_fd=None):
         self.filename = filename
@@ -619,6 +634,8 @@ def _opener(path, flags):
             raise
 
     def _close(self):
+        assert self._fd is not None
+
         self._depth -= 1
         if self._depth:
             return
@@ -682,6 +699,7 @@ class AtomicPublishFile:
     """
     Temporary file name, also relative to _dir_fd.
     """
+    _fd: Optional[TextIO]
 
     def __init__(self, filename, *args, dir_fd=None, **kwargs):
         self.filename = filename
@@ -707,10 +725,12 @@ def _opener(path, flags):
             return os.open(path, flags, mode=0o666, dir_fd=self._dir_fd)
 
         # pylint: disable=unspecified-encoding
-        self._fd = open(self._tmpname, *self._args, opener=_opener, **self._kwargs)
+        self._fd = open(self._tmpname, *self._args, opener=_opener, **self._kwargs)  # type: ignore[call-overload]
         return self._fd
 
     def __exit__(self, exc_type, exc_value, tb):
+        assert self._fd is not None
+
         self._fd.close()
 
         if exc_type is None: