test(config_flow): protocol cert/PIN pairing, reauth-after-hard-reset, and v1→v2 entry migration (scoped stubs, deterministic PIN/MAC fake client)
This commit is contained in:
James Mackay committed
1 parent
d57c4eaa40
commit
906126b0da
1 file changed
+385
@@ -0,0 +1,385 @@
|
||||
"""Phase 2 step 2 tests: protocol cert/PIN pairing, reauth, entry migration.
|
||||
|
||||
Self-contained scoped-stub variant of the config-flow test harness (same
|
||||
pattern as test_config_flow_pairing.py, with pair-hygiene capabilities added:
|
||||
FakeHass records unique_id sets, entry updates and reload scheduling). The
|
||||
androidtvremote2 client is faked with a deterministic pairing code and MAC.
|
||||
"""
|
||||
|
||||
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_ADB_ENABLED,
|
||||
CONF_ADB_HOST,
|
||||
CONF_ADB_PORT,
|
||||
CONF_ADB_STORE_APP_INFO,
|
||||
CONF_ANDROIDTV_ENTITY,
|
||||
CONF_JELLYFIN_ENTITY,
|
||||
CONF_PAIRING_CODE,
|
||||
CONF_POWER_ENTITY,
|
||||
CONF_PROTOCOL_ENABLED,
|
||||
CONF_PROTOCOL_HOST,
|
||||
CONF_RECEIVER_ENTITY,
|
||||
CONF_RECEIVER_INPUT,
|
||||
DEFAULT_ADB_PORT,
|
||||
)
|
||||
|
||||
_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",
|
||||
)
|
||||
|
||||
FAKE_PAIRING_CODE = "8421"
|
||||
FAKE_DEVICE_NAME = "mdarcy"
|
||||
FAKE_MAC = "aa:bb:cc:dd:ee:ff"
|
||||
|
||||
|
||||
class FakeInvalidAuthError(Exception):
|
||||
"""androidtvremote2.exceptions.InvalidAuth stand-in."""
|
||||
|
||||
|
||||
class FakeProtocolClient:
|
||||
"""Proto client used by the flow pairing path (deterministic)."""
|
||||
|
||||
def __init__(self, fail_code=False):
|
||||
self.cert_generated = False
|
||||
self.pairing_started = False
|
||||
self.finished_codes: list[str] = []
|
||||
self.disconnected = 0
|
||||
self._fail_code = fail_code
|
||||
|
||||
async def async_generate_cert_if_missing(self):
|
||||
self.cert_generated = True
|
||||
return True
|
||||
|
||||
async def async_start_pairing(self):
|
||||
self.pairing_started = True
|
||||
return FAKE_PAIRING_CODE # deterministic "device-side PIN"
|
||||
|
||||
async def async_finish_pairing(self, pairing_code: str) -> None:
|
||||
if self._fail_code or pairing_code != FAKE_PAIRING_CODE:
|
||||
raise FakeInvalidAuthError("device rejected the pairing code")
|
||||
self.finished_codes.append(pairing_code)
|
||||
|
||||
async def async_get_name_and_mac(self):
|
||||
return FAKE_DEVICE_NAME, FAKE_MAC
|
||||
|
||||
def disconnect(self):
|
||||
self.disconnected += 1
|
||||
|
||||
|
||||
def _install_stubs() -> None:
|
||||
"""Import-time stand-ins (idempotent, scoped by the autouse fixture)."""
|
||||
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 _form(step_id, errors or {})
|
||||
|
||||
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):
|
||||
if getattr(self, "_unique_id", None) in active_unique_ids:
|
||||
raise AbortIfConfiguredError()
|
||||
|
||||
def _form(step_id, errors):
|
||||
return {"type": "form", "step_id": step_id, "errors": dict(errors)}
|
||||
|
||||
class AbortIfConfiguredError(Exception):
|
||||
pass
|
||||
|
||||
active_unique_ids = set()
|
||||
|
||||
class OptionsFlow:
|
||||
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")
|
||||
|
||||
class _Registry:
|
||||
entities: dict = {}
|
||||
|
||||
def async_get(self, entity_id):
|
||||
return None
|
||||
|
||||
entity_registry.async_get = lambda hass: _Registry()
|
||||
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 getattr(self, f"__entry_{entry_id}", 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"):
|
||||
setattr(entry, key, value)
|
||||
else:
|
||||
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):
|
||||
import types as t
|
||||
|
||||
self.config = t.SimpleNamespace(config_dir=tmp_dir)
|
||||
self.config_entries = FakeConfigEntries()
|
||||
|
||||
@staticmethod
|
||||
async def async_add_executor_job(fn, *args):
|
||||
return fn(*args)
|
||||
|
||||
|
||||
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"]
|
||||
|
||||
|
||||
@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 make_flow(tmp_dir: str, monkeypatch: pytest.MonkeyPatch, client: FakeProtocolClient):
|
||||
"""ConfigFlow with the real client factory mocked to our fake."""
|
||||
module = _form_module()
|
||||
flow = module.NvidiaShieldConfigFlow()
|
||||
flow.hass = FakeHass(tmp_dir)
|
||||
flow._user_data = {
|
||||
"name": "mdarcy",
|
||||
CONF_ANDROIDTV_ENTITY: "media_player.shield",
|
||||
CONF_JELLYFIN_ENTITY: "",
|
||||
CONF_RECEIVER_ENTITY: "",
|
||||
CONF_RECEIVER_INPUT: "",
|
||||
CONF_POWER_ENTITY: "",
|
||||
CONF_ADB_ENABLED: False,
|
||||
CONF_ADB_STORE_APP_INFO: True,
|
||||
CONF_ADB_HOST: "",
|
||||
CONF_ADB_PORT: DEFAULT_ADB_PORT,
|
||||
CONF_PROTOCOL_ENABLED: True,
|
||||
CONF_PROTOCOL_HOST: "shield.local",
|
||||
}
|
||||
from nsh.connection import ShieldConnectionManager
|
||||
|
||||
monkeypatch.setattr(
|
||||
ShieldConnectionManager,
|
||||
"build_protocol_client",
|
||||
lambda *args, **kwargs: client,
|
||||
)
|
||||
monkeypatch.setattr(
|
||||
module,
|
||||
"protocol_cert_paths",
|
||||
lambda hass: ("/tmp/nsh-proto-test.pem", "/tmp/nsh-proto-test.key"),
|
||||
)
|
||||
return flow
|
||||
|
||||
|
||||
class TestProtocolPairing:
|
||||
async def test_first_visit_starts_pairing_and_shows_code_form(self, tmp_path, monkeypatch):
|
||||
client = FakeProtocolClient()
|
||||
flow = make_flow(str(tmp_path), monkeypatch, client)
|
||||
|
||||
result = await flow.async_step_protocol_pair(None)
|
||||
|
||||
assert client.cert_generated is True
|
||||
assert client.pairing_started is True
|
||||
assert result["type"] == "form"
|
||||
assert result["step_id"] == "protocol_pair"
|
||||
assert result["errors"] == {}
|
||||
|
||||
async def test_submit_finishes_with_code_and_records_mac_unique_id(self, tmp_path, monkeypatch):
|
||||
client = FakeProtocolClient()
|
||||
flow = make_flow(str(tmp_path), monkeypatch, client)
|
||||
|
||||
await flow.async_step_protocol_pair(None)
|
||||
result = await flow.async_step_protocol_pair(
|
||||
{CONF_PAIRING_CODE: FAKE_PAIRING_CODE}
|
||||
)
|
||||
|
||||
assert client.finished_codes == [FAKE_PAIRING_CODE]
|
||||
assert flow._unique_id == FAKE_MAC
|
||||
assert client.disconnected == 1
|
||||
assert result["type"] == "create_entry"
|
||||
assert result["options"][CONF_PROTOCOL_ENABLED] is True
|
||||
assert result["options"][CONF_PROTOCOL_HOST] == "shield.local"
|
||||
|
||||
async def test_wrong_code_asks_again_then_succeeds(self, tmp_path, monkeypatch):
|
||||
client = FakeProtocolClient()
|
||||
flow = make_flow(str(tmp_path), monkeypatch, client)
|
||||
|
||||
await flow.async_step_protocol_pair(None)
|
||||
failed = await flow.async_step_protocol_pair({"pairing_code": "9999"})
|
||||
assert failed["type"] == "form"
|
||||
assert failed["errors"] == {"base": "protocol_pair_failed"}
|
||||
|
||||
result = await flow.async_step_protocol_pair(
|
||||
{CONF_PAIRING_CODE: FAKE_PAIRING_CODE}
|
||||
)
|
||||
assert result["type"] == "create_entry"
|
||||
assert client.finished_codes == [FAKE_PAIRING_CODE]
|
||||
|
||||
|
||||
class TestReauthFlow:
|
||||
async def test_reauth_repairs_entry_and_schedules_reload(self, tmp_path, monkeypatch):
|
||||
client = FakeProtocolClient()
|
||||
flow = make_flow(str(tmp_path), monkeypatch, client)
|
||||
entry = SimpleNamespace(
|
||||
entry_id="e1", title="mdarcy",
|
||||
options={CONF_PROTOCOL_ENABLED: True, CONF_PROTOCOL_HOST: "shield.local"},
|
||||
data={},
|
||||
)
|
||||
flow.hass.config_entries.__dict__["__entry_e1"] = entry
|
||||
flow.context = {"entry_id": "e1"}
|
||||
|
||||
shown = await flow.async_step_reauth({CONF_ADB_PORT: DEFAULT_ADB_PORT})
|
||||
assert shown["type"] == "form"
|
||||
assert shown["step_id"] == "reauth_confirm"
|
||||
|
||||
pair_form = await flow.async_step_reauth_confirm({"confirm": True})
|
||||
assert pair_form["step_id"] == "protocol_pair" # PIN form open on device
|
||||
|
||||
result = await flow.async_step_protocol_pair(
|
||||
{CONF_PAIRING_CODE: FAKE_PAIRING_CODE}
|
||||
)
|
||||
assert result == {"type": "abort", "reason": "reauth_successful"}
|
||||
assert entry.options[CONF_PROTOCOL_ENABLED] is True
|
||||
assert flow.hass.config_entries.reloaded_entry_ids == ["e1"]
|
||||
|
||||
|
||||
class TestEntryMigration:
|
||||
async def test_v1_gains_protocol_defaults_and_keeps_adb_options(self, tmp_path, monkeypatch):
|
||||
module = _entry_module()
|
||||
hass = FakeHass(str(tmp_path))
|
||||
entry = SimpleNamespace(
|
||||
version=1,
|
||||
options={"adb_enabled": True, "adb_port": 5555, "jellyfin_entity": "jf.x"},
|
||||
title="mdarcy",
|
||||
)
|
||||
|
||||
ok = await module.async_migrate_entry(hass, entry)
|
||||
|
||||
assert ok is True
|
||||
# ADB options carried over byte-compatibly
|
||||
assert entry.options["adb_enabled"] is True
|
||||
assert entry.options["adb_port"] == 5555
|
||||
# Protocol fields default; host discovery (ADB fallback) stays active
|
||||
assert entry.options[CONF_PROTOCOL_ENABLED] is False
|
||||
assert entry.options[CONF_PROTOCOL_HOST] == ""
|
||||
assert hass.config_entries.updating_entry_calls == [
|
||||
(entry, {"options": entry.options, "version": 2})
|
||||
]
|
||||
|
||||
async def test_v2_is_a_noop(self, tmp_path, monkeypatch):
|
||||
module = _entry_module()
|
||||
hass = FakeHass(str(tmp_path))
|
||||
entry = SimpleNamespace(
|
||||
version=2,
|
||||
options={CONF_PROTOCOL_ENABLED: True},
|
||||
title="mdarcy",
|
||||
)
|
||||
assert await module.async_migrate_entry(hass, entry) is True
|
||||
assert hass.config_entries.updating_entry_calls == []
|
||||
|
||||
async def test_downgrade_rejected(self, tmp_path, monkeypatch):
|
||||
module = _entry_module()
|
||||
hass = FakeHass(str(tmp_path))
|
||||
entry = SimpleNamespace(version=3, options={}, title="x")
|
||||
assert await module.async_migrate_entry(hass, entry) is False
|
||||
Reference in new issue
Block a user