tinc-vpn.org Git - tinc/blob - test/integration/systemd.py

   1 #!/usr/bin/env python3
   2
   3 """Test systemd integration."""
   4
   5 import os
   6 import socket
   7 import tempfile
   8 import time
   9
  10 from testlib import check, path
  11 from testlib.log import log
  12 from testlib.feature import Feature
  13 from testlib.const import MAXSOCKETS
  14 from testlib.proc import Tinc, Script
  15 from testlib.test import Test
  16
  17
  18 def tincd_start_socket(foo: Tinc, pass_pid: bool) -> int:
  19     """Start tincd as systemd socket activation does it."""
  20
  21     pid = os.fork()
  22     if not pid:
  23         env = {**os.environ, "LISTEN_FDS": str(MAXSOCKETS + 1)}
  24         if pass_pid:
  25             env["LISTEN_PID"] = str(os.getpid())
  26         args = [
  27             path.TINCD_PATH,
  28             "-c",
  29             foo.work_dir,
  30             "--pidfile",
  31             foo.pid_file,
  32             "--logfile",
  33             foo.sub("log"),
  34         ]
  35         assert not os.execve(path.TINCD_PATH, args, env)
  36
  37     assert pid > 0
  38
  39     _, status = os.waitpid(pid, 0)
  40     assert os.WIFEXITED(status)
  41     return os.WEXITSTATUS(status)
  42
  43
  44 def test_listen_fds(foo: Tinc) -> None:
  45     """Test systemd socket activation."""
  46
  47     foo_log = foo.sub("log")
  48
  49     log.info("foreground tincd fails with too high LISTEN_FDS")
  50     status = tincd_start_socket(foo, pass_pid=True)
  51     check.failure(status)
  52     check.in_file(foo_log, "Too many listening sockets")
  53
  54     foo.add_script(Script.TINC_UP)
  55     foo.add_script(Script.TINC_DOWN)
  56     os.remove(foo_log)
  57
  58     log.info("LISTEN_FDS is ignored without LISTEN_PID")
  59     status = tincd_start_socket(foo, pass_pid=False)
  60     foo[Script.TINC_UP].wait()
  61     foo.cmd("stop")
  62     foo[Script.TINC_DOWN].wait()
  63     check.success(status)
  64     check.not_in_file(foo_log, "Too many listening sockets")
  65
  66
  67 def recv_until(sock: socket.socket, want: bytes) -> None:
  68     """Receive from a datagram socket until a specific value is found."""
  69
  70     while True:
  71         msg = sock.recv(65000)
  72         log.info("received %s", msg)
  73         if msg == want:
  74             break
  75
  76
  77 def test_watchdog(foo: Tinc) -> None:
  78     """Test systemd watchdog."""
  79
  80     address = tempfile.mktemp()
  81     foo_log = foo.sub("log")
  82
  83     log.info("watchdog is disabled if no env vars are passed")
  84     foo.cmd("start", "--logfile", foo_log)
  85     foo.cmd("stop")
  86     check.in_file(foo_log, "Watchdog is disabled")
  87
  88     log.info("watchdog is enabled by systemd env vars")
  89     foo.add_script(Script.TINC_UP)
  90     foo.add_script(Script.TINC_DOWN)
  91
  92     with socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM) as sock:
  93         sock.bind(address)
  94
  95         watchdog = 0.5
  96         env = {"NOTIFY_SOCKET": address, "WATCHDOG_USEC": str(int(watchdog * 1e6))}
  97         proc = foo.tincd("-D", env=env)
  98         recv_until(sock, b"READY=1")
  99
 100         for _ in range(6):
 101             before = time.monotonic()
 102             recv_until(sock, b"WATCHDOG=1")
 103             spent = time.monotonic() - before
 104             assert spent < watchdog
 105
 106         foo.cmd("stop")
 107         recv_until(sock, b"STOPPING=1")
 108
 109         check.success(proc.wait())
 110
 111
 112 with Test("socket activation") as context:
 113     test_listen_fds(context.node(init=True))
 114
 115 with Test("watchdog") as context:
 116     node = context.node(init=True)
 117     if Feature.WATCHDOG in node.features:
 118         test_watchdog(node)