From a3331b96133d795aff0ef2f100bea45566734e99 Mon Sep 17 00:00:00 2001 From: David Lamparter Date: Mon, 30 Sep 2024 19:02:20 +0200 Subject: [PATCH] topotato: clean up `BaseNS.__init__` behavior There's a cycle in the inheritance graph for the OS-specific RouterNS/SwitchNS classes, which combines with "dual-use" of the `name` parameter to make a bit of a mess. This previously worked because topobase.BaseNS.__init__ was never called, but that's not great either. Use a "self-or-kwarg" helper instead. Signed-off-by: David Lamparter --- topotato/frr/core.py | 2 +- topotato/jailwrap.py | 9 ++++++--- topotato/network.py | 2 +- topotato/nswrap.py | 13 +++++++++---- topotato/topobase.py | 13 +++++++++---- topotato/topofreebsd.py | 5 ++--- topotato/topolinux.py | 9 ++++----- topotato/utils.py | 25 +++++++++++++++++++++++++ 8 files changed, 57 insertions(+), 21 deletions(-) diff --git a/topotato/frr/core.py b/topotato/frr/core.py index 5c714d4..188c0d0 100644 --- a/topotato/frr/core.py +++ b/topotato/frr/core.py @@ -503,7 +503,7 @@ def __init__( frr: FRRSetup, configs: _FRRConfigProtocol, ): - super().__init__(instance, name) + super().__init__(instance=instance, name=name) self._configs = configs self.frr = frr self.logfiles = {} diff --git a/topotato/jailwrap.py b/topotato/jailwrap.py index 38904b4..a46de90 100644 --- a/topotato/jailwrap.py +++ b/topotato/jailwrap.py @@ -8,14 +8,17 @@ import subprocess import time +from .utils import self_or_kwarg + class FreeBSDJail: name: str process: subprocess.Popen jid: int - def __init__(self, name): - self.name = name + def __init__(self, **kw): + self_or_kwarg(self, kw, "name") + super().__init__(**kw) def start(self): # pylint: disable=consider-using-with @@ -65,7 +68,7 @@ def check_output(self, cmdline, *args, **kwargs): # pylint: disable=duplicate-code if __name__ == "__main__": - ns = FreeBSDJail("test") + ns = FreeBSDJail(name="test") ns.start() ns.check_call(["ifconfig", "-a"]) ns.check_call(["/bin/sh", "-c", "sleep 3"]) diff --git a/topotato/network.py b/topotato/network.py index dbd6bcb..695a66b 100644 --- a/topotato/network.py +++ b/topotato/network.py @@ -169,4 +169,4 @@ class Host(TopotatoParams): def instantiate(self): # pylint: disable=abstract-class-instantiated - return HostNS(self.instance, self.name) + return HostNS(instance=self.instance, name=self.name) diff --git a/topotato/nswrap.py b/topotato/nswrap.py index 28606eb..35ee8ea 100644 --- a/topotato/nswrap.py +++ b/topotato/nswrap.py @@ -19,7 +19,7 @@ ) from .defer import subprocess -from .utils import LockedFile, PathDict +from .utils import LockedFile, PathDict, self_or_kwarg _libc = ctypes.CDLL(ctypes.util.find_library("c"), use_errno=True) @@ -123,9 +123,12 @@ class LinuxNamespace: ) taskdir: ClassVar[str] = "/tmp/topotato" + process: Optional[subprocess.Popen] + + def __init__(self, **kw): + self_or_kwarg(self, kw, "name") + super().__init__(**kw) - def __init__(self, name): - self.name = name self.process = None def start(self): @@ -216,6 +219,8 @@ def end(self): if self.process is None: return + assert self.process.stdin + self.process.stdin.write(b"\n") self.process.stdin.close() self.process.wait() @@ -286,7 +291,7 @@ def __exit__(self, type_, value, traceback): # pylint: disable=duplicate-code def test(): - ns = LinuxNamespace("test") + ns = LinuxNamespace(name="test") ns.start() ns.check_call(["ip", "addr", "list"]) with ns: diff --git a/topotato/topobase.py b/topotato/topobase.py index d252960..e504cea 100644 --- a/topotato/topobase.py +++ b/topotato/topobase.py @@ -27,6 +27,8 @@ ) from typing_extensions import Protocol +from .utils import self_or_kwarg + if typing.TYPE_CHECKING: from typing import ( Self, @@ -52,8 +54,11 @@ class BaseNS: instance: "NetworkInstance" - def __init__(self, _instance: "NetworkInstance", name: str) -> None: - pass + def __init__(self, *, instance: "NetworkInstance", **kw) -> None: + self.instance = instance + self_or_kwarg(self, kw, "name") + + super().__init__(**kw) def tempfile(self, name: str) -> str: """ @@ -231,7 +236,7 @@ def make(self, name: str) -> RouterNS: for specific virtual routers. This enables that. """ # pylint: disable=abstract-class-instantiated - return self.RouterNS(self, name) # type: ignore + return self.RouterNS(instance=self, name=name) # type: ignore @abstractmethod def tempfile(self, name: str) -> str: @@ -244,7 +249,7 @@ def prepare(self) -> "Self": Execute setup (create switch & router objects) for this network instance. """ # pylint: disable=abstract-class-instantiated - self.switch_ns = self.SwitchyNS(self, "switch-ns") # type: ignore + self.switch_ns = self.SwitchyNS(instance=self, name="switch-ns") # self.routers is immutable, assign as a whole routers = {} diff --git a/topotato/topofreebsd.py b/topotato/topofreebsd.py index cf37973..cfab58b 100644 --- a/topotato/topofreebsd.py +++ b/topotato/topofreebsd.py @@ -47,9 +47,8 @@ class BaseNS(topobase.CallableEnvMixin, FreeBSDJail, topobase.BaseNS): instance: "NetworkInstance" - def __init__(self, instance, name): - super().__init__(name) - self.instance = instance + def __init__(self, *, instance: "NetworkInstance", name: str): + super().__init__(instance=instance, name=name) self.tempdir = instance.tempfile(name) os.mkdir(self.tempdir) diff --git a/topotato/topolinux.py b/topotato/topolinux.py index ba6d944..4927335 100644 --- a/topotato/topolinux.py +++ b/topotato/topolinux.py @@ -134,13 +134,12 @@ class BaseNS(topobase.CallableEnvMixin, LinuxNamespace, topobase.BaseNS): rb'(? Callable[P, T]: return functools.partial(func, **existing) +def self_or_kwarg( + self: object, kw: Dict[str, Any], argname: str, fieldname: Optional[str] = None +) -> None: + """ + Cooperative `__init__` helper, for multiple constructors setting the same field. + + Handles an `__init__` parameter named `argname` assigned to `fieldname` + (default same as `argname`.) It should either already be set on `self` + (and not be in `kw` anymore), or be in `kw` (in which case it's removed + from `kw` and assigned on `self`.) + """ + + fieldname = fieldname if fieldname is not None else argname + + if hasattr(self, fieldname): + if argname in kw: + raise TypeError( + f"parameter {argname!r} given argument {kw[argname]!r} but already set on self (to {getattr(self, fieldname)!r}" + ) + else: + if argname not in kw: + raise TypeError(f"missing argument for parameter {argname!r}") + setattr(self, fieldname, kw.pop(argname)) + + class PathDict(dict[str, Optional[str]]): def __call__(self, k: str) -> str: return self.get(k) or k