James Mackay
37b2accfda
`homeassistant.components.media_player` has not exported `BrowseMediaSource` for years: the class lives in `homeassistant.components.media_source.models`, where it exists to build `media_source://<domain>` ids and is the only place a `domain` parameter belongs (t30 G18's note was about THAT class). The local import therefore raised ImportError, and with it every real browse failed — while the suite stayed green because the harness stubbed a `BrowseMediaSource` that real HA does not expose from there. `async_browse_media` now builds the standard tree exactly as core's own androidtv_remote/media_player.py does: `BrowseMedia` imported from homeassistant.components.media_player at module scope, every node carrying an explicit media_content_id/media_content_type (MediaType.APP/APPS), `children_media_class=MediaClass.APP` on the directory, and no `domain`. Verified against a real install: the old import raises ImportError, the old `domain=` kwarg raises TypeError, and the new node constructs. The harness is now faithful so this cannot hide again: * `entity_stublib` exposes a `BrowseMedia` with the real keyword-only signature, `as_dict()` and `calculate_children_class()`, plus `MediaType`, and deliberately NOT `BrowseMediaSource` — so both the drifted import and the drifted kwarg fail in-suite; * the app-directory test asserts the standard shape (classes, explicit ids and types, children_media_class, the serialised payload) instead of the removed `domain` attribute; * TestFaithfulHaStubs pins the harness contract on the AST (no production module imports BrowseMediaSource) rather than on prose. Verified: 284 passed, 2 skipped (D1 stage); ruff clean. Full gate re-run on the completed branch: 289 passed / 2 skipped and 290 passed / 1 skipped.
460 lines
16 KiB
Python
460 lines
16 KiB
Python
"""Home-Assistant object stubs sized to unit-test the entity modules (t10).
|
|
|
|
Import-only stubs: MediaPlayerEntity/RemoteEntity bases, the feature/state
|
|
enum values the integration actually uses, and the event helpers. Tests load
|
|
their entity module through the synthetic ``nsh`` package AFTER calling
|
|
:func:`install` (and the per-file fixture pops everything at teardown).
|
|
"""
|
|
|
|
import asyncio
|
|
import enum
|
|
import sys
|
|
import types
|
|
|
|
import voluptuous_mirror
|
|
|
|
ENTITIES_STUB_NAMES = (
|
|
"voluptuous",
|
|
"homeassistant",
|
|
"homeassistant.config_entries",
|
|
"homeassistant.exceptions",
|
|
"homeassistant.core",
|
|
"homeassistant.const",
|
|
"homeassistant.components",
|
|
"homeassistant.components.media_player",
|
|
"homeassistant.components.remote",
|
|
"homeassistant.components.button",
|
|
"homeassistant.helpers",
|
|
"homeassistant.helpers.entity_registry",
|
|
"homeassistant.helpers.device_registry",
|
|
"homeassistant.helpers.selector",
|
|
"homeassistant.helpers.event",
|
|
"homeassistant.helpers.entity",
|
|
"homeassistant.helpers.entity_platform",
|
|
"nsh.helpers",
|
|
"nsh.power",
|
|
"nsh.jellyfin",
|
|
"nsh.media_player",
|
|
"nsh.diagnostics",
|
|
"nsh.remote",
|
|
)
|
|
|
|
|
|
class MediaPlayerState(enum.StrEnum):
|
|
OFF = "off"
|
|
ON = "on"
|
|
IDLE = "idle"
|
|
PLAYING = "playing"
|
|
PAUSED = "paused"
|
|
|
|
|
|
class MediaPlayerEntityFeature(enum.IntFlag):
|
|
PLAY = 1
|
|
PAUSE = 2
|
|
STOP = 4
|
|
VOLUME_MUTE = 8
|
|
VOLUME_STEP = 16
|
|
VOLUME_SET = 32
|
|
BROWSE_MEDIA = 4096
|
|
TURN_ON = 64
|
|
TURN_OFF = 128
|
|
PLAY_MEDIA = 256
|
|
NEXT_TRACK = 512
|
|
PREVIOUS_TRACK = 1024
|
|
|
|
|
|
class MediaClass:
|
|
"""homeassistant.components.media_player.MediaClass (values we use)."""
|
|
|
|
APP = "app"
|
|
DIRECTORY = "directory"
|
|
|
|
|
|
class MediaType:
|
|
"""homeassistant.components.media_player.MediaType (values we use)."""
|
|
|
|
APP = "app"
|
|
APPS = "apps"
|
|
|
|
|
|
class BrowseMedia:
|
|
"""The REAL media_player.BrowseMedia shape (t37 D1).
|
|
|
|
Keyword-only with the six arguments real HA requires, the optional ones it
|
|
accepts, a faithful ``as_dict`` and — crucially — NO ``domain`` parameter:
|
|
``BrowseMediaSource`` lives in
|
|
``homeassistant.components.media_source.models`` (its contract is the
|
|
``media_source://<domain>`` id) and is NOT importable from
|
|
``homeassistant.components.media_player`` at all. The old import and the
|
|
old ``domain=`` kwarg therefore fail IN-SUITE instead of hiding behind a
|
|
permissive stub.
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
*,
|
|
media_class,
|
|
media_content_id,
|
|
media_content_type,
|
|
title,
|
|
can_play,
|
|
can_expand,
|
|
children=None,
|
|
children_media_class=None,
|
|
thumbnail=None,
|
|
not_shown=0,
|
|
can_search=False,
|
|
search_media_classes=None,
|
|
):
|
|
self.media_class = media_class
|
|
self.media_content_id = media_content_id
|
|
self.media_content_type = media_content_type
|
|
self.title = title
|
|
self.can_play = can_play
|
|
self.can_expand = can_expand
|
|
self.children = children
|
|
self.children_media_class = children_media_class
|
|
self.thumbnail = thumbnail
|
|
self.not_shown = not_shown
|
|
self.can_search = can_search
|
|
self.search_media_classes = search_media_classes
|
|
|
|
def as_dict(self, *, parent=True):
|
|
"""Real payload conversion — what the frontend actually consumes."""
|
|
if self.children_media_class is None and self.children:
|
|
self.calculate_children_class()
|
|
response = {
|
|
"title": self.title,
|
|
"media_class": self.media_class,
|
|
"media_content_type": self.media_content_type,
|
|
"media_content_id": self.media_content_id,
|
|
"children_media_class": self.children_media_class,
|
|
"can_play": self.can_play,
|
|
"can_expand": self.can_expand,
|
|
"can_search": self.can_search,
|
|
"search_media_classes": self.search_media_classes,
|
|
"thumbnail": self.thumbnail,
|
|
}
|
|
if not parent:
|
|
return response
|
|
response["not_shown"] = self.not_shown
|
|
response["children"] = (
|
|
[child.as_dict(parent=False) for child in self.children]
|
|
if self.children
|
|
else []
|
|
)
|
|
return response
|
|
|
|
def calculate_children_class(self):
|
|
"""Real helper: the children's common class, else DIRECTORY."""
|
|
self.children_media_class = MediaClass.DIRECTORY
|
|
assert self.children is not None
|
|
proposed_class = self.children[0].media_class
|
|
if all(child.media_class == proposed_class for child in self.children):
|
|
self.children_media_class = proposed_class
|
|
|
|
|
|
class MediaEntityManager:
|
|
"""trivial stand-in for the HA platform internals we don't exercise."""
|
|
|
|
|
|
class MediaPlayerEntity:
|
|
"""Base entity: attributes only; state writing is a no-op."""
|
|
|
|
def async_write_ha_state(self) -> None:
|
|
return None
|
|
|
|
|
|
class RemoteEntity:
|
|
def async_write_ha_state(self) -> None:
|
|
return None
|
|
|
|
|
|
def install() -> None:
|
|
"""Register the stub modules (idempotent)."""
|
|
if "homeassistant.components.media_player" in sys.modules:
|
|
return
|
|
|
|
vol = types.ModuleType("voluptuous")
|
|
# t33 F-A: the faithful mirror keeps schema defaults visible — a bare-key
|
|
# stub cannot tell a schema WITH a default from one without.
|
|
voluptuous_mirror.install(vol)
|
|
sys.modules["voluptuous"] = vol
|
|
|
|
ha = types.ModuleType("homeassistant")
|
|
sys.modules["homeassistant"] = ha
|
|
|
|
config_entries = types.ModuleType("homeassistant.config_entries")
|
|
config_entries.ConfigEntry = type("ConfigEntry", (), {})
|
|
async def _stub_set_unique_id(self, unique_id):
|
|
# hub shape: record; a full flow test file scopes its own version
|
|
self._unique_id = unique_id
|
|
|
|
def _noop_abort(self, *, error=None, **kwargs):
|
|
return None
|
|
|
|
class _ConfigFlowStub:
|
|
"""ConfigFlow shape used by entity tests (t29 G3a API included).
|
|
|
|
Late loads under the hub stubs (e.g. test_init designs run AFTER the
|
|
other flow files) still get the base plumbing the flow modules rely
|
|
on — show/create/abort/unique-id — so the flow classes never import
|
|
against a bare base.
|
|
"""
|
|
|
|
async_set_unique_id = _stub_set_unique_id
|
|
_abort_if_unique_id_configured = _noop_abort
|
|
|
|
def _async_handle_discovery_without_unique_id(self):
|
|
# t32 F2 (audited helper): recording no-op at hub scope
|
|
pass
|
|
|
|
def async_show_form(
|
|
self, *, step_id, data_schema=None, errors=None,
|
|
description_placeholders=None,
|
|
):
|
|
return {
|
|
"type": "form",
|
|
"step_id": step_id,
|
|
"errors": dict(errors or {}),
|
|
"data_schema": data_schema,
|
|
"description_placeholders": dict(description_placeholders or {}),
|
|
}
|
|
|
|
def async_create_entry(self, *, title, data, options=None):
|
|
return {"type": "create_entry", "title": title, "data": data}
|
|
|
|
def async_abort(self, *, reason):
|
|
return {"type": "abort", "reason": reason}
|
|
|
|
def _get_reauth_entry(self, entry=None):
|
|
if entry is not None:
|
|
return entry
|
|
return None # the caller's cached entry is used
|
|
|
|
def _get_reconfigure_entry(self, entry=None):
|
|
if entry is not None:
|
|
return entry
|
|
return None # the caller's cached entry is used
|
|
|
|
# NOTE: the real HA signature is async_update_reload_and_abort(entry, ...)
|
|
# which updates, schedules the reload and aborts; the stub records both.
|
|
|
|
def async_update_reload_and_abort(self, entry, **kwargs):
|
|
"""Real-HA semantics: update, schedule the reload, abort."""
|
|
updates = kwargs.get("options_updates") or kwargs.get("data_updates") or {}
|
|
for key, value in updates.items():
|
|
if key in ("options", "data", "unique_id"):
|
|
setattr(entry, key, value)
|
|
else:
|
|
getattr(entry, "options", {}).setdefault(key, value) # best effort
|
|
scheduler = getattr(self.hass, "config_entries", None)
|
|
if scheduler is not None and hasattr(
|
|
scheduler, "async_schedule_reload"
|
|
):
|
|
scheduler.async_schedule_reload(entry.entry_id)
|
|
return self.async_abort(reason="reauth_successful")
|
|
|
|
def _config_flow_init(self, *args, **kwargs):
|
|
self.context = getattr(self, "context", {}) or {}
|
|
self.hass = getattr(self, "hass", None)
|
|
|
|
_config_flow_base = type(
|
|
"ConfigFlow",
|
|
(_ConfigFlowStub,),
|
|
{
|
|
"__init_subclass__": classmethod(
|
|
lambda cls, domain=None, **kwargs: None
|
|
),
|
|
"__init__": _config_flow_init,
|
|
},
|
|
)
|
|
config_entries.ConfigFlow = _config_flow_base
|
|
class _OptionsFlowStub:
|
|
def __init__(self):
|
|
self.config_entry = None
|
|
|
|
def add_suggested_values_to_schema(self, schema, suggested):
|
|
# real HA merges hints; the stub records them on the schema dict
|
|
schema_dict = getattr(schema, "schema", None) or (
|
|
schema if isinstance(schema, dict) else {}
|
|
)
|
|
schema_dict.setdefault("__suggested__", dict(suggested))
|
|
return schema
|
|
|
|
class _OptionsFlowWithReload(_OptionsFlowStub):
|
|
"""Options flow that reloads after a successful save (real HA)."""
|
|
|
|
def async_create_entry(self, *, title, data, options=None):
|
|
entry_id = getattr(self.config_entry, "entry_id", None)
|
|
scheduler = getattr(getattr(self, "hass", None), "config_entries", None)
|
|
if entry_id is not None and scheduler is not None and hasattr(
|
|
scheduler, "async_schedule_reload"
|
|
):
|
|
scheduler.async_schedule_reload(entry_id)
|
|
return {"type": "create_entry", "title": title, "data": data}
|
|
|
|
config_entries.OptionsFlow = _OptionsFlowStub
|
|
config_entries.SOURCE_IMPORT = "import"
|
|
config_entries.OptionsFlowWithReload = _OptionsFlowWithReload
|
|
sys.modules["homeassistant.config_entries"] = config_entries
|
|
|
|
exceptions = types.ModuleType("homeassistant.exceptions")
|
|
|
|
class ServiceValidationErrorStubError(Exception):
|
|
def __init__(self, *args, **kwargs):
|
|
super().__init__(*args or (kwargs.get("translation_key"),))
|
|
self.translation_domain = kwargs.get("translation_domain")
|
|
self.translation_key = kwargs.get("translation_key")
|
|
|
|
exceptions.ServiceValidationError = ServiceValidationErrorStubError
|
|
sys.modules["homeassistant.exceptions"] = exceptions
|
|
|
|
core = types.ModuleType("homeassistant.core")
|
|
core.HomeAssistant = object
|
|
core.callback = lambda fn: fn
|
|
core.CALLBACK_TYPE = object
|
|
core.Event = object
|
|
sys.modules["homeassistant.core"] = core
|
|
|
|
ha_const = types.ModuleType("homeassistant.const")
|
|
ha_const.EVENT_HOMEASSISTANT_STOP = "homeassistant_stop"
|
|
ha_const.STATE_PLAYING = "playing"
|
|
ha_const.STATE_PAUSED = "paused"
|
|
ha_const.Platform = types.SimpleNamespace(
|
|
MEDIA_PLAYER="media_player",
|
|
SENSOR="sensor",
|
|
REMOTE="remote",
|
|
BUTTON="button",
|
|
)
|
|
sys.modules["homeassistant.const"] = ha_const
|
|
|
|
components = types.ModuleType("homeassistant.components")
|
|
components.__path__ = []
|
|
sys.modules["homeassistant.components"] = components
|
|
|
|
mp = types.ModuleType("homeassistant.components.media_player")
|
|
mp.MediaPlayerEntity = MediaPlayerEntity
|
|
mp.MediaPlayerDeviceClass = types.SimpleNamespace(TV="tv")
|
|
mp.MediaPlayerEntityFeature = MediaPlayerEntityFeature
|
|
mp.BrowseMedia = BrowseMedia
|
|
mp.MediaClass = MediaClass
|
|
mp.MediaType = MediaType
|
|
# NOTE (t37 D1): homeassistant.components.media_player exposes NO
|
|
# BrowseMediaSource — deliberately not stubbed, so the drifted import
|
|
# fails here exactly as it does in a real install.
|
|
mp.MediaPlayerState = MediaPlayerState
|
|
sys.modules["homeassistant.components.media_player"] = mp
|
|
|
|
remote = types.ModuleType("homeassistant.components.remote")
|
|
remote.RemoteEntity = RemoteEntity
|
|
sys.modules["homeassistant.components.remote"] = remote
|
|
|
|
button = types.ModuleType("homeassistant.components.button")
|
|
button.ButtonEntity = type("ButtonEntity", (), {})
|
|
sys.modules["homeassistant.components.button"] = button
|
|
sensor_platform = types.ModuleType("homeassistant.components.sensor")
|
|
sensor_platform.SensorEntity = type("SensorEntity", (), {})
|
|
sensor_platform.SensorDeviceClass = types.SimpleNamespace(APP="app")
|
|
sys.modules["homeassistant.components.sensor"] = sensor_platform
|
|
|
|
|
|
helpers = types.ModuleType("homeassistant.helpers")
|
|
entity_registry = types.ModuleType("homeassistant.helpers.entity_registry")
|
|
|
|
class _Registry:
|
|
entities: dict = {}
|
|
|
|
def async_get(self, entity_id):
|
|
return None
|
|
|
|
entity_registry.async_get = lambda hass: _Registry()
|
|
device_registry = types.ModuleType("homeassistant.helpers.device_registry")
|
|
|
|
class _DeviceRegistry:
|
|
def __init__(self):
|
|
self.devices: dict[str, object] = {}
|
|
|
|
def async_get(self, device_id):
|
|
return self.devices.get(device_id)
|
|
|
|
_device_registry_shared = _DeviceRegistry()
|
|
|
|
device_registry.async_get = lambda hass: _device_registry_shared
|
|
sys.modules["homeassistant.helpers.device_registry"] = device_registry
|
|
|
|
helpers.entity_registry = entity_registry
|
|
sys.modules["homeassistant.helpers.entity_registry"] = entity_registry
|
|
|
|
helpers.device_registry = device_registry
|
|
sys.modules["homeassistant.helpers.device_registry"] = device_registry
|
|
|
|
event = types.ModuleType("homeassistant.helpers.event")
|
|
event.async_call_later = lambda hass, delay, callback_fn: (lambda: None)
|
|
event.async_track_state_change_event = lambda *args, **kwargs: (lambda: None)
|
|
helpers.event = event
|
|
helpers.selector = types.SimpleNamespace(
|
|
EntitySelector=lambda *a, **k: None,
|
|
EntitySelectorConfig=lambda *a, **k: "",
|
|
SelectSelector=lambda *a, **k: None,
|
|
SelectSelectorConfig=lambda **k: "",
|
|
SelectSelectorMode=types.SimpleNamespace(DROPDOWN="dropdown"),
|
|
)
|
|
entity = types.ModuleType("homeassistant.helpers.entity")
|
|
entity.DeviceInfo = dict
|
|
entity.EntityCategory = types.SimpleNamespace(
|
|
CONFIG="config", DIAGNOSTIC="diagnostic"
|
|
)
|
|
helpers.entity = entity
|
|
entity_platform = types.ModuleType("homeassistant.helpers.entity_platform")
|
|
entity_platform.AddEntitiesCallback = object
|
|
helpers.entity_platform = entity_platform
|
|
|
|
sys.modules["homeassistant.helpers.event"] = event
|
|
sys.modules["homeassistant.helpers.entity"] = entity
|
|
sys.modules["homeassistant.helpers.entity_platform"] = entity_platform
|
|
sys.modules["homeassistant.helpers"] = helpers
|
|
|
|
|
|
def load_entity_modules() -> dict:
|
|
"""Load the integration's HA-free-ish entity modules under nsh."""
|
|
install()
|
|
from conftest import _SHIELD_DIR, _load_module
|
|
|
|
for name in (
|
|
"helpers",
|
|
"power",
|
|
"jellyfin",
|
|
"sensor",
|
|
"media_player",
|
|
"remote",
|
|
"diagnostics",
|
|
"init",
|
|
): # "init" maps the integration __init__.py (t29 G8 async_setup)
|
|
if name == "init" and "nsh.init" in sys.modules:
|
|
continue
|
|
if f"nsh.{name}" not in sys.modules:
|
|
file_name = "__init__.py" if name == "init" else f"{name}.py"
|
|
_load_module(f"nsh.{name}", _SHIELD_DIR / file_name)
|
|
return {
|
|
name: sys.modules[f"nsh.{name}"]
|
|
for name in (
|
|
"power",
|
|
"jellyfin",
|
|
"sensor",
|
|
"media_player",
|
|
"remote",
|
|
"diagnostics",
|
|
)
|
|
}
|
|
|
|
|
|
class FakeHass:
|
|
"""Sized to entity unit tests: background tasks + no-op call_later."""
|
|
|
|
def async_create_task(self, coro):
|
|
return asyncio.create_task(coro)
|
|
|
|
def async_call_later(self, delay, callback_fn):
|
|
return lambda: None
|