1 """Classes related to creation and control of tincd scripts."""
8 from .event import Notification
9 from .notification import notifications
13 """A list of supported tincd scripts.
14 hosts/XXX-{up,down} are missing because we generate node names at runtime.
18 TINC_DOWN = "tinc-down"
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"
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]
33 """Control created tincd scripts and receive notifications from them."""
39 def __init__(self, node: str, script: str, path: str) -> None:
45 return f"{self._node}/{self._script}"
48 def wait(self) -> Notification:
49 """Wait for the script to finish, returning the notification sent by the script."""
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)
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)
62 return notifications.get(self._node, self._script)
63 return notifications.get(self._node, self._script, timeout)
66 def enabled(self) -> bool:
67 """Check if script is enabled."""
69 return os.path.exists(self._path)
70 return os.access(self._path, os.X_OK)
72 def disable(self) -> None:
73 """Disable the script by renaming it."""
74 log.debug("disabling script %s/%s", self._node, self._script)
76 os.rename(self._path, self._disabled_name)
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)
85 def _disabled_name(self) -> str:
86 return f"{self._path}.disabled"