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

   1 #!$PYTHON_PATH
   2
   3 import os
   4 import sys
   5 import multiprocessing.connection as mpc
   6 import typing as T
   7 import time
   8 import signal
   9
  10 def on_error(*args):
  11     try:
  12         log.error('Uncaught exception', exc_info=args)
  13     except NameError:
  14         print('Uncaught exception', args)
  15     os.kill(0, signal.SIGTERM)
  16
  17 sys.excepthook = on_error
  18
  19 os.chdir(r'$CWD')
  20 sys.path.append(r'$SRC_ROOT')
  21
  22 from testlib.proc import Tinc
  23 from testlib.event import Notification
  24 from testlib.log import new_logger
  25 from testlib.const import MPC_FAMILY
  26
  27 this = Tinc('$NODE_NAME')
  28 log = new_logger(this.name)
  29
  30 def notify_test(args: T.Dict[str, T.Any] = {}, error: T.Optional[Exception] = None):
  31     log.debug(f'sending notification to %s', $NOTIFICATIONS_ADDR)
  32
  33     evt = Notification()
  34     evt.test = '$TEST_NAME'
  35     evt.node = '$NODE_NAME'
  36     evt.script = '$SCRIPT_NAME'
  37     evt.args = args
  38     evt.error = error
  39
  40     for retry in range(1, 10):
  41         try:
  42             with mpc.Client($NOTIFICATIONS_ADDR, family=MPC_FAMILY, authkey=$AUTH_KEY) as conn:
  43                 conn.send(evt)
  44             log.debug(f'sent notification')
  45             break
  46         except Exception as ex:
  47             log.error(f'notification failed', exc_info=ex)
  48             time.sleep(0.5)
  49
  50 try:
  51     log.debug('running user code')
  52 $SCRIPT_SOURCE
  53     log.debug('user code finished')
  54 except Exception as ex:
  55     log.error('user code failed', exc_info=ex)
  56     notify_test(error=ex)
  57     sys.exit(1)
  58
  59 notify_test()