tinc-vpn.org Git - tinc/blob - test/integration/testlib/external.py

   1 """Wrappers for running external commands."""
   2
   3 import subprocess as subp
   4 import atexit
   5 import typing as T
   6
   7 from .log import log
   8
   9 _netns_created: T.Set[str] = set()
  10
  11
  12 def _netns_cleanup() -> None:
  13     for namespace in _netns_created.copy():
  14         netns_delete(namespace)
  15
  16
  17 atexit.register(_netns_cleanup)
  18
  19
  20 def _netns_action(action: str, namespace: str) -> bool:
  21     log.debug("%s network namespace %s", action, namespace)
  22
  23     res = subp.run(["ip", "netns", action, namespace], check=False)
  24     if res.returncode:
  25         log.error("could not %s netns %s", action, namespace)
  26     else:
  27         log.debug("OK %s netns %s", action, namespace)
  28
  29     return not res.returncode
  30
  31
  32 def netns_delete(namespace: str) -> bool:
  33     """Remove a previously created network namespace."""
  34     success = _netns_action("delete", namespace)
  35     if success:
  36         _netns_created.remove(namespace)
  37     return success
  38
  39
  40 def netns_add(namespace: str) -> bool:
  41     """Add a network namespace (which can be removed manually or automatically at exit)."""
  42     success = _netns_action("add", namespace)
  43     if success:
  44         _netns_created.add(namespace)
  45     return success