feat(media): SmtcController Skeleton mit circuit-breaker
Circuit-breaker-Pattern: Nach erstem Fehler beim SMTC-Manager-Zugriff bleibt _broken=true und verhindert alle weiteren Zugriff-Versuche. Logs Warnung einmalig und cleart _paused. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,75 @@
|
|||||||
|
"""Tests für SmtcController (Windows/SMTC)."""
|
||||||
|
|
||||||
|
import sys
|
||||||
|
from unittest.mock import AsyncMock, MagicMock
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
|
||||||
|
pytestmark = pytest.mark.skipif(
|
||||||
|
sys.platform != "win32", reason="SMTC is Windows-only"
|
||||||
|
)
|
||||||
|
|
||||||
|
from winrt.windows.media.control import (
|
||||||
|
GlobalSystemMediaTransportControlsSessionPlaybackStatus as Status,
|
||||||
|
)
|
||||||
|
|
||||||
|
PLAYING = Status.PLAYING
|
||||||
|
PAUSED = Status.PAUSED
|
||||||
|
|
||||||
|
|
||||||
|
def _make_session(aumid: str, status) -> MagicMock:
|
||||||
|
"""Erzeugt eine gemockte SMTC-Session mit gegebenem PlaybackStatus."""
|
||||||
|
session = MagicMock()
|
||||||
|
session.source_app_user_model_id = aumid
|
||||||
|
info = MagicMock()
|
||||||
|
info.playback_status = status
|
||||||
|
session.get_playback_info = MagicMock(return_value=info)
|
||||||
|
session.try_pause_async = AsyncMock()
|
||||||
|
session.try_play_async = AsyncMock()
|
||||||
|
return session
|
||||||
|
|
||||||
|
|
||||||
|
def _make_manager(sessions: list) -> MagicMock:
|
||||||
|
"""Erzeugt einen gemockten SMTC-Manager mit gegebenen Sessions."""
|
||||||
|
manager = MagicMock()
|
||||||
|
manager.get_sessions = MagicMock(return_value=sessions)
|
||||||
|
return manager
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_pause_is_noop_when_smtc_unreachable(monkeypatch, caplog):
|
||||||
|
from whisper_local.media._smtc import SmtcController
|
||||||
|
|
||||||
|
controller = SmtcController()
|
||||||
|
monkeypatch.setattr(
|
||||||
|
controller,
|
||||||
|
"_ensure_manager",
|
||||||
|
AsyncMock(side_effect=RuntimeError("kein SMTC")),
|
||||||
|
)
|
||||||
|
|
||||||
|
with caplog.at_level("WARNING"):
|
||||||
|
await controller.pause()
|
||||||
|
|
||||||
|
assert controller._paused == []
|
||||||
|
assert any("SMTC" in r.message or "smtc" in r.message.lower() for r in caplog.records)
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_pause_skips_reconnect_after_smtc_failure(monkeypatch):
|
||||||
|
from whisper_local.media._smtc import SmtcController
|
||||||
|
|
||||||
|
call_count = 0
|
||||||
|
|
||||||
|
async def failing_ensure():
|
||||||
|
nonlocal call_count
|
||||||
|
call_count += 1
|
||||||
|
raise RuntimeError("kein SMTC")
|
||||||
|
|
||||||
|
controller = SmtcController()
|
||||||
|
monkeypatch.setattr(controller, "_ensure_manager", failing_ensure)
|
||||||
|
|
||||||
|
await controller.pause()
|
||||||
|
await controller.pause()
|
||||||
|
await controller.pause()
|
||||||
|
|
||||||
|
assert call_count == 1
|
||||||
@@ -0,0 +1,42 @@
|
|||||||
|
"""Windows SMTC-Implementierung via pywinrt."""
|
||||||
|
|
||||||
|
import logging
|
||||||
|
from typing import Any
|
||||||
|
|
||||||
|
logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
|
||||||
|
class SmtcController:
|
||||||
|
def __init__(self) -> None:
|
||||||
|
self._paused: list[str] = []
|
||||||
|
self._manager: Any = None
|
||||||
|
self._broken: bool = False
|
||||||
|
|
||||||
|
async def _ensure_manager(self) -> Any:
|
||||||
|
if self._broken:
|
||||||
|
raise RuntimeError("SMTC nicht verfügbar")
|
||||||
|
if self._manager is None:
|
||||||
|
from winrt.windows.media.control import (
|
||||||
|
GlobalSystemMediaTransportControlsSessionManager,
|
||||||
|
)
|
||||||
|
self._manager = (
|
||||||
|
await GlobalSystemMediaTransportControlsSessionManager.request_async()
|
||||||
|
)
|
||||||
|
return self._manager
|
||||||
|
|
||||||
|
async def pause(self) -> None:
|
||||||
|
if self._broken:
|
||||||
|
self._paused = []
|
||||||
|
return
|
||||||
|
try:
|
||||||
|
await self._ensure_manager()
|
||||||
|
except Exception as e:
|
||||||
|
logger.warning(
|
||||||
|
"SMTC nicht erreichbar, Media-Pause dauerhaft deaktiviert: %s", e
|
||||||
|
)
|
||||||
|
self._broken = True
|
||||||
|
self._paused = []
|
||||||
|
return
|
||||||
|
|
||||||
|
async def resume(self) -> None:
|
||||||
|
pass
|
||||||
Reference in New Issue
Block a user