Coverage for src / ocarina / dsl / testing / watcher.py: 100.00%
50 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-05-18 20:23 +0200
1"""Concurrent observer for test scenarios.
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.
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.
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
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.
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.
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 ... )
49"""
51import threading
52from contextlib import suppress
53from typing import TYPE_CHECKING, final
55if TYPE_CHECKING:
56 from collections.abc import Callable
58 from ocarina.ports.ilogger import ILogger
59 from ocarina.ports.itake_screenshot import ITakeScreenshot
62@final
63class Watcher[Driver]:
64 """Concurrent observer that enriches test reports with live page signals.
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().
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.
83 """
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
109 @property
110 def driver(self) -> Driver:
111 """The live WebDriver instance for this test attempt.
113 Injected by TestSuite at start() time.
114 Available inside the callback for the full duration of the test.
116 Raises:
117 RuntimeError: If accessed before start() has been called.
119 Returns:
120 The WebDriver instance passed to start().
122 """
123 if self._driver is None:
124 msg = "Watcher.driver accessed before start() was called."
125 raise RuntimeError(msg)
126 return self._driver
128 @property
129 def cache(self) -> set[str]:
130 """Mutable set of DOM fingerprints already reported.
132 Owned by the callback — the Watcher never reads or modifies it.
133 Use it to deduplicate signals across poll cycles.
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}")
142 Returns:
143 A mutable set[str] scoped to this Watcher instance.
145 """
146 return self._cache
148 def report(self, message: str, *, label: str = "WATCHER") -> None:
149 """Emit a log entry and capture a screenshot.
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.
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.
162 Example:
163 >>> watcher.report("Unexpected modal appeared", label="MODAL")
165 """
166 if (
167 self._driver is None
168 or self._logger is None
169 or self._take_screenshot is None
170 ):
171 return
173 with suppress(Exception):
174 self._logger.info(message)
175 self._take_screenshot(self._driver, self._logger, label)
177 def start(
178 self,
179 driver: Driver,
180 logger: ILogger,
181 take_screenshot: ITakeScreenshot[Driver],
182 ) -> None:
183 """Start the observation loop in a background daemon thread.
185 Called by TestSuite before test_chain executes.
186 Injects the runtime dependencies (driver, logger, screenshot)
187 that the callback needs, then spawns the polling thread.
189 The thread is a daemon — it will not prevent process exit if
190 TestSuite forgets to call stop().
192 Args:
193 driver: The live WebDriver for this test attempt.
194 logger: The ILogger scoped to the current test taxonomy.
195 take_screenshot: The ITakeScreenshot callable bound to the driver.
197 """
198 # Defensively terminate any thread leaked from a previous start()
199 # whose stop()/join() timed out. Its loop holds the previous Event
200 # in its local frame; setting it here guarantees the leaked thread
201 # exits at the end of its current callback iteration instead of
202 # racing with the new thread on shared self._driver / self._logger.
203 if self._stop_event is not None:
204 self._stop_event.set()
206 self._driver = driver
207 self._logger = logger
208 self._take_screenshot = take_screenshot
209 stop_event = threading.Event()
210 self._stop_event = stop_event
211 self._thread = threading.Thread(
212 target=self._loop, args=(stop_event,), daemon=True
213 )
214 self._thread.start()
216 def stop(self) -> None:
217 """Signal the observation loop to stop and wait for the thread to exit.
219 Called by TestSuite after test_chain completes, regardless of whether
220 the test passed or failed.
221 to finish its current cycle and exit cleanly.
223 Does not raise even if the thread is slow or unresponsive — the
224 daemon flag ensures it is collected by the process regardless.
226 """
227 if self._stop_event is not None:
228 self._stop_event.set()
229 if self._thread is not None:
230 self._thread.join(timeout=self._poll_interval * 2)
232 def _loop(self, stop_event: threading.Event) -> None:
233 """Poll loop executed in the background thread.
235 Repeatedly calls callback(self) at poll_interval, suppressing any
236 exception that escapes the callback. Exits cleanly when stop_event
237 is set by stop().
239 Never raises. Never blocks the main thread.
241 ``stop_event`` is bound to *this* thread's lifetime. A subsequent
242 start() will set this Event before swapping in a fresh one for the
243 new thread — see ``start()`` for the rationale.
245 """
246 while not stop_event.is_set():
247 with suppress(Exception):
248 self._callback(self)
249 stop_event.wait(self._poll_interval)