diff --git a/topotato/assertions.py b/topotato/assertions.py index f87d057..acbbb8b 100644 --- a/topotato/assertions.py +++ b/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 diff --git a/topotato/base.py b/topotato/base.py index d272d2d..4cedd6c 100644 --- a/topotato/base.py +++ b/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,7 +881,7 @@ 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) @@ -877,8 +889,8 @@ def do_start(self): 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)) diff --git a/topotato/frr/livelog.py b/topotato/frr/livelog.py index 0a838ef..359d38b 100644 --- a/topotato/frr/livelog.py +++ b/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) diff --git a/topotato/generatorwrap.py b/topotato/generatorwrap.py index cf5685d..0adba16 100644 --- a/topotato/generatorwrap.py +++ b/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 diff --git a/topotato/livescapy.py b/topotato/livescapy.py index 786d827..31e56fc 100644 --- a/topotato/livescapy.py +++ b/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 diff --git a/topotato/nswrap.py b/topotato/nswrap.py index f1b63be..68a2b29 100644 --- a/topotato/nswrap.py +++ b/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) diff --git a/topotato/parse.py b/topotato/parse.py index 681b67d..fc13025 100644 --- a/topotato/parse.py +++ b/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 diff --git a/topotato/scapyext/pim.py b/topotato/scapyext/pim.py index f265cf1..819520f 100644 --- a/topotato/scapyext/pim.py +++ b/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 diff --git a/topotato/toponom.py b/topotato/toponom.py index 61c1ce7..dcc570f 100644 --- a/topotato/toponom.py +++ b/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)