374 lines
13 KiB
Python
374 lines
13 KiB
Python
"""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 = []
|
|
backoffs = []
|
|
backoff = 5.0
|
|
for _ in range(7):
|
|
backoffs.append(backoff)
|
|
delays.append(await manager._watchdog_cycle(backoff))
|
|
backoff = min(backoff * 2, 300.0)
|
|
|
|
# F5 soundness: the previous ±25 % jitter could flake the claim
|
|
# delays[1] > delays[0] * 1.5: BACKOFF=10 jittered to 9.17 vs
|
|
# BACKOFF=5 jittered to 6.25 → 6.25*1.5 = 9.375). Assert only what
|
|
# jitter guarantees: each delay sits inside ITS OWN backoff's band
|
|
# (non-overlapping consecutive bands ⇒ delays[1] >= delays[0],
|
|
# since min next = 7.5 >= max first = 6.25), and never above the cap.
|
|
for backoff, delay in zip(backoffs, delays, strict=True):
|
|
assert 0.75 * backoff <= delay <= 1.25 * backoff
|
|
assert delays[1] >= delays[0]
|
|
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_failed_wake_poll_forces_reconnect(self):
|
|
"""health_check drives the reconnect decision: a cycle where is_awake
|
|
fails closes the socket so the next cycle reconnects from scratch."""
|
|
poller = FakePoller()
|
|
poller.connected = True
|
|
poller.awake = None # wake poll fails while the flag says "connected"
|
|
manager = _manager(poller)
|
|
|
|
delay = await manager._watchdog_cycle(5.0)
|
|
|
|
assert 5.0 * 0.75 <= delay <= 5.0 * 1.25 # F5-style symmetric bound
|
|
assert poller.disconnect_calls == 1
|
|
assert poller.connected is False
|
|
assert manager._failures == 1
|
|
|
|
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
|