"""Controlled real-entry-point startup and module-readiness benchmark.
This module is imported only when ``SPACR_BENCHMARK_JSON`` names an output
file. The ordinary application therefore pays no import or QObject cost for
it. The benchmark still runs the ordinary ``spacr.qt.run`` path: once Home
has really painted, it presses each live sidebar button and waits for the
production timing probe to observe a painted, enabled control. After the
last registry key it writes one JSON artifact and exits Qt deliberately.
"""
from __future__ import annotations
import json
import os
import sys
import threading
import time
import weakref
from typing import Callable, Iterable, Optional, Union
from PySide6.QtCore import QObject, QTimer
from . import timing
from . import _after
OUTPUT_ENV = "SPACR_BENCHMARK_JSON"
RUN_LABEL_ENV = "SPACR_BENCHMARK_RUN"
TIMEOUT_ENV = "SPACR_BENCHMARK_TIMEOUT_S"
DEFAULT_TIMEOUT_S = 30.0
SETTLE_MS = 32
PREFERENCES_BUDGET_S = 3.0
PREFERENCES_HANG_TIMEOUT_S = 10.0
HARD_TIMEOUT_ENV = "SPACR_BENCHMARK_HARD_TIMEOUT"
HARD_TIMEOUT_GRACE_S = 1.0
def _timeout_seconds() -> float:
"""How long the startup benchmark waits before giving up.
:returns: the timeout in seconds.
"""
try:
return max(1.0, min(300.0, float(os.environ.get(
TIMEOUT_ENV, DEFAULT_TIMEOUT_S))))
except (TypeError, ValueError):
return DEFAULT_TIMEOUT_S
[docs]
class BenchmarkController(QObject):
"""Advance through Home and an exact snapshot of the live app registry.
:param app: the running :class:`QApplication`, also this object's parent.
:param window: the main window to drive.
:param keys: the module keys to visit, in order.
:param output: where to write the measurements.
:param timeout_s: how long one module may take before the run is
abandoned. ``None`` uses the default.
:param live_keys: called for the registry as it stands NOW, so a run can
check the snapshot in ``keys`` still matches the app it is driving.
:param measure_preferences: whether to open Preferences and time it.
``None`` decides from the environment.
:param preferences_factory: builds the Preferences dialog, so a test can
supply one without the real dialog.
"""
def __init__(self, app, window, keys: Iterable[str], output: str, *,
timeout_s: Optional[float] = None,
live_keys: Optional[Callable[[], Iterable[str]]] = None,
measure_preferences: Optional[bool] = None,
preferences_factory: Optional[Callable[[], object]] = None,
) -> None:
"""Drive the benchmark: open each module in turn and time it.
:param app: the running QApplication.
:param window: the main window to drive.
:param keys: the modules to open, in order.
:param output: where to write the artifact.
:param timeout_s: how long one module may take before the run fails.
:param live_keys: modules whose readiness is signalled rather than polled.
:param measure_preferences: whether to time the Preferences dialog too.
:param preferences_factory: how to build that dialog.
"""
super().__init__(app)
self.app = app
self.window = window
self.keys = tuple(str(key) for key in keys)
self._live_keys = live_keys
#: Which door the CURRENT module was opened through -- "sidebar" for
#: a row the user can click, "command palette" for a `TILELESS_APPS`
#: module that has no row and is reached from a host's button, from
#: Help or from Ctrl+K. Recorded on every result, because a number
#: that does not say how it was obtained is exactly what let fourteen
#: modules go unmeasured while the artifact reported violations.
self._door = "sidebar"
self.output = str(output)
self.timeout_s = _timeout_seconds() if timeout_s is None else float(
timeout_s)
self.results: list[dict] = []
self.phase = "home"
self.current_key: Optional[str] = None
self.index = 0
self._pending: Optional[dict] = None
self._timeout_pending = False
self._finished = False
self._written = False
self._measure_preferences = (
bool(os.environ.get(OUTPUT_ENV, "").strip())
if measure_preferences is None else bool(measure_preferences)
)
self._preferences_factory = preferences_factory
self._preferences_dialog = None
self._preferences_started_elapsed = 0.0
self._preferences_ready_at: Optional[float] = None
self._hard_timeout = None
self._armed_timeout_s = self.timeout_s
self._attempt_started = time.perf_counter()
self._attempt_started_elapsed = timing.elapsed()
self.timeout = QTimer(self)
self.timeout.setSingleShot(True)
self.timeout.timeout.connect(self._timed_out)
timing.subscribe_readiness(self._ready)
app.aboutToQuit.connect(self._application_quit)
self._arm_timeout()
def _arm_timeout(self, timeout_s: Optional[float] = None) -> None:
"""Start the clock that fails the run if a module hangs.
A HANG MUST NOT BE A HANG. Without this a module that never signals
ready leaves the benchmark waiting for ever, which in CI is an hour of
a runner rather than a failure anyone can read.
:param timeout_s: how long to allow.
"""
self._disarm_timeout()
self._timeout_pending = False
self._attempt_started = time.perf_counter()
self._attempt_started_elapsed = timing.elapsed()
self._armed_timeout_s = (
self.timeout_s if timeout_s is None else float(timeout_s)
)
self.timeout.start(max(1, int(self._armed_timeout_s * 1000.0)))
if os.environ.get(HARD_TIMEOUT_ENV, "").strip() == "1":
wall = threading.Timer(
self._armed_timeout_s + HARD_TIMEOUT_GRACE_S,
self._hard_timed_out,
)
wall.daemon = True
self._hard_timeout = wall
wall.start()
def _disarm_timeout(self) -> None:
"""Stop both the event-loop deadline and its wall-clock backstop."""
self.timeout.stop()
wall = self._hard_timeout
self._hard_timeout = None
if wall is not None:
wall.cancel()
def _hard_timed_out(self) -> None:
"""Terminate a benchmark worker whose GUI thread cannot run QTimer.
This path is enabled only in the dedicated benchmark subprocess. It
deliberately uses ``os._exit``: a normal Qt shutdown needs the very
event loop that is wedged. The parent preserves the last checkpoint
and records exit status 124 as a ratchet failure.
"""
try:
import faulthandler
print(
"spaCR benchmark hard timeout: GUI thread did not return "
f"within {self._armed_timeout_s:.1f} seconds",
file=sys.stderr,
flush=True,
)
faulthandler.dump_traceback(file=sys.stderr, all_threads=True)
finally:
os._exit(124)
def _ready(self, entry: dict) -> None:
"""Record that one module finished opening.
:param entry: the module's timing entry.
"""
if (self._finished or self._pending is not None
or self._timeout_pending):
return
expected = "__home__" if self.phase == "home" else self.current_key
if entry.get("detail") != expected:
return
expected_name = (
"interactive Home" if self.phase == "home"
else "interactive module"
)
if entry.get("name") != expected_name:
return
self._disarm_timeout()
self._pending = dict(entry)
QTimer.singleShot(SETTLE_MS, self._settle_ready)
@staticmethod
def _sealed_stall_window_end(
observed_at: float, retry: Callable[[], None]) -> Optional[float]:
"""Return the first observed watchdog beat after readiness.
A wall-clock snapshot between watchdog beats is not a stable boundary:
the next beat begins at the preceding beat and therefore spans backward
across that snapshot. Sealing at a beat instead makes every later raw
interval begin exactly at or after the closed result window. Unit
environments without the production watchdog retain the current clock
fallback because no later watchdog trace can appear there.
"""
latest = timing.last_gui_beat_at()
if latest is None:
return timing.elapsed()
if latest <= float(observed_at):
QTimer.singleShot(SETTLE_MS, retry)
return None
return float(latest)
def _settle_ready(self) -> None:
"""Let the event loop drain before the timing is taken.
A MODULE IS NOT READY WHEN ITS CONSTRUCTOR RETURNS: deferred work is
still queued, and timing it there measures the constructor rather than
the wait a user actually sees.
"""
if self._pending is None or self._finished:
return
entry = self._pending
end = self._sealed_stall_window_end(
float(entry.get("at", 0.0)), self._settle_ready)
if end is None:
return
self._pending = None
state = timing.snapshot()
start = float(entry.get("started_at", 0.0))
interval_stalls = timing.stalls_between(start, end, state["stalls"])
entry["stall_window_started_at"] = start
entry["stall_window_ended_at"] = end
entry["worst_event_loop_stall_ms"] = max(
(float(row["overlap_ms"]) for row in interval_stalls), default=0.0)
entry["worst_overlapping_frame_interval_ms"] = max(
(float(row["late_ms"]) for row in interval_stalls), default=0.0)
entry["event_loop_stall_budget_met"] = (
entry["worst_event_loop_stall_ms"] < timing.STALL_BUDGET_MS
)
entry["stall_samples"] = len(interval_stalls)
if self.phase == "module":
entry.setdefault("door", self._door)
self.results.append(entry)
self._checkpoint()
print(
f"benchmark ready: {entry['detail']} "
f"{entry['duration_s']:.3f}s, worst gap "
f"{entry['worst_event_loop_stall_ms']:.0f}ms",
flush=True,
)
if self.phase == "home":
self.phase = (
"preferences" if self._measure_preferences else "module"
)
else:
self.index += 1
self.current_key = None
self._advance_after_watchdog()
def _advance_after_watchdog(self) -> None:
"""Do not let checkpoint/prewarm work contaminate the next click."""
checkpoint_ended = timing.elapsed()
def _after_beat() -> None:
"""Continue once the GUI thread has answered again."""
latest = timing.last_gui_beat_at()
if latest is not None and latest <= checkpoint_ended:
QTimer.singleShot(SETTLE_MS, _after_beat)
return
self._advance()
QTimer.singleShot(SETTLE_MS, _after_beat)
def _advance(self) -> None:
"""Move on to the next module."""
if self._finished:
return
if self.phase == "preferences":
self._open_preferences()
return
if self.index >= len(self.keys):
QTimer.singleShot(SETTLE_MS, self._finish)
return
key = self.keys[self.index]
self.current_key = key
self._arm_timeout()
buttons = [
button for button in getattr(self.window._sidebar, "_items", ())
if str(button.property("navKey") or "") == key
]
if len(buttons) > 1:
registry_rows = [b for b in buttons
if not b.property("isFoldChild")]
if len(registry_rows) == 1:
buttons = registry_rows
if len(buttons) > 1:
self._record_error(
key, f"expected one live sidebar button, found {len(buttons)}")
return
if not buttons:
self._door = "command palette"
try:
self.window._on_nav_selected(key)
except BaseException as error: # noqa: BLE001
self._record_error(
key, f"{type(error).__name__}: {error}",
already_stopped=True)
return
self._door = "sidebar"
button = buttons[0]
if not button.isEnabled():
self._record_error(key, "the live sidebar button is disabled")
return
try:
button.click()
except BaseException as error: # noqa: BLE001
self._record_error(
key, f"{type(error).__name__}: {error}", already_stopped=True)
def _open_preferences(self) -> None:
"""Construct and paint the real Preferences dialog under a budget."""
self._preferences_started_elapsed = timing.elapsed()
self._preferences_ready_at = None
self._arm_timeout(min(self.timeout_s, PREFERENCES_HANG_TIMEOUT_S))
try:
if self._preferences_factory is None:
from .preferences import PreferencesDialog
dialog = PreferencesDialog(self.window)
else:
dialog = self._preferences_factory()
self._preferences_dialog = dialog
dialog.show()
except BaseException as error: # noqa: BLE001
self._record_error(
"__preferences__",
f"{type(error).__name__}: {error}",
)
return
QTimer.singleShot(SETTLE_MS * 2, self._settle_preferences)
def _settle_preferences(self) -> None:
"""Let the Preferences dialog finish laying out before it is timed."""
if self._finished or self.phase != "preferences":
return
if self._preferences_ready_at is None:
self._preferences_ready_at = timing.elapsed()
ready_at = self._preferences_ready_at
ended = self._sealed_stall_window_end(
ready_at, self._settle_preferences)
if ended is None:
return
self._disarm_timeout()
state = timing.snapshot()
stalls = timing.stalls_between(
self._preferences_started_elapsed, ended, state["stalls"])
duration = max(0.0, ready_at - self._preferences_started_elapsed)
worst = max(
(float(row["overlap_ms"]) for row in stalls), default=0.0)
raw_worst = max(
(float(row["late_ms"]) for row in stalls), default=0.0)
self.results.append({
"name": "interactive preferences",
"detail": "__preferences__",
"at": ready_at,
"started_at": self._preferences_started_elapsed,
"event_loop_started_at": state.get("event_loop_started_at"),
"duration_s": duration,
"budget_s": PREFERENCES_BUDGET_S,
"within_budget": duration <= PREFERENCES_BUDGET_S,
"stall_window_started_at": self._preferences_started_elapsed,
"stall_window_ended_at": ended,
"worst_event_loop_stall_ms": worst,
"worst_overlapping_frame_interval_ms": raw_worst,
"event_loop_stall_budget_met": worst < timing.STALL_BUDGET_MS,
"stall_samples": len(stalls),
})
self._close_preferences_dialog()
self.phase = "module"
self._checkpoint()
self._advance_after_watchdog()
def _close_preferences_dialog(self) -> None:
"""Close the timed Preferences dialog."""
dialog = self._preferences_dialog
self._preferences_dialog = None
self._preferences_ready_at = None
if dialog is None:
return
try:
dialog.close()
dialog.deleteLater()
except RuntimeError:
pass
def _timed_out(self) -> None:
"""Fail the run, naming the module that did not become ready."""
if self._finished or self._timeout_pending:
return
if self.phase == "home":
detail = "__home__"
elif self.phase == "preferences":
detail = "__preferences__"
else:
detail = str(self.current_key)
self._timeout_pending = True
timing.cancel_interactive(detail=str(detail))
_after(
SETTLE_MS,
self,
lambda: self._record_error(
detail,
f"no painted usable state within {self._armed_timeout_s:.1f} seconds",
already_stopped=True,
),
)
def _record_error(self, detail: str, message: str, *,
already_stopped: bool = False) -> None:
"""Record a failure without losing the timings already taken.
:param detail: what was being done.
:param message: what went wrong.
:param already_stopped: whether the run had already been halted.
"""
del already_stopped
self._disarm_timeout()
self._timeout_pending = False
timing.cancel_interactive(detail=str(detail))
duration = (
timing.elapsed() if self.phase == "home"
else max(0.0, time.perf_counter() - self._attempt_started)
)
attempt_started_elapsed = (
0.0 if self.phase == "home" else self._attempt_started_elapsed
)
state = timing.snapshot()
interval_stalls = timing.stalls_between(
attempt_started_elapsed, float(state["elapsed_s"]), state["stalls"])
worst_stall = max(
(float(row["overlap_ms"]) for row in interval_stalls), default=0.0)
raw_worst = max(
(float(row["late_ms"]) for row in interval_stalls), default=0.0)
record = {
"name": (
"interactive Home" if self.phase == "home"
else "interactive preferences" if self.phase == "preferences"
else "interactive module"
),
"detail": str(detail),
"duration_s": duration,
"budget_s": (
timing.HOME_BUDGET_S if self.phase == "home"
else PREFERENCES_BUDGET_S if self.phase == "preferences"
else timing.MODULE_BUDGET_S
),
"within_budget": False,
"stall_window_started_at": attempt_started_elapsed,
"stall_window_ended_at": float(state["elapsed_s"]),
"worst_event_loop_stall_ms": worst_stall,
"worst_overlapping_frame_interval_ms": raw_worst,
"event_loop_stall_budget_met": worst_stall < timing.STALL_BUDGET_MS,
"stall_samples": len(interval_stalls),
"error": str(message),
}
if self.phase == "module":
record["door"] = self._door
self.results.append(record)
self._checkpoint()
print(f"benchmark failed: {detail}: {message}", flush=True)
if self.phase == "home":
self._finish("Home never became interactive")
return
if self.phase == "preferences":
self._close_preferences_dialog()
self.phase = "module"
self._advance_after_watchdog()
return
self.index += 1
self.current_key = None
self._advance_after_watchdog()
def _current_registry_keys(self) -> tuple[str, ...]:
"""The modules the registry holds right now.
:returns: the keys.
"""
if self._live_keys is None:
return self.keys
return tuple(str(key) for key in self._live_keys())
def _violations(self, current_keys: Iterable[str]) -> list[str]:
"""Modules that appeared or vanished during the run.
A REGISTRY THAT MOVES INVALIDATES THE COMPARISON: the benchmark times
a fixed list, and a module registered halfway through means the
numbers describe two different applications.
:param current_keys: the keys as they are now.
:returns: the discrepancies.
"""
violations: list[str] = []
final_keys = tuple(str(key) for key in current_keys)
if final_keys != self.keys:
violations.append(
"the live registry changed during the benchmark sweep")
measured = [
str(row.get("detail")) for row in self.results
if row.get("detail") not in {"__home__", "__preferences__"}
]
if measured != list(self.keys):
violations.append(
"measured app sequence does not equal the live registry exactly")
by_detail = {str(row.get("detail")): row for row in self.results}
expected = (
("__home__", "__preferences__", *self.keys)
if self._measure_preferences else ("__home__", *self.keys)
)
for detail in expected:
row = by_detail.get(detail)
if row is None:
violations.append(f"{detail}: missing readiness record")
continue
if row.get("error"):
violations.append(f"{detail}: {row['error']}")
if row.get("within_budget") is not True:
violations.append(
f"{detail}: no readiness record meeting the "
f"{row.get('budget_s')} s budget")
if row.get("event_loop_stall_budget_met") is not True:
violations.append(
f"{detail}: event-loop stall reached the 500 ms ceiling")
return violations
def _artifact(self, exit_reason: str) -> dict:
"""Assemble the run's results into the artifact structure.
:param exit_reason: why the run ended.
:returns: the artifact.
"""
artifact = timing.snapshot()
final_keys = self._current_registry_keys()
measured = [
str(row.get("detail")) for row in self.results
if row.get("detail") not in {"__home__", "__preferences__"}
]
artifact["benchmark"] = {
"run": os.environ.get(RUN_LABEL_ENV, "benchmark"),
"exit_reason": str(exit_reason),
"registry_keys": list(self.keys),
"registry_count": len(self.keys),
"final_registry_keys": list(final_keys),
"registry_stable": final_keys == self.keys,
"measured_keys": measured,
"measured_count": len(measured),
"registry_matches_measurements": measured == list(self.keys),
"preferences_measured": self._measure_preferences,
"preferences_budget_s": PREFERENCES_BUDGET_S,
"results": list(self.results),
"violations": self._violations(final_keys),
}
return artifact
def _persist(self, exit_reason: str) -> str:
"""Atomically replace the artifact; return an error message or ``""``."""
temporary = f"{self.output}.{os.getpid()}.tmp"
try:
os.makedirs(os.path.dirname(os.path.abspath(self.output)),
exist_ok=True)
with open(temporary, "w", encoding="utf-8") as handle:
json.dump(self._artifact(exit_reason), handle,
indent=2, sort_keys=True)
handle.write("\n")
os.replace(temporary, self.output)
except Exception as error: # noqa: BLE001
try:
os.unlink(temporary)
except OSError:
pass
return str(error)
return ""
def _write(self, exit_reason: str) -> None:
"""Write the artifact to disk.
:param exit_reason: why the run ended.
"""
if self._written:
return
error = self._persist(exit_reason)
if error:
print(f"could not write spaCR benchmark artifact: {error}")
return
self._written = True
def _checkpoint(self) -> None:
"""Preserve completed states even if a later screen kills the process."""
self._persist("registry sweep in progress")
def _finish(self, reason: str = "registry sweep complete") -> None:
"""Write the artifact and stop the application.
:param reason: why the run ended.
"""
if self._finished:
return
self._finished = True
self._disarm_timeout()
self._close_preferences_dialog()
timing.unsubscribe_readiness(self._ready)
self._write(reason)
self.app.quit()
def _application_quit(self) -> None:
"""Quit, whether the run succeeded or failed."""
if not self._written:
self._write("application quit before registry sweep completed")
[docs]
def maybe_start(app, window) -> Optional[Union[BenchmarkController, _DistributionSmokeController]]:
"""Install the controller named by the environment, or return ``None``.
:param app: the running ``QApplication``, passed to the controller.
:param window: the main window the benchmark drives.
:returns: the registry benchmark controller, or the installed-application
controller when ``SPACR_DISTRIBUTION_SMOKE=1`` is explicitly selected;
``None`` when no output path is configured.
"""
output = os.environ.get(OUTPUT_ENV, "").strip()
if not output:
return None
if os.environ.get("SPACR_DISTRIBUTION_SMOKE") == "1":
return _DistributionSmokeController(app, window, output)
from .app import APPS
def _live_keys() -> tuple[str, ...]:
"""Every registered app key, read when the benchmark runs."""
return tuple(key for key, _name, _description, _section in APPS)
keys = _live_keys()
if len(keys) != len(set(keys)):
raise ValueError("the live application registry contains duplicate keys")
return BenchmarkController(
app, window, keys, output, live_keys=_live_keys)
_SMOKE_STACK_DUMP_SECONDS = 540
_SMOKE_STACK_DUMP_ARMED = False
def _arm_the_smoke_stack_dump(record):
"""Write every thread's stack to the application log if the smoke still runs at 540 s.
The CI watchdog kills the application at 660 s, and a GUI thread blocked
inside one call never gets back to the 600-second check in the smoke's
``_advance``, so without this a hang left no trace of where it stood.
Armed only in a distribution smoke run (``SPACR_DISTRIBUTION_SMOKE=1``).
:param record: the smoke receipt; the log's path is added to it.
:returns: the open log file the dump goes to, or ``None``.
"""
global _SMOKE_STACK_DUMP_ARMED
import faulthandler
if os.environ.get('SPACR_DISTRIBUTION_SMOKE') != '1':
return None
try:
from .verbose_logger import current_log_file
path = current_log_file()
path.parent.mkdir(parents=True, exist_ok=True)
handle = open(path, 'a', encoding='utf-8')
faulthandler.dump_traceback_later(
_SMOKE_STACK_DUMP_SECONDS, exit=False, file=handle)
except Exception:
return None
_SMOKE_STACK_DUMP_ARMED = True
record['stack_dump_log'] = str(path)
return handle
def _cancel_the_smoke_stack_dump():
"""Cancel the smoke's pending stack dump once it has an outcome."""
global _SMOKE_STACK_DUMP_ARMED
import faulthandler
if _SMOKE_STACK_DUMP_ARMED:
_SMOKE_STACK_DUMP_ARMED = False
faulthandler.cancel_dump_traceback_later()
def _name_the_blocking_modal(controller, reason):
"""Name the blocking modal in the smoke receipt and in its failure message.
The class, object name, window title and any message-box text go to
``record['blocking_modal']``, so a native run that cannot be repeated
interactively still says which dialog appeared. A failed cache review's
traceback goes with them, since its warning is one such dialog.
:param controller: the distribution smoke controller, or a stand-in with
``app``, ``record`` and ``_write``.
:param reason: what the dialog blocked.
:returns: the failure message, ``reason`` followed by the dialog's details.
"""
widget = controller.app.activeModalWidget()
detail = dict(cls=type(widget).__name__)
for name in ('objectName', 'windowTitle', 'text', 'informativeText'):
reader = getattr(widget, name, None)
if callable(reader):
try:
value = reader()
except Exception:
continue
if isinstance(value, str) and value:
detail[name] = value
described = ', '.join(f'{key}={value!r}' for key, value in detail.items())
failure = getattr(getattr(controller, 'window', None), '_cache_cleanup_error', None)
if isinstance(failure, str) and failure:
detail['cache_review_error'] = failure[-4000:]
controller.record['blocking_modal'] = detail
controller._write()
return f'{reason}: {described}'
def _let_the_layouts_settle():
"""Let the queued re-layouts run."""
from PySide6.QtWidgets import QApplication
for _turn in range(6):
QApplication.processEvents()
def _collapsed_heading_faults_of(controller):
"""Judge the smoke's collapsed containers on screen, then every one collapsed.
:param controller: the smoke controller, or a stand-in with no window.
:returns: the faults, ``[]`` for none, or None when there is no
window on screen to judge.
"""
from PySide6.QtWidgets import QWidget
window = getattr(controller, 'window', None)
if not isinstance(window, QWidget):
return None
try:
from .widgets.collapsible_splitter import (
_collapsed_header_faults, _probe_collapsed_headers)
screen = getattr(controller, 'screen', None)
root = screen if isinstance(screen, QWidget) else window
faults = _collapsed_header_faults(window)
faults += _probe_collapsed_headers(root, _let_the_layouts_settle)
except Exception as exc: # noqa: BLE001
faults = [{'problem': f'collapsed-heading check failed: {exc}'}]
return faults
class _DistributionSmokeController(QObject):
"""Exercise an installed application's real Measure screen and Run action."""
def __init__(self, app, window, output):
"""Retain the controller until a real run finishes or its deadline expires."""
from pathlib import Path
super().__init__(app)
self.app, self.window = app, window
self.output = Path(output).absolute()
self.output.parent.mkdir(parents=True, exist_ok=True)
self.root = self.output.parent / 'experiment'
self.screen = None
self.phase = 'launch'
self.started = time.monotonic()
self.record = {'schema': 'spacr-distribution-smoke-v1', 'status': 'running'}
self.timer = QTimer(self)
self.timer.setInterval(200)
self.timer.timeout.connect(self._tick)
app.aboutToQuit.connect(self._quitting)
self._hang_log = _arm_the_smoke_stack_dump(self.record)
self.timer.start()
self._write()
def _write(self):
"""Atomically retain progress even when native loading later fails."""
temporary = self.output.with_suffix('.tmp')
temporary.write_text(json.dumps(self.record, indent=2, default=str) + '\n',
encoding='utf-8')
temporary.replace(self.output)
def _provenance(self):
"""Refuse source/build-environment imports in artifact acceptance."""
import platform
from pathlib import Path
import numpy
import PySide6
import spacr
import torch
from PySide6.QtGui import QImage
from spacr.version import get_version
from spacr.qt.ai.providers import ClaudeCliProvider
if os.environ.get('SPACR_DEVICE') != 'cpu':
raise RuntimeError('Distribution acceptance must explicitly select CPU')
if any(os.environ.get(name) for name in
('CUDA_VISIBLE_DEVICES', 'HIP_VISIBLE_DEVICES', 'ROCR_VISIBLE_DEVICES')):
raise RuntimeError('Accelerator visibility must be empty')
kind = os.environ.get('SPACR_DISTRIBUTION_KIND', '')
frozen = bool(getattr(sys, 'frozen', False))
if kind not in ('frozen', 'debian-frozen') or not frozen:
raise RuntimeError('The running installation is not the requested artifact family')
executable = Path(sys.executable).resolve()
root = self._installed_root(executable, Path(sys._MEIPASS), sys.platform)
for entry in sys.path:
self._require_installed_origin(Path(entry), root)
origins = {}
for module in (spacr, numpy, PySide6, torch):
path = Path(module.__file__).resolve()
self._require_installed_origin(path, root)
origins[module.__name__] = str(path)
if get_version() != spacr.__version__:
raise RuntimeError('Installed distribution metadata does not match spaCR')
if sys.platform != 'darwin' and torch.version.cuda is not None:
raise RuntimeError('CPU acceptance contains CUDA torch')
resources = Path(spacr.__file__).parent / 'resources'
for resource in (resources / 'layout_policy.json', resources / 'icons' / 'measure.png'):
self._require_installed_origin(resource, root)
policy = json.loads((resources / 'layout_policy.json').read_text(encoding='utf-8'))
if not policy or QImage(str(resources / 'icons' / 'measure.png')).isNull():
raise RuntimeError('Installed layout policy or Measure icon is missing')
font = resources / 'font' / 'open_sans' / 'OpenSans-VariableFont_wdth,wght.ttf'
self._require_installed_origin(font, root)
if not font.is_file() or font.stat().st_size == 0:
raise RuntimeError('Installed Open Sans font is missing')
self.record.update(
platform=platform.platform(), machine=platform.machine(),
python=sys.version, version=spacr.__version__, torch=torch.__version__,
kind=kind, frozen=frozen, executable=sys.executable,
bundle_root=str(root), import_origins=origins, sys_path=list(sys.path),
cwd=str(Path.cwd()), qt_platform=self.app.platformName(),
source_commit=os.environ.get('SPACR_ACCEPTANCE_SOURCE_COMMIT', ''),
device='cpu', packaged_resources_verified=True,
claude_install_hint=ClaudeCliProvider.install_hint)
if self.app.platformName() in ('offscreen', 'minimal'):
raise RuntimeError('Native artifact acceptance requires the native Qt platform')
def _start_up_warm_running(self):
"""Whether the start-up import warm is still running on its thread."""
app_module = sys.modules.get('spacr.qt.app')
running = getattr(app_module, '_settings_prewarm_running', None)
try:
if running is None or not running():
return False
except Exception:
return False
self.record['waited_for_start_up_warm'] = True
return True
@staticmethod
def _installed_root(executable, bundle_root, platform):
"""Permit macOS bundle siblings without widening other platforms' roots."""
executable = executable.resolve()
root = executable.parent
if platform == 'darwin':
root = executable.parents[2]
if (root.suffix != '.app' or executable.parent.name != 'MacOS'
or executable.parent.parent.name != 'Contents'):
raise RuntimeError('The macOS executable is not inside its installed app bundle')
_DistributionSmokeController._require_installed_origin(bundle_root, root)
return root
@staticmethod
def _require_installed_origin(path, root):
"""Resolve symlinks so an artifact cannot borrow a source checkout."""
if not path.resolve().is_relative_to(root.resolve()):
raise RuntimeError(f'Import outside the installed artifact: {path}')
def _tick(self):
"""Record the phase, hold the first pass for the start-up warm, then advance.
The phase reaches the receipt before each step, so a step that never
returns leaves the phase it blocked in. The provenance check of the
first pass imports torch, NumPy and spaCR on this thread; it waits on
the event loop till the start-up warm thread is done importing, within
the same 600-second deadline.
"""
if self.record.get('phase') != self.phase:
self.record['phase'] = self.phase
self._write()
if (self.phase == 'launch' and time.monotonic() - self.started <= 600
and self._start_up_warm_running()):
return
self._advance()
def _advance(self):
"""Drive production widgets without replacing pipeline or worker callbacks."""
try:
if time.monotonic() - self.started > 600:
raise RuntimeError('Distribution smoke exceeded its 600-second deadline')
if (self.phase == 'waiting-installation-review'
and time.monotonic() - self._installation_review_started > 30):
raise RuntimeError('Installation review acceptance exceeded its deadline')
if (self.phase != 'launch' and not self.phase.startswith('native-menu-')
and self._keep_installer_cleanup()):
return
from .first_run import _TourOverlay
for overlay in self.window.findChildren(_TourOverlay):
if overlay.isVisible():
overlay._skip_btn.click()
self.record['tours_skipped'] = self.record.get('tours_skipped', 0) + 1
if self.phase == 'launch':
if self.root.exists():
raise RuntimeError('Smoke experiment already exists; refusing stale analysis output')
self._provenance()
if not self.window.isVisible():
return
self.phase = 'onboarding'
elif self.phase == 'onboarding':
if self.app.activeModalWidget() is not None:
raise RuntimeError(_name_the_blocking_modal(self,
'An unexpected modal dialog blocks first-launch onboarding'))
if self._onboarding_running():
return
self.record['onboarding_settled_s'] = round(time.monotonic() - self.started, 1)
self.window.open_module('measure')
self.phase = 'screen'
elif self.phase == 'screen':
self.screen = self.window._screens.get('measure')
if self.screen is None or not self.screen._btn_run.isEnabled():
return
if self.app.activeModalWidget() is not None:
raise RuntimeError(_name_the_blocking_modal(self,
'An unexpected modal dialog blocks the installed application'))
from .synthetic import demo_settings, generate_measure_demo
layout = generate_measure_demo(self.root, wells=('A01',), fields=1,
channels=(0, 1, 2, 3))
settings = demo_settings('measure', str(layout.src))
settings.update(save_png=False, representative_images=False,
n_jobs=1, verbose=False, plot=False)
self.database = layout.src / 'measurements' / 'measurements.db'
self.screen.apply_settings_dict(settings)
self.screen = self.window._screens['measure']
self._expected_src = str(layout.src)
self.phase = 'configured'
elif self.phase == 'configured':
if not self.screen._btn_run.isEnabled():
return
self.record['settings'] = self.screen._settings_model.collect()
if self.record['settings'].get('src') != self._expected_src:
raise RuntimeError('The live Measure screen did not receive the smoke input')
if self.record['settings'].get('n_jobs') != 1:
raise RuntimeError('The live Measure screen did not retain its one-worker limit')
if self.screen._crop_choice_warnings(self.record['settings']):
raise RuntimeError('The smoke fixture unexpectedly requires crop confirmation')
self.record['module_constructed'] = True
self.phase = 'starting'
self.screen._btn_run.click()
self._attach_to_the_worker()
elif self.phase == 'starting':
self._attach_to_the_worker()
elif self.phase == 'running':
if self.screen._thread is not None:
return
status, cells = self._read_result(self.database)
if status != ('complete', 1, 0) or cells < 1:
raise RuntimeError(f'Incomplete measured field: {status!r}, cells={cells}')
self.record.update(run_status=list(status), cells=cells,
database=str(self.database), worker_finished=True,
analysis_status='passed',
elapsed_seconds=time.monotonic() - self.started)
self._start_layout_check()
elif self.phase == 'settling-layout':
self._poll_layout_check()
elif self.phase == 'checking-layout-reachability':
self._poll_layout_reachability()
elif self.phase == 'waiting-installation-review':
self._poll_installation_review()
elif self.phase.startswith('native-menu-'):
self._poll_native_menu_check()
except Exception as exc:
self._pipeline_failed(str(exc))
def _layout_snapshot(self):
"""Measure actual pane/control geometry without resizing or activating layouts."""
from PySide6.QtCore import QPoint, QRect
from PySide6.QtGui import QTextCursor
from PySide6.QtWidgets import (QAbstractButton, QAbstractSpinBox, QComboBox,
QLabel, QLineEdit, QPlainTextEdit, QTextEdit)
from .gui_scale import _actual_minimum_size
def rectangle(widget):
"""Map actual geometry into the window.
:param widget: the real child widget being measured.
"""
return QRect(widget.mapTo(self.window, QPoint()), widget.size())
def coordinates(rect):
"""Retain integer geometry in a JSON-compatible representation.
:param rect: the observed Qt rectangle.
"""
return [rect.x(), rect.y(), rect.width(), rect.height()]
window_rect = self.window.rect()
controls = []
panes = []
interactive = (QAbstractButton, QAbstractSpinBox, QComboBox, QLineEdit)
measured = interactive + (QLabel, QPlainTextEdit, QTextEdit)
splitter = self.screen._runtime_splitter
for index in range(splitter.count()):
pane = splitter.widget(index)
registered = splitter._pane_of(pane)
name = registered.name if registered is not None else pane.objectName()
if name not in ('Console', 'System', 'Actions') or not pane.isVisible():
continue
pane_rect = rectangle(pane)
panes.append({'name': name, 'rect': coordinates(pane_rect),
'minimum': [pane.minimumWidth(), pane.minimumHeight()],
'minimum_hint': [pane.minimumSizeHint().width(),
pane.minimumSizeHint().height()],
'height_for_width': pane.heightForWidth(pane.width())
if pane.hasHeightForWidth() else None})
for position, widget in enumerate(pane.findChildren(QObject)):
if not isinstance(widget, measured) or not widget.isVisibleTo(pane):
continue
actual = rectangle(widget)
clip = window_rect.intersected(pane_rect)
ancestor = widget.parentWidget()
while ancestor is not None and ancestor is not self.window:
clip = clip.intersected(rectangle(ancestor))
ancestor = ancestor.parentWidget()
hint = widget.minimumSizeHint()
explicit = _actual_minimum_size(widget)
embedded_editor = False
if isinstance(widget, QLineEdit):
owner = widget.parentWidget()
while owner is not None and owner is not pane:
if isinstance(owner, (QAbstractSpinBox, QComboBox)):
embedded_editor = True
break
owner = owner.parentWidget()
required_height = max(explicit.height(), hint.height())
required_width = max(explicit.width(), hint.width())
clipped = actual.isEmpty() or not clip.contains(actual)
undersized = not embedded_editor and isinstance(widget, interactive) and (
actual.height() < required_height or actual.width() < required_width)
document_edges = None
document_clip = None
if isinstance(widget, (QPlainTextEdit, QTextEdit)) and widget.isReadOnly():
document_edges = {}
document_clip = coordinates(clip.intersected(rectangle(widget.viewport())))
for edge, move in (('start', QTextCursor.Start), ('end', QTextCursor.End)):
cursor = widget.textCursor()
cursor.movePosition(move)
caret = widget.cursorRect(cursor)
caret.moveTopLeft(widget.viewport().mapTo(self.window, caret.topLeft()))
document_edges[edge] = coordinates(caret)
controls.append({'pane': name, 'position': position,
'class': type(widget).__name__,
'object_name': widget.objectName(),
'rect': coordinates(actual), 'visible_clip': coordinates(clip),
'minimum_hint': [hint.width(), hint.height()],
'minimum': [explicit.width(), explicit.height()],
'embedded_editor': embedded_editor,
'acceptance_control': not embedded_editor,
'document_edges': document_edges,
'document_clip': document_clip,
'clipped': clipped, 'undersized': undersized})
settings = self.screen._settings_panel
return {'window_size': [self.window.width(), self.window.height()],
'device_pixel_ratio': self.window.devicePixelRatioF(),
'available_screen': coordinates(self.window.screen().availableGeometry()),
'body_sizes': self.screen._body_splitter.sizes(),
'runtime_sizes': splitter.sizes(),
'settings_rect': coordinates(rectangle(settings)),
'panes': panes, 'controls': controls}
def _save_layout_image(self, filename):
"""Save the unmodified native window at its current dimensions.
:param filename: the screenshot filename beside the receipt.
"""
image = self.window.grab()
if not image.save(str(self.output.parent / filename)):
raise RuntimeError('Could not retain the native application screenshot')
return [image.width(), image.height()]
def _start_layout_check(self):
"""Retain completion geometry, then observe later event-loop turns."""
self._layout_previous = self._layout_snapshot()
self._layout_stable = 0
self._layout_started = time.monotonic()
self.record['layout'] = {
'window_resized_by_witness': False,
'completion_image': 'measure-complete.png',
'completion_image_size': self._save_layout_image('measure-complete.png'),
'samples': [self._layout_previous], 'status': 'settling'}
self.phase = 'settling-layout'
self._write()
def _poll_layout_check(self):
"""Keep truthful settled/clipped evidence; never resize to obtain acceptance."""
snapshot = self._layout_snapshot()
self._layout_stable = self._layout_stable + 1 if snapshot == self._layout_previous else 0
self._layout_previous = snapshot
record = self.record['layout']
record['samples'].append(snapshot)
elapsed = time.monotonic() - self._layout_started
settled = self._layout_stable >= 2
if not settled and elapsed < 5:
self._write()
return
record.update(settled=settled, elapsed_seconds=elapsed,
settled_image='measure-settled.png',
settled_image_size=self._save_layout_image('measure-settled.png'))
violations = [control for control in snapshot['controls']
if control.get('acceptance_control', True)
and (control['clipped'] or control['undersized'])]
unchanged = snapshot['window_size'] == record['samples'][0]['window_size']
observed = {pane['name'] for pane in snapshot['panes']}
complete = observed == {'Console', 'System', 'Actions'} and all(
any(control['pane'] == name for control in snapshot['controls'])
for name in observed)
record.update(violations=violations, window_size_unchanged=unchanged,
required_panes_observed=complete,
status='passed' if settled and unchanged and complete
and not violations else 'failed')
record['raw_viewport_status'] = record['status']
from PySide6.QtWidgets import QAbstractScrollArea, QScrollArea
viewport = getattr(self.screen, '_runtime_viewport', None)
if settled and unchanged and complete and isinstance(viewport, QScrollArea):
self._layout_scroll = viewport
self._layout_scroll_states = [
(scroll, [scroll.horizontalScrollBar().value(), scroll.verticalScrollBar().value()])
for scroll in [viewport] + viewport.findChildren(QAbstractScrollArea)]
self._layout_scroll_origin = [viewport.horizontalScrollBar().value(),
viewport.verticalScrollBar().value()]
self._layout_targets = []
self._layout_document_cursors = []
for control in snapshot['controls']:
if not control.get('acceptance_control', True):
continue
if control.get('document_edges') is not None:
document = self._control_for_layout_record(control)
if document is not None:
self._layout_document_cursors.append((document, document.textCursor()))
self._layout_targets.extend(dict(control, document_edge=edge)
for edge in ('start', 'end'))
else:
self._layout_targets.append(control)
self._layout_target = None
self._layout_reachability_started = time.monotonic()
self._layout_restoring = False
record['scroll_origin'] = self._layout_scroll_origin
record['scroll_origins'] = [dict(index=index, name=scroll.objectName(), position=position)
for index, (scroll, position)
in enumerate(self._layout_scroll_states)]
record['reachability'] = []
record['status'] = 'checking-reachability'
self.phase = 'checking-layout-reachability'
self._write()
return
self._finish_layout_check()
def _control_for_layout_record(self, record):
"""Resolve a measured control without inventing a replacement widget.
:param record: the pane, child position, class, and name measured earlier.
:returns: the same observable control, or ``None`` if it disappeared.
"""
split = self.screen._runtime_splitter
for index in range(split.count()):
pane = split.widget(index)
registered = split._pane_of(pane)
name = registered.name if registered is not None else pane.objectName()
if name != record['pane']:
continue
children = pane.findChildren(QObject)
position = record['position']
if position >= len(children):
return None
widget = children[position]
if (type(widget).__name__ == record['class']
and widget.objectName() == record['object_name']
and widget.isVisibleTo(pane)):
return widget
return None
def _scroll_layout_target(self, widget, record):
"""Drive each real enclosing scroll area, starting with the innermost.
:param widget: the measured control or read-only document.
:param record: its geometry identity and optional document endpoint.
"""
from PySide6.QtGui import QTextCursor
from PySide6.QtWidgets import QScrollArea
edge = record.get('document_edge')
if edge is not None:
cursor = widget.textCursor()
cursor.movePosition(QTextCursor.Start if edge == 'start' else QTextCursor.End)
widget.setTextCursor(cursor)
widget.ensureCursorVisible()
ancestor = widget.parentWidget()
while ancestor is not None and ancestor is not self.window:
if isinstance(ancestor, QScrollArea):
if edge is None:
ancestor.ensureWidgetVisible(widget, 0, 0)
else:
cursor = widget.textCursor()
cursor.movePosition(QTextCursor.Start if edge == 'start' else QTextCursor.End)
caret = widget.cursorRect(cursor)
point = widget.viewport().mapTo(ancestor.widget(), caret.center())
ancestor.ensureVisible(point.x(), point.y(), (caret.width() + 1) // 2 + 2,
(caret.height() + 1) // 2 + 2)
ancestor = ancestor.parentWidget()
def _poll_layout_reachability(self):
"""Actually scroll each control into view and retain its settled rectangle."""
record = self.record['layout']
if time.monotonic() - self._layout_reachability_started > 120:
record.update(status='failed', reachability_timeout=True)
self._finish_layout_check()
return
viewport = self._layout_scroll
if self._layout_target is None and not self._layout_restoring:
if self._layout_targets:
before = self._layout_targets.pop(0)
widget = self._control_for_layout_record(before)
if widget is None or not viewport.isAncestorOf(widget):
record['reachability'].append({'before': before, 'reachable': False,
'error': 'Observed control disappeared'})
self._write()
return
self._scroll_layout_target(widget, before)
self._layout_target = before
else:
for document, cursor in self._layout_document_cursors:
document.setTextCursor(cursor)
for scroll, origin in self._layout_scroll_states:
scroll.horizontalScrollBar().setValue(origin[0])
scroll.verticalScrollBar().setValue(origin[1])
self._layout_restoring = True
self._layout_reach_previous = None
self._layout_target_started = time.monotonic()
return
snapshot = self._layout_snapshot()
unchanged = snapshot['window_size'] == record['samples'][0]['window_size']
position = [viewport.horizontalScrollBar().value(), viewport.verticalScrollBar().value()]
positions = [[scroll.horizontalScrollBar().value(), scroll.verticalScrollBar().value()]
for scroll, _origin in self._layout_scroll_states]
if self._layout_restoring:
observed = {'snapshot': snapshot, 'scroll_position': position,
'scroll_positions': positions}
else:
before = self._layout_target
after = next((control for control in snapshot['controls']
if all(control[key] == before[key]
for key in ('pane', 'position', 'class', 'object_name'))), None)
observed = {'after': after, 'scroll_position': position,
'scroll_positions': positions,
'window_size': snapshot['window_size']}
settled = observed == self._layout_reach_previous
self._layout_reach_previous = observed
if not settled and time.monotonic() - self._layout_target_started < 3:
return
if self._layout_restoring:
cursors_restored = all(document.textCursor().position() == cursor.position()
and document.textCursor().anchor() == cursor.anchor()
for document, cursor in self._layout_document_cursors)
restored = settled and unchanged and cursors_restored and positions == [
origin for _scroll, origin in self._layout_scroll_states]
record.update(restored=observed, scroll_origin_restored=restored,
document_cursors_restored=cursors_restored,
restored_image='measure-restored.png',
restored_image_size=self._save_layout_image('measure-restored.png'),
acceptance_scope='actual scroll reachability of controls and read-only document endpoints',
status='passed' if restored and record['reachability'] and all(
row['reachable'] for row in record['reachability']) else 'failed')
self._finish_layout_check()
return
contained = after is not None and not after['clipped'] and not after['undersized']
if after is not None and before.get('document_edge') is not None:
from PySide6.QtCore import QRect
edge = after['document_edges'][before['document_edge']]
contained = QRect(*after['document_clip']).contains(QRect(*edge))
observed['document_endpoint'] = edge
row = dict(before=before, **observed, settled=settled,
window_size_unchanged=unchanged,
reachable=bool(settled and unchanged and contained))
if before['clipped'] or not row['reachable']:
filename = f"measure-scroll-{len(record['reachability']) + 1:03d}.png"
row.update(image=filename, image_size=self._save_layout_image(filename))
record['reachability'].append(row)
self._layout_target = None
self._write()
def _finish_layout_check(self):
"""Finish visual acceptance separately from the retained scientific result.
Every collapsible container on the screen is also collapsed once and
judged (:func:`spacr.qt.widgets.collapsible_splitter._probe_collapsed_headers`):
a collapsed heading that is cut, or one with no handle left to drag,
fails the layout as a clipped button does.
"""
record = self.record['layout']
faults = _collapsed_heading_faults_of(self)
if faults is not None:
record['collapsed_heading_faults'] = faults
if faults and record.get('status') == 'passed':
record['status'] = 'failed'
self.record['visual_layout_status'] = record['status']
self._write()
if record['status'] != 'passed':
raise RuntimeError('Installed UI layout did not settle with fully contained controls; '
'completion and settled geometry/screenshots were retained')
if os.environ.get('SPACR_NATIVE_MENU_SMOKE') == '1':
self._installation_review_started = time.monotonic()
self.phase = 'waiting-installation-review'
return
self.record['status'] = 'passed'
self._finish(0)
def _keep_installer_cleanup(self):
"""Choose the actual default-off cleanup dialog's Keep button without removing files.
Keep is pressed on a later tick than the one that first sees the
dialog, and only once its window is exposed. On macOS a modal's
show runs a modal session that delivers this timer, and Keep
pressed from inside that show hid the window while Cocoa was still
showing it ("Already setting window visible!", first-run smoke).
"""
from PySide6.QtCore import Qt
from PySide6.QtTest import QTest
from PySide6.QtWidgets import QDialog, QPushButton
from .install_consent import _InstallerCleanupDialog
dialog = self.app.activeModalWidget()
if not isinstance(dialog, _InstallerCleanupDialog):
return False
handle = dialog.windowHandle()
if (getattr(self, '_cleanup_review_seen', None) is not dialog
or handle is None or not handle.isExposed()):
self._cleanup_review_seen = dialog
return True
keep = dialog.findChild(QPushButton, 'InstallerCleanupKeep')
remove = dialog.findChild(QPushButton, 'InstallerCleanupRemove')
if (not dialog.isVisible() or dialog.choices()['keys']
or any(item.checkState(0) != Qt.CheckState.Unchecked
for _key, item, _eligible in dialog._folder_choices)
or remove is None or remove.isEnabled()
or keep is None or not keep.isVisible() or not keep.isEnabled()
or not keep.isDefault()):
raise RuntimeError('Installation cleanup is not in its default Keep state')
QTest.mouseClick(keep, Qt.MouseButton.LeftButton)
if dialog.isVisible() or dialog.result() != QDialog.DialogCode.Rejected:
raise RuntimeError('Keeping installation files did not close the cleanup review')
self.record['installation_review'] = dict(choice='keep', selected_count=0)
self._write()
return True
def _poll_installation_review(self):
"""Wait for real startup review completion before driving Preferences and Quit."""
if time.monotonic() - self._installation_review_started > 30:
raise RuntimeError('Installation review acceptance exceeded its deadline')
if self.app.activeModalWidget() is not None:
raise RuntimeError(_name_the_blocking_modal(self,
'An unexpected modal dialog blocks installation review'))
if self._onboarding_running():
return
self._start_native_menu_check()
def _onboarding_running(self):
"""Whether first-launch onboarding may still open a dialog.
The installer consent, the optional cache review and the report
notice open from the start-up timers on their own schedule. The
smoke waits for them before it drives a screen, so none of them
opens in the middle of a screen being built or configured, and a
dialog that is left is one the smoke did not expect.
"""
consent = getattr(self.window, '_consent_timer', None)
startup = getattr(self.window, '_startup_sequence_running', None)
return bool((consent is not None and consent.isActive())
or (startup is not None and startup())
or getattr(self.window, '_startup_deferred', ())
or getattr(self.window, '_cache_cleanup_busy', False))
@staticmethod
def _cocoa_message(receiver, selector, result_type, *arguments):
"""Call the real Objective-C menu object with an explicit native ABI."""
import ctypes
library = ctypes.CDLL('/usr/lib/libobjc.A.dylib')
library.sel_registerName.argtypes = [ctypes.c_char_p]
library.sel_registerName.restype = ctypes.c_void_p
method = library.sel_registerName(selector.encode('ascii'))
address = ctypes.cast(library.objc_msgSend, ctypes.c_void_p).value
function = ctypes.CFUNCTYPE(result_type, ctypes.c_void_p, ctypes.c_void_p,
*(kind for kind, _ in arguments))(address)
return function(receiver, method, *(value for _, value in arguments))
def _start_native_menu_check(self):
"""Locate actual Mac actions in the configured window or system menu."""
import ctypes
if sys.platform != 'darwin' or self.app.platformName() != 'cocoa':
raise RuntimeError('Native menu acceptance requires macOS Cocoa')
if not self.window.menuBar().isNativeMenuBar():
self._start_window_menu_check()
return
library = ctypes.CDLL('/usr/lib/libobjc.A.dylib')
library.objc_getClass.argtypes = [ctypes.c_char_p]
library.objc_getClass.restype = ctypes.c_void_p
pointer, integer = ctypes.c_void_p, ctypes.c_long
message = self._cocoa_message
application = message(library.objc_getClass(b'NSApplication'), 'sharedApplication', pointer)
bar = message(application, 'mainMenu', pointer)
if not bar or message(bar, 'numberOfItems', integer) < 1:
raise RuntimeError('Cocoa has no application menu')
first = message(bar, 'itemAtIndex:', pointer, (integer, 0))
app_menu = message(first, 'submenu', pointer)
if not app_menu:
raise RuntimeError('Cocoa application submenu is absent')
message(app_menu, 'update', None)
def text(obj, selector):
"""Read a native NSString without inventing an action caption."""
value = message(obj, selector, pointer)
raw = message(value, 'UTF8String', ctypes.c_char_p) if value else None
return raw.decode('utf-8') if raw else ''
rows = []
for index in range(message(app_menu, 'numberOfItems', integer)):
item = message(app_menu, 'itemAtIndex:', pointer, (integer, index))
rows.append(dict(index=index, title=text(item, 'title'),
key=text(item, 'keyEquivalent'),
modifiers=message(item, 'keyEquivalentModifierMask', ctypes.c_ulong),
enabled=bool(message(item, 'isEnabled', ctypes.c_bool)),
hidden=bool(message(item, 'isHidden', ctypes.c_bool))))
command = 1 << 20
selected = {}
for label, key in (('preferences', ','), ('quit', 'q')):
candidates = [row for row in rows if row['key'] == key
and row['modifiers'] & command and row['enabled'] and not row['hidden']]
if len(candidates) != 1:
raise RuntimeError(f'Native application menu lacks a unique enabled {label}')
selected[label] = candidates[0]['index']
self._native_menu = app_menu
self._native_actions = selected
self._native_menu_started = time.monotonic()
self.record['native_menu'] = dict(application_menu=text(first, 'title'),
items=rows, selected=selected, preferences_opened=False,
preferences_closed=False, quit_dispatched=False, quit_observed=False,
mode='system-menu-on-cocoa')
self.phase = 'native-menu-opening'
self._write()
QTimer.singleShot(0, self._invoke_native_preferences)
def _start_window_menu_check(self):
"""Verify the requested unified Mac bar and its visible window controls."""
from PySide6.QtCore import Qt
from PySide6.QtGui import QAction
from PySide6.QtWidgets import QMenu, QToolButton
bar = self.window.menuBar()
if not bar.isVisible() or bar.visibleRegion().isEmpty():
raise RuntimeError('The in-window Mac menu bar is not visible')
menus = [menu for menu in bar.findChildren(QMenu)
if menu.title().replace('&', '') == 'spaCR']
if len(menus) != 1:
raise RuntimeError('The Mac window does not have one spaCR menu')
menu = menus[0]
actions = {'preferences': self.window._act_preferences,
'quit': self.window._act_quit}
roles = {'preferences': QAction.MenuRole.PreferencesRole,
'quit': QAction.MenuRole.QuitRole}
for label, action in actions.items():
if (action not in menu.actions() or action.menuRole() != roles[label]
or not action.isEnabled() or not action.isVisible()):
raise RuntimeError(f'The Mac window lacks its actual {label} action')
corner = bar.cornerWidget(Qt.Corner.TopRightCorner)
controls = {}
for name in ('MinimiseWindow', 'FullScreenToggle', 'CloseWindow'):
button = corner.findChild(QToolButton, name) if corner else None
if (button is None or not button.isVisible() or not button.isEnabled()
or not button.visibleRegion().contains(button.rect())):
raise RuntimeError(f'The Mac window control is not fully visible: {name}')
controls[name] = [button.width(), button.height()]
self._window_menu = menu
self._window_menu_actions = actions
self._native_menu_started = time.monotonic()
self.record['native_menu'] = dict(mode='in-window-on-cocoa',
application_menu=menu.title(), window_controls=controls,
items=[dict(title=action.text(), role=action.menuRole().name)
for action in menu.actions() if not action.isSeparator()],
preferences_opened=False, preferences_closed=False,
quit_dispatched=False, quit_observed=False)
self.phase = 'native-menu-opening'
self._write()
QTimer.singleShot(0, self._invoke_native_preferences)
def _click_window_menu_action(self, label):
"""Click the real bar and popup action using their actual visible geometry.
:param label: verified ``preferences`` or ``quit`` action to click.
"""
from PySide6.QtCore import Qt
from PySide6.QtTest import QTest
bar, menu = self.window.menuBar(), self._window_menu
action = self._window_menu_actions[label]
rectangle = bar.actionGeometry(menu.menuAction())
if rectangle.isEmpty() or not bar.rect().contains(rectangle):
raise RuntimeError('The actual spaCR menu entry is outside its bar')
QTest.mouseClick(bar, Qt.MouseButton.LeftButton, pos=rectangle.center())
if not menu.isVisible():
raise RuntimeError('Clicking the Mac menu bar did not open spaCR')
rectangle = menu.actionGeometry(action)
if rectangle.isEmpty() or not menu.rect().contains(rectangle):
raise RuntimeError(f'The actual {label} menu action is outside its popup')
if not menu.grab().save(str(self.output.parent / f'native-menu-{label}.png')):
raise RuntimeError('Could not retain the actual Mac menu screenshot')
QTest.mouseClick(menu, Qt.MouseButton.LeftButton, pos=rectangle.center())
def _invoke_native_preferences(self):
"""Activate the actual Mac Preferences item and observe its modal dialog."""
import ctypes
try:
if self.record['native_menu'].get('mode') == 'in-window-on-cocoa':
self._click_window_menu_action('preferences')
else:
self._cocoa_message(self._native_menu, 'performActionForItemAtIndex:', None,
(ctypes.c_long, self._native_actions['preferences']))
if self.phase != 'native-menu-closing':
raise RuntimeError('Native Preferences did not open the verified dialog')
self.record['native_menu']['preferences_closed'] = True
self.phase = 'native-menu-ready-to-quit'
except Exception as exc:
self._pipeline_failed(str(exc))
def _poll_native_menu_check(self):
"""Observe actual Preferences, then terminate through the native Quit item."""
import ctypes
from .preferences import _preferences_window_class
if time.monotonic() - self._native_menu_started > 30:
raise RuntimeError('Native menu acceptance exceeded its deadline')
if self.phase == 'native-menu-opening':
dialog = self.app.activeModalWidget()
if dialog is None:
return
if not isinstance(dialog, _preferences_window_class()) or not dialog.isVisible():
raise RuntimeError('Native Preferences opened a different dialog')
if not dialog.grab().save(str(self.output.parent / 'native-preferences.png')):
raise RuntimeError('Could not retain the actual Preferences screenshot')
self.record['native_menu']['preferences_opened'] = True
self.record['native_menu']['dialog_class'] = type(dialog).__name__
self._native_preferences_ref = weakref.ref(dialog)
self.phase = 'native-menu-closing'
dialog.reject()
elif self.phase == 'native-menu-ready-to-quit':
modal = self.app.activeModalWidget()
if modal is not None:
reference = getattr(self, '_native_preferences_ref', None)
observed = reference() if reference is not None else None
self.record['native_menu']['blocking_modal'] = dict(
class_name=type(modal).__name__, visible=modal.isVisible(),
same_observed_wrapper=(modal is observed if observed is not None else None))
self._write()
raise RuntimeError('Preferences did not close before native Quit')
self.phase = 'native-menu-quitting'
self.record['native_menu']['quit_dispatched'] = True
self._write()
if self.record['native_menu'].get('mode') == 'in-window-on-cocoa':
self._click_window_menu_action('quit')
else:
self._cocoa_message(self._native_menu, 'performActionForItemAtIndex:', None,
(ctypes.c_long, self._native_actions['quit']))
@staticmethod
def _read_result(database):
"""Require a real terminal database and release its handle on Windows."""
from ..database_concurrency import connect
from contextlib import closing
if not database.is_file():
raise RuntimeError('Measure did not produce measurements.db')
with closing(connect(database, readonly=True)) as connection:
status = connection.execute(
'SELECT status,n_succeeded,n_failed FROM run_status ORDER BY rowid DESC LIMIT 1'
).fetchone()
cells = connection.execute('SELECT COUNT(*) FROM cell').fetchone()[0]
return status, cells
def _attach_to_the_worker(self):
"""Follow the worker the real Run started, once it has started.
A session's first Run imports its pipeline off the GUI thread and
starts the worker only when that import ends; until then the screen
holds ``_entry_import`` and the controller waits for the next tick.
:raises RuntimeError: when the Run ended without a worker.
"""
if getattr(self.screen, '_entry_import', None) is not None:
return
worker = getattr(self.screen, '_worker', None)
if worker is None:
raise RuntimeError('The real Run action did not start a pipeline worker')
worker.error.connect(self._pipeline_failed)
self.record['real_run_clicked'] = True
self.phase = 'running'
self._write()
def _pipeline_failed(self, error):
"""Retain the failure before requesting ordinary Qt shutdown."""
self.record.update(status='failed', error=str(error))
self._finish(1)
def _finish(self, code):
"""Stop polling, write the receipt, and leave the real Qt event loop."""
self.phase = 'finished'
self.timer.stop()
_cancel_the_smoke_stack_dump()
self._write()
self.app.exit(code)
def _quitting(self):
"""A premature user/application exit cannot become a successful smoke."""
_cancel_the_smoke_stack_dump()
if self.phase == 'native-menu-quitting':
menu = self.record.get('native_menu', {})
if all(menu.get(key) for key in ('preferences_opened', 'preferences_closed', 'quit_dispatched')):
menu['quit_observed'] = True
self.record['status'] = 'passed'
self.phase = 'finished'
self.timer.stop()
self._write()
return
if self.phase != 'finished':
self.record.update(status='failed', error='Application exited before smoke completion')
self._write()