From ec2db2ae5827987eea6f4f70f20c9fe02f9a3e07 Mon Sep 17 00:00:00 2001 From: James Mackay Date: Thu, 8 Oct 2026 08:58:00 +0100 Subject: [PATCH] =?UTF-8?q?test(connection):=20manager=20unit=20tests=20wi?= =?UTF-8?q?th=20a=20fake=20poller=20=E2=80=94=20connect/reconnect=20backof?= =?UTF-8?q?f,=20TTL=20expiry,=20listeners,=20lifecycle?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- tests/conftest.py | 40 +++++ tests/test_connection.py | 351 +++++++++++++++++++++++++++++++++++++++ 2 files changed, 391 insertions(+) create mode 100644 tests/test_connection.py diff --git a/tests/conftest.py b/tests/conftest.py index 55401ec..6dd2a1f 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -3,10 +3,50 @@ Puts the vendored ha_adb_common package on sys.path so helpers can be imported directly (their code is stdlib-only) without installing Home Assistant or executing the integration's __init__ module. + +The integration-package modules that stay Home-Assistant-free at import time +(currently connection.py) are also made importable — via a synthetic package +("nsh") that wraps custom_components/nvidia_shield WITHOUT executing its +__init__.py (that one imports Home Assistant). Only the stdlib-only modules +are loaded into it explicitly; their relative imports resolve normally. """ +import importlib.util import sys +import types from pathlib import Path REPO_ROOT = Path(__file__).resolve().parent.parent sys.path.insert(0, str(REPO_ROOT / "custom_components" / "nvidia_shield")) + +_SHIELD_DIR = REPO_ROOT / "custom_components" / "nvidia_shield" +_PKG = "nsh" + + +def _load_module(name: str, path: Path, search_paths: list[str] | None = None): + """Import a source file under ``name`` without touching real packages.""" + spec = importlib.util.spec_from_file_location( + name, path, submodule_search_locations=search_paths + ) + module = importlib.util.module_from_spec(spec) + sys.modules[name] = module + spec.loader.exec_module(module) + return module + + +# Synthetic integration package — no __init__.py execution (HA-free). +_root = types.ModuleType(_PKG) +_root.__path__ = [str(_SHIELD_DIR)] +sys.modules[_PKG] = _root + +# Real subpackage __init__ (stdlib-only) so `from .ha_adb_common import ...` +# works through the synthetic package. +_load_module( + f"{_PKG}.ha_adb_common", + _SHIELD_DIR / "ha_adb_common" / "__init__.py", + [str(_SHIELD_DIR / "ha_adb_common")], +) + +# Stdlib-only integration modules. +for _mod in ("const", "adb_session", "connection"): + _load_module(f"{_PKG}.{_mod}", _SHIELD_DIR / f"{_mod}.py") diff --git a/tests/test_connection.py b/tests/test_connection.py new file mode 100644 index 0000000..1a04042 --- /dev/null +++ b/tests/test_connection.py @@ -0,0 +1,351 @@ +"""Unit tests for ShieldConnectionManager + StateStore (Phase 1). + +Driven with a fake poller — no real ADB, no Home Assistant. Covers the +connect/reconnect backoff ladder, state expiration, listener notification, +command delegation, and full task lifecycle. +""" + +import asyncio +import time +import types + +import pytest +from ha_adb_common.media_session import MediaSessionInfo +from nsh import connection +from nsh.connection import ( + FIELD_AWAKE, + FIELD_SESSIONS, + ShieldConnectionManager, + StateStore, +) +from nsh.const import ADB_POLL_INTERVAL + + +def test_connection_module_depends_on_no_homeassistant(): + """The manager module must stay importable without HA installed.""" + with open(connection.__file__, encoding="utf-8") as fh: + source = fh.read() + assert "homeassistant" not in source + + +class FakePoller: + """AdbMediaSessionPoller shape — canned answers, no socket.""" + + def __init__(self): + self.connected = False + self.connect_failures = 0 + self.disconnect_calls = 0 + self.awake = None + self.sessions: list[MediaSessionInfo] = [] + self.apps: list[dict] = [] + self.audio_dump = "" + self.head_lines_used: int | None = "unset" + + async def connect(self) -> bool: + if self.connect_failures > 0: + self.connect_failures -= 1 + self.connected = False + return False + self.connected = True + return True + + async def disconnect(self) -> None: + self.disconnect_calls += 1 + self.connected = False + + async def health_check(self) -> bool: + return self.connected + + async def is_awake(self) -> bool | None: + return self.awake if self.connected else None + + async def poll_media_sessions( + self, foreground_package=None, head_lines: int | None = 200 + ) -> MediaSessionInfo | None: + self.head_lines_used = head_lines + return self.sessions[0] if self.sessions else None + + @property + def all_sessions(self) -> list[MediaSessionInfo]: + return self.sessions + + @property + def last_dump(self) -> str | None: + return self.audio_dump + + async def poll_running_apps(self) -> list[dict]: + return list(self.apps) + + async def send_media_command(self, command: str) -> bool: + return self.connected + + async def bring_to_foreground(self, package: str) -> bool: + return self.connected + + +def _manager(poller=None, **kwargs) -> ShieldConnectionManager: + return ShieldConnectionManager(hass=None, entry=None, poller=poller, **kwargs) + + +class TestStateStore: + def test_set_get_roundtrip(self): + store = StateStore(ttl=60, clock=time.monotonic) + store.set("awake", True) + assert store.get("awake") is True + + def test_expired_field_returns_default(self): + """B5: a field older than the TTL reads as expired, not last-known.""" + now = [1000.0] + store = StateStore(ttl=60.0, clock=lambda: now[0]) + store.set(FIELD_AWAKE, True) + store.set(FIELD_SESSIONS, [MediaSessionInfo(package="app.a", playing=True)]) + + assert store.get(FIELD_AWAKE) is True + now[0] += 60.1 + assert store.get(FIELD_AWAKE) is None + assert store.get(FIELD_SESSIONS, []) == [] + + def test_missing_field_returns_default(self): + store = StateStore(ttl=60, clock=time.monotonic) + assert store.get(FIELD_AWAKE) is None + assert store.get(FIELD_SESSIONS, []) == [] + + def test_clear_immediately_expires(self): + store = StateStore(ttl=60, clock=time.monotonic) + store.set(FIELD_AWAKE, False) + store.clear(FIELD_AWAKE) + assert store.get(FIELD_AWAKE) is None + + +class TestConnectAndRefresh: + async def test_cycle_connects_and_populates_store(self): + poller = FakePoller() + poller.awake = True + poller.sessions = [MediaSessionInfo(package="app.a", playing=True)] + poller.apps = [{"package": "app.a"}] + poller.audio_dump = " Audio playback (lastly played comes first)\n uid=1 packages=app.a\n" + + manager = _manager(poller) + notified = [] + unsub = manager.register_listener(lambda: notified.append(True)) + + delay = await manager._watchdog_cycle(5.0) + + assert delay == ADB_POLL_INTERVAL + assert manager.connected is True + assert manager.awake is True + assert manager.best_session("app.a").package == "app.a" + assert manager.running_apps == [{"package": "app.a"}] + assert manager.audio_playback == ["app.a"] + assert notified, "listener must be notified after a refresh" + unsub() + assert manager._listeners == [] + + async def test_query_is_untruncated(self): + """The manager always asks for the full dump (plan blocker B4).""" + poller = FakePoller() + poller.awake = True + manager = _manager(poller) + await manager._watchdog_cycle(5.0) + assert poller.head_lines_used is None + + async def test_standby_expires_playback_fields(self): + """Asleep → sessions/audio/apps are expired, awake stays False.""" + poller = FakePoller() + poller.awake = False + manager = _manager(poller) + manager.store.set(FIELD_SESSIONS, [MediaSessionInfo(package="old.app")]) + manager.store.set(FIELD_AWAKE, True) # stale truth, about to flip + + await manager._watchdog_cycle(5.0) + + assert manager.awake is False + assert manager.active_sessions == [] + assert manager.audio_playback == [] + assert manager.running_apps == [] + + async def test_awake_none_is_a_cycle_failure(self): + """Wake check returning None must back off instead of storing unknown.""" + poller = FakePoller() + poller.connected = True + manager = _manager(poller) + manager.store.set(FIELD_AWAKE, True) + + delay = await manager._watchdog_cycle(5.0) + + assert delay is not None + assert delay < 10.0 # backoff-scaled, not the heartbeat interval + assert manager.awake is None # stale True dropped + + async def test_foreground_preference_and_playing_order(self): + poller = FakePoller() + poller.awake = True + # parse order convention: playing first + poller.sessions = [ + MediaSessionInfo(package="app.playing", playing=True), + MediaSessionInfo(package="app.paused", paused=True), + ] + manager = _manager(poller) + await manager._watchdog_cycle(5.0) + + assert manager.best_session("app.paused").package == "app.paused" + assert manager.best_session().package == "app.playing" + assert manager.best_session("app.unknown").package == "app.playing" + + +class TestReconnectBackoff: + async def test_failed_connect_returns_backoff_scaled_delay(self): + poller = FakePoller() + poller.connect_failures = 1 + manager = _manager(poller) + + delay = await manager._watchdog_cycle(5.0) + + assert not poller.connected + assert 5.0 * 0.75 <= delay <= 5.0 * 1.25 # ±jitter around the backoff + assert manager._failures == 1 + + async def test_backoff_doubles_up_to_the_cap(self): + poller = FakePoller() + poller.connect_failures = 1 + manager = _manager(poller) + + delays = [] + backoff = 5.0 + for _ in range(7): + delays.append(await manager._watchdog_cycle(backoff)) + backoff = min(backoff * 2, 300.0) + + # 5→10→20→40→80→160→300 growth with easing; the last must not exceed + # the cap (with jitter). + assert max(delays) <= 300.0 * 1.25 + assert delays[1] > delays[0] * 1.5 + assert delays[-1] < 300.0 * 1.25 + + async def test_repeated_failures_do_not_spam_listeners(self): + poller = FakePoller() + poller.connect_failures = 99 + manager = _manager(poller) + notified = [] + manager.register_listener(lambda: notified.append(True)) + + for _ in range(3): + await manager._watchdog_cycle(5.0) + + assert notified.count(None) == 0 + assert len(notified) == 1, "down-notified exactly once until recovery" + + async def test_recovery_resets_failures_and_backoff(self): + poller = FakePoller() + poller.connect_failures = 0 + poller.awake = True + manager = _manager(poller) + + await manager._watchdog_cycle(5.0) # fresh connect → ok + assert manager._failures == 0 + assert await manager._watchdog_cycle(5.0) == ADB_POLL_INTERVAL + + async def test_connect_failure_clears_stale_store_state(self): + poller = FakePoller() + poller.connect_failures = 1 + manager = _manager(poller) + manager.store.set(FIELD_AWAKE, True) + sessions = [MediaSessionInfo(package="app.a", playing=True)] + manager.store.set(FIELD_SESSIONS, sessions) + + await manager._watchdog_cycle(5.0) + + assert manager.awake is None + assert manager.active_sessions == [] + + +class TestCommandDelegation: + async def test_send_media_command_requires_poller(self): + manager = _manager(poller=None) + assert manager.available is False + assert await manager.send_media_command("play") is False + + async def test_send_media_command_delegates(self): + poller = FakePoller() + poller.awake = True + manager = _manager(poller) + await manager._watchdog_cycle(5.0) + assert await manager.send_media_command("play") is True + + async def test_bring_to_foreground(self): + poller = FakePoller() + poller.awake = True + manager = _manager(poller) + await manager._watchdog_cycle(5.0) + assert await manager.bring_to_foreground("app.a") is True + + +class TestLifecycle: + async def test_passive_manager_exits_immediately(self): + """No factory → no watchdog loop; entry stays functional without ADB.""" + manager = ShieldConnectionManager(hass=None, entry=types.SimpleNamespace()) + await manager.async_start() + await asyncio.sleep(0.01) + assert manager.available is False + assert manager.connected is False + assert await manager.send_media_command("play") is False + await manager.async_stop() + + async def test_async_stop_disconnects_and_cancels_tasks(self): + poller = FakePoller() + poller.awake = True + manager = _manager(poller) + await manager.async_start() + await asyncio.sleep(0) # let the watchdog task start + + assert manager.available + await manager.async_stop() + + assert poller.disconnect_calls == 1 + assert manager._poller is None + assert manager._tasks == [] + assert manager._stopping is True + + async def test_stop_is_idempotent(self): + manager = _manager(FakePoller()) + await manager.async_stop() + await manager.async_stop() + + async def test_factory_used_when_no_poller_given(self): + poller = FakePoller() + poller.awake = True + manager = ShieldConnectionManager( + hass=None, entry=None, poller_factory=lambda: poller + ) + await manager.async_start() + await asyncio.sleep(0) + await manager.async_stop() + assert poller.disconnect_calls == 1 + + +class TestStoreTtlInManager: + async def test_expired_session_reads_as_none(self): + """Full manager view: TTL expiry means 'state unavailable', not + 'last known playing'.""" + now = [1000.0] + store = StateStore(ttl=60.0, clock=lambda: now[0]) + manager = _manager(store=store) + manager.store.set( + FIELD_SESSIONS, [MediaSessionInfo(package="app.a", playing=True)] + ) + now[0] += 60.1 + assert manager.best_session() is None + + +@pytest.mark.parametrize("owned", [True, False]) +async def test_watchdog_cycle_never_raises(owned): + """A poller can throw anything — the cycle converts to a backoff delay.""" + poller = FakePoller() + + class ExplodingPoller(FakePoller): + async def is_awake(self): + raise RuntimeError("boom") + + manager = _manager(ExplodingPoller() if owned else poller) + delay = await manager._watchdog_cycle(5.0) + assert delay > 0