test: Test Harness & CI — url/transcode/config/escalation coverage (Epic 3, #8)

.github/workflows/repo-guards.yml

@@ -1,24 +1,24 @@
 name: repo-guards
 
-# Runs the no-dependency repo guards (tests/run_repo_guards.py) on every push
-# and PR. Guard #8 (test_08_no_versioned_exe_literals) fails the build if the
-# hyperwall_v8 / hardcoded-version drift ever creeps back — the permanent gate
-# for Epic 1 (Identity Unification). Full unit tests arrive in Epic 3 (#8).
+# Runs all Hyperwall test suites (tests/run_all.py) on every push and PR:
+#   - repo guards (structure + version-drift guard, Epic 1)
+#   - reliability unit tests (Epic 2)
+#   - url / transcode unit tests incl. load-bearing static=true (Epic 3)
+#   - config round-trip (Epic 3)
+# No pytest / PyQt / mpv needed — pure-logic suites run headless.
 
 on:
   push:
     branches: [main]
   pull_request:
 
 jobs:
-  guards:
+  tests:
     runs-on: ubuntu-latest
     steps:
       - uses: actions/checkout@v4
       - uses: actions/setup-python@v5
         with:
           python-version: "3.12"
-      - name: Run repo guards
-        run: python tests/run_repo_guards.py
-      - name: Run reliability unit tests
-        run: python tests/test_reliability.py
+      - name: Run all test suites
+        run: python tests/run_all.py

README.md

@@ -130,6 +130,23 @@ Produces `hyperwall.exe` — a versionless basename. G-Sync isolation is gated
 on the `hyperwall*.exe` prefix (or `HYPERWALL_ISOLATED=1`), so the exe name is
 stable across version bumps and the NVIDIA profile keeps matching.
 
+## Testing
+
+The test suites are pure-logic and dependency-light — no PyQt/mpv/Emby, no
+pytest. They run headless in CI (`.github/workflows/repo-guards.yml`) and
+locally:
+
+```bash
+python tests/run_all.py
+```
+
+| Suite | Covers |
+|---|---|
+| `run_repo_guards` | Package structure + version-drift guard (no `hyperwall_v<N>` / hardcoded version literals) |
+| `test_reliability` | Stall watchdog, crash-loop guard, cache-budget scaling, retry→transcode→skip escalation |
+| `test_urls` | Emby URL construction (incl. load-bearing `static=true`) + transcode heuristic boundaries |
+| `test_config` | `config.ini` save/load round-trip, typed fields, frozen dataclass |
+
 ## License
 
 MIT

hyperwall/cell.py

@@ -67,7 +67,7 @@
     WATCHDOG_INTERVAL_MS,
     apply_env_overrides,
 )
-from .reliability import is_stalled, should_park
+from .reliability import escalation_plan, is_stalled, should_park
 
 logger = logging.getLogger("HyperWall")
 
@@ -899,12 +899,13 @@ def _on_error(self) -> None:
         logger.warning(
             "Playback error (attempt %d/%d)", self._retry_count, MAX_RETRIES
         )
-        if self._retry_count <= MAX_RETRIES:
-            if self._retry_count >= 2 and not self._force_transcode:
+        plan = escalation_plan(self._retry_count, MAX_RETRIES)
+        if plan["action"] == "retry":
+            if plan["transcode"] and not self._force_transcode:
                 self._force_transcode = True
                 logger.info("Escalating to server transcode.")
             QTimer.singleShot(
-                (2 ** self._retry_count) * 1000,
+                plan["delay_s"] * 1000,
                 lambda: self._request_next_throttled(True),
             )
         else:

hyperwall/emby.py

@@ -26,15 +26,14 @@
 
 
 def needs_transcode(item: dict[str, Any]) -> bool:
-    """Heuristic: return True if the source exceeds 1080p."""
-    if not _AUTO_TRANSCODE:
-        return False
-    src = (item.get("MediaSources") or [{}])[0]
-    streams = src.get("MediaStreams") or item.get("MediaStreams") or []
-    v = next((s for s in streams if s.get("Type") == "Video"), {}) or {}
-    w = v.get("Width") or 0
-    h = v.get("Height") or 0
-    return w > 1920 or h > 1080
+    """Heuristic: return True if the source exceeds 1080p.
+
+    Thin wrapper over the pure helper in urls.py, binding the resolved
+    HYPERWALL_AUTO_TRANSCODE flag. Kept here for backward-compatible imports.
+    """
+    from .urls import needs_transcode as _needs_transcode
+
+    return _needs_transcode(item, auto_transcode=_AUTO_TRANSCODE)
 
 
 class EmbyClient:

hyperwall/reliability.py

@@ -71,3 +71,26 @@ def scale_demuxer_mb(
     n = max(1, int(n_cells))
     per = min(per_cell_mb, total_budget_mb / n)
     return int(max(floor_mb, per))
+
+
+def escalation_plan(attempt: int, max_retries: int) -> dict:
+    """Decide what to do after a playback failure at 1-based `attempt`.
+
+    Returns a small plan dict describing the retry/escalation policy, so the
+    behavior is testable without QTimer/mpv:
+
+      - action: "retry" (try again) or "skip" (give up, advance playlist)
+      - transcode: True once attempt >= 2 (escalate to server transcode)
+      - delay_s: backoff before the next attempt (2**attempt for retries, 0
+                 for a skip)
+
+    attempt 1 → retry direct (2s), attempt 2 → retry transcode (4s),
+    attempt 3 → retry transcode (8s), attempt 4 (> max_retries=3) → skip.
+    """
+    if attempt <= max_retries:
+        return {
+            "action": "retry",
+            "transcode": attempt >= 2,
+            "delay_s": 2 ** attempt,
+        }
+    return {"action": "skip", "transcode": False, "delay_s": 0}

hyperwall/urls.py

@@ -0,0 +1,66 @@
+"""
+Hyperwall — pure Emby URL construction + transcode heuristic (no PyQt / mpv).
+
+Extracted so the genuinely bug-prone playback-URL logic is unit-testable
+without a display server or a live Emby instance (Epic 3 / #8). `emby.py` and
+`wall.py` delegate here.
+
+⚠️ LOAD-BEARING: the DIRECT url MUST carry `static=true`. Emby 4.9.5.0's
+`/Videos/{id}/stream` without `static=true` returns HTTP 500 for every item
+(ffmpeg remux writes a temp file with no extension). `static=true` serves the
+raw file. Do not remove it without live-curl proof against the target instance.
+The transcode path uses `master.m3u8` (HLS) which is a different, working code
+path on the server.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+
+def exceeds_1080p(item: dict[str, Any]) -> bool:
+    """True if the item's primary video stream is wider than 1920 or taller
+    than 1080. Tolerant of Emby's two shapes (MediaSources[0].MediaStreams or
+    a top-level MediaStreams) and missing/None dimensions."""
+    src = (item.get("MediaSources") or [{}])[0]
+    streams = src.get("MediaStreams") or item.get("MediaStreams") or []
+    v = next((s for s in streams if s.get("Type") == "Video"), {}) or {}
+    w = v.get("Width") or 0
+    h = v.get("Height") or 0
+    return w > 1920 or h > 1080
+
+
+def needs_transcode(item: dict[str, Any], *, auto_transcode: bool = True) -> bool:
+    """Whether the auto-transcode heuristic wants a server-side downscale.
+
+    `auto_transcode` is the resolved HYPERWALL_AUTO_TRANSCODE flag; when False
+    the heuristic is disabled and everything tries DIRECT first.
+    """
+    if not auto_transcode:
+        return False
+    return exceeds_1080p(item)
+
+
+def build_stream_url(
+    *,
+    base: str,
+    item_id: str,
+    api_key: str,
+    session_id: str,
+    transcode: bool,
+) -> str:
+    """Build the Emby stream URL for an item.
+
+    transcode=True  → HLS master playlist (server-side transcode to 1080p h264).
+    transcode=False → DIRECT raw file via static=true (load-bearing — see
+                      module docstring).
+    """
+    if transcode:
+        return (
+            f"{base}/Videos/{item_id}/master.m3u8?api_key={api_key}"
+            f"&VideoCodec=h264&AudioCodec=aac&MaxAudioChannels=2"
+            f"&MaxHeight=1080&MaxWidth=1920"
+            f"&MaxFramerate=30&VideoBitrate=12000000"
+            f"&PlaySessionId={session_id}"
+        )
+    return f"{base}/Videos/{item_id}/stream?api_key={api_key}&static=true"

hyperwall/wall.py

@@ -46,6 +46,7 @@
     SCRIPT_DIR,
 )
 from .emby import EmbyClient, ContentLoader, needs_transcode
+from .urls import build_stream_url
 
 logger = logging.getLogger("HyperWall")
 
@@ -231,18 +232,15 @@ def _build_url(
         sid = uuid.uuid4().hex
 
         auto_transcode = needs_transcode(item)
-        if force_transcode or auto_transcode:
-            url = (
-                f"{base}/Videos/{iid}/master.m3u8?api_key={key}"
-                f"&VideoCodec=h264&AudioCodec=aac&MaxAudioChannels=2"
-                f"&MaxHeight=1080&MaxWidth=1920"
-                f"&MaxFramerate=30&VideoBitrate=12000000"
-                f"&PlaySessionId={sid}"
-            )
+        transcode = bool(force_transcode or auto_transcode)
+        url = build_stream_url(
+            base=base, item_id=iid, api_key=key,
+            session_id=sid, transcode=transcode,
+        )
+        if transcode:
             tag = "TRANSCODE/retry" if force_transcode else "TRANSCODE/auto"
             logger.info("[%s] %s", tag, item.get("Name"))
         else:
-            url = f"{base}/Videos/{iid}/stream?api_key={key}&static=true"
             logger.info("[DIRECT] %s", item.get("Name"))
         return url, sid
 

tests/run_all.py

@@ -0,0 +1,44 @@
+"""
+Run every Hyperwall test suite and aggregate the result.
+
+Single entry point for CI and local dev — no pytest dependency. Discovers the
+sibling test modules, runs each module's run_all(), and returns nonzero if any
+suite has a failure.
+
+Run: python tests/run_all.py
+"""
+
+from __future__ import annotations
+
+import importlib
+import os
+import sys
+
+REPO_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+sys.path.insert(0, REPO_ROOT)
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+
+SUITES = [
+    "run_repo_guards",
+    "test_reliability",
+    "test_urls",
+    "test_config",
+]
+
+
+def main() -> int:
+    total_failures = 0
+    for name in SUITES:
+        print(f"\n=== {name} ===")
+        mod = importlib.import_module(name)
+        total_failures += mod.run_all()
+    print("\n" + "=" * 48)
+    if total_failures:
+        print(f"OVERALL: FAIL ({total_failures} failing test(s))")
+    else:
+        print("OVERALL: PASS (all suites green)")
+    return total_failures
+
+
+if __name__ == "__main__":
+    sys.exit(main())

tests/test_config.py

@@ -0,0 +1,116 @@
+"""
+Unit tests for hyperwall.config (Epic 3) — config save/load round-trip.
+
+No PyQt / mpv / Emby. Uses a temp dir. Run: python tests/test_config.py
+"""
+
+from __future__ import annotations
+
+import os
+import sys
+import tempfile
+
+REPO_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+sys.path.insert(0, REPO_ROOT)
+
+from hyperwall.config import HyperwallConfig, ConfigMissingError  # noqa: E402
+
+
+def test_missing_config_creates_template_and_raises():
+    with tempfile.TemporaryDirectory() as d:
+        path = os.path.join(d, "config.ini")
+        raised = False
+        try:
+            HyperwallConfig.load(path)
+        except ConfigMissingError:
+            raised = True
+        assert raised, "expected ConfigMissingError on first load"
+        assert os.path.exists(path), "template should have been written"
+        # Template must contain both sections.
+        with open(path) as f:
+            content = f.read()
+        assert "[Login]" in content
+        assert "[Settings]" in content
+
+
+def test_save_then_load_round_trips():
+    with tempfile.TemporaryDirectory() as d:
+        path = os.path.join(d, "config.ini")
+        cfg = HyperwallConfig(
+            server_url="http://emby:8096",
+            username="alice",
+            password="s3cret",
+            verify_ssl=False,
+            last_screens="DP-1,DP-2",
+            last_libraries="Movies,Music Videos",
+            last_grid_rows=3,
+            last_grid_cols=4,
+            cleanup_on_startup=True,
+        )
+        cfg.save(path)
+        loaded = HyperwallConfig.load(path)
+        assert loaded.server_url == "http://emby:8096"
+        assert loaded.username == "alice"
+        assert loaded.password == "s3cret"
+        assert loaded.verify_ssl is False
+        assert loaded.last_screens == "DP-1,DP-2"
+        assert loaded.last_libraries == "Movies,Music Videos"
+        assert loaded.last_grid_rows == 3
+        assert loaded.last_grid_cols == 4
+        assert loaded.cleanup_on_startup is True
+
+
+def test_typed_fields_are_correct_types():
+    with tempfile.TemporaryDirectory() as d:
+        path = os.path.join(d, "config.ini")
+        HyperwallConfig(
+            server_url="http://h", username="u", password="p",
+            last_grid_rows=2, last_grid_cols=2,
+        ).save(path)
+        loaded = HyperwallConfig.load(path)
+        # Ints must load as ints, bools as bools — not strings.
+        assert isinstance(loaded.last_grid_rows, int)
+        assert isinstance(loaded.last_grid_cols, int)
+        assert isinstance(loaded.verify_ssl, bool)
+        assert isinstance(loaded.cleanup_on_startup, bool)
+
+
+def test_config_is_frozen():
+    cfg = HyperwallConfig(server_url="http://h", username="u", password="p")
+    frozen = False
+    try:
+        cfg.server_url = "mutated"  # type: ignore[misc]
+    except Exception:
+        frozen = True
+    assert frozen, "HyperwallConfig should be an immutable (frozen) dataclass"
+
+
+def test_defaults_applied_for_absent_settings():
+    with tempfile.TemporaryDirectory() as d:
+        path = os.path.join(d, "config.ini")
+        # Write a minimal config with only [Login].
+        with open(path, "w") as f:
+            f.write("[Login]\nserver_url = http://h\nusername = u\npassword = p\n")
+        loaded = HyperwallConfig.load(path)
+        assert loaded.last_grid_rows == 2   # fallback default
+        assert loaded.last_grid_cols == 2
+        assert loaded.cleanup_on_startup is False
+
+
+def run_all() -> int:
+    tests = [v for k, v in sorted(globals().items()) if k.startswith("test_")]
+    passed = failed = 0
+    for t in tests:
+        try:
+            t()
+            passed += 1
+            print(f"  PASS  {t.__name__}")
+        except Exception as e:  # noqa: BLE001
+            failed += 1
+            print(f"  FAIL  {t.__name__}: {e}")
+    print(f"\n{passed} passed, {failed} failed out of {len(tests)} tests.")
+    return failed
+
+
+if __name__ == "__main__":
+    sys.exit(run_all())