test(connection): manager unit tests with a fake poller — connect/reconnect backoff, TTL expiry, listeners, lifecycle

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")

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