topotato: bulk-fix mypy warnings

This is 99% of getting `mypy --check-untyped-defs` clean.  `CallableNS`
is left to fix... somehow.

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

topotato/assertions.py

@@ -15,6 +15,7 @@
 import inspect
 from collections import OrderedDict
 
+import typing
 from typing import (
     cast,
     Any,
@@ -46,6 +47,10 @@
     TopotatoUnhandledArgs,
 )
 
+if typing.TYPE_CHECKING:
+    from .frr.core import FRRRouterNS
+    from . import toponom, topobase
+
 __all__ = [
     "AssertKernelRoutesV4",
     "AssertKernelRoutesV6",
@@ -226,7 +231,7 @@ class AssertVtysh(TimedMixin, TopotatoAssertion):
 
     commands: OrderedDict
 
-    _rtr: str
+    _rtr: "toponom.Router"
     _daemon: str
     _command: str
     _compare: Optional[str]
@@ -267,7 +272,7 @@ def __init__(
         self._compare = compare
 
     def __call__(self):
-        router = self.instance.routers[self._rtr.name]
+        router = cast("FRRRouterNS", self.instance.routers[self._rtr.name])
 
         for _ in self.timeline.run_tick(self._timing):
             _, out, rc = router.vtysh_polled(self.timeline, self._daemon, self._command)
@@ -443,7 +448,7 @@ def __call__(self):
 
 
 class _DaemonControl(TopotatoModifier):
-    _rtr: str
+    _rtr: "toponom.Router"
     _daemon: str
 
     op_name: ClassVar[str]
@@ -479,7 +484,7 @@ def do(self, router):
 
 class ModifyLinkStatus(TopotatoModifier):
     _rtr: Any
-    _iface: str
+    _iface: "toponom.LinkIface"
     _state: bool
 
     posargs = ["rtr", "iface", "state"]
@@ -509,6 +514,8 @@ class BackgroundCommand:
     run sth in bg
     """
 
+    _rtr: "toponom.Router"
+
     tmpfile: Any
     proc: Any
 
@@ -517,7 +524,7 @@ def __init__(self, rtr, cmd):
         self._cmd = cmd
 
     class Action(TopotatoModifier):
-        _rtr: str
+        _rtr: "toponom.Router"
         _cmdobj: "BackgroundCommand"
 
         def __init__(self, *, name, cmdobj, **kwargs):
@@ -535,7 +542,7 @@ def __init__(self, *, name, cmdobj, **kwargs):
     class Start(Action):
         # pylint: disable=consider-using-with
         def __call__(self):
-            router = self.instance.routers[self._rtr.name]
+            router = cast("topobase.CallableNS", self.instance.routers[self._rtr.name])
 
             ifd = open("/dev/null", "rb")
             self._cmdobj.tmpfile = tmpfile = tempfile.TemporaryFile()

topotato/base.py

@@ -439,7 +439,7 @@ def __init__(self, codeloc, nexttb):
         # see comment in pytest
         Path = _pytest.pathlib.Path
         try:
-            abspath = Path(os.getcwd()) != Path(str(self.config.invocation_dir))
+            abspath = Path(os.getcwd()) != Path(str(self.config.invocation_params.dir))
         except OSError:
             abspath = True
 
@@ -744,7 +744,6 @@ class TopotatoFunction(nodes.Collector, _pytest.python.PyobjMixin):
     def from_hook(cls, obj, collector, name, include_startup=False):
         self = super().from_parent(collector, name=name)
         self._obj = obj._call
-        self._obj_raw = obj
         self.include_startup = include_startup
 
         return self
@@ -878,7 +877,7 @@ def collect(self) -> Iterable[Union[nodes.Item, nodes.Collector]]:
         if not first:
             yield InstanceShutdown.from_parent(self)
 
-    def do_start(self):
+    def do_start(self) -> None:
         self.starting_ts = time.time()
 
         netinst = self.netinst
@@ -887,7 +886,7 @@ def do_start(self):
         tcname = "".join(
             ch if ch in string.ascii_letters + string.digits else "_" for ch in tcname
         )
-        netinst.lcov_args.extend(
+        netinst.lcov_args.extend(  # type: ignore[attr-defined]
             [
                 "-t",
                 tcname,
@@ -904,7 +903,7 @@ def do_start(self):
         netinst.timeline.sleep(0.2)
         # netinst.status()
 
-        failed = []  # List[Tuple[str, str]]
+        failed: List[Tuple[str, str]] = []
         for rtr in netinst.network.routers.keys():
             router = netinst.routers[rtr]
             router.start_post(netinst.timeline, failed)
@@ -921,7 +920,7 @@ def do_start(self):
 
         self.started_ts = time.time()
 
-        for ifname, sock in netinst.scapys.items():
+        for ifname, sock in netinst.scapys.items():  # type: ignore[attr-defined]
             netinst.timeline.install(LiveScapy(ifname, sock))
 
     @staticmethod

topotato/exabgp/_exabgp.py

@@ -11,14 +11,19 @@
 import tempfile
 import time
 
-from typing import Optional
+import typing
+from typing import Optional, cast
 
 from jinja2 import Environment
 
 from topotato.assertions import TopotatoModifier
 from topotato.base import skiptrace
 from topotato.toponom import Router
 
+if typing.TYPE_CHECKING:
+    from ..topobase import CallableNS
+    import subprocess
+
 
 jenv = Environment(
     line_comment_prefix="#" + "#",
@@ -39,6 +44,12 @@ class ExaBGP:
     >>> yield from self.peer.stop()
     """
 
+    _rtr: Router
+
+    proc: "subprocess.Popen"
+    proc_cli: "subprocess.Popen"
+    path: str
+
     def __init__(
         self, rtr: Router, conf: str, custom_environment: Optional[str] = None
     ):
@@ -48,7 +59,7 @@ def __init__(
         self._env = custom_environment
 
     class Action(TopotatoModifier):
-        _rtr: str
+        _rtr: "Router"
         _cmdobj: "ExaBGP"
 
         def __init__(self, *, name, cmdobj, **kwargs):
@@ -80,7 +91,7 @@ def __call__(self):
             self.validate()
             self.run()
 
-        def prepare_files(self):
+        def prepare_files(self) -> str:
             timestr = time.strftime("%Y%m%d-%H%M%S")
             tempdir = tempfile.mkdtemp(prefix=f"exabgp-{timestr}-")
             cli_dir = os.path.join(tempdir, "run")
@@ -139,7 +150,7 @@ def create_config():
             return tempdir
 
         def validate(self):
-            router = self.instance.routers[self._rtr.name]
+            router = cast("CallableNS", self.instance.routers[self._rtr.name])
             path = self._cmdobj.path
             self._cmdobj.proc_cli = router.popen(
                 [
@@ -158,7 +169,7 @@ def validate(self):
             self.is_cli_ok()
 
         def run(self):
-            router = self.instance.routers[self._rtr.name]
+            router = cast("CallableNS", self.instance.routers[self._rtr.name])
             path = self._cmdobj.path
             self._cmdobj.proc = router.popen(
                 [
@@ -186,7 +197,7 @@ def __init__(self, *, cmd: str, **kwargs):
         def __call__(self):
             self.is_bgp_daemon_running()
 
-            router = self.instance.routers[self._rtr.name]
+            router = cast("CallableNS", self.instance.routers[self._rtr.name])
             path = self._cmdobj.path
 
             self._cmdobj.proc_cli = router.popen(
@@ -207,7 +218,7 @@ class Stop(Action):
         def __call__(self):
             self.is_bgp_daemon_running()
 
-            router = self.instance.routers[self._rtr.name]
+            router = cast("CallableNS", self.instance.routers[self._rtr.name])
 
             path = self._cmdobj.path
             self._cmdobj.proc_cli = router.popen(

topotato/exceptions.py

@@ -170,7 +170,9 @@ def __init__(self, failed_node, *args, **kwargs):
         self.failed_node = failed_node
 
     def __str__(self):
-        sub_id = self.failed_node.nodeid.removeprefix(self.failed_node.parent.nodeid)
+        fno = self.failed_node
+        parentnodeid = fno.parent.nodeid if fno.parent else ""
+        sub_id = fno.nodeid.removeprefix(parentnodeid)
         try:
             cause = self.__cause__.__class__.__name__
         except AttributeError:

topotato/frr/core.py

@@ -308,6 +308,7 @@ def _xrefs_load(self):
 
 class _FRRConfigProtocol(Protocol):
     daemons: Collection[str]
+    configs: Dict[str, str]
 
     def want_daemon(self, daemon: str) -> bool: ...
 
@@ -607,9 +608,7 @@ def _load_config(self, daemon=None):
             )
         if vtysh.returncode != 0:
             # terminated by signal
-            raise subprocess.CalledProcessError(
-                vtysh.returncode, vtysh.args, vtysh.stdout, vtysh.stderr
-            )
+            raise subprocess.CalledProcessError(vtysh.returncode, vtysh.args, out, err)
 
     def adjust_cmdline(self, daemon: str, args: List[str]):
         pass

topotato/interactive.py

@@ -356,7 +356,10 @@ def _topotato_stop(self, item: "nodes.Item", context=None, show_context=None):
             self.show_instance_for_stop(item.instance)
 
         def _yield_from(iterator):
-            tests = item.getparent(TopotatoFunction).collect_iter(iter(iterator))
+            assert isinstance(item, TopotatoItem)
+            func_node = item.getparent(TopotatoFunction)
+            assert func_node is not None
+            tests = func_node.collect_iter(iter(iterator))
 
             for value in tests:
                 value.instance = item.instance

topotato/jailwrap.py

@@ -16,8 +16,6 @@ class FreeBSDJail:
 
     def __init__(self, name):
         self.name = name
-        self.process = None
-        self.jid = None
 
     def start(self):
         # pylint: disable=consider-using-with
@@ -37,10 +35,14 @@ def start(self):
             stdout=subprocess.PIPE,
             shell=False,
         )
+        assert self.process.stdout is not None
+
         self.jid = int(self.process.stdout.readline())
         self.process.stdout.readline()
 
     def end(self):
+        assert self.process.stdin is not None
+
         subprocess.check_call(["jail", "-r", "%d" % self.jid])
 
         self.process.stdin.close()

topotato/multicast.py

@@ -9,11 +9,15 @@
 import socket
 import struct
 
+import typing
 from typing import Any, Optional
 
 from .base import skiptrace
 from .assertions import TopotatoModifier
 
+if typing.TYPE_CHECKING:
+    from . import toponom
+
 
 __all__ = [
     "MulticastReceiver",
@@ -58,6 +62,9 @@ class MulticastReceiver:
     Join an IP (v4/v6) multicast group on a specified host and interface.
     """
 
+    _rtr: "toponom.Router"
+    _iface: "toponom.LinkIface"
+
     def __init__(self, rtr, iface):
         self._rtr = rtr
         self._iface = iface
@@ -74,7 +81,7 @@ def _get_sock_ifindex(self, router, af):
         return self._sock, self._ifindex
 
     class Action(TopotatoModifier):
-        _rtr: str
+        _rtr: "toponom.Router"
         _cmdobj: "MulticastReceiver"
         _group: Any
         _source: Any

topotato/pcapng.py

@@ -34,7 +34,7 @@
 )
 
 
-_EndianType = Union[Literal["<"], Literal[">"]]
+_EndianType = Union[Literal["<"], Literal[">"], Literal["="]]
 
 
 def _pad(data: bytes) -> bytes:

topotato/potatool.py

@@ -23,6 +23,7 @@
 
 import typing
 from typing import (
+    cast,
     Any,
     Callable,
     ClassVar,
@@ -93,9 +94,9 @@ def reselect(cls):
         if cls.want_sess is None:
             if len(cls.running) == 1:
                 # if there is only 1 running session, it is selected by default
-                cls.sel_sess = list(cls.running.values())[0]
+                cls.sel_sess = cast(PotatoolSession, list(cls.running.values())[0])
         elif cls.want_sess in cls.running:
-            cls.sel_sess = cls.running[cls.want_sess]
+            cls.sel_sess = cast(PotatoolSession, cls.running[cls.want_sess])
 
         if cls.sel_sess is not None:
             if cls.want_rtr in cls.sel_sess.routers:
@@ -164,8 +165,8 @@ def do(cls, command: str):
         return cls.apply(cmdname, cmd, words)
 
     @classmethod
-    def _fillkwargs(cls, cmdname, argspec, hints):
-        kwargs = {}
+    def _fillkwargs(cls, cmdname, argspec, hints) -> Dict[str, Any]:
+        kwargs: Dict[str, Any] = {}
         for kwname in argspec.kwonlyargs:
             hint = hints.get(kwname)
             hint, optional = typing_is_optional(hint)
@@ -452,7 +453,7 @@ def main():
         readline.read_history_file(histfile)
         hist_len = readline.get_current_history_length()
     except FileNotFoundError:
-        with open(histfile, "wb") as fd:
+        with open(histfile, "wb") as _:
             pass
         hist_len = 0
 
@@ -500,7 +501,7 @@ def rl_prompt():
             )
             if PotatoolSession.sel_rtr is not None:
                 state += (
-                    " \033[38;5;118m%s \033[38;5;246m(\033[38;5;252m%d\033[38;5;246m)"
+                    " \033[38;5;118m%s \033[38;5;246m(\033[38;5;252m%r\033[38;5;246m)"
                     % (PotatoolSession.sel_rtr.name, PotatoolSession.sel_rtr.pid)
                 )
             elif PotatoolSession.want_rtr is not None: