365 lines
13 KiB
Python
365 lines
13 KiB
Python
"""Unit tests for the protocol transport (androidtvremote2, Phase 2, t8).
|
|
|
|
Fake client mirrors AndroidTVRemote's surface (callback registration,
|
|
``keep_reconnecting`` that owns its own reconnect task, sync ``disconnect``);
|
|
no real gRPC / no Home Assistant. Also covers `volume_tuple` normalization
|
|
against the real library's VolumeInfo TypedDict when the package is present
|
|
(`uvx --with androidtvremote2 pytest`).
|
|
"""
|
|
|
|
import asyncio
|
|
import sys
|
|
import time
|
|
|
|
import pytest
|
|
from nsh.connection import (
|
|
FIELD_AWAKE,
|
|
FIELD_IS_AVAILABLE,
|
|
FIELD_IS_ON,
|
|
FIELD_SESSIONS,
|
|
FIELD_VOLUME_INFO,
|
|
ShieldConnectionManager,
|
|
StateStore,
|
|
volume_tuple,
|
|
)
|
|
|
|
|
|
class FakeProtocolClient:
|
|
"""AndroidTVRemote-shaped fake: callbacks + self-owned reconnect task."""
|
|
|
|
def __init__(self):
|
|
self._is_on_callbacks: list = []
|
|
self._is_available_callbacks: list = []
|
|
self._current_app_callbacks: list = []
|
|
self._volume_callbacks: list = []
|
|
self._invalid_auth_callback = None
|
|
self._reconnect_task = None
|
|
self.disconnect_calls = 0
|
|
|
|
# -- registration (mirrors the real API) --
|
|
def add_is_on_updated_callback(self, callback):
|
|
self._is_on_callbacks.append(callback)
|
|
|
|
def add_is_available_updated_callback(self, callback):
|
|
self._is_available_callbacks.append(callback)
|
|
|
|
def add_current_app_updated_callback(self, callback):
|
|
self._current_app_callbacks.append(callback)
|
|
|
|
def add_volume_info_updated_callback(self, callback):
|
|
self._volume_callbacks.append(callback)
|
|
|
|
# -- pushes (test helper: simulates device-side messages) --
|
|
def push_is_on(self, on):
|
|
for cb in list(self._is_on_callbacks):
|
|
cb(on)
|
|
|
|
def push_is_available(self, available):
|
|
for cb in list(self._is_available_callbacks):
|
|
cb(available)
|
|
|
|
def push_current_app(self, app_package):
|
|
for cb in list(self._current_app_callbacks):
|
|
cb(app_package)
|
|
|
|
def push_volume_info(self, info):
|
|
for cb in list(self._volume_callbacks):
|
|
cb(info)
|
|
|
|
# -- lifecycle --
|
|
def keep_reconnecting(self, invalid_auth_callback=None):
|
|
"""Own task like the real library (0.1 s → 30 s backoff internally)."""
|
|
self._invalid_auth_callback = invalid_auth_callback
|
|
loop = asyncio.get_event_loop()
|
|
self._reconnect_task = loop.create_task(self._reconnect_loop())
|
|
|
|
async def _reconnect_loop(self):
|
|
while True: # the real task sleeps through its internal backoff
|
|
await asyncio.sleep(0.05)
|
|
|
|
def fire_invalid_auth(self):
|
|
self._invalid_auth_callback()
|
|
|
|
def disconnect(self):
|
|
self.disconnect_calls += 1
|
|
if self._reconnect_task is not None:
|
|
self._reconnect_task.cancel()
|
|
|
|
|
|
def _protocol_manager(client=None, **kwargs) -> ShieldConnectionManager:
|
|
return ShieldConnectionManager(hass=None, entry=None, protocol=client, **kwargs)
|
|
|
|
|
|
async def _drain_loop():
|
|
"""Let call_soon_threadsafe callbacks scheduled via the running loop run."""
|
|
await asyncio.sleep(0.01)
|
|
|
|
|
|
class TestPushCallbackWiring:
|
|
async def test_is_on_and_availability_push_into_store(self):
|
|
client = FakeProtocolClient()
|
|
manager = _protocol_manager(client)
|
|
notified = []
|
|
manager.register_listener(lambda: notified.append(True))
|
|
|
|
await manager._start_protocol()
|
|
client.push_is_on(True)
|
|
client.push_is_available(False)
|
|
await _drain_loop()
|
|
|
|
assert manager.is_on is True
|
|
assert manager.is_available is False
|
|
assert manager.store.get(FIELD_IS_ON) is True
|
|
assert notified, "listeners pinged from protocol pushes"
|
|
|
|
await manager.async_stop()
|
|
|
|
async def test_current_app_push_stored_raw_package(self):
|
|
client = FakeProtocolClient()
|
|
manager = _protocol_manager(client)
|
|
await manager._start_protocol()
|
|
client.push_current_app("com.netflix.ninja")
|
|
await _drain_loop()
|
|
assert manager.current_app == "com.netflix.ninja" # name resolution deferred
|
|
await manager.async_stop()
|
|
|
|
async def test_volume_push_normalized_to_level_max_muted(self):
|
|
client = FakeProtocolClient()
|
|
manager = _protocol_manager(client)
|
|
await manager._start_protocol()
|
|
client.push_volume_info({"level": 7, "max": 15, "muted": True})
|
|
await _drain_loop()
|
|
assert manager.volume_info == (7.0, 15.0, True)
|
|
assert manager.store.get(FIELD_VOLUME_INFO)[2] is True
|
|
await manager.async_stop()
|
|
|
|
|
|
class TestInvalidAuthHook:
|
|
async def test_invalid_auth_invokes_entry_callback(self):
|
|
client = FakeProtocolClient()
|
|
fired = []
|
|
manager = _protocol_manager(
|
|
client, invalid_auth_callback=lambda: fired.append(True)
|
|
)
|
|
await manager._start_protocol()
|
|
|
|
client.fire_invalid_auth()
|
|
|
|
assert fired == [True], "reauth hook fired, not a retry storm (t9 delegates)"
|
|
await manager.async_stop()
|
|
|
|
async def test_invalid_auth_without_callback_is_safe(self):
|
|
client = FakeProtocolClient()
|
|
manager = _protocol_manager(client)
|
|
await manager._start_protocol()
|
|
client.fire_invalid_auth() # no callback registered — must not raise
|
|
await manager.async_stop()
|
|
|
|
|
|
class TestLifecycle:
|
|
async def test_stop_disconnects_lib_owned_reconnect_task(self):
|
|
"""The library owns keep_reconnecting's task; closing the client
|
|
tears it down — the manager never wraps that task itself."""
|
|
client = FakeProtocolClient()
|
|
manager = _protocol_manager(client)
|
|
await manager._start_protocol()
|
|
task = client._reconnect_task
|
|
assert task is not None
|
|
|
|
await manager.async_stop()
|
|
await _drain_loop() # let the lib-owned task settle its cancellation
|
|
|
|
assert client.disconnect_calls == 1
|
|
assert task.cancelled()
|
|
assert manager._protocol is None
|
|
|
|
async def test_protocol_with_factory_settings_builds_client(self, monkeypatch):
|
|
client = FakeProtocolClient()
|
|
built = []
|
|
|
|
def fake_builder(*args, **kwargs):
|
|
built.append(kwargs)
|
|
return client
|
|
|
|
monkeypatch.setattr(ShieldConnectionManager, "build_protocol_client", fake_builder)
|
|
manager = ShieldConnectionManager(
|
|
hass=None,
|
|
entry=None,
|
|
protocol_settings={
|
|
"host": "shield.local",
|
|
"client_name": "Home Assistant",
|
|
"certfile": "/tmp/cert.pem",
|
|
"keyfile": "/tmp/key.pem",
|
|
},
|
|
)
|
|
await manager._start_protocol()
|
|
|
|
assert manager._protocol is client
|
|
assert built
|
|
kwargs = built[0]
|
|
assert kwargs["host"] == "shield.local"
|
|
assert kwargs["client_name"] == "Home Assistant"
|
|
assert kwargs["certfile"] == "/tmp/cert.pem"
|
|
assert kwargs["keyfile"] == "/tmp/key.pem"
|
|
await manager.async_stop()
|
|
|
|
async def test_passive_without_protocol_configuration(self):
|
|
manager = _protocol_manager(client=None)
|
|
await manager._start_protocol() # no factory, no settings → no-op
|
|
|
|
assert manager._protocol is None
|
|
assert manager.is_on is None
|
|
assert manager.is_available is None
|
|
assert manager.current_app is None
|
|
assert manager.volume_info is None
|
|
await manager.async_stop()
|
|
|
|
|
|
class TestPushEdgeCases:
|
|
async def test_push_after_loop_close_falls_back_to_direct_notify(self):
|
|
"""Push callbacks that outlive a closed loop must not crash."""
|
|
client = FakeProtocolClient()
|
|
manager = _protocol_manager(client)
|
|
notified = []
|
|
manager.register_listener(lambda: notified.append(True))
|
|
|
|
await manager._start_protocol()
|
|
|
|
loop = asyncio.new_event_loop()
|
|
loop.close()
|
|
manager._protocol_loop = loop # stale/closed loop (mid-shutdown state)
|
|
|
|
client.push_is_on(True)
|
|
await _drain_loop()
|
|
|
|
assert manager.is_on is True # store stamped via direct-notify path
|
|
assert notified # listeners still pinged synchronously
|
|
await manager.async_stop()
|
|
|
|
async def test_store_ttl_applies_to_protocol_fields(self):
|
|
now = [1000.0]
|
|
store = StateStore(ttl=60.0, clock=lambda: now[0])
|
|
client = FakeProtocolClient()
|
|
manager = _protocol_manager(client, store=store)
|
|
await manager._start_protocol()
|
|
client.push_is_on(True)
|
|
await _drain_loop()
|
|
|
|
now[0] += 60.1
|
|
assert manager.is_on is None # expired like every other store field
|
|
assert manager.store.get(FIELD_IS_AVAILABLE) is None
|
|
await manager.async_stop()
|
|
|
|
async def test_pushes_are_stamped_with_transition_time(self):
|
|
client = FakeProtocolClient()
|
|
manager = _protocol_manager(client)
|
|
await manager._start_protocol()
|
|
before = time.monotonic()
|
|
client.push_is_available(True)
|
|
await _drain_loop()
|
|
after = time.monotonic()
|
|
stamped_at = manager.store._values[FIELD_IS_AVAILABLE][0]
|
|
assert before <= stamped_at <= after
|
|
await manager.async_stop()
|
|
|
|
|
|
class FakeAdbForOutage:
|
|
"""ADB poller stub for the outage test (wake poll fails)."""
|
|
|
|
def __init__(self):
|
|
self.connected = True
|
|
self.awake = None
|
|
self.disconnect_calls = 0
|
|
|
|
async def connect(self):
|
|
self.connected = True
|
|
return True
|
|
|
|
async def disconnect(self):
|
|
self.disconnect_calls += 1
|
|
self.connected = False
|
|
|
|
async def is_awake(self):
|
|
return self.awake if self.connected else None
|
|
|
|
|
|
class TestAdbOutageKeepsProtocolTruth:
|
|
async def test_note_failure_clears_only_adb_fields(self):
|
|
"""F2: an ADB outage must not clobber device-reported protocol truth."""
|
|
poller = FakeAdbForOutage()
|
|
poller.connected = True
|
|
poller.awake = None # wake poll fails → cycle failure path
|
|
client = FakeProtocolClient()
|
|
manager = ShieldConnectionManager(
|
|
hass=None, entry=None, poller=poller, protocol=client
|
|
)
|
|
await manager._start_protocol()
|
|
|
|
client.push_is_on(True)
|
|
client.push_is_available(True)
|
|
client.push_current_app("com.netflix.ninja")
|
|
client.push_volume_info({"level": 5, "max": 15, "muted": False})
|
|
manager.store.set(FIELD_AWAKE, True)
|
|
manager.store.set(FIELD_SESSIONS, ["session-data"])
|
|
|
|
await manager._watchdog_cycle(5.0) # ADB outage → _note_failure
|
|
|
|
# ADB-derived fields cleared (regenerated by the watchdog)
|
|
assert manager.store.get(FIELD_AWAKE) is None
|
|
assert manager.store.get(FIELD_SESSIONS, []) == []
|
|
# protocol push truth survives the ADB outage (TTL expiry rules it)
|
|
assert manager.is_on is True
|
|
assert manager.is_available is True
|
|
assert manager.current_app == "com.netflix.ninja"
|
|
assert manager.volume_info == (5.0, 15.0, False)
|
|
await manager.async_stop()
|
|
|
|
|
|
class TestImportGuard:
|
|
def test_missing_library_degrades_to_adb_only(self, monkeypatch, caplog):
|
|
"""F1: a protocol-enabled entry without androidtvremote2 keeps
|
|
running ADB-only — the guarded factory logs a warning and builds
|
|
nothing instead of raising at setup."""
|
|
import logging
|
|
|
|
monkeypatch.setitem(sys.modules, "androidtvremote2", None)
|
|
with caplog.at_level(logging.WARNING):
|
|
client = ShieldConnectionManager.build_protocol_client(
|
|
"shield.local", "Home Assistant", "/tmp/c.pem", "/tmp/k.pem"
|
|
)
|
|
assert client is None
|
|
assert any("degrades to ADB-only" in r.message for r in caplog.records)
|
|
|
|
async def test_settings_entry_without_library_stays_passive(self, monkeypatch):
|
|
manager = ShieldConnectionManager(
|
|
hass=None,
|
|
entry=None,
|
|
protocol_settings={
|
|
"host": "shield.local",
|
|
"client_name": "HA",
|
|
"certfile": "/tmp/c.pem",
|
|
"keyfile": "/tmp/k.pem",
|
|
},
|
|
)
|
|
monkeypatch.setitem(sys.modules, "androidtvremote2", None)
|
|
await manager._start_protocol()
|
|
assert manager._protocol is None # passive, no crash
|
|
await manager.async_stop()
|
|
|
|
|
|
class TestVolumeTuple:
|
|
def test_normalizes_typed_dict(self):
|
|
assert volume_tuple({"level": 7, "max": 15, "muted": True}) == (7.0, 15.0, True)
|
|
|
|
def test_defaults_on_partial_dict(self):
|
|
assert volume_tuple({"muted": False}) == (0.0, 1.0, False)
|
|
|
|
def test_none_and_garbage_yield_none(self):
|
|
assert volume_tuple(None) is None
|
|
assert volume_tuple("garbage") is None
|
|
|
|
def test_accepts_real_library_volume_info(self):
|
|
androidtvremote2 = pytest.importorskip("androidtvremote2")
|
|
info = androidtvremote2.VolumeInfo(level=3, max=15, muted=True)
|
|
assert volume_tuple(info) == (3.0, 15.0, True)
|