From 2a74694ab1a8ba0482afda9d4fc02cbdaccadcd5 Mon Sep 17 00:00:00 2001 From: Hermes Agent Date: Thu, 2 Jul 2026 08:41:12 -0500 Subject: [PATCH] test: mock-free unit tests for URL/transcode/config/escalation + unified runner (Epic 3, #8) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds real coverage for the genuinely bug-prone logic, all dependency-free so it runs headless in CI (no PyQt/mpv/Emby, no pytest). Follows the Epic 2 pattern: extract pure logic, test it in isolation, delegate the widget/IO code to it. New pure module hyperwall/urls.py: - exceeds_1080p(), needs_transcode(item, auto_transcode=), build_stream_url(). - emby.needs_transcode() and wall._build_url() now delegate here — behavior identical, verified for parity (incl. HYPERWALL_AUTO_TRANSCODE=0). - ⚠️ DIRECT url still carries static=true (Emby 4.9.5.0 500 workaround); a test asserts it so the load-bearing param can never silently regress. reliability.escalation_plan(attempt, max_retries): - extracts the retry→transcode→skip policy from cell._on_error (which was coupled to QTimer). _on_error now calls it, so the tested code is the code that runs. attempt1=retry/direct/2s, attempt2=retry/transcode/4s, attempt3=retry/transcode/8s, attempt4=skip. Tests (48 total, all green): - tests/test_urls.py (14): static=true assertion, HLS master path, 1080p boundaries, missing/None/absent-stream safety, flag binding. - tests/test_config.py (5): save/load round-trip, typed fields, frozen dataclass, template creation, fallback defaults. - tests/test_reliability.py (+5 = 21): full escalation ladder. - tests/run_all.py: single runner aggregating all 4 suites; CI workflow now invokes it. README documents the suite. Refs #8 --- .github/workflows/repo-guards.yml | 18 ++-- README.md | 17 ++++ hyperwall/cell.py | 9 +- hyperwall/emby.py | 17 ++-- hyperwall/reliability.py | 23 +++++ hyperwall/urls.py | 66 ++++++++++++++ hyperwall/wall.py | 16 ++-- tests/run_all.py | 44 ++++++++++ tests/test_config.py | 116 +++++++++++++++++++++++++ tests/test_reliability.py | 40 +++++++++ tests/test_urls.py | 140 ++++++++++++++++++++++++++++++ 11 files changed, 475 insertions(+), 31 deletions(-) create mode 100644 hyperwall/urls.py create mode 100644 tests/run_all.py create mode 100644 tests/test_config.py create mode 100644 tests/test_urls.py diff --git a/.github/workflows/repo-guards.yml b/.github/workflows/repo-guards.yml index b266228..cfb1b25 100644 --- a/.github/workflows/repo-guards.yml +++ b/.github/workflows/repo-guards.yml @@ -1,9 +1,11 @@ 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: @@ -11,14 +13,12 @@ on: 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 diff --git a/README.md b/README.md index e1a33ac..d3066ab 100644 --- a/README.md +++ b/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` / 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 diff --git a/hyperwall/cell.py b/hyperwall/cell.py index 6b2c146..c685877 100644 --- a/hyperwall/cell.py +++ b/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: diff --git a/hyperwall/emby.py b/hyperwall/emby.py index 5f840fe..2fdcbef 100644 --- a/hyperwall/emby.py +++ b/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: diff --git a/hyperwall/reliability.py b/hyperwall/reliability.py index 780307b..f3af98e 100644 --- a/hyperwall/reliability.py +++ b/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} diff --git a/hyperwall/urls.py b/hyperwall/urls.py new file mode 100644 index 0000000..ba4995a --- /dev/null +++ b/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" diff --git a/hyperwall/wall.py b/hyperwall/wall.py index 37880e7..8f1a2a1 100644 --- a/hyperwall/wall.py +++ b/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 diff --git a/tests/run_all.py b/tests/run_all.py new file mode 100644 index 0000000..e4d10be --- /dev/null +++ b/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()) diff --git a/tests/test_config.py b/tests/test_config.py new file mode 100644 index 0000000..ab02515 --- /dev/null +++ b/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()) diff --git a/tests/test_reliability.py b/tests/test_reliability.py index 5a3e139..01227fa 100644 --- a/tests/test_reliability.py +++ b/tests/test_reliability.py @@ -15,6 +15,7 @@ from hyperwall.reliability import ( # noqa: E402 count_recent, + escalation_plan, is_stalled, scale_demuxer_mb, should_park, @@ -101,6 +102,45 @@ def test_zero_cells_is_safe(): assert scale_demuxer_mb(0, per_cell_mb=512, total_budget_mb=3072) == 512 +# ── escalation_plan (retry → transcode → skip) ──────────────────────────────── + +def test_attempt1_retries_direct(): + p = escalation_plan(1, max_retries=3) + assert p["action"] == "retry" + assert p["transcode"] is False # first attempt stays DIRECT + assert p["delay_s"] == 2 # 2**1 + + +def test_attempt2_escalates_to_transcode(): + p = escalation_plan(2, max_retries=3) + assert p["action"] == "retry" + assert p["transcode"] is True # escalate at attempt >= 2 + assert p["delay_s"] == 4 # 2**2 + + +def test_attempt3_still_retries_transcode(): + p = escalation_plan(3, max_retries=3) + assert p["action"] == "retry" + assert p["transcode"] is True + assert p["delay_s"] == 8 # 2**3 + + +def test_attempt_over_max_skips(): + p = escalation_plan(4, max_retries=3) + assert p["action"] == "skip" + assert p["transcode"] is False + assert p["delay_s"] == 0 + + +def test_full_escalation_sequence(): + # The exact retry→transcode→skip ladder a dead stream walks. + seq = [escalation_plan(a, 3) for a in range(1, 5)] + actions = [p["action"] for p in seq] + transcodes = [p["transcode"] for p in seq] + assert actions == ["retry", "retry", "retry", "skip"] + assert transcodes == [False, True, True, False] + + # ── constants env clamping (integration of _int_env) ────────────────────────── def test_constants_defaults_load(): diff --git a/tests/test_urls.py b/tests/test_urls.py new file mode 100644 index 0000000..fc164ad --- /dev/null +++ b/tests/test_urls.py @@ -0,0 +1,140 @@ +""" +Unit tests for hyperwall.urls (Epic 3) — pure Emby URL + transcode logic. + +No PyQt / mpv / Emby. Runnable anywhere Python is. +Run: python tests/test_urls.py + +The critical assertion here is that the DIRECT path carries static=true — a +load-bearing workaround for Emby 4.9.5.0's HTTP 500 on /stream without it. +""" + +from __future__ import annotations + +import os +import sys + +REPO_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +sys.path.insert(0, REPO_ROOT) + +from hyperwall.urls import ( # noqa: E402 + build_stream_url, + exceeds_1080p, + needs_transcode, +) + + +def _item(width=None, height=None, top_level=False): + stream = {"Type": "Video"} + if width is not None: + stream["Width"] = width + if height is not None: + stream["Height"] = height + if top_level: + return {"Id": "x", "MediaStreams": [stream]} + return {"Id": "x", "MediaSources": [{"MediaStreams": [stream]}]} + + +# ── exceeds_1080p ───────────────────────────────────────────────────────────── + +def test_1080p_source_does_not_exceed(): + assert not exceeds_1080p(_item(1920, 1080)) + + +def test_4k_source_exceeds(): + assert exceeds_1080p(_item(3840, 2160)) + + +def test_width_just_over_exceeds(): + assert exceeds_1080p(_item(1921, 1080)) + + +def test_height_just_over_exceeds(): + assert exceeds_1080p(_item(1920, 1081)) + + +def test_top_level_mediastreams_shape(): + assert exceeds_1080p(_item(3840, 2160, top_level=True)) + + +def test_missing_dimensions_safe(): + # No Width/Height keys → treated as 0 → not exceeding, no crash. + assert not exceeds_1080p({"Id": "x", "MediaSources": [{"MediaStreams": [{"Type": "Video"}]}]}) + + +def test_none_dimensions_safe(): + assert not exceeds_1080p(_item(None, None)) + + +def test_no_streams_safe(): + assert not exceeds_1080p({"Id": "x"}) + + +# ── needs_transcode (flag binding) ──────────────────────────────────────────── + +def test_needs_transcode_true_for_4k_when_enabled(): + assert needs_transcode(_item(3840, 2160), auto_transcode=True) + + +def test_needs_transcode_false_when_disabled(): + # Even a 4K source must not transcode when the flag is off. + assert not needs_transcode(_item(3840, 2160), auto_transcode=False) + + +def test_needs_transcode_false_for_1080p(): + assert not needs_transcode(_item(1920, 1080), auto_transcode=True) + + +# ── build_stream_url ────────────────────────────────────────────────────────── + +def test_direct_url_has_static_true(): + # LOAD-BEARING: Emby 4.9.5.0 returns 500 on /stream without static=true. + url = build_stream_url( + base="http://emby:8096", item_id="ID1", api_key="KEY", + session_id="SID", transcode=False, + ) + assert "static=true" in url, url + assert "/Videos/ID1/stream?" in url + assert "api_key=KEY" in url + assert "master.m3u8" not in url + + +def test_transcode_url_is_hls_master(): + url = build_stream_url( + base="http://emby:8096", item_id="ID1", api_key="KEY", + session_id="SID", transcode=True, + ) + assert "/Videos/ID1/master.m3u8?" in url + assert "static=true" not in url # HLS path must NOT use static + assert "VideoCodec=h264" in url + assert "MaxHeight=1080" in url + assert "MaxWidth=1920" in url + assert "PlaySessionId=SID" in url + + +def test_urls_carry_item_and_key(): + for transcode in (True, False): + url = build_stream_url( + base="http://h", item_id="abc123", api_key="tok", + session_id="s", transcode=transcode, + ) + assert "abc123" in url + assert "api_key=tok" in url + + +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())