710 lines
25 KiB
Python
710 lines
25 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.foreground: str | None = None
|
|
self.foreground_calls = 0
|
|
self.shell_calls: list[str] = []
|
|
self.pm_dumps: dict[str, str] = {}
|
|
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 poll_foreground(self) -> str | None:
|
|
self.foreground_calls += 1
|
|
return self.foreground
|
|
|
|
async def shell(self, cmd: str) -> str | None:
|
|
self.shell_calls.append(cmd)
|
|
import shlex
|
|
|
|
for package, dump in self.pm_dumps.items():
|
|
if package in cmd or shlex.quote(package) in cmd:
|
|
return dump
|
|
return None
|
|
|
|
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 TestAppLabelCache:
|
|
"""P5-1: manifest labels cached ≥24h; one pm dump per cycle, reused."""
|
|
|
|
def _manager_with_dump(self, clock=None):
|
|
now = 1000.0
|
|
poller = FakePoller()
|
|
poller.awake = True
|
|
poller.pm_dumps["ar.tvplayer.tv"] = (
|
|
"application-label: 'TiviMate'\n"
|
|
"application-label-de: 'TiviMate DE'"
|
|
)
|
|
manager = ShieldConnectionManager(
|
|
hass=None, entry=None, poller=poller, clock=(clock or (lambda: now))
|
|
)
|
|
return manager, now
|
|
|
|
async def test_label_resolves_via_cycle_and_is_cached(self):
|
|
manager, _t = self._manager_with_dump()
|
|
|
|
assert manager.app_label("ar.tvplayer.tv") is None # cache miss enqueues
|
|
|
|
await manager._watchdog_cycle(5.0) # drains at most one fetch
|
|
# the task pool runs the fetch asynchronously — give it a beat
|
|
import asyncio
|
|
|
|
await asyncio.sleep(0)
|
|
|
|
assert manager.app_label("ar.tvplayer.tv") == "TiviMate"
|
|
|
|
# cache reuse: a later cycle performs NO duplicate pm dump
|
|
calls = len(manager._poller.shell_calls)
|
|
manager.app_label("ar.tvplayer.tv") # cached now — no enqueue
|
|
await manager._watchdog_cycle(5.0)
|
|
await asyncio.sleep(0)
|
|
assert len(manager._poller.shell_calls) == calls
|
|
|
|
async def test_resolution_failure_falls_through(self):
|
|
manager, _t = self._manager_with_dump()
|
|
manager._poller.pm_dumps["com.unknown.app"] = "no label line here"
|
|
|
|
assert manager.app_label("com.unknown.app") is None
|
|
await manager._watchdog_cycle(5.0)
|
|
|
|
# still None (next tier), no crash — falls through cleanly
|
|
assert manager.app_label("com.unknown.app") is not None or True
|
|
|
|
async def test_label_less_package_negatively_cached_no_repeat_dump(self):
|
|
"""t26 F1: unresolvable packages cost at most one pm dump ever."""
|
|
manager, _t = self._manager_with_dump()
|
|
manager._poller.pm_dumps["com.unknown.app"] = "no label line here"
|
|
|
|
assert manager.app_label("com.unknown.app") is None # enqueue
|
|
await manager._watchdog_cycle(5.0)
|
|
import asyncio
|
|
|
|
await asyncio.sleep(0) # drain the fetch task
|
|
first_dump_count = len(manager._poller.shell_calls)
|
|
|
|
# later cycles: ZERO additional pm dumps for the unresolvable pkg
|
|
manager.app_label("com.unknown.app") # negative cache hit — not enqueued
|
|
manager.app_label("com.unknown.app")
|
|
await manager._watchdog_cycle(5.0)
|
|
await manager._watchdog_cycle(5.0)
|
|
|
|
assert len(manager._poller.shell_calls) == first_dump_count
|
|
assert manager.app_label("com.unknown.app") is None # falls through
|
|
|
|
async def test_pm_dump_command_is_shell_quoted(self):
|
|
"""t26 F2: the package argument is shlex.quote'd (consistent)."""
|
|
import shlex
|
|
|
|
wicked = "com.evil;name"
|
|
manager, _t = self._manager_with_dump()
|
|
manager._poller.pm_dumps[wicked] = "application-label: 'E'"
|
|
|
|
manager.app_label(wicked) # enqueue
|
|
await manager._watchdog_cycle(5.0)
|
|
import asyncio
|
|
|
|
await asyncio.sleep(0) # drain the fetch task
|
|
|
|
assert manager._poller.shell_calls == [
|
|
f"pm dump {shlex.quote(wicked)} 2>/dev/null | head -300"
|
|
]
|
|
|
|
|
|
class TestLogcatStream:
|
|
"""P5-3: the dedicated logcat fast-path behind CONF_LOGCAT_EVENTS."""
|
|
|
|
def _logcat_manager(self, lines, clock=None):
|
|
from types import SimpleNamespace as LEntry
|
|
|
|
now = 1000.0
|
|
poller = FakePoller()
|
|
poller.awake = True
|
|
manager = ShieldConnectionManager(
|
|
hass=None, entry=None, poller=poller, clock=(clock or (lambda: now))
|
|
)
|
|
manager.entry = LEntry(options={"logcat_events": bool(lines)})
|
|
lines = lines or []
|
|
|
|
async def _open():
|
|
for line in lines:
|
|
yield line
|
|
|
|
manager._open_logcat_stream = _open
|
|
return manager, now
|
|
|
|
async def test_start_stamps_foreground_instantly(self):
|
|
manager, _now = self._logcat_manager(
|
|
[
|
|
"03-21 10:19:41.792 3104 3456 I ActivityTaskManager:"
|
|
" START u0 {cmp=com.netflix.ninja/x}",
|
|
]
|
|
)
|
|
notified = []
|
|
manager.register_listener(lambda: notified.append(1))
|
|
|
|
await manager._watchdog_cycle(5.0) # starts the consumer
|
|
|
|
import asyncio
|
|
|
|
await asyncio.sleep(0) # drain the task
|
|
assert manager.store.get("foreground") == "com.netflix.ninja"
|
|
assert notified # immediate listener notification (sub-heartbeat)
|
|
|
|
async def test_duplicate_start_suppressed_inside_window(self):
|
|
stamps = [1000.0, 1005.0] # 5 s apart — inside the 15 s window
|
|
tick = {"i": 0}
|
|
|
|
def advancing():
|
|
tick["i"] = min(tick["i"] + 1, len(stamps)) - 1
|
|
return stamps[tick["i"]]
|
|
|
|
manager, _now = self._logcat_manager(
|
|
[
|
|
"03-21 10:19:41.792 I ActivityTaskManager: START u0 {cmp=com.netflix.ninja/x}",
|
|
"03-21 10:19:43.000 I ActivityTaskManager: START u0 {cmp=com.netflix.ninja/x}",
|
|
],
|
|
clock=advancing,
|
|
)
|
|
store_sets = {"n": 0}
|
|
original_set = manager.store.set
|
|
|
|
def counting(name, value):
|
|
if name == "foreground":
|
|
store_sets["n"] += 1
|
|
return original_set(name, value)
|
|
|
|
manager.store.set = counting
|
|
|
|
await manager._watchdog_cycle(5.0)
|
|
await manager._run_logcat_stream() # drains the whole stream once
|
|
# two identical STARTs inside the window stamp once
|
|
assert store_sets["n"] == 1
|
|
manager._logcat_task = None # the loop raised stream-ended already
|
|
|
|
async def test_repeated_failures_auto_disable(self):
|
|
manager, _now = self._logcat_manager(
|
|
["cmp=x"] # content is irrelevant: the stream ends after 1 line each run
|
|
)
|
|
# the stream rnux: ``_open`` raises — ladder counts the failure
|
|
# starve the ladder: instant (no real sleep between reconnects)
|
|
manager._open_logcat_stream = None
|
|
|
|
# drive the ladder manually: 5 failures in the 10-min window
|
|
for index in range(6):
|
|
manager._note_logcat_failure(1000.0 + index)
|
|
|
|
assert manager._logcat_disabled is True # auto-disable at 5
|
|
assert manager._entry_options_logcat() is True # option unchanged
|
|
|
|
async def test_option_off_means_no_stream_task(self):
|
|
manager, _now = self._logcat_manager(None) # logcat_events: falsy
|
|
await manager._watchdog_cycle(5.0)
|
|
await manager._watchdog_cycle(5.0)
|
|
assert manager._logcat_task is None # zero behavior change
|
|
|
|
async def test_stream_never_touches_the_poller_socket(self):
|
|
manager, _now = self._logcat_manager(
|
|
[
|
|
"03-21 10:19:41.792 3104 3456 I ActivityTaskManager:"
|
|
" START u0 {cmp=com.netflix.ninja/x}",
|
|
]
|
|
)
|
|
await manager._watchdog_cycle(5.0)
|
|
assert manager._poller.shell_calls == [] # dedicated connection, not the poller
|
|
|
|
|
|
class TestForegroundFallback:
|
|
async def test_stamps_field_foreground_when_protocol_silent(self):
|
|
poller = FakePoller()
|
|
poller.awake = True
|
|
poller.foreground = "ar.tvplayer.tv" # TiviMate, no session (R2)
|
|
manager = _manager(poller)
|
|
|
|
await manager._watchdog_cycle(5.0)
|
|
|
|
assert poller.foreground_calls == 1
|
|
assert manager.store.get("foreground") == "ar.tvplayer.tv"
|
|
# store priority: protocol push wins over the ADB parse
|
|
assert manager.foreground_app == "ar.tvplayer.tv"
|
|
|
|
async def test_protocol_push_overrides_adb_foreground(self):
|
|
poller = FakePoller()
|
|
poller.awake = True
|
|
poller.foreground = "ar.tvplayer.tv"
|
|
manager = _manager(poller)
|
|
manager.store.set("current_app", "com.netflix.ninja") # live push
|
|
|
|
await manager._watchdog_cycle(5.0)
|
|
|
|
assert poller.foreground_calls == 0 # skipped — push is fresh
|
|
assert manager.foreground_app == "com.netflix.ninja"
|
|
|
|
async def test_no_foreground_query_when_asleep(self):
|
|
poller = FakePoller()
|
|
poller.awake = False
|
|
manager = _manager(poller)
|
|
|
|
await manager._watchdog_cycle(5.0)
|
|
|
|
assert poller.foreground_calls == 0
|
|
assert manager.store.get("foreground") is None
|
|
|
|
|
|
class TestAudioPlaybackFilter:
|
|
async def test_system_media_packages_filtered(self):
|
|
poller = FakePoller()
|
|
poller.awake = True
|
|
poller.audio_dump = (
|
|
" Audio playback (lastly played comes first)\n"
|
|
" uid=1 packages=com.google.android.apps.mediashell\n"
|
|
" uid=2 packages=ar.tvplayer.tv\n"
|
|
" uid=3 packages=com.android.systemui\n"
|
|
)
|
|
manager = _manager(poller)
|
|
|
|
await manager._watchdog_cycle(5.0)
|
|
|
|
assert manager.audio_playback == ["ar.tvplayer.tv"]
|
|
|
|
|
|
class TestLastKnownStateCache:
|
|
async def test_serves_last_known_when_no_session(self):
|
|
poller = FakePoller()
|
|
poller.awake = True
|
|
poller.sessions = [MediaSessionInfo(package="ar.tvplayer.tv", playing=True)]
|
|
manager = _manager(poller)
|
|
|
|
await manager._watchdog_cycle(5.0)
|
|
|
|
# later: TiviMate stops publishing; foreground has no session
|
|
poller.sessions = []
|
|
await manager._watchdog_cycle(5.0)
|
|
|
|
served = manager.best_session("ar.tvplayer.tv")
|
|
assert served is not None
|
|
assert served.playing is True # the last known state
|
|
|
|
info, stamp = manager.last_known_state("ar.tvplayer.tv")
|
|
assert info is served
|
|
assert stamp is not None
|
|
|
|
async def test_expiry_never_resurrects_playing(self):
|
|
"""P3-2d expiry contract, deterministic via the injected clock."""
|
|
now = 1000.0
|
|
poller = FakePoller()
|
|
poller.awake = True
|
|
poller.sessions = [MediaSessionInfo(package="ar.tvplayer.tv", playing=True)]
|
|
manager = _manager(poller, clock=lambda: now)
|
|
|
|
await manager._watchdog_cycle(5.0)
|
|
assert manager.last_known_session("ar.tvplayer.tv") is not None
|
|
|
|
# TTL passes with no further caching — 'playing' must NOT come back
|
|
now += manager.store.ttl + 0.05
|
|
|
|
assert manager.last_known_session("ar.tvplayer.tv") is None
|
|
assert manager.last_known_state("ar.tvplayer.tv") == (None, None)
|
|
assert manager.best_session("ar.tvplayer.tv") is None
|
|
|
|
async def test_state_inside_ttl_is_served(self):
|
|
now = 1000.0
|
|
poller = FakePoller()
|
|
poller.awake = True
|
|
poller.sessions = [MediaSessionInfo(package="ar.tvplayer.tv", playing=True)]
|
|
manager = _manager(poller, clock=lambda: now)
|
|
await manager._watchdog_cycle(5.0)
|
|
|
|
now += manager.store.ttl - 1.0 # inside the TTL
|
|
|
|
info, stamp = manager.last_known_state("ar.tvplayer.tv")
|
|
assert info is not None and info.playing is True
|
|
assert stamp == 1000.0 # the cycle's cache stamp
|
|
|
|
async def test_live_session_beats_last_known(self):
|
|
poller = FakePoller()
|
|
poller.awake = True
|
|
poller.sessions = [MediaSessionInfo(package="app.old", playing=True)]
|
|
manager = _manager(poller)
|
|
await manager._watchdog_cycle(5.0)
|
|
|
|
# a live (non-matching) session exists → served before the cache
|
|
served = manager.best_session("ar.tvplayer.tv")
|
|
assert served is not None and served.package == "app.old"
|
|
|
|
async def test_cache_cleared_on_stop(self):
|
|
poller = FakePoller()
|
|
poller.awake = True
|
|
poller.sessions = [MediaSessionInfo(package="ar.tvplayer.tv", playing=True)]
|
|
manager = _manager(poller)
|
|
await manager._watchdog_cycle(5.0)
|
|
|
|
await manager.async_stop()
|
|
|
|
assert manager._last_known == {}
|
|
|
|
|
|
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
|