James Mackay
0c7e405108
The Current App sensor and media_player.app_name showed a package id in
the live run. Root cause was in the manifest-label tier: `_fetch_app_label`
returned early (no poller, ADB down) or logged and returned on a shell
error WITHOUT touching the cache, but `app_label` had already put the
package in `_app_label_requests`; the drain then threw the request away,
so a render between the failure and the next cycle re-enqueued — while
nothing distinguished "the device answered, this package has no label"
from "we never got an answer". Nothing was visibly retried, and the label
tier could stay silent indefinitely.
The cache is now (label, stamp, resolved) and records the FETCH, not the
wishful result:
* a fetch that RAN caches its outcome for APP_LABEL_TTL — the label, or
the empty string for a genuinely label-less package. Definitive: no
re-enqueue, so the t24 F1 per-heartbeat round-trip tax stays closed.
* a fetch that could NOT run stamps nothing, logs at debug (so a
package id in the UI is diagnosable), and the next render retries.
* a queued or in-flight fetch is not enqueued twice, so a burst of
renders still costs one `pm dump` per watchdog cycle.
`cache_app_label()` is the one public writer of the cache (tests and any
future out-of-band reader no longer poke the private tuple), and
`app_label_requests`/`app_label_inflight` expose the queue for tests.
The old 2-tuple seeds in the label tests move to `cache_app_label`.
863 lines
31 KiB
Python
863 lines
31 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)
|
|
|
|
# the dump RAN and carried no label: definitive negative, and the
|
|
# caller's next tier (the raw package) takes over — still None here,
|
|
# never a crash or a fabricated label
|
|
assert manager.app_label("com.unknown.app") is None
|
|
|
|
async def test_label_less_package_is_a_definitive_negative(self):
|
|
"""t24 F1: a label-less package's ANSWER costs one pm dump, not one/cycle.
|
|
|
|
The old code additionally stamped this answer on FAILURE paths; t55
|
|
separates the two (see the retry tests below), but the negative cache
|
|
for a dump that really ran stays — the round-trip tax must not return.
|
|
"""
|
|
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_failed_fetch_is_not_cached_and_retries(self):
|
|
"""t55 root cause: a fetch that could NOT run must not become a miss.
|
|
|
|
The live defect was a metadata fetch failing and the failure being
|
|
written as ("", stamp) — an eternal "no label" for the whole 24 h TTL,
|
|
which left app_name showing the raw package id with no way back. Here
|
|
the failed fetch stamps nothing, the next render re-enqueues, and the
|
|
retry resolves the real label.
|
|
"""
|
|
manager, _t = self._manager_with_dump()
|
|
real_shell = manager._poller.shell
|
|
calls: list[str] = []
|
|
|
|
async def _flaky(cmd):
|
|
calls.append(cmd)
|
|
if "pm dump" in cmd:
|
|
raise RuntimeError("adb shell blew up")
|
|
return await real_shell(cmd)
|
|
|
|
manager._poller.shell = _flaky
|
|
|
|
assert manager.app_label("ar.tvplayer.tv") is None # enqueue
|
|
await manager._watchdog_cycle(5.0)
|
|
import asyncio
|
|
|
|
await asyncio.sleep(0)
|
|
|
|
# the fetch ran and failed → NOTHING is stamped as an answer
|
|
assert manager._app_labels.get("ar.tvplayer.tv") is None
|
|
assert [c for c in calls if "pm dump" in c]
|
|
|
|
# the next render enqueues again (the miss is retried, not silenced)
|
|
assert manager.app_label("ar.tvplayer.tv") is None
|
|
assert manager.app_label_requests == {"ar.tvplayer.tv"}
|
|
|
|
# the retry reaches the device and resolves the label
|
|
manager._poller.shell = real_shell
|
|
await manager._watchdog_cycle(5.0)
|
|
await asyncio.sleep(0)
|
|
assert manager.app_label("ar.tvplayer.tv") == "TiviMate"
|
|
|
|
async def test_shell_failure_is_not_cached_and_retries(self):
|
|
manager, _t = self._manager_with_dump()
|
|
manager._poller.pm_dumps["ar.tvplayer.tv"] = "application-label: 'TiviMate'"
|
|
|
|
async def _boom(_cmd):
|
|
raise RuntimeError("adb shell blew up")
|
|
|
|
manager._poller.shell = _boom
|
|
manager.app_label("ar.tvplayer.tv")
|
|
await manager._watchdog_cycle(5.0)
|
|
import asyncio
|
|
|
|
await asyncio.sleep(0)
|
|
|
|
assert manager._app_labels.get("ar.tvplayer.tv") is None
|
|
assert manager.app_label_requests == set()
|
|
assert manager.app_label("ar.tvplayer.tv") is None
|
|
assert manager.app_label_requests == {"ar.tvplayer.tv"} # re-enqueued
|
|
|
|
async def test_in_flight_fetch_is_not_enqueued_twice(self):
|
|
manager, _t = self._manager_with_dump()
|
|
|
|
assert manager.app_label("ar.tvplayer.tv") is None
|
|
assert manager.app_label("ar.tvplayer.tv") is None # same render loop
|
|
assert manager.app_label_requests == {"ar.tvplayer.tv"}
|
|
assert manager.app_label_inflight == set()
|
|
|
|
async def test_one_fetch_per_cycle_even_under_a_burst(self):
|
|
manager, _t = self._manager_with_dump()
|
|
manager._poller.pm_dumps["com.a"] = "application-label: 'A'"
|
|
manager._poller.pm_dumps["com.b"] = "application-label: 'B'"
|
|
|
|
manager.app_label("com.a")
|
|
manager.app_label("com.b")
|
|
await manager._watchdog_cycle(5.0)
|
|
await manager._watchdog_cycle(5.0)
|
|
import asyncio
|
|
|
|
await asyncio.sleep(0)
|
|
|
|
dumps = [c for c in manager._poller.shell_calls if "pm dump" in c]
|
|
assert len(dumps) == 2 # one per cycle, both eventually resolved
|
|
assert sorted([manager.app_label("com.a"), manager.app_label("com.b")]) == [
|
|
"A",
|
|
"B",
|
|
]
|
|
|
|
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
|
|
|
|
async def fake_ladder_sleep(seconds):
|
|
return None # deterministic: no real ladder sleeps (t26)
|
|
|
|
manager._logcat_ladder_sleep = fake_ladder_sleep
|
|
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
|
|
|
|
def test_streaming_call_carries_no_idle_timeout(self):
|
|
"""t27 F-NEW1: quiet logcat must not trip the transport timeout."""
|
|
import inspect
|
|
|
|
src = inspect.getsource(ShieldConnectionManager._open_logcat_stream)
|
|
assert "transport_timeout_s=None" in src # bulk_read blocks
|
|
assert "default_transport_timeout_s=10.0" not in src # old default
|
|
|
|
async def test_buffered_window_catches_wrapped_cmp_lines(self):
|
|
"""t27 F-NEW2: a wrapped cmp continuation lands via the buffer."""
|
|
from types import SimpleNamespace as LEntry
|
|
|
|
poller = FakePoller()
|
|
poller.awake = True
|
|
manager = ShieldConnectionManager(
|
|
hass=None, entry=None, poller=poller, clock=lambda: 1000.0
|
|
)
|
|
manager.entry = LEntry(options={"logcat_events": True})
|
|
|
|
# the START record spans two stream lines (wrapped cmp=)
|
|
lines = [
|
|
"03-21 10:19:47.000 3104 3456 I"
|
|
" ActivityTaskManager: START u0 {cmp=ar.tvplayer.tv",
|
|
" /.ui.MainActivity flg=0x1}",
|
|
]
|
|
|
|
async def _open():
|
|
for line in lines:
|
|
yield line
|
|
|
|
manager._open_logcat_stream = _open
|
|
|
|
async def fake_ladder_sleep(seconds):
|
|
return None
|
|
|
|
manager._logcat_ladder_sleep = fake_ladder_sleep
|
|
|
|
raised = None
|
|
try:
|
|
await manager._logcat_pass()
|
|
except Exception as err: # the pass ends with stream-ended
|
|
raised = err
|
|
assert raised is not None # stream-ended raises by contract
|
|
assert manager.store.get("foreground") == "ar.tvplayer.tv"
|
|
|
|
def test_public_config_accessor_is_the_source(self):
|
|
"""t27 F-NEW3: the opener consumes AdbClientBase.config publicly."""
|
|
import inspect
|
|
|
|
src = inspect.getsource(ShieldConnectionManager._open_logcat_stream)
|
|
assert "poller.config" in src # public accessor
|
|
assert '"_key_path"' not in src and '"_host"' not in src # no reach-ins
|
|
|
|
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
|