feat(adb_client): exact-value mWakefulness= parse (R5), health_check(), optional media-session truncation (B4); wakefulness fixtures + tests

tests/test_adb_client.py

@@ -0,0 +1,145 @@
+"""Tests for AdbClientBase additions: exact wakefulness parse (R5),
+health_check(), and optional media-session truncation (plan blocker B4).
+
+The client runs against a FakeDevice recording shell commands — no real ADB
+socket or adb_shell import is needed.
+"""
+
+from pathlib import Path
+
+from ha_adb_common.adb_client import AdbClientBase, parse_wakefulness
+
+FIXTURES = Path(__file__).parent / "fixtures"
+
+CMD_WAKEFULNESS = "dumpsys power 2>/dev/null | grep mWakefulness"
+CMD_MEDIA_DEFAULT = "dumpsys media_session 2>/dev/null | head -200"
+CMD_MEDIA_FULL = "dumpsys media_session 2>/dev/null"
+CMD_ECHO_OK = "echo ok"
+
+
+class FakeDevice:
+    """Records shell commands and returns canned responses per command."""
+
+    def __init__(self, responses=None):
+        self._responses = dict(responses or {})
+        self.commands: list[str] = []
+
+    async def shell(self, cmd: str) -> str | None:
+        self.commands.append(cmd)
+        response = self._responses.get(cmd)
+        if isinstance(response, Exception):
+            raise response
+        return response
+
+
+def _client_response(cmd: str, response: str | None) -> AdbClientBase:
+    """A client whose (fake) device is already connected and answers cmd."""
+    client = AdbClientBase("shield.local", 5555, "/dev/null/adbkey")
+    client._device = FakeDevice({cmd: response})
+    client._connected = True  # skip connect(): no adb_shell in the test env
+    return client
+
+
+class TestParseWakefulness:
+    """Exact-value mWakefulness= parse; mWakefulnessChanging excluded."""
+
+    def test_awake_fixture(self):
+        text = (FIXTURES / "dumpsys_power_wakefulness_awake.txt").read_text()
+        assert parse_wakefulness(text) is True
+
+    def test_asleep_fixture(self):
+        text = (FIXTURES / "dumpsys_power_wakefulness_asleep.txt").read_text()
+        assert parse_wakefulness(text) is False
+
+    def test_known_asleep_values(self):
+        assert parse_wakefulness("mWakefulness=Asleep") is False
+        assert parse_wakefulness("mWakefulness=Dozing") is False
+        assert parse_wakefulness("mWakefulness=Dozing-suspend") is False
+
+    def test_unknown_value_is_unknown_not_awake(self):
+        """The old substring heuristic read any future value containing no
+        'Awake' as asleep; unrecognized values must be None (captain's note)."""
+        assert parse_wakefulness("mWakefulness=Dozing-ext") is None
+        assert parse_wakefulness("mWakefulness=123") is None
+
+    def test_changing_and_setting_lines_are_ignored(self):
+        assert (
+            parse_wakefulness("  mWakefulnessChanging=false") is None
+        )
+        assert parse_wakefulness("  mWakefulnessSetting=2") is None
+
+    def test_assignment_wins_over_changing_line(self):
+        text = "  mWakefulnessChanging=false\nmWakefulness=Awake\n"
+        assert parse_wakefulness(text) is True
+
+    def test_empty_input(self):
+        assert parse_wakefulness("") is None
+        assert parse_wakefulness(None) is None
+        assert parse_wakefulness("no match here") is None
+
+
+class TestIsAwake:
+    async def test_awake(self):
+        client = _client_response(
+            CMD_WAKEFULNESS,
+            (FIXTURES / "dumpsys_power_wakefulness_awake.txt").read_text(),
+        )
+        assert await client.is_awake() is True
+        assert client._device.commands == [CMD_WAKEFULNESS]
+
+    async def test_asleep(self):
+        client = _client_response(
+            CMD_WAKEFULNESS,
+            (FIXTURES / "dumpsys_power_wakefulness_asleep.txt").read_text(),
+        )
+        assert await client.is_awake() is False
+
+    async def test_unknown_reports_none(self):
+        """A garbage/empty grep result parses to None, not False."""
+        client = _client_response(CMD_WAKEFULNESS, "")
+        assert await client.is_awake() is None
+
+
+class TestHealthCheck:
+    async def test_echo_ok_round_trip_is_healthy(self):
+        client = _client_response(CMD_ECHO_OK, "ok\n")
+        assert await client.health_check() is True
+
+    async def test_failed_shell_is_unhealthy(self):
+        client = _client_response(CMD_ECHO_OK, Exception("socket closed"))
+        assert await client.health_check() is False
+        assert client.connected is False  # shell() marked the socket broken
+
+    async def test_wrong_body_is_unhealthy(self):
+        client = _client_response(CMD_ECHO_OK, "something else\n")
+        assert await client.health_check() is False
+
+
+class TestPollMediaSessionsTruncation:
+    async def test_default_truncates_at_200(self):
+        client = _client_response(CMD_MEDIA_DEFAULT, "")
+        assert await client.poll_media_sessions() is None
+        assert client._device.commands == [CMD_MEDIA_DEFAULT]
+
+    async def test_explicit_head_lines(self):
+        client = _client_response("dumpsys media_session 2>/dev/null | head -80", "")
+        await client.poll_media_sessions(head_lines=80)
+        assert client._device.commands == [
+            "dumpsys media_session 2>/dev/null | head -80"
+        ]
+
+    async def test_none_head_lines_fetches_full_dump(self):
+        """None = no truncation — the manager later passes None (B4)."""
+        wholphin = (FIXTURES / "media_session_wholphin.txt").read_text()
+        client = _client_response(CMD_MEDIA_FULL, wholphin)
+        info = await client.poll_media_sessions(head_lines=None)
+        assert client._device.commands == [CMD_MEDIA_FULL]
+        assert info is not None
+        assert info.package == "com.github.damontecres.wholphin"
+        assert info.position == 17237
+        assert client.all_sessions[0].playing is True
+
+    async def test_failed_shell_returns_none_without_reparsing(self):
+        client = _client_response(CMD_MEDIA_DEFAULT, Exception("dead"))
+        assert await client.poll_media_sessions() is None
+        assert client.last_info is None