804 lines
29 KiB
Python
804 lines
29 KiB
Python
"""P3-1 tests: standalone setup — optional androidtv entity + app fallback.
|
|
|
|
The integration no longer hard-requires an external androidtv media_player:
|
|
the protocol current-app push can be the only state source. Covers the flow
|
|
validation (no state source → form error), the standalone pairing route,
|
|
entry data omitting the field, media_player app_id/app_name fallback, and
|
|
migration validity for v2 entries without the field.
|
|
"""
|
|
|
|
import asyncio # noqa: F401 (async steps need the event loop)
|
|
import sys
|
|
import types
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
from conftest import _SHIELD_DIR, _load_module
|
|
from nsh.const import (
|
|
CONF_ANDROIDTV_ENTITY,
|
|
CONF_PROTOCOL_ENABLED,
|
|
CONF_PROTOCOL_HOST,
|
|
)
|
|
|
|
_STUB_NAMES = (
|
|
"voluptuous",
|
|
"homeassistant",
|
|
"homeassistant.config_entries",
|
|
"homeassistant.core",
|
|
"homeassistant.const",
|
|
"homeassistant.helpers",
|
|
"homeassistant.helpers.entity_registry",
|
|
"homeassistant.helpers.selector",
|
|
"nsh.helpers",
|
|
"nsh.config_flow",
|
|
"nsh.entry",
|
|
)
|
|
|
|
|
|
def _install_stubs() -> None:
|
|
"""Import-time stand-ins (idempotent; same shape as the pairing tests)."""
|
|
if sys.modules.get("voluptuous") is not None:
|
|
return
|
|
|
|
vol = types.ModuleType("voluptuous")
|
|
vol.Required = lambda key, default=None: key
|
|
vol.Optional = lambda key, default=None: key
|
|
vol.Schema = lambda schema: schema
|
|
sys.modules["voluptuous"] = vol
|
|
|
|
ha = types.ModuleType("homeassistant")
|
|
sys.modules["homeassistant"] = ha
|
|
|
|
config_entries = types.ModuleType("homeassistant.config_entries")
|
|
|
|
class ConfigFlow:
|
|
VERSION = 1
|
|
|
|
def __init_subclass__(cls, **kwargs):
|
|
kwargs.pop("domain", None)
|
|
super().__init_subclass__(**kwargs)
|
|
|
|
def async_show_form(self, *, step_id, data_schema=None, errors=None):
|
|
return {
|
|
"type": "form",
|
|
"step_id": step_id,
|
|
"errors": dict(errors or {}),
|
|
"data_schema": data_schema,
|
|
}
|
|
|
|
def async_create_entry(self, *, title, data, options=None):
|
|
return {"type": "create_entry", "title": title, "data": data,
|
|
"options": options}
|
|
|
|
def async_abort(self, *, reason):
|
|
return {"type": "abort", "reason": reason}
|
|
|
|
async def async_set_unique_id(self, unique_id):
|
|
self._unique_id = unique_id
|
|
|
|
def _abort_if_unique_id_configured(self, *, error=None, **kwargs):
|
|
return None
|
|
|
|
def _async_current_entries(self):
|
|
return _zc_holder.entries
|
|
|
|
class OptionsFlow:
|
|
def async_show_form(self, *, step_id, data_schema=None, errors=None):
|
|
return {
|
|
"type": "form",
|
|
"step_id": step_id,
|
|
"errors": dict(errors or {}),
|
|
"data_schema": data_schema,
|
|
}
|
|
|
|
def async_create_entry(self, *, title, data):
|
|
return {"type": "create_entry", "title": title, "data": data}
|
|
|
|
# Shared holder so tests can swap the current-entry list through the
|
|
# module object (the stub closure reads this same object).
|
|
_zc_holder = types.SimpleNamespace(entries=[])
|
|
sys.modules[__name__]._zc_holder = _zc_holder
|
|
|
|
class AbortIfConfiguredError(Exception):
|
|
pass
|
|
|
|
config_entries.ConfigFlow = ConfigFlow
|
|
config_entries.OptionsFlow = OptionsFlow
|
|
config_entries.ConfigEntry = type("ConfigEntry", (), {})
|
|
sys.modules["homeassistant.config_entries"] = config_entries
|
|
|
|
core = types.ModuleType("homeassistant.core")
|
|
core.HomeAssistant = object
|
|
core.callback = lambda fn: fn
|
|
sys.modules["homeassistant.core"] = core
|
|
|
|
ha_const = types.ModuleType("homeassistant.const")
|
|
ha_const.EVENT_HOMEASSISTANT_STOP = "homeassistant_stop"
|
|
|
|
class _Platform:
|
|
MEDIA_PLAYER = "media_player"
|
|
SENSOR = "sensor"
|
|
REMOTE = "remote"
|
|
BUTTON = "button"
|
|
|
|
ha_const.Platform = _Platform
|
|
sys.modules["homeassistant.const"] = ha_const
|
|
|
|
helpers = types.ModuleType("homeassistant.helpers")
|
|
entity_registry = types.ModuleType("homeassistant.helpers.entity_registry")
|
|
entity_registry.async_get = lambda hass: SimpleNamespace(entities={})
|
|
helpers.entity_registry = entity_registry
|
|
sys.modules["homeassistant.helpers.entity_registry"] = entity_registry
|
|
sys.modules["homeassistant.helpers"] = helpers
|
|
|
|
selector = types.ModuleType("homeassistant.helpers.selector")
|
|
selector.EntitySelector = lambda *a, **k: None
|
|
selector.EntitySelectorConfig = lambda *a, **k: ""
|
|
selector.SelectSelector = lambda *a, **k: None
|
|
selector.SelectSelectorConfig = lambda **k: ""
|
|
selector.SelectSelectorMode = types.SimpleNamespace(DROPDOWN="dropdown")
|
|
sys.modules["homeassistant.helpers.selector"] = selector
|
|
|
|
|
|
class FakeConfigEntries:
|
|
def __init__(self):
|
|
self.updating_entry_calls: list = []
|
|
self.reloaded_entry_ids: list[str] = []
|
|
|
|
def async_get_entry(self, entry_id):
|
|
return None
|
|
|
|
def async_update_entry(self, entry, **kwargs):
|
|
self.updating_entry_calls.append((entry, kwargs))
|
|
for key, value in kwargs.items():
|
|
if key in ("options", "data", "unique_id"):
|
|
setattr(entry, key, value)
|
|
|
|
def async_schedule_reload(self, entry_id):
|
|
self.reloaded_entry_ids.append(entry_id)
|
|
|
|
|
|
class FakeHass:
|
|
def __init__(self, tmp_dir: str, states: dict | None = None):
|
|
self.config = SimpleNamespace(config_dir=tmp_dir)
|
|
self.config_entries = FakeConfigEntries()
|
|
self._states = states or {}
|
|
|
|
class _States:
|
|
def get(self, entity_id):
|
|
return self._states.get(entity_id)
|
|
|
|
self._states_ref = _States()
|
|
self._states_ref._states = self._states
|
|
self.states = self._states_ref
|
|
|
|
@staticmethod
|
|
async def async_add_executor_job(fn, *args):
|
|
return fn(*args)
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def scoped_stubs():
|
|
"""Keep the import stubs inside this module's tests (t7 discipline)."""
|
|
snapshot = {name: sys.modules.get(name) for name in _STUB_NAMES}
|
|
_install_stubs()
|
|
yield
|
|
for name in _STUB_NAMES:
|
|
original = snapshot[name]
|
|
if original is None:
|
|
sys.modules.pop(name, None)
|
|
else:
|
|
sys.modules[name] = original
|
|
|
|
|
|
def _form_module():
|
|
_install_stubs()
|
|
if "nsh.helpers" not in sys.modules:
|
|
_load_module("nsh.helpers", _SHIELD_DIR / "helpers.py")
|
|
if "nsh.config_flow" not in sys.modules:
|
|
_load_module("nsh.config_flow", _SHIELD_DIR / "config_flow.py")
|
|
return sys.modules["nsh.config_flow"]
|
|
|
|
|
|
def _entry_module():
|
|
_install_stubs()
|
|
if "nsh.entry" not in sys.modules:
|
|
_load_module("nsh.entry", _SHIELD_DIR / "__init__.py")
|
|
return sys.modules["nsh.entry"]
|
|
|
|
|
|
def _user_input(**overrides) -> dict:
|
|
base = {
|
|
"name": "mdarcy",
|
|
CONF_ANDROIDTV_ENTITY: "media_player.shield",
|
|
CONF_PROTOCOL_ENABLED: False,
|
|
CONF_PROTOCOL_HOST: "",
|
|
}
|
|
base.update(overrides)
|
|
return base
|
|
|
|
|
|
class TestStandaloneValidation:
|
|
async def test_neither_source_shows_form_error(self, tmp_path):
|
|
module = _form_module()
|
|
flow = module.NvidiaShieldConfigFlow()
|
|
flow.hass = FakeHass(str(tmp_path))
|
|
|
|
result = await flow.async_step_user(_user_input(**{CONF_ANDROIDTV_ENTITY: ""}))
|
|
|
|
assert result["type"] == "form"
|
|
assert result["step_id"] == "user"
|
|
assert result["errors"] == {"base": "no_state_source"}
|
|
|
|
async def test_androidtv_entity_alone_is_a_valid_source(self, tmp_path):
|
|
module = _form_module()
|
|
flow = module.NvidiaShieldConfigFlow()
|
|
flow.hass = FakeHass(str(tmp_path))
|
|
|
|
result = await flow.async_step_user(_user_input())
|
|
|
|
assert result["type"] == "create_entry"
|
|
assert result["data"] == {CONF_ANDROIDTV_ENTITY: "media_player.shield"}
|
|
|
|
async def test_standalone_protocol_goes_to_pairing(self, tmp_path, monkeypatch):
|
|
module = _form_module()
|
|
flow = module.NvidiaShieldConfigFlow()
|
|
flow.hass = FakeHass(str(tmp_path))
|
|
flow._user_data = {}
|
|
from nsh.connection import ShieldConnectionManager
|
|
|
|
class _Client:
|
|
async def async_generate_cert_if_missing(self):
|
|
return True
|
|
|
|
async def async_start_pairing(self):
|
|
return "8421"
|
|
|
|
monkeypatch.setattr(
|
|
ShieldConnectionManager, "build_protocol_client",
|
|
lambda *a, **k: _Client(),
|
|
)
|
|
monkeypatch.setattr(
|
|
module, "protocol_cert_paths", lambda hass: ("/tmp/a.pem", "/tmp/a.key")
|
|
)
|
|
|
|
result = await flow.async_step_user(
|
|
_user_input(**{CONF_ANDROIDTV_ENTITY: "", CONF_PROTOCOL_ENABLED: True})
|
|
)
|
|
|
|
assert result["type"] == "form"
|
|
assert result["step_id"] == "protocol_pair" # routed into pairing, no error
|
|
|
|
async def test_standalone_entry_data_omits_the_field(self, tmp_path, monkeypatch):
|
|
"""No androidtv entity → the key is absent from entry data (not "")."""
|
|
module = _form_module()
|
|
flow = module.NvidiaShieldConfigFlow()
|
|
flow.hass = FakeHass(str(tmp_path))
|
|
flow._user_data = {
|
|
"name": "mdarcy",
|
|
CONF_ANDROIDTV_ENTITY: "",
|
|
CONF_PROTOCOL_ENABLED: True,
|
|
CONF_PROTOCOL_HOST: "shield.local",
|
|
}
|
|
|
|
# skip the pairing route: drive _create_entry directly
|
|
result = flow._create_entry()
|
|
|
|
assert result["type"] == "create_entry"
|
|
assert CONF_ANDROIDTV_ENTITY not in result["data"]
|
|
|
|
|
|
class TestAdbOnlyStandalone:
|
|
async def test_adb_only_standalone_is_valid(self, tmp_path):
|
|
"""F2: ADB media-session mode is a third accepted state source."""
|
|
module = _form_module()
|
|
flow = module.NvidiaShieldConfigFlow()
|
|
flow.hass = FakeHass(str(tmp_path))
|
|
|
|
result = await flow.async_step_user(
|
|
_user_input(
|
|
**{
|
|
CONF_ANDROIDTV_ENTITY: "",
|
|
CONF_PROTOCOL_ENABLED: False,
|
|
"adb_enabled": True,
|
|
}
|
|
)
|
|
)
|
|
|
|
# routed into ADB pairing — no no_state_source error
|
|
assert result["type"] == "form"
|
|
assert result["step_id"] == "adb_pair"
|
|
assert result["errors"] == {}
|
|
|
|
async def test_error_still_fires_with_no_source_at_all(self, tmp_path):
|
|
module = _form_module()
|
|
flow = module.NvidiaShieldConfigFlow()
|
|
flow.hass = FakeHass(str(tmp_path))
|
|
|
|
result = await flow.async_step_user(
|
|
_user_input(**{CONF_ANDROIDTV_ENTITY: "", "adb_enabled": False})
|
|
)
|
|
|
|
assert result["errors"] == {"base": "no_state_source"}
|
|
|
|
|
|
class TestMigrationKeepsOrAllowsAbsence:
|
|
async def test_v1_to_v2_keeps_the_androidtv_field(self, tmp_path):
|
|
module = _entry_module()
|
|
hass = FakeHass(str(tmp_path))
|
|
entry = SimpleNamespace(
|
|
version=1,
|
|
data={CONF_ANDROIDTV_ENTITY: "media_player.shield"},
|
|
options={},
|
|
title="mdarcy",
|
|
)
|
|
|
|
ok = await module.async_migrate_entry(hass, entry)
|
|
|
|
assert ok is True
|
|
assert entry.data[CONF_ANDROIDTV_ENTITY] == "media_player.shield" # kept
|
|
# options were rewritten with the protocol defaults; the data dict —
|
|
# including the androidtv field — was never touched
|
|
assert hass.config_entries.updating_entry_calls == [
|
|
(
|
|
entry,
|
|
{
|
|
"options": {CONF_PROTOCOL_ENABLED: False, CONF_PROTOCOL_HOST: ""},
|
|
"version": 2,
|
|
},
|
|
)
|
|
]
|
|
|
|
async def test_v2_without_the_field_is_valid(self, tmp_path):
|
|
"""Standalone entries born v2 must pass through migration untouched."""
|
|
module = _entry_module()
|
|
hass = FakeHass(str(tmp_path))
|
|
entry = SimpleNamespace(
|
|
version=2,
|
|
data={},
|
|
options={CONF_PROTOCOL_ENABLED: True, CONF_PROTOCOL_HOST: "shield.local"},
|
|
title="mdarcy",
|
|
)
|
|
|
|
ok = await module.async_migrate_entry(hass, entry)
|
|
|
|
assert ok is True
|
|
assert hass.config_entries.updating_entry_calls == [] # no rewrite needed
|
|
|
|
|
|
def _cmd_player(modules):
|
|
"""Standalone media_player entity via the command-routing harness."""
|
|
import test_command_routing as routing
|
|
|
|
player, _served, _intents = routing.make_player(modules)
|
|
return player, None, None
|
|
|
|
|
|
class TestAppIconColumn:
|
|
"""P5-2: the editor's optional icon column (package=name|icon)."""
|
|
|
|
async def test_two_column_line_stores_the_icon(self, tmp_path):
|
|
module = _form_module()
|
|
entry = SimpleNamespace(options={}, title="Shield")
|
|
flow = module.NvidiaShieldOptionsFlow(entry)
|
|
flow.hass = FakeHass(str(tmp_path))
|
|
flow.async_show_form = lambda **kwargs: {"type": "form", **kwargs}
|
|
flow.async_create_entry = lambda **kwargs: {"type": "create_entry", **kwargs}
|
|
|
|
await flow.async_step_init({"app_name_map": "ar.tvplayer.tv=TiviMate|http://i.local/t.png"})
|
|
|
|
assert flow._options["app_name_map"] == {"ar.tvplayer.tv": "TiviMate"}
|
|
assert flow._options["app_icon_map"] == {"ar.tvplayer.tv": "http://i.local/t.png"}
|
|
|
|
async def test_single_column_map_stays_valid(self, tmp_path):
|
|
module = _form_module()
|
|
entry = SimpleNamespace(options={}, title="Shield")
|
|
flow = module.NvidiaShieldOptionsFlow(entry)
|
|
flow.hass = FakeHass(str(tmp_path))
|
|
flow.async_show_form = lambda **kwargs: {"type": "form", **kwargs}
|
|
flow.async_create_entry = lambda **kwargs: {"type": "create_entry", **kwargs}
|
|
|
|
await flow.async_step_init({"app_name_map": "ar.tvplayer.tv=TiviMate"})
|
|
|
|
assert flow._options["app_name_map"] == {"ar.tvplayer.tv": "TiviMate"}
|
|
assert flow._options["app_icon_map"] == {} # blank-safe (no icons)
|
|
|
|
async def test_prefill_round_trips_the_icon_column(self):
|
|
module = _form_module()
|
|
|
|
# the serializer writes the second column through on prefill
|
|
text = module._app_map_to_text(
|
|
{"ar.tvplayer.tv": "TiviMate"},
|
|
{"ar.tvplayer.tv": "http://i.local/t.png"},
|
|
)
|
|
assert text == "ar.tvplayer.tv=TiviMate|http://i.local/t.png"
|
|
|
|
# and the parser round-trips it back
|
|
icons: dict[str, str] = {}
|
|
app_map = module._app_map_from_text(text, icons)
|
|
assert app_map == {"ar.tvplayer.tv": "TiviMate"}
|
|
assert icons == {"ar.tvplayer.tv": "http://i.local/t.png"}
|
|
|
|
async def test_malformed_icon_invalidates_the_row(self, tmp_path):
|
|
module = _form_module()
|
|
entry = SimpleNamespace(options={}, title="Shield")
|
|
flow = module.NvidiaShieldOptionsFlow(entry)
|
|
flow.hass = FakeHass(str(tmp_path))
|
|
flow.async_show_form = lambda **kwargs: {"type": "form", **kwargs}
|
|
flow.async_create_entry = lambda **kwargs: {"type": "create_entry", **kwargs}
|
|
|
|
result = await flow.async_step_init({"app_name_map": "p=Name|has spaces"})
|
|
|
|
assert result["type"] == "form"
|
|
assert result["errors"] == {"base": "invalid_app_map"}
|
|
|
|
|
|
class TestChannelEntryOptions:
|
|
"""P5-2: digit-delay + optional send-enter on the options page."""
|
|
|
|
def _options_flow(self, tmp_path, options):
|
|
from types import SimpleNamespace as DefaultEntry
|
|
|
|
module = _form_module()
|
|
entry = DefaultEntry(options=dict(options) if options else {})
|
|
flow = module.NvidiaShieldOptionsFlow(entry)
|
|
hass = FakeHass(str(tmp_path))
|
|
flow.hass = hass
|
|
flow._options = dict(options) if options else {}
|
|
return module, flow
|
|
|
|
async def test_digit_delay_and_send_enter_store(self, tmp_path):
|
|
module, flow = self._options_flow(tmp_path, options={})
|
|
|
|
flow.async_show_form = lambda **kwargs: {"type": "form", **kwargs}
|
|
flow.async_create_entry = lambda **kwargs: {"type": "create_entry", **kwargs}
|
|
|
|
await flow.async_step_init({"channel_digit_delay": "0.3", "channel_send_enter": True})
|
|
|
|
assert flow._options["channel_digit_delay"] == 0.3
|
|
assert flow._options["channel_send_enter"] is True
|
|
|
|
async def test_digit_delay_out_of_window_rejects(self, tmp_path):
|
|
module, flow = self._options_flow(tmp_path, options={})
|
|
|
|
flow.async_show_form = lambda **kwargs: {"type": "form", **kwargs}
|
|
flow.async_create_entry = lambda **kwargs: {"type": "create_entry", **kwargs}
|
|
|
|
result = await flow.async_step_init({"channel_digit_delay": "2.0"})
|
|
|
|
assert result["type"] == "form"
|
|
assert result["errors"] == {"base": "invalid_digit_delay"}
|
|
|
|
async def test_digit_delay_not_a_number_rejects(self, tmp_path):
|
|
module, flow = self._options_flow(tmp_path, options={})
|
|
|
|
flow.async_show_form = lambda **kwargs: {"type": "form", **kwargs}
|
|
flow.async_create_entry = lambda **kwargs: {"type": "create_entry", **kwargs}
|
|
|
|
result = await flow.async_step_init({"channel_digit_delay": "slow"})
|
|
|
|
assert result["type"] == "form"
|
|
assert result["errors"] == {"base": "invalid_digit_delay"}
|
|
|
|
|
|
class TestLegacyUniqueIdMigration:
|
|
"""P5-1: one-time legacy androidtv-entity-id unique_id migration."""
|
|
|
|
async def test_migrates_to_entry_id_without_protocol(self, tmp_path):
|
|
import entity_stublib
|
|
|
|
entity_stublib.load_entity_modules()
|
|
modules = sys.modules["nsh.media_player"]
|
|
from types import SimpleNamespace as MigrationEntry
|
|
|
|
hass = FakeHass(str(tmp_path))
|
|
entry = MigrationEntry(
|
|
entry_id="cfgid123",
|
|
data={CONF_ANDROIDTV_ENTITY: "media_player.shield"},
|
|
options={},
|
|
title="Shield",
|
|
unique_id="media_player.shield", # legacy
|
|
runtime_data=None,
|
|
)
|
|
|
|
await modules._migrate_legacy_unique_id(hass, entry)
|
|
|
|
assert entry.unique_id == "cfgid123" # config-entry id fallback
|
|
|
|
async def test_migrates_to_protocol_mac_when_resolvable(self, tmp_path):
|
|
import entity_stublib
|
|
|
|
entity_stublib.load_entity_modules()
|
|
modules = sys.modules["nsh.media_player"]
|
|
from types import SimpleNamespace as MigrationEntry
|
|
|
|
hass = FakeHass(str(tmp_path))
|
|
entry = MigrationEntry(
|
|
entry_id="cfgid123",
|
|
data={CONF_ANDROIDTV_ENTITY: "media_player.shield"},
|
|
options={},
|
|
title="Shield",
|
|
unique_id="media_player.shield",
|
|
runtime_data=None,
|
|
)
|
|
|
|
class _Manager:
|
|
async def async_protocol_device_info(self):
|
|
return "Shield TV", "aa:bb:cc:dd:ee:ff"
|
|
|
|
entry.runtime_data = _Manager()
|
|
|
|
await modules._migrate_legacy_unique_id(hass, entry)
|
|
|
|
assert entry.unique_id == "aa:bb:cc:dd:ee:ff" # MAC preferred
|
|
|
|
async def test_never_runs_twice_for_a_migrated_entry(self, tmp_path):
|
|
"""The legacy pattern no longer matches after the one-time move."""
|
|
import entity_stublib
|
|
|
|
entity_stublib.load_entity_modules()
|
|
modules = sys.modules["nsh.media_player"]
|
|
from types import SimpleNamespace as MigrationEntry
|
|
|
|
hass = FakeHass(str(tmp_path))
|
|
entry = MigrationEntry(
|
|
entry_id="cfgid123",
|
|
data={CONF_ANDROIDTV_ENTITY: "media_player.shield"},
|
|
options={},
|
|
title="Shield",
|
|
unique_id="aa:bb:cc:dd:ee:ff", # already migrated
|
|
runtime_data=None,
|
|
)
|
|
|
|
await modules._migrate_legacy_unique_id(hass, entry)
|
|
|
|
assert hass.config_entries.updating_entry_calls == [] # untouched
|
|
|
|
|
|
class TestZeroconfDiscovery:
|
|
"""P4-1a: _androidtvremote2._tcp discovery → pre-filled user step."""
|
|
|
|
def _discovery(self, host="192.168.1.50", props=None):
|
|
return SimpleNamespace(
|
|
ip_address=host, host=host, properties=props or {}
|
|
)
|
|
|
|
async def test_discovery_prefills_the_user_step(self, tmp_path):
|
|
module = _form_module()
|
|
flow = module.NvidiaShieldConfigFlow()
|
|
flow.hass = FakeHass(str(tmp_path))
|
|
|
|
result = await flow.async_step_zeroconf(self._discovery())
|
|
|
|
assert result["type"] == "form"
|
|
assert result["step_id"] == "user"
|
|
# pre-fill landed: the discovered host is recorded for the schema
|
|
assert flow._discovered_host == "192.168.1.50"
|
|
assert result["data_schema"] is not None
|
|
|
|
async def test_discovery_without_address_aborts(self, tmp_path):
|
|
module = _form_module()
|
|
flow = module.NvidiaShieldConfigFlow()
|
|
flow.hass = FakeHass(str(tmp_path))
|
|
|
|
result = await flow.async_step_zeroconf(
|
|
SimpleNamespace(ip_address=None, host=None, properties={})
|
|
)
|
|
|
|
assert result == {"type": "abort", "reason": "zeroconf_no_host"}
|
|
|
|
async def test_known_mac_ip_change_updates_entry_hosts(self, tmp_path):
|
|
module = _form_module()
|
|
flow = module.NvidiaShieldConfigFlow()
|
|
hass = FakeHass(str(tmp_path))
|
|
flow.hass = hass
|
|
|
|
entry = SimpleNamespace(
|
|
unique_id="aa:bb:cc:dd:ee:ff",
|
|
options={CONF_PROTOCOL_ENABLED: True, CONF_PROTOCOL_HOST: "10.0.0.1"},
|
|
entry_id="e1",
|
|
title="Shield",
|
|
)
|
|
sys.modules[__name__]._zc_holder.entries[:] = [entry]
|
|
|
|
result = await flow.async_step_zeroconf(
|
|
self._discovery(props={"mac": "aa:bb:cc:dd:ee:ff"})
|
|
)
|
|
|
|
assert result == {"type": "abort", "reason": "already_configured"}
|
|
# the hosts moved to the discovery address
|
|
assert entry.options[CONF_PROTOCOL_HOST] == "192.168.1.50"
|
|
assert entry.options["adb_host"] == "192.168.1.50"
|
|
assert hass.config_entries.reloaded_entry_ids == ["e1"]
|
|
|
|
async def test_known_mac_same_ip_skips_the_rewrite(self, tmp_path):
|
|
module = _form_module()
|
|
flow = module.NvidiaShieldConfigFlow()
|
|
hass = FakeHass(str(tmp_path))
|
|
flow.hass = hass
|
|
|
|
# both hosts already hold the discovered address: nothing changes
|
|
entry = SimpleNamespace(
|
|
unique_id="aa:bb:cc:dd:ee:ff",
|
|
options={
|
|
CONF_PROTOCOL_ENABLED: True,
|
|
CONF_PROTOCOL_HOST: "192.168.1.50",
|
|
"adb_host": "192.168.1.50",
|
|
},
|
|
entry_id="e1",
|
|
title="Shield",
|
|
)
|
|
sys.modules[__name__]._zc_holder.entries[:] = [entry]
|
|
|
|
result = await flow.async_step_zeroconf(
|
|
self._discovery(props={"mac": "aa:bb:cc:dd:ee:ff"})
|
|
)
|
|
|
|
assert result == {"type": "abort", "reason": "already_configured"}
|
|
assert hass.config_entries.updating_entry_calls == [] # no IP change
|
|
assert hass.config_entries.reloaded_entry_ids == []
|
|
|
|
async def test_blank_adb_host_self_heals_from_the_discovery_address(self, tmp_path):
|
|
"""F2: a blank (auto-detect) adb host follows a same-IP rediscovery."""
|
|
module = _form_module()
|
|
flow = module.NvidiaShieldConfigFlow()
|
|
hass = FakeHass(str(tmp_path))
|
|
flow.hass = hass
|
|
|
|
entry = SimpleNamespace(
|
|
unique_id="aa:bb:cc:dd:ee:ff",
|
|
options={CONF_PROTOCOL_ENABLED: True, CONF_PROTOCOL_HOST: "192.168.1.50"},
|
|
entry_id="e1",
|
|
title="Shield",
|
|
)
|
|
sys.modules[__name__]._zc_holder.entries[:] = [entry]
|
|
|
|
result = await flow.async_step_zeroconf(
|
|
self._discovery(props={"mac": "aa:bb:cc:dd:ee:ff"})
|
|
)
|
|
|
|
assert result == {"type": "abort", "reason": "already_configured"}
|
|
# only the adb host was filled (self-heal), protocol already current:
|
|
update = hass.config_entries.updating_entry_calls[0][1]["options"]
|
|
assert update["adb_host"] == "192.168.1.50"
|
|
assert update["protocol_host"] == "192.168.1.50"
|
|
assert hass.config_entries.reloaded_entry_ids == ["e1"]
|
|
|
|
async def test_deliberately_different_adb_host_is_never_clobbered(self, tmp_path):
|
|
"""F2: an explicit ADB override survives a protocol-host IP change."""
|
|
module = _form_module()
|
|
flow = module.NvidiaShieldConfigFlow()
|
|
hass = FakeHass(str(tmp_path))
|
|
flow.hass = hass
|
|
|
|
entry = SimpleNamespace(
|
|
unique_id="aa:bb:cc:dd:ee:ff",
|
|
options={
|
|
CONF_PROTOCOL_ENABLED: True,
|
|
CONF_PROTOCOL_HOST: "10.0.0.1", # stale (IP moved)
|
|
"adb_host": "10.0.0.9", # deliberate ADB override
|
|
},
|
|
entry_id="e1",
|
|
title="Shield",
|
|
)
|
|
sys.modules[__name__]._zc_holder.entries[:] = [entry]
|
|
|
|
result = await flow.async_step_zeroconf(
|
|
self._discovery(props={"mac": "aa:bb:cc:dd:ee:ff"})
|
|
)
|
|
|
|
assert result == {"type": "abort", "reason": "already_configured"}
|
|
update = hass.config_entries.updating_entry_calls[0][1]["options"]
|
|
assert update["protocol_host"] == "192.168.1.50" # protocol followed
|
|
assert update["adb_host"] == "10.0.0.9" # override untouched
|
|
assert hass.config_entries.reloaded_entry_ids == ["e1"]
|
|
|
|
|
|
class TestAppsMapOptionsEditor:
|
|
"""P4-1b: the apps name/icon map editor in the options flow."""
|
|
|
|
def _options_flow(self, tmp_path, options):
|
|
module = _form_module()
|
|
entry = SimpleNamespace(options=dict(options), title="Shield")
|
|
flow = module.NvidiaShieldOptionsFlow(entry)
|
|
flow.hass = FakeHass(str(tmp_path))
|
|
return flow
|
|
|
|
async def test_map_text_stored_as_dict(self, tmp_path):
|
|
flow = self._options_flow(tmp_path, options={})
|
|
|
|
result = await flow.async_step_init(
|
|
{"app_name_map": "ar.tvplayer.tv=TiviMate\ncom.plex.hdx=Plex"}
|
|
)
|
|
|
|
assert result["type"] == "create_entry"
|
|
assert flow._options["app_name_map"] == {
|
|
"ar.tvplayer.tv": "TiviMate",
|
|
"com.plex.hdx": "Plex",
|
|
}
|
|
|
|
async def test_invalid_map_shows_form_error(self, tmp_path):
|
|
flow = self._options_flow(tmp_path, options={})
|
|
|
|
result = await flow.async_step_init({"app_name_map": "no equals sign here"})
|
|
|
|
assert result["type"] == "form"
|
|
assert result["errors"] == {"base": "invalid_app_map"}
|
|
|
|
async def test_blank_map_clears_the_entries(self, tmp_path):
|
|
flow = self._options_flow(
|
|
tmp_path, options={"app_name_map": {"ar.tvplayer.tv": "TiviMate"}}
|
|
)
|
|
|
|
result = await flow.async_step_init({"app_name_map": " "})
|
|
|
|
assert result["type"] == "create_entry"
|
|
assert flow._options["app_name_map"] == {}
|
|
|
|
async def test_existing_map_prefills_the_text_field(self, tmp_path):
|
|
flow = self._options_flow(
|
|
tmp_path, options={"app_name_map": {"ar.tvplayer.tv": "TiviMate"}}
|
|
)
|
|
|
|
# first visit: the form carries the serialized current map as the
|
|
# field default
|
|
result = await flow.async_step_init(None)
|
|
|
|
assert result["type"] == "form"
|
|
# the editor carried the existing map through the defaults path
|
|
assert flow._options["app_name_map"] == {"ar.tvplayer.tv": "TiviMate"}
|
|
|
|
|
|
class TestAppNameMapResolution:
|
|
"""app_name resolves through the map after the entity/push tiers."""
|
|
|
|
async def test_mapped_package_resolves_the_friendly_name(self):
|
|
import entity_stublib
|
|
|
|
modules = entity_stublib.load_entity_modules()
|
|
player, _s, _i = _cmd_player(modules)
|
|
player._entry.options["app_name_map"] = {"ar.tvplayer.tv": "TiviMate"}
|
|
player._androidtv_entity = "" # standalone
|
|
|
|
manager = player._manager
|
|
manager._on_current_app("ar.tvplayer.tv") # protocol push
|
|
await asyncio.sleep(0.05) # drain the call_soon_threadsafe push
|
|
|
|
assert player.app_name == "TiviMate"
|
|
assert player.app_id == "ar.tvplayer.tv" # app_id stays raw
|
|
|
|
async def test_unmapped_package_stays_raw(self):
|
|
import entity_stublib
|
|
|
|
modules = entity_stublib.load_entity_modules()
|
|
player, _s, _i = _cmd_player(modules)
|
|
player._entry.options["app_name_map"] = {"ar.tvplayer.tv": "TiviMate"}
|
|
player._androidtv_entity = ""
|
|
|
|
manager = player._manager
|
|
manager._on_current_app("com.google.android.youtube.tv")
|
|
await asyncio.sleep(0.05)
|
|
|
|
assert player.app_name == "com.google.android.youtube.tv"
|
|
|
|
async def test_entity_tier_serves_when_no_higher_tier_matches(self):
|
|
"""P5-1: the entity tier is below map/manifest — it serves when the
|
|
package has no explicit override and no manifest label yet."""
|
|
from types import SimpleNamespace as SimpleNS
|
|
|
|
import entity_stublib
|
|
|
|
modules = entity_stublib.load_entity_modules()
|
|
player, _s, _i = _cmd_player(modules)
|
|
player._entry.options["app_name_map"] = {} # no override
|
|
player._androidtv_entity = "media_player.atv"
|
|
state = SimpleNS(attributes={"app_id": "ar.tvplayer.tv", "app_name": "ATV Tivi"})
|
|
player.hass = SimpleNS(states=SimpleNS(get=lambda eid: state))
|
|
player._read = lambda eid: state if eid == "media_player.atv" else None
|
|
|
|
player._manager._on_current_app("ar.tvplayer.tv")
|
|
await asyncio.sleep(0.05)
|
|
|
|
# no cached label yet → falls through to the entity tier
|
|
assert player.app_name == "ATV Tivi"
|