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

   1 """Classes related to creation and control of tincd scripts."""
   2
   3 import os
   4 import typing as T
   5 from enum import Enum
   6
   7 from .log import log
   8 from .event import Notification
   9 from .notification import notifications
  10
  11
  12 class Script(Enum):
  13     """A list of supported tincd scripts.
  14     hosts/XXX-{up,down} are missing because we generate node names at runtime.
  15     """
  16
  17     TINC_UP = "tinc-up"
  18     TINC_DOWN = "tinc-down"
  19     HOST_UP = "host-up"
  20     HOST_DOWN = "host-down"
  21     SUBNET_UP = "subnet-up"
  22     SUBNET_DOWN = "subnet-down"
  23     INVITATION_CREATED = "invitation-created"
  24     INVITATION_ACCEPTED = "invitation-accepted"
  25
  26
  27 # Since we rely on dynamically created node names, we cannot put 'hosts/XXX-up' in an enum.
  28 # This is the reason we sometimes need strings to type script variables.
  29 ScriptType = T.Union[Script, str]
  30
  31
  32 class TincScript:
  33     """Control created tincd scripts and receive notifications from them."""
  34
  35     _node: str
  36     _path: str
  37     _script: str
  38
  39     def __init__(self, node: str, script: str, path: str) -> None:
  40         self._node = node
  41         self._script = script
  42         self._path = path
  43
  44     def __str__(self):
  45         return f"{self._node}/{self._script}"
  46
  47     @T.overload
  48     def wait(self) -> Notification:
  49         """Wait for the script to finish, returning the notification sent by the script."""
  50         return self.wait()
  51
  52     @T.overload
  53     def wait(self, timeout: float) -> T.Optional[Notification]:
  54         """Wait for the script to finish, returning the notification sent by the script.
  55         If nothing arrives before timeout expires, None is returned."""
  56         return self.wait(timeout)
  57
  58     def wait(self, timeout: T.Optional[float] = None) -> T.Optional[Notification]:
  59         """Wait for the script to finish. See overloads above."""
  60         log.debug("waiting for script %s/%s", self._node, self._script)
  61         if timeout is None:
  62             return notifications.get(self._node, self._script)
  63         return notifications.get(self._node, self._script, timeout)
  64
  65     @property
  66     def enabled(self) -> bool:
  67         """Check if script is enabled."""
  68         if os.name == "nt":
  69             return os.path.exists(self._path)
  70         return os.access(self._path, os.X_OK)
  71
  72     def disable(self) -> None:
  73         """Disable the script by renaming it."""
  74         log.debug("disabling script %s/%s", self._node, self._script)
  75         assert self.enabled
  76         os.rename(self._path, self._disabled_name)
  77
  78     def enable(self) -> None:
  79         """Enable the script by renaming it back."""
  80         log.debug("enabling script %s/%s", self._node, self._script)
  81         assert not self.enabled
  82         os.rename(self._disabled_name, self._path)
  83
  84     @property
  85     def _disabled_name(self) -> str:
  86         return f"{self._path}.disabled"