146 lines
5.7 KiB
Python
146 lines
5.7 KiB
Python
"""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
|