Make native flash, snapshot and token-ID tests deterministic

pyproject.toml

@@ -33,7 +33,7 @@ dependencies = [
     "psutil>=5.9",
     "Pygments>=2.15",
     "pyzmq>=22.0",
-    "python-introspect>=0.2.3",
+    "python-introspect>=0.2.5",
     "zmqruntime>=0.3.0,<0.5",
 ]
 

tests/conftest.py

@@ -14,3 +14,32 @@ def qapp():
     ThemeManager(color_scheme).apply_color_scheme(color_scheme)
     yield app
     # Don't quit - may cause issues with other tests
+
+
+class FrozenClock:
+    """Flash clock that moves only when a test advances it.
+
+    Flash phases change only with this clock, so the event loop can paint at
+    any speed without moving a playback past the phase a test checks.
+    """
+
+    def __init__(self) -> None:
+        self.now = 1000.0
+
+    def perf_counter(self) -> float:
+        return self.now
+
+    def advance(self, seconds: float) -> None:
+        self.now += seconds
+
+
+@pytest.fixture
+def flash_clock(monkeypatch):
+    """Drive the flash coordinator's clock explicitly."""
+    from types import SimpleNamespace
+
+    from pyqt_reactive.animation import flash_mixin
+
+    clock = FrozenClock()
+    monkeypatch.setattr(flash_mixin, "time", SimpleNamespace(perf_counter=clock.perf_counter))
+    return clock

tests/test_flash_playback.py

@@ -63,7 +63,7 @@ def test_phase_leaves_own_duration_and_opacity(config, phase):
 
 
 @pytest.fixture
-def actual_surfaces(qapp):
+def actual_surfaces(qapp, qtbot, flash_clock):
     from PyQt6 import sip
     from PyQt6.QtWidgets import QDialog, QPushButton, QVBoxLayout
     from pyqt_reactive.animation.flash_mixin import WindowFlashOverlay, _GlobalFlashCoordinator, create_widget_rect_element
@@ -79,16 +79,18 @@ def create(*keys):
         layout = QVBoxLayout(window)
         for key in keys:
             layout.addWidget(QPushButton(key))
-        window.show()
-        qapp.processEvents()
+        # Only natively exposed windows receive the flash, and native
+        # platforms expose a shown window asynchronously.
+        with qtbot.waitExposed(window):
+            window.show()
         overlay = WindowFlashOverlay.get_for_window(window)
         for key, button in zip(keys, window.findChildren(QPushButton)):
             overlay.register_element(create_widget_rect_element(key, button))
         windows.append(window)
         overlays.append(overlay)
         return window, overlay
 
-    yield coordinator, create
+    yield coordinator, create, flash_clock
     if coordinator._timer is not None:
         coordinator._timer.stop()
     coordinator._playbacks.clear()
@@ -103,7 +105,7 @@ def create(*keys):
 
 
 def test_native_subset_retrigger_drops_former_only_recipients_from_old_cohort(actual_surfaces, qtbot):
-    coordinator, create = actual_surfaces
+    coordinator, create, clock = actual_surfaces
     window, overlay = create("first", "second")
     painted = []
     overlay.frame_painted.connect(painted.append)
@@ -119,6 +121,7 @@ def test_native_subset_retrigger_drops_former_only_recipients_from_old_cohort(ac
     coordinator._on_global_tick()
     assert len(old.pending_recipients) == 1
     assert len(retrigger.pending_recipients) == 1
+    clock.advance(coordinator._config.fade_in_s)
     qtbot.waitUntil(lambda: old.phase is FlashPhase.HOLD and retrigger.phase is FlashPhase.HOLD, timeout=1500)
     assert any(element.key == "second" and element.rgba[3] == 255
                for frame in painted for element in frame.elements)
@@ -127,7 +130,7 @@ def test_native_subset_retrigger_drops_former_only_recipients_from_old_cohort(ac
 @pytest.mark.parametrize("disappears", ("hidden", "disposed"))
 def test_native_hidden_or_disposed_window_cannot_block_exposed_peer(actual_surfaces, qtbot, qapp, disappears):
     from PyQt6.QtCore import QEvent
-    coordinator, create = actual_surfaces
+    coordinator, create, clock = actual_surfaces
     first, overlay = create("first")
     second, _ = create("second")
     painted = []
@@ -142,6 +145,7 @@ def test_native_hidden_or_disposed_window_cannot_block_exposed_peer(actual_surfa
     else:
         second.deleteLater()
         qapp.sendPostedEvents(None, QEvent.Type.DeferredDelete)
+    clock.advance(coordinator._config.fade_in_s)
     qtbot.waitUntil(lambda: playback.phase is FlashPhase.HOLD, timeout=1500)
     assert "second" not in coordinator._playbacks
     assert any(frame.has_maximum_alpha for frame in painted)

tests/test_function_list_editor_tokens.py

@@ -305,7 +305,7 @@ class Step(ScopeTokenTarget):
     assert vars(target) == {"number": 3, "_alternative_token": "step_9"}
 
 
-def test_native_reused_object_id_does_not_override_restored_token(monkeypatch):
+def test_reused_object_id_does_not_override_restored_token(monkeypatch):
     @dataclass
     class Step(ScopeTokenTarget):
         number: int = 3
@@ -314,10 +314,14 @@ class Step(ScopeTokenTarget):
     monkeypatch.setattr(ScopeTokenService, "_scope_id_cache", {})
     previous = Step()
     assert ScopeTokenService.build_scope_id("plate", previous) == "plate::step_0"
-    departed_id = id(previous)
+    departed_key = ("plate", id(previous))
     del previous
     detached = Step(99)
-    assert id(detached) == departed_id, "Native CPython reproduction must actually reuse the departed ID"
+    # CPython may give the new object the departed object's address, but the
+    # allocator does not promise it. Leave the departed entry under the new
+    # object's id, which is the state an address reuse produces.
+    cache = ScopeTokenService._scope_id_cache
+    cache[("plate", id(detached))] = cache.pop(departed_key)
     generator = ScopeTokenService.get_generator("plate", "step")
     counter_before = generator._counter
     cache_before = dict(ScopeTokenService._scope_id_cache)

tests/test_window_snapshot.py

@@ -20,7 +20,7 @@
 
 
 @pytest.fixture
-def observed_form(qapp):
+def observed_form(qapp, qtbot):
     import objectstate.config as config_module
     from objectstate import ObjectState, ObjectStateRegistry, set_base_config_type
     from PyQt6.QtCore import QEvent
@@ -64,8 +64,8 @@ def grab(self, *args):
         ),
     )
     QVBoxLayout(window).addWidget(form)
-    window.show()
-    qapp.processEvents()
+    with qtbot.waitExposed(window):
+        window.show()
     yield window, form
     coordinator = _GlobalFlashCoordinator.get()
     if coordinator._timer is not None:
@@ -128,7 +128,7 @@ class SubjectCapture(WindowSnapshotCaptureSpec):
 
 
 @pytest.fixture
-def rendered_canvas(qapp):
+def rendered_canvas(qapp, qtbot):
     """A real Qt painter and completion receipt, never a pixel/content mock."""
     from PyQt6.QtCore import QEvent, pyqtSignal
     from PyQt6.QtGui import QColor, QPainter
@@ -150,8 +150,10 @@ def paintEvent(self, event):
 
     canvas = Canvas()
     canvas.resize(80, 60)
-    canvas.show()
-    qapp.processEvents()
+    # Native platforms expose a shown window asynchronously; Qt paints only
+    # exposed windows.
+    with qtbot.waitExposed(canvas):
+        canvas.show()
     yield canvas
     from PyQt6 import sip
 
@@ -731,6 +733,7 @@ def test_two_actual_windows_both_present_maximum_before_shared_hold(
     qtbot,
     observed_form,
     tmp_path,
+    flash_clock,
 ):
     from PyQt6.QtWidgets import QDialog, QVBoxLayout
     from pyqt_reactive.animation.flash_config import FlashPhase
@@ -750,8 +753,8 @@ def test_two_actual_windows_both_present_maximum_before_shared_hold(
         ),
     )
     QVBoxLayout(second).addWidget(second_form)
-    second.show()
-    qtbot.wait(20)
+    with qtbot.waitExposed(second):
+        second.show()
     coordinator = _GlobalFlashCoordinator.get()
     overlay = WindowFlashOverlay.get_for_window(first)
     phases_after_first_maximum = []
@@ -773,6 +776,8 @@ def actual_first_paint(frame):
                 failed.append,
             )
         form.update_parameter("number", 8)
+        qtbot.waitUntil(lambda: "number" in coordinator._playbacks, timeout=2000)
+        flash_clock.advance(coordinator._config.fade_in_s)
         qtbot.waitUntil(lambda: len(completed) == 2 or bool(failed), timeout=2000)
         assert not failed and len(completed) == 2
         assert {result.observation.frame.window_identity for result in completed} == {