|
| 1 | +from __future__ import annotations |
| 2 | + |
| 3 | +""" |
| 4 | +External notifications for events that occur during testsuite execution. |
| 5 | +
|
| 6 | +Each time a tracked event occurs during the testsuite execution (see the |
| 7 | +``TestNotification`` subclasses below for the kinds of events that are |
| 8 | +tracked), a user-provided command is executed with arguments that give details |
| 9 | +about the event that occurred. |
| 10 | +""" |
| 11 | + |
| 12 | +import abc |
| 13 | +import dataclasses |
| 14 | +import importlib |
| 15 | +import logging |
| 16 | +import re |
| 17 | +import shlex |
| 18 | +import subprocess |
| 19 | +from typing import Callable, TYPE_CHECKING |
| 20 | + |
| 21 | +from e3.testsuite.result import TestResultSummary |
| 22 | + |
| 23 | +if TYPE_CHECKING: |
| 24 | + from e3.testsuite import TestsuiteCore |
| 25 | + |
| 26 | + |
| 27 | +logger = logging.getLogger("testsuite") |
| 28 | + |
| 29 | + |
| 30 | +@dataclasses.dataclass(frozen=True) |
| 31 | +class TestNotification: |
| 32 | + """Notification for a given test.""" |
| 33 | + |
| 34 | + test_name: str |
| 35 | + """ |
| 36 | + Name of the test that this notification refers to. |
| 37 | + """ |
| 38 | + |
| 39 | + @abc.abstractmethod |
| 40 | + def to_args(self) -> list[str]: |
| 41 | + raise NotImplementedError |
| 42 | + |
| 43 | + |
| 44 | +@dataclasses.dataclass(frozen=True) |
| 45 | +class TestQueueNotification(TestNotification): |
| 46 | + """Notification to signal that a test was queued.""" |
| 47 | + |
| 48 | + def to_args(self) -> list[str]: |
| 49 | + return ["--queue", self.test_name] |
| 50 | + |
| 51 | + |
| 52 | +@dataclasses.dataclass(frozen=True) |
| 53 | +class TestStartNotification(TestNotification): |
| 54 | + """Notification to signal that the execution of a test has just started.""" |
| 55 | + |
| 56 | + def to_args(self) -> list[str]: |
| 57 | + return ["--start", self.test_name] |
| 58 | + |
| 59 | + |
| 60 | +@dataclasses.dataclass(frozen=True) |
| 61 | +class TestResultNotification(TestNotification): |
| 62 | + """Notification to signal the addition of a new test result.""" |
| 63 | + |
| 64 | + result: TestResultSummary |
| 65 | + """Summary for the added test result.""" |
| 66 | + |
| 67 | + yaml_result_filename: str |
| 68 | + """Absolute filename for the YAML file that stores the test result. |
| 69 | +
|
| 70 | + This YAML file must be loaded through the ``TestResult.load`` static |
| 71 | + method. |
| 72 | + """ |
| 73 | + |
| 74 | + def to_args(self) -> list[str]: |
| 75 | + return ["--result", self.test_name, self.yaml_result_filename] |
| 76 | + |
| 77 | + |
| 78 | +@dataclasses.dataclass(frozen=True) |
| 79 | +class TestEndNotification(TestNotification): |
| 80 | + """Notification to signal that the execution of a test has just ended.""" |
| 81 | + |
| 82 | + def to_args(self) -> list[str]: |
| 83 | + return ["--end", self.test_name] |
| 84 | + |
| 85 | + |
| 86 | +class InvalidNotifyCommand(Exception): |
| 87 | + pass |
| 88 | + |
| 89 | + |
| 90 | +class EventNotifier: |
| 91 | + """Abstraction to send notifications.""" |
| 92 | + |
| 93 | + python_cmd_re = re.compile( |
| 94 | + r"(?P<module>[a-zA-Z_][a-zA-Z0-9_]*(\.[a-zA-Z_][a-zA-Z0-9_]*)*)" |
| 95 | + r":" |
| 96 | + r"(?P<callback>[a-zA-Z_][a-zA-Z0-9_]*)" |
| 97 | + ) |
| 98 | + |
| 99 | + def __init__(self, testsuite: TestsuiteCore, notify_cmd: str | None): |
| 100 | + """Initialize an EventNotifier. |
| 101 | +
|
| 102 | + :param testsuite: Testsuite instance for which this event notifier is |
| 103 | + created. |
| 104 | + :param notify_cmd: Base command line arguments for the notification |
| 105 | + command. Notification specifics are added as extra arguments for |
| 106 | + each notification that is sent. If ``None``, notifications are |
| 107 | + discarded. |
| 108 | +
|
| 109 | + This raises an `InvalidNotifyCommand` if `notify_cmd` cannot be |
| 110 | + decoded/loaded. |
| 111 | + """ |
| 112 | + self.notify_callback: Callable[[TestNotification], None] | None = None |
| 113 | + self.notify_cmd: list[str] | None = None |
| 114 | + self._parse_notify_cmd(testsuite, notify_cmd) |
| 115 | + |
| 116 | + def _parse_notify_cmd( |
| 117 | + self, |
| 118 | + testsuite: TestsuiteCore, |
| 119 | + cmd: str | None, |
| 120 | + ) -> None: |
| 121 | + """Decode the given notify command.""" |
| 122 | + if cmd is None: |
| 123 | + return |
| 124 | + |
| 125 | + python_prefix = "python:" |
| 126 | + if cmd.startswith(python_prefix): |
| 127 | + m = self.python_cmd_re.match(cmd[len(python_prefix) :]) |
| 128 | + if not m: |
| 129 | + logger.exception(f"Wrong format: {cmd!r}") |
| 130 | + raise InvalidNotifyCommand() |
| 131 | + |
| 132 | + module_name = m.group("module") |
| 133 | + callback_name = m.group("callback") |
| 134 | + |
| 135 | + try: |
| 136 | + module = importlib.import_module(module_name) |
| 137 | + except ImportError as exc: |
| 138 | + logger.exception(f"Cannot load module {module_name!r}") |
| 139 | + raise InvalidNotifyCommand from exc |
| 140 | + |
| 141 | + try: |
| 142 | + callback = getattr(module, callback_name) |
| 143 | + except AttributeError as exc: |
| 144 | + logger.exception(f"Cannot load callback {callback_name!r}") |
| 145 | + raise InvalidNotifyCommand from exc |
| 146 | + |
| 147 | + try: |
| 148 | + self.notify_callback = callback(testsuite) |
| 149 | + except Exception as exc: |
| 150 | + logger.exception("Cannot create notification callback") |
| 151 | + raise InvalidNotifyCommand from exc |
| 152 | + return |
| 153 | + |
| 154 | + self.notify_cmd = shlex.split(cmd) |
| 155 | + |
| 156 | + def notify(self, notification: TestNotification) -> None: |
| 157 | + """Send a given notification.""" |
| 158 | + if self.notify_callback is not None: |
| 159 | + try: |
| 160 | + self.notify_callback(notification) |
| 161 | + except Exception: |
| 162 | + logger.exception( |
| 163 | + "Error while executing the event notification callback" |
| 164 | + ) |
| 165 | + elif self.notify_cmd is not None: |
| 166 | + try: |
| 167 | + subprocess.check_call( |
| 168 | + self.notify_cmd + notification.to_args(), |
| 169 | + stdin=subprocess.DEVNULL, |
| 170 | + ) |
| 171 | + except (OSError, subprocess.SubprocessError): |
| 172 | + logger.exception( |
| 173 | + "Error while running the event notification command" |
| 174 | + ) |
| 175 | + |
| 176 | + def notify_test_queue(self, test_name: str) -> None: |
| 177 | + self.notify(TestQueueNotification(test_name)) |
| 178 | + |
| 179 | + def notify_test_start(self, test_name: str) -> None: |
| 180 | + self.notify(TestStartNotification(test_name)) |
| 181 | + |
| 182 | + def notify_test_result( |
| 183 | + self, |
| 184 | + test_name: str, |
| 185 | + result: TestResultSummary, |
| 186 | + yaml_result_filename: str, |
| 187 | + ) -> None: |
| 188 | + self.notify( |
| 189 | + TestResultNotification(test_name, result, yaml_result_filename) |
| 190 | + ) |
| 191 | + |
| 192 | + def notify_test_end(self, test_name: str) -> None: |
| 193 | + self.notify(TestEndNotification(test_name)) |
0 commit comments