From 9a1e1300ddd027e2e93bc89e6f81d3327454fdb3 Mon Sep 17 00:00:00 2001 From: James Mackay Date: Thu, 8 Oct 2026 09:27:38 +0100 Subject: [PATCH] =?UTF-8?q?test(connection):=20protocol-transport=20unit?= =?UTF-8?q?=20tests=20=E2=80=94=20push=20wiring,=20InvalidAuth=20reauth=20?= =?UTF-8?q?hook,=20lib-owned=20keep=5Freconnecting=20lifecycle,=20passive?= =?UTF-8?q?=20paths,=20volume=20normalization?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- tests/test_protocol_transport.py | 277 +++++++++++++++++++++++++++++++ 1 file changed, 277 insertions(+) create mode 100644 tests/test_protocol_transport.py diff --git a/tests/test_protocol_transport.py b/tests/test_protocol_transport.py new file mode 100644 index 0000000..58843f7 --- /dev/null +++ b/tests/test_protocol_transport.py @@ -0,0 +1,277 @@ +"""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)