diff --git a/CLAUDE.md b/CLAUDE.md index 0803a3f..1f3995a 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -43,7 +43,7 @@ pip install -r requirements-dev.txt ## Architecture ### Entry point -`src/pytest_fly/__main__.py` → `main.py` initializes the stdlib-based logger (`logger.py`) and launches the Qt app. +`src/pytest_fly/__main__.py` → `main.py` initializes the stdlib-based logger (`logger.py`) and launches the Qt app. The GUI package is imported lazily inside `main.fly_main` and `pytest_runner/__init__.py` resolves its exports lazily: every spawn child re-imports `__main__`, and must not load PySide6 or the orchestration layer as a side effect (`tests/test_main_app.py::test_spawn_child_entry_does_not_import_gui`). ### GUI layer (`src/pytest_fly/gui/`) - `gui_main.py` — `FlyAppMainWindow`: 8-tab Qt window with a periodic timer (default 3 s) that pulls updates from the runner and refreshes all tabs. @@ -64,7 +64,7 @@ pip install -r requirements-dev.txt - `system_monitor.py` — `SystemMonitor` subprocess: system-wide CPU/memory/commit/disk/network sampling for the Run tab charts. - `commit_memory.py` — Windows commit-charge readers and psutil subtree helpers. - `coverage.py` — merges per-process coverage data (one `cov.report()` pass yields both the percentage and the TOTAL line). -- `coverage_aggregator.py` — `CoverageAggregator` spawn child + `aggregate_coverage()`: runs `calculate_coverage()` out of the GUI process (it has crashed the interpreter natively); a dead/hung child is a logged warning, not a crash. +- `coverage_aggregator.py` — `CoverageAggregator` spawn child + `aggregate_coverage()` → `CoverageResult`: runs `calculate_coverage()` and the per-test executed-line counts out of the GUI process (constructing `coverage.Coverage` in-process has crashed the interpreter natively — the GUI never does it); a dead/hung child is a logged warning, not a crash. - `ordering.py` — applies the user's test-ordering aspects; `live_output.py` — per-test live-output file paths. ### Crash diagnostics diff --git a/pyproject.toml b/pyproject.toml index 80f5b3e..f704470 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,7 +1,7 @@ [project] name = "pytest-fly" description = "pytest runner and observer" -version = "0.9.0" +version = "0.9.1" readme = "README.md" requires-python = ">=3.12" authors = [ diff --git a/src/pytest_fly/gui/coverage_tracker.py b/src/pytest_fly/gui/coverage_tracker.py index 4a59256..99850d3 100644 --- a/src/pytest_fly/gui/coverage_tracker.py +++ b/src/pytest_fly/gui/coverage_tracker.py @@ -21,13 +21,9 @@ from pathlib import Path from threading import Event, Lock, Thread -from coverage import Coverage - -from ..file_util import sanitize_test_name from ..interfaces import PytestRunnerState from ..logger import get_logger from ..preferences import get_pref -from ..pytest_runner.coverage import COVERAGE_READ_ERRORS from ..pytest_runner.coverage_aggregator import aggregate_coverage from ..tick_data import TickData @@ -177,30 +173,18 @@ def _calculate(self, completed: set[str], generation: int, run_start: float | No previously published values stand and this pass is skipped. """ timeout = self._timeout_seconds if self._timeout_seconds is not None else get_pref().coverage_timeout_seconds - result = aggregate_coverage("current", self._data_dir, write_report=False, timeout=timeout) + # One child does both the combine/report pass and the per-test executed-line counts, so + # no coverage.Coverage object is ever constructed in this (GUI) process. + result = aggregate_coverage("current", self._data_dir, write_report=False, timeout=timeout, per_test_names=sorted(completed)) with self._lock: self._last_calculation_time = time.monotonic() # counts failed attempts too, so a crashing child is not re-spawned every tick if result is None: return - coverage_pct, covered_lines, total_lines = result + coverage_pct, covered_lines, total_lines = result.totals - # Recompute per-test coverage for ALL completed tests since the denominator + # Per-test coverage for ALL completed tests is recomputed each pass since the denominator # (total_lines) may have changed as new tests discover new source files. - per_test_coverage: dict[str, float] = {} - if total_lines > 0: - coverage_dir = Path(self._data_dir, "coverage") - for test_name in completed: - safe_name = sanitize_test_name(test_name) - cov_file = coverage_dir / f"{safe_name}.coverage" - if cov_file.exists(): - try: - cov = Coverage(cov_file) - cov.load() - data = cov.get_data() - executed = sum(len(data.lines(f) or []) for f in data.measured_files()) - per_test_coverage[test_name] = executed / total_lines - except COVERAGE_READ_ERRORS as e: - log.info(f"per-test coverage for {test_name} failed: {e}") + per_test_coverage = result.per_test_fractions(total_lines) with self._lock: if generation != self._generation: diff --git a/src/pytest_fly/gui/run_tab/control_window.py b/src/pytest_fly/gui/run_tab/control_window.py index 9e6bc66..d0674ea 100644 --- a/src/pytest_fly/gui/run_tab/control_window.py +++ b/src/pytest_fly/gui/run_tab/control_window.py @@ -29,7 +29,7 @@ from ...preferences import ParallelismControl, duration_to_seconds, get_ordering_aspects_ordered, get_pref from ...put_version import detect_put_version from ...pytest_runner.admission import AdmissionGateConfig -from ...pytest_runner.coverage import compute_per_test_coverage +from ...pytest_runner.coverage_aggregator import aggregate_coverage from ...pytest_runner.ordering import OrderingContext, apply_ordering_aspects from ...pytest_runner.pytest_runner import PytestRunner from ...pytest_runner.resource_guard import ResourceGuardConfig @@ -60,6 +60,7 @@ class _RunPrepConfig: gate_config: AdmissionGateConfig stall_config: StallConfig resource_guard_config: ResourceGuardConfig + coverage_timeout_seconds: float @dataclass @@ -298,6 +299,7 @@ def run(self): min_free_disk_gb=pref.resource_guard_min_free_disk_gb, commit_threshold=pref.resource_guard_commit_threshold, ), + coverage_timeout_seconds=pref.coverage_timeout_seconds, ) self._run_prep_abort.clear() self._run_prep_thread = Thread(target=self._prepare_run, args=(config, self.pytest_runner), name="run_prep", daemon=True) @@ -430,7 +432,11 @@ def _build_runner(self, config: _RunPrepConfig, prior_runner: PytestRunner | Non # means "rerun every test," not "forget the durations/failures we know about." per_test_cov: dict[str, float] = {} if OrderingAspect.COVERAGE_EFFICIENCY in config.enabled_aspects: - per_test_cov = compute_per_test_coverage(self.data_dir, [t.node_id for t in tests]) + # Out of process: constructing coverage.Coverage objects inside the GUI process has + # crashed it natively. A dead or hung child just means no coverage-based ordering + # for this run. + result = aggregate_coverage("ordering", self.data_dir, write_report=False, timeout=config.coverage_timeout_seconds, per_test_names=[t.node_id for t in tests]) + per_test_cov = result.per_test_fractions() if result is not None else {} # Coverage-efficiency reads duration/coverage off the ScheduledTest # itself, so rebuild the list with those fields populated. tests = [ diff --git a/src/pytest_fly/main.py b/src/pytest_fly/main.py index ea88bb0..9791f68 100644 --- a/src/pytest_fly/main.py +++ b/src/pytest_fly/main.py @@ -6,7 +6,6 @@ from .__version__ import application_name from .faults import enable_faulthandler, report_previous_crashes -from .gui import fly_main from .logger import get_logger, init_parent_logger from .paths import get_default_data_dir, get_workspace_dir, init_workspace from .platform.wer import report_previous_crash_dumps @@ -17,6 +16,22 @@ log = get_logger(application_name) +def fly_main(data_dir: Path, *, auto_start: bool = False, auto_quit_on_done: bool = False) -> None: + """Launch the Qt GUI (lazy wrapper around :func:`pytest_fly.gui.fly_main`). + + pytest-fly's own GUI package is imported here, at call time, rather than at module + level: every spawn child (test process, process monitor, system monitor, coverage + aggregator, test discovery) re-imports ``pytest_fly.__main__`` and therefore this module, + and a module-level ``from .gui import fly_main`` loaded PySide6 and all of pytest-fly's + widgets into every one of them — ~0.3 s and tens of MB of commit per child, and + pytest-fly's Qt binding resident in the test process before the program under test's own + tests (which may use a different Qt binding) ever run. The PUT's own imports are unaffected. + """ + from .gui import fly_main as gui_fly_main + + gui_fly_main(data_dir, auto_start=auto_start, auto_quit_on_done=auto_quit_on_done) + + def _parse_args(argv: list[str] | None = None) -> argparse.Namespace: parser = argparse.ArgumentParser(prog=application_name, description="pytest-fly: pytest runner and observer GUI") parser.add_argument( diff --git a/src/pytest_fly/pytest_runner/__init__.py b/src/pytest_fly/pytest_runner/__init__.py index 72e4e3a..9005814 100644 --- a/src/pytest_fly/pytest_runner/__init__.py +++ b/src/pytest_fly/pytest_runner/__init__.py @@ -1,3 +1,27 @@ -from .pytest_runner import PytestRunner as PytestRunner -from .pytest_runner import PytestRunState as PytestRunState -from .test_list import GetTests as GetTests +"""pytest-fly test runner package. + +The runner classes are resolved lazily (PEP 562): spawn children unpickle +:class:`ProcessMonitor` / :class:`PytestProcess` / :class:`GetTests` through this package, and +an eager ``from .pytest_runner import PytestRunner`` here would drag the whole orchestration +layer (and, transitively, pytest) into every child. +""" + +from typing import Any + +__all__ = ["GetTests", "PytestRunState", "PytestRunner"] + +_LAZY = { + "PytestRunner": ("pytest_fly.pytest_runner.pytest_runner", "PytestRunner"), + "PytestRunState": ("pytest_fly.pytest_runner.pytest_runner", "PytestRunState"), + "GetTests": ("pytest_fly.pytest_runner.test_list", "GetTests"), +} + + +def __getattr__(name: str) -> Any: + try: + module_name, attribute = _LAZY[name] + except KeyError: + raise AttributeError(f"module {__name__!r} has no attribute {name!r}") from None + from importlib import import_module + + return getattr(import_module(module_name), attribute) diff --git a/src/pytest_fly/pytest_runner/coverage.py b/src/pytest_fly/pytest_runner/coverage.py index 5c351e6..ca81674 100644 --- a/src/pytest_fly/pytest_runner/coverage.py +++ b/src/pytest_fly/pytest_runner/coverage.py @@ -4,6 +4,12 @@ Combines individual ``.coverage`` files produced by each :class:`PytestProcess` into a unified report and calculates per-test coverage fractions used for coverage-efficiency ordering. + +Everything here constructs :class:`coverage.Coverage` objects, which has crashed the +interpreter natively when run on a thread inside the GUI process. The GUI therefore never +calls these functions directly — it goes through +:func:`pytest_fly.pytest_runner.coverage_aggregator.aggregate_coverage`, which runs them in a +short-lived spawn child. Direct calls are for that child and for tests. """ import io @@ -155,22 +161,19 @@ def calculate_coverage(test_identifier: str, coverage_parent_directory: Path, wr return coverage_value, covered_statements, total_statements -def compute_per_test_coverage(data_dir: Path, test_names: list[str]) -> dict[str, float]: - """Compute per-test coverage fractions from stored per-test coverage files. - - Loads each test's individual ``.coverage`` file, counts executed lines, - and divides by the total lines across all tests to produce a fraction. +def per_test_executed_lines(data_dir: Path, test_names: list[str]) -> tuple[dict[str, int], int]: + """Count executed lines per test from the stored per-test ``.coverage`` files. :param data_dir: The application data directory containing the ``coverage/`` subdirectory. :param test_names: List of test node_ids (e.g. ``"tests/test_foo.py"``). - :return: Mapping of test name to coverage fraction (0.0--1.0). Tests - without a coverage file are omitted. + :return: ``(executed_lines_by_test, union_executed_lines)`` — the executed line count of + every test that has a readable coverage file, and the size of the union of executed + lines across all of them (the denominator for coverage-efficiency ordering). """ coverage_dir = Path(data_dir, "coverage") if not coverage_dir.exists(): - return {} + return {}, 0 - # Load each test's coverage data and count executed lines per_test_lines: dict[str, int] = {} all_file_lines: dict[str, set[int]] = {} # source_file -> set of executed line numbers (union across all tests) @@ -192,8 +195,23 @@ def compute_per_test_coverage(data_dir: Path, test_names: list[str]) -> dict[str except COVERAGE_READ_ERRORS as e: log.info(f"per-test coverage load for {test_name} failed: {e}") - total_lines = sum(len(lines) for lines in all_file_lines.values()) - if total_lines == 0: - return {} + union_lines = sum(len(lines) for lines in all_file_lines.values()) + return per_test_lines, union_lines - return {name: executed / total_lines for name, executed in per_test_lines.items()} + +def compute_per_test_coverage(data_dir: Path, test_names: list[str]) -> dict[str, float]: + """Compute per-test coverage fractions from stored per-test coverage files. + + Each test's executed line count divided by the union of executed lines across all the + given tests. Runs in-process — see the module docstring; the GUI obtains the same numbers + via the aggregator child. + + :param data_dir: The application data directory containing the ``coverage/`` subdirectory. + :param test_names: List of test node_ids (e.g. ``"tests/test_foo.py"``). + :return: Mapping of test name to coverage fraction (0.0--1.0). Tests + without a coverage file are omitted. + """ + per_test_lines, union_lines = per_test_executed_lines(data_dir, test_names) + if union_lines == 0: + return {} + return {name: executed / union_lines for name, executed in per_test_lines.items()} diff --git a/src/pytest_fly/pytest_runner/coverage_aggregator.py b/src/pytest_fly/pytest_runner/coverage_aggregator.py index f8957ee..81f993d 100644 --- a/src/pytest_fly/pytest_runner/coverage_aggregator.py +++ b/src/pytest_fly/pytest_runner/coverage_aggregator.py @@ -6,40 +6,82 @@ ``python.dll``) while running on a thread inside the GUI process, taking the orchestrator and the run down with it. -Nothing about that work needs to be in-process: it reads files and returns three numbers. So it +Nothing about that work needs to be in-process: it reads files and returns numbers. So it runs in a short-lived spawn child, like :class:`ProcessMonitor` and :class:`GetTests`. A child that dies or hangs is a logged warning and the last good values stand; the run continues. + +The same child also counts per-test executed lines (for the Coverage tab's per-test view and +for coverage-efficiency ordering) when asked, so that no :class:`coverage.Coverage` object is +ever constructed in the GUI process at all. """ import time +from dataclasses import dataclass, field from multiprocessing import Process, Queue from pathlib import Path from queue import Empty from ..faults import enable_faulthandler from ..logger import EVENT_EXTRA, configure_child_logger, get_logger -from .coverage import calculate_coverage +from .coverage import calculate_coverage, per_test_executed_lines log = get_logger() -CoverageResult = tuple[float | None, int, int] # (coverage 0.0-1.0 or None, covered statements, total statements) + +@dataclass(frozen=True) +class CoverageResult: + """What one aggregation pass produced. + + ``per_test_executed_lines`` and ``union_executed_lines`` are only populated when the + caller supplied ``per_test_names``; they describe those tests only. + """ + + coverage: float | None # overall coverage 0.0-1.0, or None when there is no data yet + covered_statements: int + total_statements: int + per_test_executed_lines: dict[str, int] = field(default_factory=dict) # test node_id -> executed line count + union_executed_lines: int = 0 # size of the union of executed lines across the requested tests + + @property + def totals(self) -> tuple[float | None, int, int]: + """The ``(coverage, covered, total)`` triple, as :func:`calculate_coverage` returns it.""" + return self.coverage, self.covered_statements, self.total_statements + + def per_test_fractions(self, denominator: int | None = None) -> dict[str, float]: + """Per-test executed lines as fractions of *denominator*. + + :param denominator: line count to divide by; ``None`` uses ``union_executed_lines`` + (the coverage-efficiency ordering convention). The Coverage tab passes + ``total_statements`` instead. + :return: empty when the denominator is zero. + """ + if denominator is None: + denominator = self.union_executed_lines + if denominator <= 0: + return {} + return {name: executed / denominator for name, executed in self.per_test_executed_lines.items()} class CoverageAggregator(Process): - """Spawn child that runs :func:`calculate_coverage` once and reports the result on a queue.""" + """Spawn child that runs :func:`calculate_coverage` once and reports a :class:`CoverageResult` on a queue.""" - def __init__(self, test_identifier: str, coverage_parent_directory: Path, write_report: bool) -> None: + def __init__(self, test_identifier: str, coverage_parent_directory: Path, write_report: bool, per_test_names: list[str] | None = None) -> None: super().__init__(name="coverage_aggregator", daemon=True) self._test_identifier = test_identifier self._coverage_parent_directory = coverage_parent_directory self._write_report = write_report + self._per_test_names = list(per_test_names) if per_test_names else [] self._result_queue: Queue = Queue() def run(self) -> None: configure_child_logger("coverage_aggregator.log") enable_faulthandler() # reads PYTEST_FLY_FAULTHANDLER from the inherited environment - result = calculate_coverage(self._test_identifier, self._coverage_parent_directory, self._write_report) - self._result_queue.put(result) + coverage, covered, total = calculate_coverage(self._test_identifier, self._coverage_parent_directory, self._write_report) + per_test: dict[str, int] = {} + union = 0 + if self._per_test_names: + per_test, union = per_test_executed_lines(self._coverage_parent_directory, self._per_test_names) + self._result_queue.put(CoverageResult(coverage, covered, total, per_test, union)) def result(self) -> CoverageResult | None: """The child's result, or ``None`` if it produced none (crashed, killed, or still running).""" @@ -49,15 +91,17 @@ def result(self) -> CoverageResult | None: return None -def aggregate_coverage(test_identifier: str, coverage_parent_directory: Path, write_report: bool, timeout: float) -> CoverageResult | None: +def aggregate_coverage(test_identifier: str, coverage_parent_directory: Path, write_report: bool, timeout: float, per_test_names: list[str] | None = None) -> CoverageResult | None: """Run :func:`calculate_coverage` in a child process and wait for it. :param timeout: seconds before the child is terminated as hung. + :param per_test_names: tests whose executed-line counts should also be returned (see + :attr:`CoverageResult.per_test_executed_lines`); ``None`` skips that pass. :return: the coverage result, or ``None`` when the child crashed, was killed, or timed out — never raises for those; the caller keeps its last good values. """ start = time.monotonic() - child = CoverageAggregator(test_identifier, coverage_parent_directory, write_report) + child = CoverageAggregator(test_identifier, coverage_parent_directory, write_report, per_test_names) child.start() child.join(timeout) if child.is_alive(): diff --git a/tests/test_coverage_aggregator.py b/tests/test_coverage_aggregator.py index 185fa4d..74d70d2 100644 --- a/tests/test_coverage_aggregator.py +++ b/tests/test_coverage_aggregator.py @@ -12,7 +12,7 @@ from pytest_fly.gui.coverage_tracker import CoverageTracker from pytest_fly.interfaces import PyTestFlyExitCode, PytestProcessInfo, PytestRunnerState from pytest_fly.pytest_runner import coverage_aggregator -from pytest_fly.pytest_runner.coverage import PytestFlyCoverage, _parse_report_totals, calculate_coverage +from pytest_fly.pytest_runner.coverage import PytestFlyCoverage, _parse_report_totals, calculate_coverage, compute_per_test_coverage from pytest_fly.pytest_runner.coverage_aggregator import CoverageAggregator, aggregate_coverage from pytest_fly.pytest_runner.run_state import PytestRunState from pytest_fly.tick_data import TickData @@ -51,12 +51,33 @@ def test_child_returns_same_numbers_as_in_process(tmp_path): _write_fixture(tmp_path, "tests/test_one.py", [1, 2]) in_process = calculate_coverage("current", tmp_path, write_report=False) out_of_process = aggregate_coverage("current", tmp_path, write_report=False, timeout=120) - assert out_of_process == in_process - assert out_of_process[1:] == (2, 4) + assert out_of_process is not None + assert out_of_process.totals == in_process + assert out_of_process.totals[1:] == (2, 4) + # No per-test names requested → no per-test pass. + assert out_of_process.per_test_executed_lines == {} and out_of_process.union_executed_lines == 0 + + +def test_child_returns_per_test_executed_lines(tmp_path): + """The same child pass yields per-test counts, so the GUI never constructs a Coverage object itself.""" + _write_fixture(tmp_path, "tests/test_one.py", [1, 2]) + _write_fixture(tmp_path, "tests/test_two.py", [2, 3, 4]) + names = ["tests/test_one.py", "tests/test_two.py", "tests/test_missing.py"] + result = aggregate_coverage("current", tmp_path, write_report=False, timeout=120, per_test_names=names) + assert result is not None + assert result.per_test_executed_lines == {"tests/test_one.py": 2, "tests/test_two.py": 3} + assert result.union_executed_lines == 4 + # Ordering convention: fraction of the union of executed lines. + assert result.per_test_fractions() == {"tests/test_one.py": 0.5, "tests/test_two.py": 0.75} + assert result.per_test_fractions() == compute_per_test_coverage(tmp_path, names) + # Coverage-tab convention: fraction of the report's total statements. + assert result.per_test_fractions(result.total_statements) == {"tests/test_one.py": 2 / result.total_statements, "tests/test_two.py": 3 / result.total_statements} + assert result.per_test_fractions(0) == {} def test_child_with_no_data_returns_empty_result(tmp_path): - assert aggregate_coverage("current", tmp_path, write_report=False, timeout=120) == (None, 0, 0) + result = aggregate_coverage("current", tmp_path, write_report=False, timeout=120) + assert result is not None and result.totals == (None, 0, 0) class _DyingAggregator(CoverageAggregator): diff --git a/tests/test_main_app.py b/tests/test_main_app.py index c51e8aa..7ee4028 100644 --- a/tests/test_main_app.py +++ b/tests/test_main_app.py @@ -1,6 +1,8 @@ """Tests for the application bootstrap in :mod:`pytest_fly.main`.""" import os +import subprocess +import sys from pathlib import Path import pytest @@ -95,3 +97,15 @@ def test_app_main_put_defaults_to_workspace_when_no_target(tmp_path, monkeypatch assert main_module.get_active_put_path() == workspace.resolve() assert captured["data_dir"] == data_dir.resolve() + + +def test_spawn_child_entry_does_not_import_gui(): + """Spawn children re-import ``pytest_fly.__main__``; that must not load PySide6 or pytest-fly's GUI. + + A module-level GUI import in main.py put Qt and every tab widget into each test process, + process monitor, system monitor, and coverage child (~0.3 s and tens of MB each), and left + pytest-fly's Qt binding resident before the program under test's own tests ran. + """ + probe = "import sys, pytest_fly.__main__; print(sorted(m for m in sys.modules if m == 'PySide6' or m.startswith('pytest_fly.gui')))" + completed = subprocess.run([sys.executable, "-c", probe], capture_output=True, text=True, check=True) + assert completed.stdout.strip() == "[]", completed.stdout diff --git a/tests/test_pyside_app.py b/tests/test_pyside_app.py new file mode 100644 index 0000000..26f966d --- /dev/null +++ b/tests/test_pyside_app.py @@ -0,0 +1,126 @@ +"""A small PySide6 application, tested the way a GUI program under test would be. + +pytest-fly runs this ``tests/`` tree as its own program under test, so this module is a +stand-in for a real Qt PUT: it creates a QApplication and widgets inside a pytest-fly test +child, drives them with ``qtbot``, and checks that pytest-fly's own GUI package has *not* +been pre-loaded into that child — a PUT must get a clean process, with only the Qt it +imports itself. +""" + +import multiprocessing +import sys +from pathlib import Path + +from PySide6.QtCore import Qt, Signal +from PySide6.QtWidgets import QLabel, QMainWindow, QPushButton, QVBoxLayout, QWidget + + +class CounterWindow(QMainWindow): + """A window with a label and two buttons that count up and reset.""" + + count_changed = Signal(int) + + def __init__(self) -> None: + super().__init__() + self.setWindowTitle("Counter") + self._count = 0 + + self.label = QLabel(self._label_text()) + self.label.setAlignment(Qt.AlignmentFlag.AlignCenter) + + self.increment_button = QPushButton("Increment") + self.increment_button.clicked.connect(self.increment) + + self.reset_button = QPushButton("Reset") + self.reset_button.clicked.connect(self.reset) + self.reset_button.setEnabled(False) + + central = QWidget() + layout = QVBoxLayout(central) + layout.addWidget(self.label) + layout.addWidget(self.increment_button) + layout.addWidget(self.reset_button) + self.setCentralWidget(central) + + @property + def count(self) -> int: + return self._count + + def increment(self) -> None: + self._set_count(self._count + 1) + + def reset(self) -> None: + self._set_count(0) + + def _set_count(self, value: int) -> None: + self._count = value + self.label.setText(self._label_text()) + self.reset_button.setEnabled(value != 0) + self.count_changed.emit(value) + + def _label_text(self) -> str: + return f"Count: {self._count}" + + +def test_click_increments(qtbot): + window = CounterWindow() + qtbot.addWidget(window) + window.show() + + assert window.count == 0 + assert window.label.text() == "Count: 0" + assert not window.reset_button.isEnabled() + + qtbot.mouseClick(window.increment_button, Qt.MouseButton.LeftButton) + qtbot.mouseClick(window.increment_button, Qt.MouseButton.LeftButton) + + assert window.count == 2 + assert window.label.text() == "Count: 2" + assert window.reset_button.isEnabled() + + +def test_reset_clears_and_disables(qtbot): + window = CounterWindow() + qtbot.addWidget(window) + window.increment() + window.increment() + + qtbot.mouseClick(window.reset_button, Qt.MouseButton.LeftButton) + + assert window.count == 0 + assert window.label.text() == "Count: 0" + assert not window.reset_button.isEnabled() + + +def test_count_changed_signal(qtbot): + window = CounterWindow() + qtbot.addWidget(window) + + with qtbot.waitSignal(window.count_changed, timeout=1000) as blocker: + qtbot.mouseClick(window.increment_button, Qt.MouseButton.LeftButton) + + assert blocker.args == [1] + + +def _running_as_pytest_fly_child() -> bool: + """True inside a :class:`PytestProcess` spawn child running this module. + + PytestProcess names its process after the test module it runs, so the child's + ``current_process().name`` is this file's path; a top-level ``pytest tests/`` run is + ``MainProcess``. (An environment variable is not a safe signal: the suite's own + ``enable_faulthandler`` tests export one into the top-level process.) + """ + process = multiprocessing.current_process() + return multiprocessing.parent_process() is not None and Path(process.name).name == Path(__file__).name + + +def test_pytest_fly_gui_not_preloaded_into_put_process(): + """Inside a pytest-fly test child, only the PUT's own Qt is present — not pytest-fly's GUI. + + Under a plain ``pytest tests/`` run the suite's own GUI tests legitimately import + ``pytest_fly.gui``, so the check applies only when this module is the program under test. + """ + if not _running_as_pytest_fly_child(): + return + preloaded = sorted(m for m in sys.modules if m.startswith("pytest_fly.gui")) + assert preloaded == [], f"pytest-fly's GUI package leaked into the program-under-test process: {preloaded}"