""" Step 25 — Headset / Media-Key Control of Session Execution `navigator.mediaSession` is replaced by a lightweight mock injected before every page load (see `_MEDIA_SESSION_MOCK`): `setActionHandler` calls are captured into `window.__mediaSessionMock.handlers`, and `metadata` / `playbackState` assignments are recorded into history arrays. Tests invoke the captured handlers directly (`window.__mediaSessionMock.handlers.nexttrack()`) to simulate a real headset button press. - test_nexttrack_advances_step_and_saves_log: `nexttrack` on a normal step calls Done, advancing the queue and logging the step - test_nexttrack_on_final_round_continues: `nexttrack` on the final-round screen calls Continue, advancing to the next session item - test_previoustrack_returns_to_previous_step: `previoustrack` steps back and is a no-op on the first step - test_onscreen_previous_button: the visible Previous button mirrors `previoustrack` and is disabled on the first step - test_play_pause_toggles_timer_and_playback_state: `play`/`pause` toggle an active timer and update `mediaSession.playbackState` - test_stop_exits_with_aborted_status: `stop` exits immediately and saves a partial log with status `aborted` - test_metadata_tracks_current_exercise: metadata title matches the current exercise and updates on advance - test_handlers_unregistered_after_exit: action handlers are cleared once the session ends - test_works_with_media_session_absent: execution works normally (red path) when `navigator.mediaSession` does not exist """ import json import pytest from playwright.sync_api import Page, expect # ── Mock injected before every page load ───────────────────────────────────── _MEDIA_SESSION_MOCK = """ window.__mediaSessionMock = { handlers: {}, metadataHistory: [], playbackStateHistory: [], }; const __mockMediaSession = { setActionHandler(action, handler) { window.__mediaSessionMock.handlers[action] = handler; }, set metadata(m) { window.__mediaSessionMock.metadataHistory.push( m ? { title: m.title, artist: m.artist, album: m.album } : null ); }, get metadata() { const h = window.__mediaSessionMock.metadataHistory; return h.length ? h[h.length - 1] : null; }, set playbackState(s) { window.__mediaSessionMock.playbackStateHistory.push(s); }, get playbackState() { const h = window.__mediaSessionMock.playbackStateHistory; return h.length ? h[h.length - 1] : 'none'; }, }; Object.defineProperty(navigator, 'mediaSession', { configurable: true, value: __mockMediaSession, }); """ _MEDIA_SESSION_ABSENT = """ delete Navigator.prototype.mediaSession; """ def _setup(page: Page, app_url: str, mock_media_session=True): if mock_media_session: page.add_init_script(_MEDIA_SESSION_MOCK) else: page.add_init_script(_MEDIA_SESSION_ABSENT) page.goto(app_url) page.wait_for_function( "document.documentElement.getAttribute('data-db-ready') === 'true'", timeout=15000, ) def _create_exercise(page: Page, title: str) -> str: return page.evaluate(f"""async () => {{ const userId = await window.__repos.settings.get('user_uuid') return window.__repos.exercises.create({{ title: {repr(title)}, description: '', asymmetric: false, alternate: false, image_urls: '[]', video_urls: '[]', default_reps: 10, default_weight: 0, created_by: userId }}) }}""") def _create_combo(page: Page, title: str, combo_type: str, exercise_ids: list, timer_config: dict = None): tc_json = json.dumps(timer_config or {}) ex_json = json.dumps([{"exerciseId": eid, "defaultReps": 10, "defaultWeight": 0} for eid in exercise_ids]) return page.evaluate(f"""async () => {{ const userId = await window.__repos.settings.get('user_uuid') const comboId = await window.__repos.combos.create({{ title: {repr(title)}, description: '', type: {repr(combo_type)}, timer_config: {tc_json}, created_by: userId }}) await window.__repos.combos.setExercises(comboId, {ex_json}) return comboId }}""") def _create_session_with_items(page: Page, title: str, items: list) -> str: items_json = json.dumps(items) return page.evaluate(f"""async () => {{ const userId = await window.__repos.settings.get('user_uuid') const sessId = await window.__repos.sessions.create({{ title: {repr(title)}, description: '', created_by: userId }}) const items = {items_json} await window.__repos.sessions.setItems(sessId, items.map((it, i) => ({{ item_id: it.item_id, item_type: it.item_type, position: i, repetitions: it.repetitions, weight: it.weight || 0 }}))) return sessId }}""") def _clean_all(page: Page): page.evaluate("""async () => { const logs = await window.__repos.executionLogs.getAll() for (const l of logs) await window.__repos.executionLogs.delete(l.id) const sessions = await window.__repos.sessions.getAll() for (const s of sessions) await window.__repos.sessions.delete(s.id) const combos = await window.__repos.combos.getAll() for (const c of combos) await window.__repos.combos.delete(c.id) const exercises = await window.__repos.exercises.getAll() for (const e of exercises) await window.__repos.exercises.delete(e.id) }""") def _navigate_to_execute(page: Page, app_url: str, sess_id: str): page.goto(f"{app_url}#/sessions/{sess_id}/execute") page.wait_for_selector('[data-testid="execute-session-title"]', timeout=10000) page.wait_for_timeout(300) def _wait_for_handlers(page: Page): page.wait_for_function( "!!(window.__mediaSessionMock && window.__mediaSessionMock.handlers.nexttrack)", timeout=10000, ) def _call_handler(page: Page, action: str): page.evaluate(f"async () => await window.__mediaSessionMock.handlers['{action}']()") page.wait_for_timeout(200) # ── Tests ───────────────────────────────────────────────────────────────────── def test_nexttrack_advances_step_and_saves_log(page: Page, app_url: str): """`nexttrack` on a normal step advances the queue and logs the step.""" _setup(page, app_url) _clean_all(page) ex1 = _create_exercise(page, "MediaSession Ex1") ex2 = _create_exercise(page, "MediaSession Ex2") sess_id = _create_session_with_items(page, "MediaSession Session", [ {"item_id": ex1, "item_type": "exercise", "repetitions": 5}, {"item_id": ex2, "item_type": "exercise", "repetitions": 5}, ]) _navigate_to_execute(page, app_url, sess_id) _wait_for_handlers(page) expect(page.locator('[data-testid="execute-exercise-title"]')).to_have_text("MediaSession Ex1") _call_handler(page, "nexttrack") expect(page.locator('[data-testid="execute-exercise-title"]')).to_have_text("MediaSession Ex2") logs = page.evaluate("window.__repos.executionLogs.getAll()") assert len(logs) == 1 items = page.evaluate(f"window.__repos.executionLogs.getItems({json.dumps(logs[0]['id'])})") assert len(items) == 1, f"Expected 1 logged item, got {items}" assert items[0]["exercise_id"] == ex1 def test_nexttrack_on_final_round_continues(page: Page, app_url: str): """`nexttrack` on the final-round screen continues to the next session item.""" _setup(page, app_url) _clean_all(page) ex1 = _create_exercise(page, "MediaSession Combo Ex1") ex2 = _create_exercise(page, "MediaSession Combo Ex2") trailing = _create_exercise(page, "MediaSession Trailing") combo_id = _create_combo(page, "MediaSession Combo", "NONE", [ex1, ex2]) sess_id = _create_session_with_items(page, "MediaSession Combo Session", [ {"item_id": combo_id, "item_type": "combo", "repetitions": 1}, {"item_id": trailing, "item_type": "exercise", "repetitions": 5}, ]) _navigate_to_execute(page, app_url, sess_id) _wait_for_handlers(page) _call_handler(page, "nexttrack") # Ex1 done -> Ex2 expect(page.locator('[data-testid="execute-exercise-title"]')).to_have_text( "MediaSession Combo Ex2" ) _call_handler(page, "nexttrack") # Ex2 done (final) -> final-round screen expect(page.locator('[data-testid="execute-final-round"]')).to_be_visible() _call_handler(page, "nexttrack") # Continue -> trailing exercise expect(page.locator('[data-testid="execute-final-round"]')).not_to_be_visible() expect(page.locator('[data-testid="execute-exercise-title"]')).to_have_text( "MediaSession Trailing" ) def test_previoustrack_returns_to_previous_step(page: Page, app_url: str): """`previoustrack` steps back to the previous step and is a no-op at index 0.""" _setup(page, app_url) _clean_all(page) ex1 = _create_exercise(page, "MediaSession Prev Ex1") ex2 = _create_exercise(page, "MediaSession Prev Ex2") sess_id = _create_session_with_items(page, "MediaSession Prev Session", [ {"item_id": ex1, "item_type": "exercise", "repetitions": 5}, {"item_id": ex2, "item_type": "exercise", "repetitions": 5}, ]) _navigate_to_execute(page, app_url, sess_id) _wait_for_handlers(page) # No-op at the first step _call_handler(page, "previoustrack") expect(page.locator('[data-testid="execute-exercise-title"]')).to_have_text( "MediaSession Prev Ex1" ) _call_handler(page, "nexttrack") expect(page.locator('[data-testid="execute-exercise-title"]')).to_have_text( "MediaSession Prev Ex2" ) _call_handler(page, "previoustrack") expect(page.locator('[data-testid="execute-exercise-title"]')).to_have_text( "MediaSession Prev Ex1" ) def test_onscreen_previous_button(page: Page, app_url: str): """The visible Previous button mirrors `previoustrack` and is disabled on the first step.""" _setup(page, app_url) _clean_all(page) ex1 = _create_exercise(page, "MediaSession Btn Ex1") ex2 = _create_exercise(page, "MediaSession Btn Ex2") sess_id = _create_session_with_items(page, "MediaSession Btn Session", [ {"item_id": ex1, "item_type": "exercise", "repetitions": 5}, {"item_id": ex2, "item_type": "exercise", "repetitions": 5}, ]) _navigate_to_execute(page, app_url, sess_id) prev_btn = page.locator('[data-testid="execute-previous-btn"]') expect(prev_btn).to_be_visible() expect(prev_btn).to_be_disabled() page.click('[data-testid="execute-done-btn"]') expect(page.locator('[data-testid="execute-exercise-title"]')).to_have_text( "MediaSession Btn Ex2" ) expect(prev_btn).to_be_enabled() prev_btn.click() expect(page.locator('[data-testid="execute-exercise-title"]')).to_have_text( "MediaSession Btn Ex1" ) expect(prev_btn).to_be_disabled() def test_play_pause_toggles_timer_and_playback_state(page: Page, app_url: str): """`play`/`pause` toggle an active timer and update `mediaSession.playbackState`.""" _setup(page, app_url) _clean_all(page) ex_id = _create_exercise(page, "MediaSession EMOM Ex") combo_id = _create_combo(page, "MediaSession EMOM", "EMOM", [ex_id]) sess_id = _create_session_with_items(page, "MediaSession EMOM Session", [ {"item_id": combo_id, "item_type": "combo", "repetitions": 1}, ]) _navigate_to_execute(page, app_url, sess_id) _wait_for_handlers(page) _call_handler(page, "pause") state = page.evaluate("navigator.mediaSession.playbackState") assert state == "paused", f"Expected 'paused', got {state}" expect(page.locator(".timer-pause-btn i")).to_have_class("bi bi-play-fill") _call_handler(page, "play") state = page.evaluate("navigator.mediaSession.playbackState") assert state == "playing", f"Expected 'playing', got {state}" expect(page.locator(".timer-pause-btn i")).to_have_class("bi bi-pause-fill") def test_stop_exits_with_aborted_status(page: Page, app_url: str): """`stop` exits the session immediately and saves a partial log with status `aborted`.""" _setup(page, app_url) _clean_all(page) ex_id = _create_exercise(page, "MediaSession Stop Ex") sess_id = _create_session_with_items(page, "MediaSession Stop Session", [ {"item_id": ex_id, "item_type": "exercise", "repetitions": 5}, ]) _navigate_to_execute(page, app_url, sess_id) _wait_for_handlers(page) _call_handler(page, "stop") page.wait_for_function( f"window.location.hash.includes('/sessions/{sess_id}') && !window.location.hash.includes('/execute')", timeout=5000, ) logs = page.evaluate("window.__repos.executionLogs.getAll()") assert len(logs) == 1 assert logs[0]["status"] == "aborted", f"Expected aborted, got {logs[0]}" def test_metadata_tracks_current_exercise(page: Page, app_url: str): """Metadata title matches the current exercise and updates on advance.""" _setup(page, app_url) _clean_all(page) ex1 = _create_exercise(page, "MediaSession Meta Ex1") ex2 = _create_exercise(page, "MediaSession Meta Ex2") sess_id = _create_session_with_items(page, "MediaSession Meta Session", [ {"item_id": ex1, "item_type": "exercise", "repetitions": 5}, {"item_id": ex2, "item_type": "exercise", "repetitions": 5}, ]) _navigate_to_execute(page, app_url, sess_id) _wait_for_handlers(page) meta = page.evaluate("navigator.mediaSession.metadata") assert meta["title"] == "MediaSession Meta Ex1", f"Unexpected metadata: {meta}" page.click('[data-testid="execute-done-btn"]') page.wait_for_timeout(200) meta = page.evaluate("navigator.mediaSession.metadata") assert meta["title"] == "MediaSession Meta Ex2", f"Unexpected metadata: {meta}" def test_handlers_unregistered_after_exit(page: Page, app_url: str): """Action handlers are cleared once the session ends.""" _setup(page, app_url) _clean_all(page) ex_id = _create_exercise(page, "MediaSession Unreg Ex") sess_id = _create_session_with_items(page, "MediaSession Unreg Session", [ {"item_id": ex_id, "item_type": "exercise", "repetitions": 5}, ]) _navigate_to_execute(page, app_url, sess_id) _wait_for_handlers(page) page.click('[data-testid="execute-exit-btn"]') page.wait_for_function( f"window.location.hash.includes('/sessions/{sess_id}') && !window.location.hash.includes('/execute')", timeout=5000, ) page.wait_for_timeout(200) handlers = page.evaluate("window.__mediaSessionMock.handlers") for action in ["play", "pause", "nexttrack", "previoustrack", "stop"]: assert handlers.get(action) is None, f"Handler for '{action}' should be null after exit" def test_works_with_media_session_absent(page: Page, app_url: str): """Execution works normally when `navigator.mediaSession` does not exist (red path).""" _setup(page, app_url, mock_media_session=False) _clean_all(page) errors = [] page.on("pageerror", lambda exc: errors.append(str(exc))) ex_id = _create_exercise(page, "MediaSession Absent Ex") sess_id = _create_session_with_items(page, "MediaSession Absent Session", [ {"item_id": ex_id, "item_type": "exercise", "repetitions": 5}, ]) _navigate_to_execute(page, app_url, sess_id) expect(page.locator('[data-testid="execute-exercise-title"]')).to_have_text( "MediaSession Absent Ex" ) page.click('[data-testid="execute-done-btn"]') page.wait_for_function( f"window.location.hash.includes('/sessions/{sess_id}') && !window.location.hash.includes('/execute')", timeout=5000, ) logs = page.evaluate("window.__repos.executionLogs.getAll()") assert len(logs) == 1 assert logs[0]["status"] == "completed" assert errors == [], f"Unexpected page errors: {errors}"