fix(connection): ADB outage clears only ADB-derived store fields — protocol push truth survives by TTL (F2)

tests/test_protocol_transport.py

@@ -8,12 +8,15 @@ against the real library's VolumeInfo TypedDict when the package is present
 """
 
 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,
@@ -260,6 +263,90 @@ class TestPushEdgeCases:
         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)