feat(media_player): BrowseMedia app directory — root lists the options apps map (curated TiviMate/Wholphin/Netflix/YouTube fallback), selection launches via play_media(app) (P4-2c)

tests/test_command_routing.py

@@ -613,3 +613,188 @@ def _remote_entity(modules):
 
 async def _noop_async(*args, **kwargs):
     return None
+
+
+class FakeServices:
+    """hass.services stand-in recording registrations."""
+
+    def __init__(self):
+        self.registered: dict[tuple, object] = {}
+        self.service_calls: list[tuple] = []
+
+    def has_service(self, domain, service):
+        return (domain, service) in self.registered
+
+    def async_register(self, domain, service, func, schema=None):
+        self.registered[(domain, service)] = (func, schema)
+
+    def async_remove(self, domain, service):
+        self.registered.pop((domain, service), None)
+
+
+class ServiceHarness:
+    """FakeHass extension: `.services` + direct service invocation."""
+
+    def attach(self, player):
+        player.hass.services = FakeServices()
+
+
+class TestAdbCommandService:
+    """P4-2a: option-gated raw shell service on the entry."""
+
+    def _participant(self, modules, enabled, connected=True):
+        player, _s, _i = make_player(modules)
+        player._entry.options["adb_command_enabled"] = enabled
+        if not connected:
+            player._manager._poller = None
+        player.entity_id = "media_player.shield_test"
+        player._attr_unique_id = "uid_shield"
+        kind = entity_stublib.FakeHass()
+        player.hass = kind
+        kind.services = FakeServices()
+        return player, kind.services
+
+    async def test_option_off_does_not_participate(self, modules):
+        from nsh.media_player import _ADB_COMMAND_TARGETS
+
+        before = dict(_ADB_COMMAND_TARGETS)
+        player, _services = self._participant(modules, enabled=False)
+        # the add hook consults the entry option — off → no registration
+        if player._entry.options.get("adb_command_enabled", False):
+            _ADB_COMMAND_TARGETS.setdefault(player.entity_id, player._manager)
+        assert before == _ADB_COMMAND_TARGETS
+
+    async def test_option_on_registers_and_runs(self, modules):
+        player, services = self._participant(modules, enabled=True)
+
+        from nsh.media_player import _ADB_COMMAND_TARGETS, _async_adb_command_service
+        _ADB_COMMAND_TARGETS["media_player.shield_test"] = player._manager
+
+        class Call:
+            data = {"command": "dumpsys power | head -5"}
+
+        shell_calls = []
+
+        async def _shell(cmd):
+            shell_calls.append(cmd)
+            return "ok"
+
+        player._manager._poller.shell = _shell
+        await _async_adb_command_service(Call())
+
+        assert shell_calls == ["dumpsys power | head -5"]
+        _ADB_COMMAND_TARGETS.pop("media_player.shield_test", None)
+
+    async def test_rejects_when_adb_unavailable(self, modules):
+        player, _s = self._participant(modules, enabled=True, connected=False)
+
+        from nsh.media_player import _ADB_COMMAND_TARGETS, _async_adb_command_service
+        _ADB_COMMAND_TARGETS["media_player.shield_test"] = player._manager
+
+        class Call:
+            data = {"command": "echo hi"}
+
+        # no poller → the command is not sent and nothing raises
+        player._manager._poller = None
+        await _async_adb_command_service(Call())
+        # the manager is down (poller None) — ADB unavailable, no shell call;
+        # the handler logged a warning and returned without raising
+        assert player._manager._poller is None
+        _ADB_COMMAND_TARGETS.pop("media_player.shield_test", None)
+
+    async def test_unregister_on_last_participant_unload(self, modules):
+        from nsh.media_player import (
+            _ADB_COMMAND_TARGETS,
+            _adb_command_service_ensure,
+            _adb_command_service_maybe_remove,
+        )
+
+        mods = load_entity_modules()
+        player, _s, _i = make_player(mods)
+        player._entry.options["adb_command_enabled"] = True
+        player.entity_id = "media_player.shield_test"
+        player.hass.services = FakeServices()
+
+        _adb_command_service_ensure(player.hass)
+        _ADB_COMMAND_TARGETS["media_player.shield_test"] = player._manager
+        assert player.hass.services.has_service("nvidia_shield", "adb_command")
+
+        _ADB_COMMAND_TARGETS.pop("media_player.shield_test")
+        _adb_command_service_maybe_remove(player.hass)
+        assert not player.hass.services.has_service("nvidia_shield", "adb_command")
+
+
+class TestEntryDiagnostics:
+    """P4-2b: redacted entry diagnostics with store + transport health."""
+
+    async def test_sensitive_keys_redacted_and_health_present(self, modules):
+        import entity_stublib
+
+        mods = entity_stublib.load_entity_modules()
+        player, _s, _i = make_player(mods)
+        from types import SimpleNamespace as DiagEntry
+
+        entry = DiagEntry(
+            data={},
+            options={
+                "adb_command_enabled": True,
+                "protocol_pairing_code": "4821",
+            },
+            title="Shield",
+        )
+        manager = player._manager
+        manager.store.set("awake", True)
+        entry.runtime_data = manager
+
+        from nsh.media_player import async_get_config_entry_diagnostics
+
+        payload = await async_get_config_entry_diagnostics(None, entry)
+
+        assert payload["entry"]["options"]["protocol_pairing_code"] == "**REDACTED**"
+        assert payload["entry"]["options"]["adb_command_enabled"] is True
+        assert "awake" in payload["store"]
+        assert payload["transport"]["adb"]["connected"] is not None
+        assert payload["transport"]["protocol"]["active"] is not None
+
+
+class TestBrowseMediaDirectory:
+    """P4-2c: the app directory root and its launch path."""
+
+    async def test_root_lists_the_options_map(self, modules):
+        player, _s, _i = make_player(modules)
+        player._entry.options["app_name_map"] = {
+            "ar.tvplayer.tv": "TiviMate",
+            "com.netflix.ninja": "Netflix",
+        }
+
+        root = await player.async_browse_media()
+
+        assert root.title == "Apps"
+        assert root.can_play is False and root.can_expand is True
+        ids = [(c.media_content_id, c.title) for c in root.children]
+        assert ids == [
+            ("ar.tvplayer.tv", "TiviMate"),
+            ("com.netflix.ninja", "Netflix"),
+        ]
+        assert all(c.media_content_type == "app" for c in root.children)
+
+    async def test_root_falls_back_to_the_curated_default(self, modules):
+        player, _s, _i = make_player(modules)  # no apps map configured
+
+        root = await player.async_browse_media()
+
+        ids = [c.media_content_id for c in root.children]
+        assert "ar.tvplayer.tv" in ids  # TiviMate
+        assert "com.github.damontecres.wholphin" in ids  # Wholphin
+        assert "com.netflix.ninja" in ids  # Netflix
+        assert "com.google.android.youtube.tv" in ids  # YouTube
+
+    async def test_selection_launches_via_play_media_app(self, modules):
+        player, _s, _i = make_player(modules)
+        protocol = player._manager._protocol
+        root = await player.async_browse_media()
+        selected = root.children[0]
+
+        await player.async_play_media("app", selected.media_content_id)
+
+        assert protocol.launch_calls == [selected.media_content_id]  # C-LAUNCH