feat(soak): function exerciser — every user action churned + invariant-checked (v10.12.0)

hyperwall/__init__.py

@@ -7,7 +7,7 @@
 
 from __future__ import annotations
 
-__version__ = "10.11.0"
+__version__ = "10.12.0"
 # Short "major.minor" form, derived — used for User-Agent / Emby auth Version /
 # window titles so a version bump touches exactly ONE line (this file).
 VERSION_SHORT = ".".join(__version__.split(".")[:2])

hyperwall/soak.py

@@ -39,6 +39,10 @@
 
 SOAK_MINUTES = int(os.environ.get("HYPERWALL_SOAK_MINUTES", "0") or 0)
 SOAK_DWELL_S = int(os.environ.get("HYPERWALL_SOAK_DWELL_S", "75") or 0)
+# Function exerciser: each churn tick drives a random USER ACTION through
+# the same handlers a real click takes (advance/prev/seek/mute/volume/
+# pause/loop/favorite), then verifies state invariants. 0 = advances only.
+SOAK_ACTIONS = os.environ.get("HYPERWALL_SOAK_ACTIONS", "1") == "1"
 
 _RES_SAMPLE_S = 60
 
@@ -102,6 +106,10 @@ def __init__(self, wall) -> None:
         self._wall = wall
         self._t0 = time.monotonic()
         self._advances = 0
+        self._action_counts: dict[str, int] = {}
+        self._invariant_violations = 0
+        self._unmuted_cell = None   # at most one audible cell during a soak
+        self._paused_cell = None    # at most one paused cell during a soak
         self._baseline = _resource_snapshot()
 
         self._res_timer = QTimer(self)
@@ -132,32 +140,131 @@ def _arm_churn(self) -> None:
         base_ms = SOAK_DWELL_S * 1000 / cells
         self._churn_timer.start(int(base_ms * random.uniform(0.75, 1.25)))
 
+    # (action, weight). The trash tag is deliberately ABSENT: soak-writing
+    # ToDelete tags feeds the cleanup-delete pipeline - a stray one could
+    # cost a real file. Favorites are exercised as immediate double-toggles
+    # (net zero; the writes are HTTP-status-verified since the audit).
+    _ACTIONS = (
+        ("advance", 40), ("seek", 15), ("audio", 15), ("volume", 10),
+        ("pause", 10), ("prev", 4), ("loop", 3), ("favorite", 3),
+    )
+
     def _churn(self) -> None:
         try:
             cell = random.choice(self._wall.cells)
-            self._advances += 1
-            self._wall.next_video(cell, False)
+            if SOAK_ACTIONS:
+                names = [a for a, _ in self._ACTIONS]
+                weights = [w for _, w in self._ACTIONS]
+                action = random.choices(names, weights=weights, k=1)[0]
+            else:
+                action = "advance"
+            self._action_counts[action] = self._action_counts.get(action, 0) + 1
+            self._do_action(action, cell)
+            self._verify_invariants(cell, action)
         except Exception as e:
-            logger.warning("SOAK churn advance failed: %s", e)
+            logger.warning("SOAK action failed: %s", e)
         self._arm_churn()
 
+    def _do_action(self, action: str, cell) -> None:
+        """Drive one user action through the real handlers (button clicks /
+        slider grabs), never through private state pokes."""
+        if action == "advance":
+            self._advances += 1
+            self._wall.next_video(cell, False)
+        elif action == "prev":
+            self._wall.prev_video(cell)
+        elif action == "seek":
+            if cell._mpv is None or cell._duration_s <= 0:
+                return
+            cell.seek_slider.setSliderDown(True)
+            cell.seek_slider.setValue(random.randint(50, 980))
+            cell.seek_slider.setSliderDown(False)
+        elif action == "audio":
+            # Cycle audibility with at most ONE cell unmuted wall-wide, at a
+            # civilized volume - an hour of testing must not be an hour of
+            # random noise. Exercises the lazy audio arm + relock seek.
+            prev = self._unmuted_cell
+            if prev is not None and prev is not cell and not prev.muted:
+                prev.btn_mute.click()          # re-mute the previous one
+            if cell.muted:
+                cell.btn_mute.click()          # unmute (restores last vol)
+                cell.vol_slider.setValue(25)
+                self._unmuted_cell = cell
+            else:
+                cell.btn_mute.click()          # mute it back
+                self._unmuted_cell = None
+        elif action == "volume":
+            if not cell.muted:
+                cell.vol_slider.setValue(random.randint(10, 60))
+            # volume drag on a muted cell would unmute it - audibility is
+            # owned by the "audio" action to keep the one-audible invariant.
+        elif action == "pause":
+            prev = self._paused_cell
+            if prev is not None and prev is not cell and prev._paused:
+                prev.btn_play.click()          # resume the previous one
+            cell.btn_play.click()
+            self._paused_cell = cell if cell._paused else None
+        elif action == "loop":
+            cell.btn_loop.click()              # on
+            cell.btn_loop.click()              # immediately off (net zero)
+        elif action == "favorite":
+            if cell.current_item is not None:
+                cell.btn_fav.click()           # toggle
+                cell.btn_fav.click()           # restore (net zero)
+
+    def _verify_invariants(self, cell, action: str) -> None:
+        """The exerciser is a TEST: after every action, cached state, button
+        state, and QSS properties must agree. A mismatch is exactly the
+        state-drift class the 2026-07-13 audit hunted."""
+        problems = []
+        if cell.muted != cell.btn_mute.isChecked():
+            problems.append(
+                f"muted={cell.muted} vs btn_checked={cell.btn_mute.isChecked()}"
+            )
+        if cell.btn_mute.property("audible") is not (not cell.muted):
+            problems.append(
+                f"audible prop={cell.btn_mute.property('audible')} "
+                f"vs muted={cell.muted}"
+            )
+        if not cell.muted and cell.vol_slider.value() == 0:
+            problems.append("unmuted with volume slider at 0")
+        if cell.looping != cell.btn_loop.isChecked():
+            problems.append(
+                f"looping={cell.looping} vs btn={cell.btn_loop.isChecked()}"
+            )
+        if problems:
+            self._invariant_violations += 1
+            logger.warning(
+                "SOAK INVARIANT violated after %r: %s",
+                action, "; ".join(problems),
+            )
+
     def _sample(self) -> None:
         snap = _resource_snapshot()
         mins = (time.monotonic() - self._t0) / 60
         logger.info(
             "SOAK res @%.0fmin: ws=%sMB private=%sMB gdi=%s user=%s "
-            "threads=%s churn_advances=%d",
+            "threads=%s actions=%s invariant_violations=%d",
             mins, snap.get("ws_mb"), snap.get("private_mb"),
             snap.get("gdi"), snap.get("user"), snap.get("threads"),
-            self._advances,
+            dict(sorted(self._action_counts.items())),
+            self._invariant_violations,
         )
 
     def _finish(self) -> None:
         snap = _resource_snapshot()
         logger.info(
-            "SOAK done after %d min: churn_advances=%d  baseline %s → final %s",
-            SOAK_MINUTES, self._advances, self._baseline, snap,
+            "SOAK done after %d min: actions=%s invariant_violations=%d  "
+            "baseline %s → final %s",
+            SOAK_MINUTES, dict(sorted(self._action_counts.items())),
+            self._invariant_violations, self._baseline, snap,
         )
         self._res_timer.stop()
         self._churn_timer.stop()
+        # Leave the wall silent regardless of where the audio cycle ended.
+        if self._unmuted_cell is not None and not self._unmuted_cell.muted:
+            try:
+                self._unmuted_cell.btn_mute.click()
+            except Exception:
+                pass
         self._wall._shutdown()

tests/run_repo_guards.py

@@ -46,10 +46,10 @@ def test_01_entry_point_imports():
 def test_02_package_identity():
     """Package has version and banner."""
     from hyperwall import __version__, runtime_banner
-    assert __version__ == "10.11.0"
+    assert __version__ == "10.12.0"
     banner = runtime_banner()
     assert "Hyperwall" in banner
-    assert "10.11.0" in banner
+    assert "10.12.0" in banner
 
 
 def test_03_config_loads():

tests/test_instrumentation.py

@@ -12,6 +12,9 @@
 sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
 os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
 os.environ["HYPERWALL_SOAK_MINUTES"] = "1"
+# Plain-wall churn test needs deterministic advances (fake cells
+# have no buttons for the function exerciser).
+os.environ["HYPERWALL_SOAK_ACTIONS"] = "0"
 
 # The ubuntu CI job is the pure-logic lane and deliberately has no PyQt
 # (and hyperwall.soak needs ctypes.wintypes anyway); these tests then skip
@@ -68,6 +71,58 @@ def test_resource_snapshot_returns_real_values():
     assert "user" in snap and "gdi" in snap
 
 
+def test_soak_function_exerciser_drives_real_cell():
+    """Every soak action runs against a real VideoCell through the real
+    handlers, and the state invariants hold after each one."""
+    from hyperwall.soak import SoakController
+    from hyperwall.cell import VideoCell
+
+    class _FakeMpv:
+        def __init__(self):
+            self.props = {"mute": True, "volume": 100.0, "aid": "no",
+                          "pause": False, "time-pos": 12.0,
+                          "eof-reached": False}
+        def __setitem__(self, k, v): self.props[k] = v
+        def __getitem__(self, k): return self.props[k]
+        def __getattr__(self, name):
+            key = name.replace("_", "-")
+            if key in self.props: return self.props[key]
+            raise AttributeError(name)
+        def seek(self, *a, **k): pass
+        def command(self, *a): pass
+
+    class _Ctl:
+        controls_visible = True
+        def update_favorite(self, *a): pass
+
+    cell = VideoCell(_Ctl())
+    cell.resize(1280, 720)
+    cell.show()
+    _app.processEvents()
+    cell._mpv = _FakeMpv()
+    cell._duration_s = 100.0
+    cell.current_item = {"Id": "x", "Name": "n",
+                         "UserData": {"IsFavorite": False}, "Tags": []}
+
+    class _W:
+        def __init__(self): self.cells = [cell]; self.nexts = 0; self.prevs = 0
+        def next_video(self, c, r=False): self.nexts += 1
+        def prev_video(self, c): self.prevs += 1
+        def _shutdown(self): pass
+
+    wall = _W()
+    soak = SoakController(wall)
+    for action in ("advance", "prev", "seek", "audio", "volume",
+                   "pause", "loop", "favorite", "audio"):
+        soak._do_action(action, cell)
+        soak._verify_invariants(cell, action)
+    assert soak._invariant_violations == 0, "state drift under exerciser"
+    assert wall.nexts == 1 and wall.prevs == 1
+    assert cell.muted is True, "second audio action must re-mute"
+    assert cell.looping is False, "loop must double-toggle to off"
+    assert cell.current_item["UserData"]["IsFavorite"] is False
+
+
 def test_soak_controller_constructs_against_plain_wall():
     from hyperwall.soak import SoakController
     wall = _PlainWall()