278 lines
9.6 KiB
Python
278 lines
9.6 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 time
|
|
|
|
import pytest
|
|
from nsh.connection import (
|
|
FIELD_IS_AVAILABLE,
|
|
FIELD_IS_ON,
|
|
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 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)
|