5 import multiprocessing.connection as mpc
12 log.error('Uncaught exception', exc_info=args)
14 print('Uncaught exception', args)
15 os.kill(0, signal.SIGTERM)
17 sys.excepthook = on_error
20 sys.path.append(r'$SRC_ROOT')
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
27 this = Tinc('$NODE_NAME')
28 log = new_logger(this.name)
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)
34 evt.test = '$TEST_NAME'
35 evt.node = '$NODE_NAME'
36 evt.script = '$SCRIPT_NAME'
40 for retry in range(1, 10):
42 with mpc.Client($NOTIFICATIONS_ADDR, family=MPC_FAMILY, authkey=$AUTH_KEY) as conn:
44 log.debug(f'sent notification')
46 except Exception as ex:
47 log.error(f'notification failed', exc_info=ex)
51 log.debug('running user code')
53 log.debug('user code finished')
54 except Exception as ex:
55 log.error('user code failed', exc_info=ex)