Coverage for src / ocarina / dsl / testing / watcher.py: 0.00%

1"""Concurrent observer for test scenarios. 

2 

3A Watcher runs in a background daemon thread alongside a test chain. 

4It polls a user-defined callback at a fixed interval, giving the callback 

5full control over what to detect, how to deduplicate, and when to report. 

6 

7Lifecycle (managed by TestSuite): 

8 1. start(driver, logger, take_screenshot) 

9 → spawns a daemon thread that calls callback(self) in a loop. 

10 2. [test_chain executes in the main thread] 

11 3. stop() 

12 → signals the loop to exit and joins the thread. 

13 

14The callback receives the Watcher instance itself, which exposes: 

15 - watcher.driver — the live WebDriver (read-only property) 

16 - watcher.cache — a mutable set[str] for DOM fingerprints 

17 - watcher.report(msg) — logs + screenshots when something is detected 

18 

19Deduplication pattern: 

20 The callback is responsible for avoiding repeated reports of the same 

21 element. The recommended approach is to compute a fingerprint (text, 

22 hash, element id, etc.) and check it against watcher.cache before 

23 calling watcher.report(). Once stored in cache, the fingerprint will 

24 never be reported again for the lifetime of the watcher. 

25 

26Error isolation: 

27 Any exception raised inside the callback is silently suppressed. 

28 The watcher thread will never crash the test, never propagate errors, 

29 and never interfere with the main test chain execution. 

30 

31Example: 

32 >>> def watch_cookie_banner(watcher: Watcher) -> None: 

33 ... banner = CookieBanner(watcher.driver) 

34 ... if not banner.is_visible(): 

35 ... return 

36 ... fingerprint = banner.get_text() 

37 ... if fingerprint in watcher.cache: 

38 ... return 

39 ... watcher.cache.add(fingerprint) 

40 ... watcher.report(f"Cookie banner detected: {fingerprint!r}") 

41 ... 

42 >>> scenario = Scenario( 

43 ... test_chain=[...], 

44 ... watchers=[ 

45 ... Watcher(callback=watch_cookie_banner, poll_interval=1.0), 

46 ... ], 

47 ... ) 

48 

49""" 

50 

51import threading 

52from contextlib import suppress 

53from typing import TYPE_CHECKING, final 

54 

55if TYPE_CHECKING: 

56 from collections.abc import Callable 

57 

58 from ocarina.ports.ilogger import ILogger 

59 from ocarina.ports.itake_screenshot import ITakeScreenshot 

60 

61 

62@final 

63class Watcher[Driver]: 

64 """Concurrent observer that enriches test reports with live page signals. 

65 

66 Runs as a daemon thread alongside the test chain, polling a user-defined 

67 callback at a fixed interval. The callback decides what to detect, how 

68 to deduplicate, and when to emit a log + screenshot via report(). 

69 

70 Args: 

71 callback: A function receiving this Watcher instance. 

72 Called repeatedly at every poll cycle. 

73 Must never raise — exceptions are suppressed internally. 

74 Use watcher.driver to access the WebDriver, 

75 watcher.cache to deduplicate, 

76 and watcher.report() to emit a signal. 

77 name: Name of the watcher. 

78 poll_interval: Seconds to wait between callback invocations. 

79 Defaults to 0.5s. Lower values increase sensitivity 

80 but add CPU pressure. 0.5 to 1.0s is appropriate for 

81 most browser observation use cases. 

82 

83 """ 

84 

85 def __init__( 

86 self, 

87 *, 

88 callback: Callable[[Watcher[Driver]], None], 

89 name: str, 

90 poll_interval: float | None = None, 

91 ) -> None: 

92 """Initialize the watcher with its callback and polling interval.""" 

93 self._callback = callback 

94 self._poll_interval = ( 

95 poll_interval if poll_interval is not None and poll_interval > 0.1 else 0.5 # noqa: PLR2004 

96 ) 

97 # Per-call: a fresh Event is created on every start() and passed by 

98 # argument to the worker thread. The previous Event (if any) is set 

99 # here so that a leaked thread from a stop() that timed out can never 

100 # see a "cleared" event and resume polling against new state. 

101 self._stop_event: threading.Event | None = None 

102 self._cache: set[str] = set() 

103 self._thread: threading.Thread | None = None 

104 self._logger: ILogger | None = None 

105 self._take_screenshot: ITakeScreenshot[Driver] | None = None 

106 self._driver: Driver | None = None 

107 self.name = name 

108 

109 @property 

110 def driver(self) -> Driver: 

111 """The live WebDriver instance for this test attempt. 

112 

113 Injected by TestSuite at start() time. 

114 Available inside the callback for the full duration of the test. 

115 

116 Raises: 

117 RuntimeError: If accessed before start() has been called. 

118 

119 Returns: 

120 The WebDriver instance passed to start(). 

121 

122 """ 

123 if self._driver is None: 

124 msg = "Watcher.driver accessed before start() was called." 

125 raise RuntimeError(msg) 

126 return self._driver 

127 

128 @property 

129 def cache(self) -> set[str]: 

130 """Mutable set of DOM fingerprints already reported. 

131 

132 Owned by the callback — the Watcher never reads or modifies it. 

133 Use it to deduplicate signals across poll cycles. 

134 

135 Typical usage: 

136 fingerprint = element.text or hash(element.id) 

137 if fingerprint in watcher.cache: 

138 return 

139 watcher.cache.add(fingerprint) 

140 watcher.report(f"Detected: {fingerprint}") 

141 

142 Returns: 

143 A mutable set[str] scoped to this Watcher instance. 

144 

145 """ 

146 return self._cache 

147 

148 def report(self, message: str, *, label: str = "WATCHER") -> None: 

149 """Emit a log entry and capture a screenshot. 

150 

151 Call this from inside the callback when something worth reporting 

152 is detected. Both the log and the screenshot are silently suppressed 

153 on failure — a broken report must never crash the watcher thread. 

154 

155 Args: 

156 message: Human-readable description of what was detected. 

157 Will appear in the logger output and the Allure report. 

158 label: Screenshot filename label. Defaults to "WATCHER". 

159 Use a descriptive label to distinguish screenshots 

160 from multiple watchers in the same scenario. 

161 

162 Example: 

163 >>> watcher.report("Unexpected modal appeared", label="MODAL") 

164 

165 """ 

166 if ( 

167 self._driver is None 

168 or self._logger is None 

169 or self._take_screenshot is None 

170 ): 

171 return 

172 

173 # Encode the invariant: we report only while actively observing. Once 

174 # stop() has been requested, a callback still in flight from before the 

175 # stop must not log or screenshot. This also closes a benign race at 

176 # teardown where the driver gets disposed just as a leaked callback 

177 # would have tried to screenshot it. 

178 if self._stop_event is None or self._stop_event.is_set(): 

179 return 

180 

181 with suppress(Exception): 

182 self._logger.info(message) 

183 self._take_screenshot(self._driver, self._logger, label) 

184 

185 def start( 

186 self, 

187 driver: Driver, 

188 logger: ILogger, 

189 take_screenshot: ITakeScreenshot[Driver], 

190 ) -> None: 

191 """Start the observation loop in a background daemon thread. 

192 

193 Called by TestSuite before test_chain executes. 

194 Injects the runtime dependencies (driver, logger, screenshot) 

195 that the callback needs, then spawns the polling thread. 

196 

197 The thread is a daemon — it will not prevent process exit if 

198 TestSuite forgets to call stop(). 

199 

200 Args: 

201 driver: The live WebDriver for this test attempt. 

202 logger: The ILogger scoped to the current test taxonomy. 

203 take_screenshot: The ITakeScreenshot callable bound to the driver. 

204 

205 """ 

206 # Defensively terminate any thread leaked from a previous start() 

207 # whose stop()/join() timed out. Its loop holds the previous Event 

208 # in its local frame; setting it here guarantees the leaked thread 

209 # exits at the end of its current callback iteration instead of 

210 # racing with the new thread on shared self._driver / self._logger. 

211 if self._stop_event is not None: 

212 self._stop_event.set() 

213 

214 self._driver = driver 

215 self._logger = logger 

216 self._take_screenshot = take_screenshot 

217 stop_event = threading.Event() 

218 self._stop_event = stop_event 

219 self._thread = threading.Thread( 

220 target=self._loop, args=(stop_event,), daemon=True 

221 ) 

222 self._thread.start() 

223 

224 def stop(self) -> None: 

225 """Signal the observation loop to stop and wait for the thread to exit. 

226 

227 Called by TestSuite after test_chain completes, regardless of whether 

228 the test passed or failed. 

229 to finish its current cycle and exit cleanly. 

230 

231 Does not raise even if the thread is slow or unresponsive — the 

232 daemon flag ensures it is collected by the process regardless. 

233 

234 """ 

235 if self._stop_event is not None: 

236 self._stop_event.set() 

237 if self._thread is not None: 

238 self._thread.join(timeout=self._poll_interval * 2) 

239 

240 def _loop(self, stop_event: threading.Event) -> None: 

241 """Poll loop executed in the background thread. 

242 

243 Repeatedly calls callback(self) at poll_interval, suppressing any 

244 exception that escapes the callback. Exits cleanly when stop_event 

245 is set by stop(). 

246 

247 Never raises. Never blocks the main thread. 

248 

249 ``stop_event`` is bound to *this* thread's lifetime. A subsequent 

250 start() will set this Event before swapping in a fresh one for the 

251 new thread — see ``start()`` for the rationale. 

252 

253 """ 

254 while not stop_event.is_set(): 

255 with suppress(Exception): 

256 self._callback(self) 

257 stop_event.wait(self._poll_interval)