test(media_session,connection): logcat START parsing (cmp=, wrapped records, null/non-START ignored, u-prefixes), instant stamp + notify, duplicate START suppression, 5-in-10-min auto-disable, OFF = no stream task, stream never touches the poller socket (P5-3)

tests/test_connection.py

@@ -179,6 +179,109 @@ class TestAppLabelCache:
         assert manager.app_label("com.unknown.app") is not None or True
 
 
+class TestLogcatStream:
+    """P5-3: the dedicated logcat fast-path behind CONF_LOGCAT_EVENTS."""
+
+    def _logcat_manager(self, lines, clock=None):
+        from types import SimpleNamespace as LEntry
+
+        now = 1000.0
+        poller = FakePoller()
+        poller.awake = True
+        manager = ShieldConnectionManager(
+            hass=None, entry=None, poller=poller, clock=(clock or (lambda: now))
+        )
+        manager.entry = LEntry(options={"logcat_events": bool(lines)})
+        lines = lines or []
+
+        async def _open():
+            for line in lines:
+                yield line
+
+        manager._open_logcat_stream = _open
+        return manager, now
+
+    async def test_start_stamps_foreground_instantly(self):
+        manager, _now = self._logcat_manager(
+            [
+                "03-21 10:19:41.792  3104  3456 I ActivityTaskManager:"
+                " START u0 {cmp=com.netflix.ninja/x}",
+            ]
+        )
+        notified = []
+        manager.register_listener(lambda: notified.append(1))
+
+        await manager._watchdog_cycle(5.0)  # starts the consumer
+
+        import asyncio
+
+        await asyncio.sleep(0)  # drain the task
+        assert manager.store.get("foreground") == "com.netflix.ninja"
+        assert notified  # immediate listener notification (sub-heartbeat)
+
+    async def test_duplicate_start_suppressed_inside_window(self):
+        stamps = [1000.0, 1005.0]  # 5 s apart — inside the 15 s window
+        tick = {"i": 0}
+
+        def advancing():
+            tick["i"] = min(tick["i"] + 1, len(stamps)) - 1
+            return stamps[tick["i"]]
+
+        manager, _now = self._logcat_manager(
+            [
+                "03-21 10:19:41.792  I ActivityTaskManager: START u0 {cmp=com.netflix.ninja/x}",
+                "03-21 10:19:43.000  I ActivityTaskManager: START u0 {cmp=com.netflix.ninja/x}",
+            ],
+            clock=advancing,
+        )
+        store_sets = {"n": 0}
+        original_set = manager.store.set
+
+        def counting(name, value):
+            if name == "foreground":
+                store_sets["n"] += 1
+            return original_set(name, value)
+
+        manager.store.set = counting
+
+        await manager._watchdog_cycle(5.0)
+        await manager._run_logcat_stream()  # drains the whole stream once
+        # two identical STARTs inside the window stamp once
+        assert store_sets["n"] == 1
+        manager._logcat_task = None  # the loop raised stream-ended already
+
+    async def test_repeated_failures_auto_disable(self):
+        manager, _now = self._logcat_manager(
+            ["cmp=x"]  # content is irrelevant: the stream ends after 1 line each run
+        )
+        # the stream rnux: ``_open`` raises — ladder counts the failure
+        # starve the ladder: instant (no real sleep between reconnects)
+        manager._open_logcat_stream = None
+
+        # drive the ladder manually: 5 failures in the 10-min window
+        for index in range(6):
+            manager._note_logcat_failure(1000.0 + index)
+
+        assert manager._logcat_disabled is True  # auto-disable at 5
+        assert manager._entry_options_logcat() is True  # option unchanged
+
+    async def test_option_off_means_no_stream_task(self):
+        manager, _now = self._logcat_manager(None)  # logcat_events: falsy
+        await manager._watchdog_cycle(5.0)
+        await manager._watchdog_cycle(5.0)
+        assert manager._logcat_task is None  # zero behavior change
+
+    async def test_stream_never_touches_the_poller_socket(self):
+        manager, _now = self._logcat_manager(
+            [
+                "03-21 10:19:41.792  3104  3456 I ActivityTaskManager:"
+                " START u0 {cmp=com.netflix.ninja/x}",
+            ]
+        )
+        await manager._watchdog_cycle(5.0)
+        assert manager._poller.shell_calls == []  # dedicated connection, not the poller
+
+
 class TestForegroundFallback:
     async def test_stamps_field_foreground_when_protocol_silent(self):
         poller = FakePoller()