topotato: typing fix executable path lookups

A lot of `| None` causing errors in type checks on these lookups. Create a tiny wrapper to help fixing it across the board. Signed-off-by: David Lamparter <[email protected]>
opensourcerouting · Sep 17, 2024 · acce7b8 · acce7b8
1 parent 3a0b7b1
commit acce7b8
Show file tree

Hide file tree

Showing 3 changed files with 26 additions and 21 deletions.
diff --git a/topotato/nswrap.py b/topotato/nswrap.py
@@ -14,7 +14,7 @@
 from typing import List, ClassVar
 
 from .defer import subprocess
-from .utils import LockedFile
+from .utils import LockedFile, PathDict
 
 _libc = ctypes.CDLL(ctypes.util.find_library("c"), use_errno=True)
 
@@ -93,11 +93,13 @@ class LinuxNamespace:
     name: str
     pid: int
 
-    _exec = {
-        "unshare": None,
-        "nsenter": None,
-        "tini": None,
-    }
+    _exec = PathDict(
+        {
+            "unshare": None,
+            "nsenter": None,
+            "tini": None,
+        }
+    )
 
     taskdir: ClassVar[str] = "/tmp/topotato"
 
@@ -123,14 +125,14 @@ def start(self):
 
         self.process = subprocess.Popen(
             [
-                self._exec.get("unshare", "unshare"),
+                self._exec("unshare"),
                 "-u",
                 "-m",
                 "-n",
                 "-p",
                 "-f",
                 "--mount-proc",
-                self._exec.get("tini", "tini"),
+                self._exec("tini"),
                 "-g",
                 sys.executable,
                 "--",
@@ -200,7 +202,7 @@ def end(self):
 
     def prefix(self, kwargs) -> List[str]:
         ret = [
-            str(self._exec.get("nsenter", "nsenter")),
+            self._exec("nsenter"),
             "-t",
             str(self.pid),
             "-m",

diff --git a/topotato/topolinux.py b/topotato/topolinux.py
@@ -98,9 +98,7 @@ def pytest_topotato_envcheck(cls, session, result: EnvcheckResult):
             if cur is None:
                 result.error("%s is required to run on Linux systems", name)
 
-        ip_ver = subprocess.check_output([cls._exec.get("ip") or "ip", "-V"]).decode(
-            "UTF-8"
-        )
+        ip_ver = subprocess.check_output([cls._exec("ip"), "-V"]).decode("UTF-8")
         ip_ver_m = re.search(r"iproute2-((?:ss)?[\d\.]+)", ip_ver)
         if ip_ver_m and ip_ver_m.group(1).startswith("ss"):
             ver = ip_ver_m.group(1)
@@ -120,7 +118,7 @@ def pytest_topotato_envcheck(cls, session, result: EnvcheckResult):
             _logger.warning(
                 "cannot parse iproute2 version %r from %r",
                 ip_ver,
-                cls._exec.get("ip") or "ip",
+                cls._exec("ip"),
             )
 
     class BaseNS(topobase.CallableEnvMixin, LinuxNamespace, topobase.BaseNS):
@@ -153,7 +151,7 @@ def tempfile(self, name: str) -> str:
 
         def start(self):
             super().start()
-            self.check_call([self._exec.get("ip", "ip"), "link", "set", "lo", "up"])
+            self.check_call([self._exec("ip"), "link", "set", "lo", "up"])
 
         def end_prep(self):
             pass
@@ -184,7 +182,7 @@ def add(arr):
 
             def ip_r_call(extra=None):
                 text = self.check_output(
-                    [self._exec.get("ip", "ip"), "-%d" % af, "-j", "route", "list"]
+                    [self._exec("ip"), "-%d" % af, "-j", "route", "list"]
                     + (extra or [])
                 )
                 text = self.iproute_json_re.sub(rb'"type":"\1"', text)
@@ -332,7 +330,7 @@ def link_set(self, iface: LinkIface, state: bool):
             ifn = ifname(self.name, iface.ifname)
             self.instance.switch_ns.check_call(
                 [
-                    str(self._exec.get("ip", "ip")),
+                    self._exec("ip"),
                     "link",
                     "set",
                     ifn,
@@ -409,7 +407,7 @@ def start(self):
                 self.bridges.append(brname)
                 self.switch_ns.check_call(
                     [
-                        self._exec.get("ip", "ip"),
+                        self._exec("ip"),
                         "link",
                         "add",
                         "name",
@@ -422,7 +420,7 @@ def start(self):
                 )
                 self.switch_ns.check_call(
                     [
-                        self._exec.get("ip", "ip"),
+                        self._exec("ip"),
                         "link",
                         "set",
                         ifname(link.a.endpoint.name, link.a.ifname),
@@ -433,7 +431,7 @@ def start(self):
                 )
                 self.switch_ns.check_call(
                     [
-                        self._exec.get("ip", "ip"),
+                        self._exec("ip"),
                         "link",
                         "set",
                         ifname(link.b.endpoint.name, link.b.ifname),
@@ -448,7 +446,7 @@ def start(self):
             self.bridges.append(brname)
             self.switch_ns.check_call(
                 [
-                    self._exec.get("ip", "ip"),
+                    self._exec("ip"),
                     "link",
                     "add",
                     "name",
@@ -462,7 +460,7 @@ def start(self):
             for iface in lan.ifaces:
                 self.switch_ns.check_call(
                     [
-                        self._exec.get("ip", "ip"),
+                        self._exec("ip"),
                         "link",
                         "set",
                         ifname(iface.other.endpoint.name, iface.other.ifname),

diff --git a/topotato/utils.py b/topotato/utils.py
@@ -140,6 +140,11 @@ def apply_kwargs_maybe(func: Callable[P, T], **args) -> Callable[P, T]:
     return functools.partial(func, **existing)
 
 
+class PathDict(dict[str, Optional[str]]):
+    def __call__(self, k: str) -> str:
+        return self.get(k) or k
+
+
 class json_cmp_result:
     "json_cmp result class for better assertion messages"