James Mackay
07682ff082
The media session parser only understood states 2 (paused) and 3 (playing), so a session refilling its buffer (Android state 6, which the Shield's own ExoPlayer/MediaSession sessions report exactly like stock Android) read as "idle" or vanished into the foreground tier. MediaSessionInfo gains an independent `buffering` flag (playing/paused/ buffering are now three flags off one integer), the session sort puts buffering right behind playing — it is mid-playback, so it outranks a paused session when the ladder picks one — and the running-apps enrichment names a buffering live/last-known session "buffering" instead of falling into "idle". The MediaPlayerState stub gains BUFFERING and STANDBY so an in-suite reference to a member real HA does not have fails here (real HA 2026.10 exposes off/on/idle/playing/paused/standby/buffering).
348 lines
14 KiB
Python
348 lines
14 KiB
Python
"""Parser tests driven by real Shield firmware captures (Android 11, RQ1A.210105.003).
|
|
|
|
Fixtures live in tests/fixtures — see docs/RESEARCH_SHIELD.md for how each
|
|
capture was produced and which expectations it pins.
|
|
"""
|
|
|
|
from pathlib import Path
|
|
|
|
from ha_adb_common.media_session import (
|
|
MediaSessionInfo,
|
|
parse_all_media_sessions,
|
|
parse_audio_playback,
|
|
parse_media_session,
|
|
)
|
|
|
|
FIXTURES = Path(__file__).parent / "fixtures"
|
|
|
|
|
|
def _fixture(name: str) -> str:
|
|
return (FIXTURES / name).read_text(encoding="utf-8")
|
|
|
|
|
|
class TestTiviMateHasNoSession:
|
|
"""TiviMate (ar.tvplayer.tv) registers no MediaSession on this firmware."""
|
|
|
|
def test_no_sessions_parsed(self):
|
|
assert parse_all_media_sessions(_fixture("media_session_tivimate.txt")) == []
|
|
|
|
def test_session_query_returns_none(self):
|
|
dump = _fixture("media_session_tivimate.txt")
|
|
assert parse_media_session(dump, "ar.tvplayer.tv") is None
|
|
|
|
def test_inactive_stale_session_is_not_surfaced(self):
|
|
"""Netflix registers an inactive, state-less session; the parser must
|
|
ignore it — an active=false block never appears in the results."""
|
|
dump = _fixture("media_session_tivimate.txt")
|
|
assert parse_media_session(dump, "com.netflix.ninja") is None
|
|
|
|
|
|
class TestWholphinSession:
|
|
"""Wholphin (com.github.damontecres.wholphin) publishes a proper session."""
|
|
|
|
def test_playing_session_parsed(self):
|
|
sessions = parse_all_media_sessions(_fixture("media_session_wholphin.txt"))
|
|
assert len(sessions) == 1
|
|
session = sessions[0]
|
|
assert session.package == "com.github.damontecres.wholphin"
|
|
assert session.playing is True
|
|
assert session.paused is False
|
|
assert session.title == "Twirlywoos"
|
|
assert session.artist is None
|
|
|
|
def test_foreground_package_is_preferred(self):
|
|
dump = _fixture("media_session_wholphin.txt")
|
|
best = parse_media_session(dump, "com.github.damontecres.wholphin")
|
|
assert best is not None
|
|
assert best.package == "com.github.damontecres.wholphin"
|
|
assert best.playing is True
|
|
|
|
|
|
class TestPlaybackStatePosition:
|
|
"""PlaybackState carries position (ms), speed, and uptime-ms `updated` (R9)."""
|
|
|
|
def test_wholphin_position_fields_parsed(self):
|
|
sessions = parse_all_media_sessions(_fixture("media_session_wholphin.txt"))
|
|
assert len(sessions) == 1
|
|
session = sessions[0]
|
|
assert session.position == 17237
|
|
assert session.speed == 1.0
|
|
assert session.updated == 1408392
|
|
|
|
def test_buffered_position_is_not_mistaken_for_position(self):
|
|
"""`buffered position=70070` sits in the same block; only the real
|
|
`position=17237` may be read."""
|
|
sessions = parse_all_media_sessions(_fixture("media_session_wholphin.txt"))
|
|
assert sessions[0].position != 70070
|
|
|
|
def test_position_at_uptime_extrapolates_with_speed(self):
|
|
session = MediaSessionInfo(
|
|
position=17237, speed=1.0, updated=1408392, playing=True
|
|
)
|
|
assert session.position_at_uptime(1408392 + 1000) == 18237
|
|
assert session.position_at_uptime(1408392) == 17237
|
|
|
|
def test_position_at_uptime_none_without_sample(self):
|
|
assert MediaSessionInfo().position_at_uptime(1_000_000) is None
|
|
|
|
def test_paused_session_parses_state_and_position(self):
|
|
dump = (
|
|
"Sessions Stack - have 1 sessions:\n"
|
|
" Some app session com.example.player/Session (userId=0)\n"
|
|
" package=com.example.player\n"
|
|
" active=true\n"
|
|
" state=PlaybackState {state=2, position=4200, buffered position=9000, "
|
|
"speed=0.0, updated=500, actions=1, custom actions=[], active item id=0, "
|
|
"error=null}\n"
|
|
" metadata: size=2, description=Song Title, Some Artist\n"
|
|
" Audio playback (lastly played comes first)\n"
|
|
" uid=10100 packages=com.example.player\n"
|
|
)
|
|
sessions = parse_all_media_sessions(dump)
|
|
assert len(sessions) == 1
|
|
session = sessions[0]
|
|
assert session.paused is True
|
|
assert session.playing is False
|
|
assert session.position == 4200
|
|
assert session.speed == 0.0
|
|
assert session.updated == 500
|
|
assert session.title == "Song Title"
|
|
assert session.artist == "Some Artist"
|
|
|
|
def test_stateless_playbackstate_line_ignored(self):
|
|
"""A non-PlaybackState `state=` line must not crash or fabricate state."""
|
|
dump = (
|
|
" Some app session com.example.player/Session (userId=0)\n"
|
|
" package=com.example.player\n"
|
|
" active=true\n"
|
|
" state=null\n"
|
|
" metadata: size=1, description=Only Title\n"
|
|
)
|
|
sessions = parse_all_media_sessions(dump)
|
|
assert len(sessions) == 1
|
|
assert sessions[0].playing is False
|
|
assert sessions[0].paused is False
|
|
assert sessions[0].position is None
|
|
|
|
|
|
class TestPlaybackStateBufferingT55:
|
|
"""t55: Android PlaybackState 6 is BUFFERING, mapped in the parser.
|
|
|
|
The Shield's own sessions report the same integer stock Android does (6),
|
|
so no vendor-specific variant is treated as buffering — a device variant
|
|
would be a template field on the same PlaybackState line, never a
|
|
different integer.
|
|
"""
|
|
|
|
def _buffering_dump(self, state=6, speed="0.0"):
|
|
return (
|
|
"Sessions Stack - have 1 sessions:\n"
|
|
" Some app session com.netflix.ninja/Session (userId=0)\n"
|
|
" package=com.netflix.ninja\n"
|
|
" active=true\n"
|
|
f" state=PlaybackState {{state={state}, position=0, "
|
|
f"buffered position=2000, speed={speed}, updated=900, actions=1, "
|
|
"custom actions=[], active item id=0, error=null}\n"
|
|
" metadata: size=2, description=Movie Title, Some Artist\n"
|
|
)
|
|
|
|
def test_state_6_is_buffering(self):
|
|
sessions = parse_all_media_sessions(self._buffering_dump())
|
|
assert len(sessions) == 1
|
|
session = sessions[0]
|
|
assert session.buffering is True
|
|
assert session.playing is False
|
|
assert session.paused is False
|
|
assert session.title == "Movie Title"
|
|
|
|
def test_state_3_and_2_stay_playing_and_paused(self):
|
|
for state, attr in ((3, "playing"), (2, "paused")):
|
|
sessions = parse_all_media_sessions(self._buffering_dump(state=state))
|
|
session = sessions[0]
|
|
assert session.buffering is False
|
|
assert getattr(session, attr) is True
|
|
|
|
def test_state_6_wins_when_the_dump_says_speed_zero_too(self):
|
|
"""The buffering signal is the integer, not the speed field."""
|
|
sessions = parse_all_media_sessions(self._buffering_dump(state=6))
|
|
assert sessions[0].buffering is True
|
|
|
|
def test_buffering_outranks_a_paused_session_in_the_pick_order(self):
|
|
sessions = parse_all_media_sessions(
|
|
self._buffering_dump()
|
|
+ " Other app session com.example.player/Session (userId=0)\n"
|
|
" package=com.example.player\n"
|
|
" active=true\n"
|
|
" state=PlaybackState {state=2, position=1, speed=0.0, "
|
|
"updated=1, actions=1, custom actions=[], active item id=0, "
|
|
"error=null}\n"
|
|
" metadata: size=1, description=Paused Tune\n"
|
|
)
|
|
assert [s.package for s in sessions] == [
|
|
"com.netflix.ninja",
|
|
"com.example.player",
|
|
]
|
|
# and parse_media_session (no foreground preference) picks buffering
|
|
dump = self._buffering_dump() + (
|
|
" Other app session com.example.player/Session (userId=0)\n"
|
|
" package=com.example.player\n"
|
|
" active=true\n"
|
|
" state=PlaybackState {state=2, position=1, speed=0.0, "
|
|
"updated=1, actions=1, custom actions=[], active item id=0, "
|
|
"error=null}\n"
|
|
" metadata: size=1, description=Paused Tune\n"
|
|
)
|
|
assert parse_media_session(dump).package == "com.netflix.ninja"
|
|
|
|
|
|
class TestAudioPlaybackSection:
|
|
"""The "Audio playback (lastly played comes first)" section lists apps
|
|
that recently played audio even with no MediaSession (R2)."""
|
|
|
|
def test_tivimate_shows_up_without_a_session(self):
|
|
packages = parse_audio_playback(_fixture("media_session_tivimate.txt"))
|
|
assert packages == [
|
|
"ar.tvplayer.tv",
|
|
"com.google.android.apps.mediashell",
|
|
]
|
|
|
|
def test_order_is_dump_order_most_recent_first(self):
|
|
packages = parse_audio_playback(_fixture("media_session_tivimate.txt"))
|
|
assert packages[0] == "ar.tvplayer.tv"
|
|
|
|
def test_wholphin_audio_playback(self):
|
|
assert parse_audio_playback(_fixture("media_session_wholphin.txt")) == [
|
|
"com.github.damontecres.wholphin"
|
|
]
|
|
|
|
def test_lines_before_the_section_are_ignored(self):
|
|
"""`... package:` listener lines precede the section and must not leak
|
|
into the audio-playback package list."""
|
|
dump = (
|
|
" Media key listener package:\n"
|
|
" OnMediaKeyEventDispatchedListener: added 0 listener(s)\n"
|
|
" Audio playback (lastly played comes first)\n"
|
|
" uid=10117 packages=ar.tvplayer.tv\n"
|
|
)
|
|
assert parse_audio_playback(dump) == ["ar.tvplayer.tv"]
|
|
|
|
def test_empty_dump(self):
|
|
assert parse_audio_playback("") == []
|
|
|
|
|
|
def test_running_apps_buffering_state_t55():
|
|
"""t55: a buffering session (and a buffering last-known) is named buffering."""
|
|
from nsh.ha_adb_common import MediaSessionInfo, running_apps_from_tasks
|
|
|
|
tasks = [
|
|
{"package": "com.netflix.ninja"},
|
|
{"package": "ar.tvplayer.tv"},
|
|
]
|
|
sessions = [MediaSessionInfo(package="com.netflix.ninja", buffering=True)]
|
|
last_known = {
|
|
"ar.tvplayer.tv": MediaSessionInfo(package="ar.tvplayer.tv", buffering=True)
|
|
}
|
|
|
|
apps = running_apps_from_tasks(
|
|
tasks, sessions, foreground_package=None, last_known=last_known
|
|
)
|
|
|
|
netflix = next(a for a in apps if a["package"] == "com.netflix.ninja")
|
|
tivi = next(a for a in apps if a["package"] == "ar.tvplayer.tv")
|
|
assert netflix["state"] == "buffering"
|
|
assert tivi["last_known_state"] == "buffering"
|
|
|
|
|
|
def test_running_apps_last_known_enrichment():
|
|
"""P3-3b: apps without a live session carry their TTL-capped last-known state."""
|
|
from nsh.ha_adb_common import MediaSessionInfo, running_apps_from_tasks
|
|
|
|
tasks = [
|
|
{"package": "ar.tvplayer.tv"},
|
|
{"package": "com.netflix.ninja"},
|
|
]
|
|
sessions = [MediaSessionInfo(package="com.netflix.ninja", playing=True, title="Movie")]
|
|
last_known = {
|
|
"ar.tvplayer.tv": MediaSessionInfo(
|
|
package="ar.tvplayer.tv", playing=False, paused=True, title="Last show"
|
|
)
|
|
}
|
|
|
|
apps = running_apps_from_tasks(tasks, sessions, foreground_package=None, last_known=last_known)
|
|
|
|
tivi = next(a for a in apps if a["package"] == "ar.tvplayer.tv")
|
|
netflix = next(a for a in apps if a["package"] == "com.netflix.ninja")
|
|
|
|
assert tivi["state"] == "running" # no live session
|
|
assert tivi["last_known_state"] == "paused"
|
|
assert tivi["last_known_title"] == "Last show"
|
|
assert "last_known_state" not in netflix # live session enriched instead
|
|
assert netflix["state"] == "playing"
|
|
|
|
|
|
def test_parse_application_label_default_locale_wins():
|
|
"""P5-1: the default-locale label wins; localized suffixes are skipped."""
|
|
from nsh.ha_adb_common.app_label import parse_application_label
|
|
|
|
dump = (
|
|
"application-label: 'TiviMate'\n"
|
|
"application-label-de: 'TiviMate DE'\n"
|
|
"application-icon-320: 'x'"
|
|
)
|
|
assert parse_application_label(dump) == "TiviMate"
|
|
|
|
|
|
def test_parse_application_label_tolerance():
|
|
from nsh.ha_adb_common.app_label import parse_application_label
|
|
|
|
# double-quoted, empty first label (multi-label block), nonexistent
|
|
assert parse_application_label('application-label:"Wholphin"') == "Wholphin"
|
|
multi_label = "application-label:''\napplication-label: 'TiviMate'"
|
|
assert parse_application_label(multi_label) == "TiviMate"
|
|
assert parse_application_label("application-label: null") is None
|
|
assert parse_application_label("") is None
|
|
assert parse_application_label("nothing here") is None
|
|
|
|
|
|
def _logcat_lines():
|
|
"""Synthetic logcat list covering START shapes + noise + wrapping."""
|
|
return [
|
|
"03-21 10:19:40.100 3104 3456 I NotificationService: noise",
|
|
"03-21 10:19:41.792 3104 3456 I ActivityTaskManager: START u0 "
|
|
"{cmp=com.netflix.ninja/com.netflix.ui.main flg=0x10000000}",
|
|
"03-21 10:19:45.100 3104 3456 I ActivityTaskManager: START u0 {cmp=null}",
|
|
"03-21 10:19:47.000 3104 3456 I ActivityTaskManager: START u10 "
|
|
"{cmp=ar.tvplayer.tv/.ui.MainActivity",
|
|
" xflg=214 abi=arm64-v8a}",
|
|
"03-21 10:19:50.000 3104 3456 I ActivityTaskManager: Displayed",
|
|
]
|
|
|
|
|
|
def test_logcat_start_records_parse_cmp():
|
|
from nsh.ha_adb_common.logcat_events import foreground_from_logcat, foreground_packages
|
|
|
|
lines = _logcat_lines()
|
|
assert foreground_packages(lines) == ["com.netflix.ninja", "ar.tvplayer.tv"]
|
|
# LAST valid START wins
|
|
assert foreground_from_logcat(lines) == "ar.tvplayer.tv"
|
|
|
|
|
|
def test_logcat_non_start_and_null_ignored():
|
|
from nsh.ha_adb_common.logcat_events import foreground_from_logcat
|
|
|
|
assert foreground_from_logcat([
|
|
"03-21 10:19:50.000 3104 3456 I ActivityTaskManager: Displayed",
|
|
"plain noise",
|
|
]) is None
|
|
lines = _logcat_lines()
|
|
assert foreground_from_logcat(lines) == "ar.tvplayer.tv" # null skipped
|
|
|
|
|
|
def test_logcat_multi_user_prefix_tolerated():
|
|
from nsh.ha_adb_common.logcat_events import foreground_from_logcat
|
|
|
|
assert foreground_from_logcat(
|
|
["03-21 10:19:41.792 3104 3456 I ActivityTaskManager: START u123 {cmp=net.whatsapp/"] + [
|
|
" flg=0x1}"]
|
|
) == "net.whatsapp"
|