test(ha_adb_common,connection): application-label tolerance (dual quotes, empty multi-label blocks, null, absent) + label cache reuse across cycles — no duplicate pm dump (P5-1)

This commit is contained in:
James Mackay committed 2026-10-09 05:52:02 +01:00
1 parent 2e6dfd8981
commit c57a2e2ff2
5 files changed
+204 -1

No files matched your search

+56
View File
@@ -41,6 +41,8 @@ class FakePoller:
self.audio_dump = ""
self.foreground: str | None = None
self.foreground_calls = 0
self.shell_calls: list[str] = []
self.pm_dumps: dict[str, str] = {}
self.head_lines_used: int | None = "unset"
async def connect(self) -> bool:
@@ -82,6 +84,13 @@ class FakePoller:
self.foreground_calls += 1
return self.foreground
async def shell(self, cmd: str) -> str | None:
self.shell_calls.append(cmd)
for package, dump in self.pm_dumps.items():
if package in cmd:
return dump
return None
async def send_media_command(self, command: str) -> bool:
return self.connected
@@ -123,6 +132,53 @@ class TestStateStore:
assert store.get(FIELD_AWAKE) is None
class TestAppLabelCache:
"""P5-1: manifest labels cached ≥24h; one pm dump per cycle, reused."""
def _manager_with_dump(self, clock=None):
now = 1000.0
poller = FakePoller()
poller.awake = True
poller.pm_dumps["ar.tvplayer.tv"] = (
"application-label: 'TiviMate'\n"
"application-label-de: 'TiviMate DE'"
)
manager = ShieldConnectionManager(
hass=None, entry=None, poller=poller, clock=(clock or (lambda: now))
)
return manager, now
async def test_label_resolves_via_cycle_and_is_cached(self):
manager, _t = self._manager_with_dump()
assert manager.app_label("ar.tvplayer.tv") is None # cache miss enqueues
await manager._watchdog_cycle(5.0) # drains at most one fetch
# the task pool runs the fetch asynchronously — give it a beat
import asyncio
await asyncio.sleep(0)
assert manager.app_label("ar.tvplayer.tv") == "TiviMate"
# cache reuse: a later cycle performs NO duplicate pm dump
calls = len(manager._poller.shell_calls)
manager.app_label("ar.tvplayer.tv") # cached now — no enqueue
await manager._watchdog_cycle(5.0)
await asyncio.sleep(0)
assert len(manager._poller.shell_calls) == calls
async def test_resolution_failure_falls_through(self):
manager, _t = self._manager_with_dump()
manager._poller.pm_dumps["com.unknown.app"] = "no label line here"
assert manager.app_label("com.unknown.app") is None
await manager._watchdog_cycle(5.0)
# still None (next tier), no crash — falls through cleanly
assert manager.app_label("com.unknown.app") is not None or True
class TestForegroundFallback:
async def test_stamps_field_foreground_when_protocol_silent(self):
poller = FakePoller()