topotato: further mypy untyped-defs cleanup

Slowly progressing through...

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

topotato/assertions.py

@@ -146,7 +146,8 @@ def finalize(self):
         return [finalize]
 
     def relative_start(self):
-        fn = self.getparent(TopotatoFunction)
+        fn = cast(TopotatoItem, self).getparent(TopotatoFunction)
+        assert fn is not None
         return fn.started_ts
 
 
@@ -300,6 +301,7 @@ def __call__(self):
                 out[-1].match_for.append(self)
                 break
         else:
+            assert result is not None
             raise result
 
     @property
@@ -430,9 +432,10 @@ def __call__(self):
             msg.match_for.append(self)
             break
         else:
-            detail = self._msg
-            if isinstance(detail, re.Pattern):
-                detail = detail.pattern
+            if isinstance(self._msg, re.Pattern):
+                detail = cast(str, self._msg.pattern)
+            else:
+                detail = cast(str, self._msg)
             raise TopotatoLogFail(detail)
 
 
@@ -530,6 +533,9 @@ class BackgroundCommand:
     run sth in bg
     """
 
+    tmpfile: Any
+    proc: Any
+
     def __init__(self, rtr, cmd):
         self._rtr = rtr
         self._cmd = cmd

topotato/base.py

@@ -48,6 +48,8 @@
 from .network import TopotatoNetwork
 
 if typing.TYPE_CHECKING:
+    from types import TracebackType
+
     from _pytest._code.code import ExceptionInfo, TracebackEntry, Traceback
     from _pytest.python import Function
 
@@ -354,6 +356,7 @@ def sleep(self, step=None, until=None):
         deadline = min(abs_until, abs_delay)
 
         tinst = self.getparent(TopotatoClass)
+        assert tinst is not None
         tinst.netinst.timeline.sleep(deadline - time.time())
 
     def reportinfo(self):  # -> Tuple[Union[py.path.local, str], int, str]:
@@ -379,6 +382,7 @@ def _traceback_filter(self, excinfo: "ExceptionInfo[BaseException]") -> "Traceba
         tb = excinfo.traceback
         newtb: List["TracebackEntry"] = []
         for entry in reversed(tb):
+            # pylint: disable=protected-access
             if entry._rawentry.tb_frame.f_code in endtrace:
                 break
             if entry._rawentry.tb_frame.f_code in skiptrace:
@@ -409,7 +413,11 @@ def __init__(self, codeloc, nexttb):
                 self.tb_lineno = codeloc.lineno
                 self.tb_next = nexttb
 
-        ftb = FakeTraceback(self._codeloc, excinfo.traceback[0]._rawentry)
+        # pylint: disable=protected-access
+        ftb = cast(
+            "TracebackType",
+            FakeTraceback(self._codeloc, excinfo.traceback[0]._rawentry),
+        )
         excinfo.traceback.insert(0, _pytest._code.code.TracebackEntry(ftb))
 
         if self.config.getoption("fulltrace", False):
@@ -476,7 +484,9 @@ def from_parent(cls, parent):
         return self
 
     def reportinfo(self):
-        fspath, _, _ = self.getparent(TopotatoClass).reportinfo()
+        tcls = self.getparent(TopotatoClass)
+        assert tcls is not None
+        fspath, _, _ = tcls.reportinfo()
         return fspath, float("-inf"), "startup"
 
     def setup(self):
@@ -522,7 +532,9 @@ def from_parent(cls, parent):
         return self
 
     def reportinfo(self):
-        fspath, _, _ = self.getparent(TopotatoClass).reportinfo()
+        tcls = self.getparent(TopotatoClass)
+        assert tcls is not None
+        fspath, _, _ = tcls.reportinfo()
         return fspath, float("inf"), "shutdown"
 
     def runtest(self):
@@ -869,16 +881,16 @@ def do_start(self):
         netinst.timeline.sleep(0.2)
         # netinst.status()
 
-        failed = []
+        failed = []  # List[Tuple[str, str]]
         for rtr in netinst.network.routers.keys():
             router = netinst.routers[rtr]
             router.start_post(netinst.timeline, failed)
 
         if len(failed) > 0:
             netinst.timeline.sleep(0)
             if len(failed) == 1:
-                router, daemon = failed[0]
-                raise TopotatoDaemonCrash(daemon=daemon, router=router)
+                rname, daemon = failed[0]
+                raise TopotatoDaemonCrash(daemon=daemon, router=rname)
 
             routers = ",".join(set(i[0] for i in failed))
             daemons = ",".join(set(i[1] for i in failed))

topotato/frr/livelog.py

@@ -307,13 +307,13 @@ def close_prep(self):
         """
         if self._wrfd is not None:
             self._wrfd.close()
-            self._wrfd = None
+            self._wrfd = None  # type: ignore[assignment]
 
     def close(self):
         assert self._wrfd is None
         if self._rdfd is not None:
             self._rdfd.close()
-            self._rdfd = None
+            self._rdfd = None  # type: ignore[assignment]
 
     def readable(self):
         """
@@ -332,7 +332,7 @@ def readable(self):
             if len(rddata) == 0:
                 yield LogClosed(self._router.name, self._daemon)
                 self._rdfd.close()
-                self._rdfd = None
+                self._rdfd = None  # type: ignore[assignment]
                 return
 
             logmsg = LogMessage(self._router, self._daemon, rddata)

topotato/generatorwrap.py

@@ -31,9 +31,9 @@ class GeneratorsUnused(Exception):
     :py:class:`GeneratorChecks` ``with`` context.
     """
 
-    generators: List[Generator]
+    generators: List["GeneratorWrapper"]
 
-    def __init__(self, generators: List[Generator]):
+    def __init__(self, generators: List["GeneratorWrapper"]):
         super().__init__()
         self.generators = generators
 

topotato/livescapy.py

@@ -73,6 +73,8 @@ def fileno(self):
 
     def readable(self):
         maxdelay = time.time() + 0.1
+        if self._sock is None:
+            return
 
         while time.time() < maxdelay:
             try:
@@ -93,6 +95,7 @@ def readable(self):
             yield TimedScapy(pkt)
 
     def close(self):
+        assert self._sock is not None
         self._sock.close()
         self._sock = None
 

topotato/nswrap.py

@@ -145,6 +145,7 @@ def start(self):
         )
         # wait for child to tell us it's ready...
         # (match sys.stdout.write("\n") below)
+        assert self.process.stdout is not None
         self.process.stdout.read(1)
 
         self.pid = find_child(self.process.pid)

topotato/parse.py

@@ -194,7 +194,7 @@ class Link(Item, metaclass=abc.ABCMeta):
         Common code for horizontal & vertical link lines.
         """
 
-        routers: List[Tuple["Topology.Item", "Topology.Router"]]
+        routers: List[Tuple["Topology.BoxMerge", Optional["Topology.Token"]]]
 
         def __init__(self, token):
             super().__init__(token)
@@ -249,6 +249,7 @@ def connect(self, items: List["Topology.Item"]):
             for r in items:
                 if not r.y1 <= self.y1 < r.y2:
                     continue
+                assert isinstance(r, Topology.BoxMerge)
                 if r.x2 == self.x1:
                     self.routers.append((r, self.left))
                 elif r.x1 == self.x2:
@@ -295,6 +296,7 @@ def connect(self, items: List["Topology.Item"]):
             for r in items:
                 if not r.x1 <= self.xmain < r.x2:
                     continue
+                assert isinstance(r, Topology.BoxMerge)
                 if r.y2 == self.y1:
                     self.routers.append((r, self.top))
                 elif r.y1 == self.y2:
@@ -412,7 +414,7 @@ def merge(start, items):
 
 
 def test():
-    topo = """
+    topostr = """
 
     [    ](eth0)------[ r2 ]
     [ r1 ]
@@ -426,7 +428,7 @@ def test():
 
     """
 
-    topo = Topology(topo)
+    topo = Topology(topostr)
 
     # pylint: disable=import-outside-toplevel
     from pprint import pprint

topotato/scapyext/pim.py

@@ -11,7 +11,11 @@
 import socket
 import struct
 from enum import Enum
-from typing import ClassVar, Dict
+from typing import (
+    ClassVar,
+    Dict,
+    Type,
+)
 
 from scapy.packet import bind_layers, Packet  # type: ignore
 from scapy.fields import (  # type: ignore
@@ -181,7 +185,7 @@ class PIM_Hello_Option(Packet):
     def extract_padding(self, s):
         return b"", s
 
-    _reg: ClassVar[Dict[int, "PIM_Hello_Option"]] = {}
+    _reg: ClassVar[Dict[int, Type["PIM_Hello_Option"]]] = {}
 
     @classmethod
     def register_variant(cls):
@@ -190,7 +194,7 @@ def register_variant(cls):
     @classmethod
     def dispatch_hook(cls, *args, pkt=None, **kargs):
         if pkt:
-            tmp_type = struct.unpack(pkt[:2], ">H")[0]
+            tmp_type = struct.unpack(">H", pkt[:2])[0]
             return cls._reg.get(tmp_type, cls)
         return cls
 

topotato/toponom.py

@@ -17,8 +17,8 @@
     Dict,
     Generator,
     List,
+    Mapping,
     Optional,
-    Sequence,
     Tuple,
     Type,
     Union,
@@ -136,7 +136,7 @@ def __init__(self, af: Union[Literal[4], Literal[6]]):
 }
 
 
-def name_to_tuple(name: str) -> Sequence[Union[str, int]]:
+def name_to_tuple(name: str) -> Tuple[Union[str, int], ...]:
     """
     convert string "abc123def456ghi" to ("abc",123,"def",456,"ghi") for sorting
 
@@ -171,7 +171,7 @@ class NOMLinked(NOMNode, metaclass=abc.ABCMeta):
     name: str
     ifaces: List["LinkIface"]
 
-    sortkey: Tuple[Any]
+    sortkey: Tuple[Any, ...]
     num_default: int
     num_explicit: Optional[int]
     dotname: str
@@ -727,7 +727,7 @@ def topo_rtr_or_lan(item):
                 j += 1
 
     def auto_num(self):
-        def do_auto(items: Dict[str, NOMLinked]):
+        def do_auto(items: Mapping[str, NOMLinked]):
             if not items:
                 return
 
@@ -760,8 +760,8 @@ def do_auto(items: Dict[str, NOMLinked]):
     def auto_ifnames(self):
         for r in self.routers.values():
             r.auto_ifnames()
-        for r in self.lans.values():
-            r.auto_ifnames()
+        for l in self.lans.values():
+            l.auto_ifnames()
 
     def auto_ip4(self):
         if self.noauto_v4:
@@ -770,8 +770,8 @@ def auto_ip4(self):
         if self.lo_v4:
             for r in self.routers.values():
                 r.auto_lo4()
-        for r in self.lans.values():
-            r.auto_ip4()
+        for l in self.lans.values():
+            l.auto_ip4()
         for links in self.links.values():
             for link in links:
                 link.a.auto_ip4()
@@ -784,8 +784,8 @@ def auto_ip6(self):
         if self.lo_v6:
             for r in self.routers.values():
                 r.auto_lo6()
-        for r in self.lans.values():
-            r.auto_ip6()
+        for l in self.lans.values():
+            l.auto_ip6()
         for links in self.links.values():
             for link in links:
                 link.a.auto_ip6()
@@ -819,8 +819,8 @@ def dot(self):
         for r in self.routers.values():
             r.dot(out)
         out += ["  } "]
-        for r in self.lans.values():
-            r.dot(out)
+        for l in self.lans.values():
+            l.dot(out)
         for links in self.links.values():
             for link in links:
                 link.dot(out)