"""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)) 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) 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) 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) ) 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) 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) 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", }, ) 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) 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)) 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) 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) 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)