test(ha_adb_common): fixture-driven foreground-parser tests — TiviMate capture, R4 launcher, wrap join, topResumed fallback ordering, duplicates tolerated (P3-2a)

custom_components/nvidia_shield/connection.py

@@ -167,10 +167,13 @@ class ShieldConnectionManager:
         protocol: Any = None,
         protocol_settings: dict | None = None,
         invalid_auth_callback: Callable[[], None] | None = None,
+        clock: Callable[[], float] = time.monotonic,
     ):
         self.hass = hass
         self.entry = entry
-        self.store = store if store is not None else StateStore(ttl=STATE_STORE_TTL)
+        self.store = (
+            store if store is not None else StateStore(ttl=STATE_STORE_TTL, clock=clock)
+        )
         self._poller_factory = poller_factory
         self._poller = poller
         # Remote-protocol transport (androidtvremote2): entry passes either a
@@ -182,6 +185,13 @@ class ShieldConnectionManager:
         self._invalid_auth_callback = invalid_auth_callback
         self._protocol_loop: asyncio.AbstractEventLoop | None = None
         self._listeners: list[Callable[[], None]] = []
+        # P3-2d last-known-state cache: package → (session info, stamp).
+        # Serves the foreground app's last known playback state when it
+        # publishes no MediaSession (TiviMate, R2). TTL-capped: 'playing'
+        # is NEVER resurrected past STATE_STORE_TTL.
+        self._last_known: dict[str, tuple[MediaSessionInfo, float]] = {}
+        # Injectable clock keeps the TTL expiry tests deterministic.
+        self._clock = clock
         self._tasks: list[asyncio.Task] = []
         self._stopping = False
         self._failures = 0
@@ -397,15 +407,58 @@ class ShieldConnectionManager:
     def best_session(
         self, foreground_package: str | None = None
     ) -> MediaSessionInfo | None:
-        """Best media session: foreground app first, then playing > paused."""
+        """Best media session: foreground app first, then playing > paused.
+
+        P3-2d: when there is no live session at all, the foreground app's
+        last known state is served from the TTL-capped cache (TiviMate
+        publishes no MediaSession — RESEARCH_SHIELD R2).
+        """
         sessions = self.active_sessions
-        if not sessions:
-            return None
         if foreground_package:
             for session in sessions:
                 if session.package == foreground_package:
                     return session
-        return sessions[0]
+        if sessions:
+            return sessions[0]
+        if foreground_package:
+            return self.last_known_session(foreground_package)
+        return None
+
+    def _cache_last_known_sessions(self) -> None:
+        """Stamp every live session's state as the package's last known."""
+        now = self._clock()
+        for session in self.active_sessions:
+            if session.package:
+                self._last_known[session.package] = (session, now)
+
+    def last_known_session(self, package: str) -> MediaSessionInfo | None:
+        """Last known session state for a package, TTL-capped.
+
+        Returns None once STATE_STORE_TTL passes — 'playing' must never be
+        resurrected from a stale cache (P3-2d expiry contract).
+        """
+        cached = self._last_known.get(package)
+        if cached is None:
+            return None
+        info, stamp = cached
+        if self._clock() - stamp > STATE_STORE_TTL:
+            self._last_known.pop(package, None)
+            return None
+        return info
+
+    def last_known_state(
+        self, package: str
+    ) -> tuple[MediaSessionInfo | None, float | None]:
+        """(last known session, freshness stamp) for a package — the stamp
+        lets callers age the served state in their UI."""
+        cached = self._last_known.get(package)
+        if cached is None:
+            return None, None
+        info, stamp = cached
+        if self._clock() - stamp > STATE_STORE_TTL:
+            self._last_known.pop(package, None)
+            return None, None
+        return info, stamp
 
     # ── Listener API (entities are thin callback-views) ──
 
@@ -506,6 +559,7 @@ class ShieldConnectionManager:
         for task in tasks:
             task.cancel()
         self._listeners.clear()
+        self._last_known.clear()
         if self._poller is not None:
             with contextlib.suppress(Exception):
                 await self._poller.disconnect()
@@ -615,6 +669,7 @@ class ShieldConnectionManager:
             # Full dump, no truncation (B4) — low sessions must not be lost.
             await poller.poll_media_sessions(head_lines=None)
             self.store.set(FIELD_SESSIONS, list(poller.all_sessions))
+            self._cache_last_known_sessions()
             # Zero-cost hint from the same dump; system processes filtered
             # (mediashell/systemui/tv settings/diagtools report "audio").
             playback = [
@@ -631,7 +686,6 @@ class ShieldConnectionManager:
                 foreground = await poller.poll_foreground()
                 if foreground:
                     self.store.set(FIELD_FOREGROUND, foreground)
-                    self._cache_last_known(foreground)
         else:
             # Standby: playback data is not current — expire it.
             self.store.set(FIELD_SESSIONS, [])

tests/test_connection.py

@@ -123,6 +123,136 @@ class TestStateStore:
         assert store.get(FIELD_AWAKE) is None
 
 
+class TestForegroundFallback:
+    async def test_stamps_field_foreground_when_protocol_silent(self):
+        poller = FakePoller()
+        poller.awake = True
+        poller.foreground = "ar.tvplayer.tv"  # TiviMate, no session (R2)
+        manager = _manager(poller)
+
+        await manager._watchdog_cycle(5.0)
+
+        assert poller.foreground_calls == 1
+        assert manager.store.get("foreground") == "ar.tvplayer.tv"
+        # store priority: protocol push wins over the ADB parse
+        assert manager.foreground_app == "ar.tvplayer.tv"
+
+    async def test_protocol_push_overrides_adb_foreground(self):
+        poller = FakePoller()
+        poller.awake = True
+        poller.foreground = "ar.tvplayer.tv"
+        manager = _manager(poller)
+        manager.store.set("current_app", "com.netflix.ninja")  # live push
+
+        await manager._watchdog_cycle(5.0)
+
+        assert poller.foreground_calls == 0  # skipped — push is fresh
+        assert manager.foreground_app == "com.netflix.ninja"
+
+    async def test_no_foreground_query_when_asleep(self):
+        poller = FakePoller()
+        poller.awake = False
+        manager = _manager(poller)
+
+        await manager._watchdog_cycle(5.0)
+
+        assert poller.foreground_calls == 0
+        assert manager.store.get("foreground") is None
+
+
+class TestAudioPlaybackFilter:
+    async def test_system_media_packages_filtered(self):
+        poller = FakePoller()
+        poller.awake = True
+        poller.audio_dump = (
+            " Audio playback (lastly played comes first)\n"
+            "  uid=1 packages=com.google.android.apps.mediashell\n"
+            "  uid=2 packages=ar.tvplayer.tv\n"
+            "  uid=3 packages=com.android.systemui\n"
+        )
+        manager = _manager(poller)
+
+        await manager._watchdog_cycle(5.0)
+
+        assert manager.audio_playback == ["ar.tvplayer.tv"]
+
+
+class TestLastKnownStateCache:
+    async def test_serves_last_known_when_no_session(self):
+        poller = FakePoller()
+        poller.awake = True
+        poller.sessions = [MediaSessionInfo(package="ar.tvplayer.tv", playing=True)]
+        manager = _manager(poller)
+
+        await manager._watchdog_cycle(5.0)
+
+        # later: TiviMate stops publishing; foreground has no session
+        poller.sessions = []
+        await manager._watchdog_cycle(5.0)
+
+        served = manager.best_session("ar.tvplayer.tv")
+        assert served is not None
+        assert served.playing is True  # the last known state
+
+        info, stamp = manager.last_known_state("ar.tvplayer.tv")
+        assert info is served
+        assert stamp is not None
+
+    async def test_expiry_never_resurrects_playing(self):
+        """P3-2d expiry contract, deterministic via the injected clock."""
+        now = 1000.0
+        poller = FakePoller()
+        poller.awake = True
+        poller.sessions = [MediaSessionInfo(package="ar.tvplayer.tv", playing=True)]
+        manager = _manager(poller, clock=lambda: now)
+
+        await manager._watchdog_cycle(5.0)
+        assert manager.last_known_session("ar.tvplayer.tv") is not None
+
+        # TTL passes with no further caching — 'playing' must NOT come back
+        now += manager.store.ttl + 0.05
+
+        assert manager.last_known_session("ar.tvplayer.tv") is None
+        assert manager.last_known_state("ar.tvplayer.tv") == (None, None)
+        assert manager.best_session("ar.tvplayer.tv") is None
+
+    async def test_state_inside_ttl_is_served(self):
+        now = 1000.0
+        poller = FakePoller()
+        poller.awake = True
+        poller.sessions = [MediaSessionInfo(package="ar.tvplayer.tv", playing=True)]
+        manager = _manager(poller, clock=lambda: now)
+        await manager._watchdog_cycle(5.0)
+
+        now += manager.store.ttl - 1.0  # inside the TTL
+
+        info, stamp = manager.last_known_state("ar.tvplayer.tv")
+        assert info is not None and info.playing is True
+        assert stamp == 1000.0  # the cycle's cache stamp
+
+    async def test_live_session_beats_last_known(self):
+        poller = FakePoller()
+        poller.awake = True
+        poller.sessions = [MediaSessionInfo(package="app.old", playing=True)]
+        manager = _manager(poller)
+        await manager._watchdog_cycle(5.0)
+
+        # a live (non-matching) session exists → served before the cache
+        served = manager.best_session("ar.tvplayer.tv")
+        assert served is not None and served.package == "app.old"
+
+    async def test_cache_cleared_on_stop(self):
+        poller = FakePoller()
+        poller.awake = True
+        poller.sessions = [MediaSessionInfo(package="ar.tvplayer.tv", playing=True)]
+        manager = _manager(poller)
+        await manager._watchdog_cycle(5.0)
+
+        await manager.async_stop()
+
+        assert manager._last_known == {}
+
+
 class TestConnectAndRefresh:
     async def test_cycle_connects_and_populates_store(self):
         poller = FakePoller()

tests/test_foreground_parser.py

@@ -0,0 +1,76 @@
+"""P3-2a tests: the mResumedActivity foreground parser (RESEARCH_SHIELD R4).
+
+Fixture-driven from the on-device captures; the launcher case comes from the
+R4-documented custom-launcher capture. Window-focus parsing is never used
+(R4 finding 2: mCurrentFocus=null trap).
+"""
+
+from pathlib import Path
+
+from nsh.ha_adb_common import parse_foreground
+
+FIXTURES = Path(__file__).parent / "fixtures"
+
+# R4: the on-device capture (grep -iE 'resumed') — the wrapped ActivityRecord
+# plus tolerated Resumed/ResumedActivity duplicates.
+_TIVIMATE_DUMP = (FIXTURES / "activity_activities_tivimate.txt").read_text()
+
+
+def test_tivimate_fixture_primary_pattern():
+    assert parse_foreground(_TIVIMATE_DUMP) == "ar.tvplayer.tv"
+
+
+def test_launcher_foreground_from_r4_capture():
+    # R4 finding 4: the user's custom launcher is a legitimate foreground app.
+    dump = (
+        "mResumedActivity: ActivityRecord{abcd1234 u0 com.spocky.projengmenu"
+        "/.ui.home.MainActivity\n"
+        "              state=RESUMED stopped=false delayedResume=false finishing=false\n"
+    )
+    assert parse_foreground(dump) == "com.spocky.projengmenu"
+
+
+def test_multiline_wrap_joined_before_matching():
+    # The wrap splits right after <pkg>/<activity>; the package must survive.
+    dump = (
+        "    mResumedActivity: ActivityRecord{4f0d030 u0 ar.tvplayer.tv/.ui.MainActivity\n"
+        "    t8152}\n"
+    )
+    assert parse_foreground(dump) == "ar.tvplayer.tv"
+
+
+def test_topresumed_only_fallback():
+    # Newer firmware: no mResumedActivity, only topResumedActivity (checked after).
+    dump = (
+        "  topResumedActivity: ActivityRecord{aa11bb22 u0 com.other.app/.Main t1}\n"
+    )
+    assert parse_foreground(dump) == "com.other.app"
+
+
+def test_topresumed_never_beats_primary():
+    dump = (
+        "  mResumedActivity: ActivityRecord{x u0 primary.pkg/.Main t1}\n"
+        "  topResumedActivity: ActivityRecord{y u0 newer.pkg/.Main t2}\n"
+    )
+    assert parse_foreground(dump) == "primary.pkg"
+
+
+def test_empty_and_garbage_return_none():
+    assert parse_foreground("") is None
+    assert parse_foreground("some random log lines\nwithout any resumed marker") is None
+
+
+def test_resumed_duplicates_tolerated():
+    # Resumed: / ResumedActivity: / mResumedActivity: all describe the same app.
+    dump = (
+        "      Resumed activities in task display areas (from top to bottom):\n"
+        "        Resumed: ActivityRecord{4f0d030 u0 ar.tvplayer.tv/.ui.MainActivity t8152}\n"
+        "      ResumedActivity: ActivityRecord{4f0d030 u0 ar.tvplayer.tv/.ui.MainActivity t8152}\n"
+        "mResumedActivity: ActivityRecord{4f0d030 u0 ar.tvplayer.tv/.ui.MainActivity t8152}\n"
+    )
+    assert parse_foreground(dump) == "ar.tvplayer.tv"
+
+
+def test_null_package_never_returned():
+    dump = "mResumedActivity: ActivityRecord{4f0d030 u0 null/null t1}\n"
+    assert parse_foreground(dump) is None