overseer
2b6b7431b5
The integration will consume the NestQuest API service (Feature 16) over HTTP instead of reading the database. This adds the one module that speaks the panel plane: api_client.py with get_snapshot() and complete_instance(), a typed NestQuestApiError for transport failures and non-2xx responses (carrying the HTTP status so 401/404 are distinguishable), one timeout per request and no retry loops. The HTTP session is injected, so tests exercise the client against a stub transport with no network and no aiohttp; the production factory hands in HA's shared aiohttp session and resolves the base URL and panel service token off the config entry (options, then entry data, then defaults). The two new settings are wired into the options flow like every other section (validation mirrored with the client's constructor rules); the token is a secret and is never logged or echoed into error messages. Coordinator wiring is the next task.
904 lines
33 KiB
Python
904 lines
33 KiB
Python
"""Tests for the options flow of NestQuest."""
|
|
from __future__ import annotations
|
|
|
|
import ast
|
|
import asyncio
|
|
import json
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
from conftest import make_config_entry, make_hass, options_flow_base as _options_flow_base
|
|
|
|
from custom_components.nestquest import config_flow
|
|
from custom_components.nestquest import options_flow
|
|
from custom_components.nestquest.const import (
|
|
CONF_ADMIN_USER_IDS,
|
|
CONF_AFTERNOON_REMINDER_ENABLED,
|
|
CONF_AFTERNOON_REMINDER_TIME,
|
|
CONF_API_BASE_URL,
|
|
CONF_CELEBRATION_ENABLED,
|
|
CONF_DAY_ROLLOVER_TIME,
|
|
CONF_END_OF_DAY_REPORT_ENABLED,
|
|
CONF_END_OF_DAY_REPORT_TIME,
|
|
CONF_HORIZON_DAYS,
|
|
CONF_MORNING_SUMMARY_ENABLED,
|
|
CONF_MORNING_SUMMARY_TIME,
|
|
CONF_NOTIFY_TARGET,
|
|
CONF_PANEL_IDLE_TIMEOUT,
|
|
CONF_PANEL_TOKEN,
|
|
CONF_UPDATE_INTERVAL,
|
|
DEFAULT_API_BASE_URL,
|
|
DEFAULT_AFTERNOON_REMINDER_TIME,
|
|
DEFAULT_AUTOMATION_ENABLED,
|
|
DEFAULT_END_OF_DAY_REPORT_TIME,
|
|
DEFAULT_MORNING_SUMMARY_TIME,
|
|
DEFAULT_PANEL_TOKEN,
|
|
DEFAULT_UPDATE_INTERVAL,
|
|
DEFAULT_DAY_ROLLOVER_TIME,
|
|
DEFAULT_HORIZON_DAYS,
|
|
DEFAULT_PANEL_IDLE_TIMEOUT,
|
|
DOMAIN,
|
|
)
|
|
|
|
PKG_DIR = Path(options_flow.__file__).parent
|
|
STRINGS_PATH = PKG_DIR / "strings.json"
|
|
TRANSLATIONS_EN_PATH = PKG_DIR / "translations" / "en.json"
|
|
|
|
VALID_INPUT = {
|
|
CONF_HORIZON_DAYS: 7,
|
|
CONF_DAY_ROLLOVER_TIME: "03:30",
|
|
CONF_PANEL_IDLE_TIMEOUT: 60,
|
|
}
|
|
|
|
#: What a schema-validated submit produces from a bare VALID_INPUT on a
|
|
#: fresh entry: the Feature 10 refresh interval, the Feature 11
|
|
#: notification section, and the Feature 18 API connection fill from
|
|
#: their defaults.
|
|
FULL_INPUT = {
|
|
**VALID_INPUT,
|
|
CONF_UPDATE_INTERVAL: DEFAULT_UPDATE_INTERVAL,
|
|
CONF_NOTIFY_TARGET: "",
|
|
CONF_MORNING_SUMMARY_TIME: DEFAULT_MORNING_SUMMARY_TIME,
|
|
CONF_AFTERNOON_REMINDER_TIME: DEFAULT_AFTERNOON_REMINDER_TIME,
|
|
CONF_END_OF_DAY_REPORT_TIME: DEFAULT_END_OF_DAY_REPORT_TIME,
|
|
CONF_MORNING_SUMMARY_ENABLED: DEFAULT_AUTOMATION_ENABLED,
|
|
CONF_AFTERNOON_REMINDER_ENABLED: DEFAULT_AUTOMATION_ENABLED,
|
|
CONF_END_OF_DAY_REPORT_ENABLED: DEFAULT_AUTOMATION_ENABLED,
|
|
CONF_CELEBRATION_ENABLED: DEFAULT_AUTOMATION_ENABLED,
|
|
CONF_API_BASE_URL: DEFAULT_API_BASE_URL,
|
|
CONF_PANEL_TOKEN: DEFAULT_PANEL_TOKEN,
|
|
}
|
|
|
|
|
|
def _run(coro):
|
|
return asyncio.new_event_loop().run_until_complete(coro)
|
|
|
|
|
|
def _make_hass():
|
|
return make_hass()
|
|
|
|
|
|
def _make_entry(entry_id: str = "test_entry", options: dict | None = None, data: dict | None = None):
|
|
return make_config_entry(entry_id=entry_id, options=options, data=data)
|
|
|
|
|
|
def _make_flow(entry, *, users=None, database=None) -> options_flow.NestQuestOptionsFlow:
|
|
"""Create an options flow instance bound to a stubbed config entry.
|
|
|
|
``users`` populates a stubbed hass.auth for the admin picker;
|
|
``database`` wires the runtime record so the flow can read and
|
|
write the allowlist like real HA's options flow does.
|
|
"""
|
|
from unittest.mock import AsyncMock
|
|
|
|
flow = options_flow.NestQuestOptionsFlow(config_entry=entry)
|
|
flow.hass = SimpleNamespace()
|
|
flow.hass.auth = SimpleNamespace(
|
|
async_get_users=AsyncMock(return_value=users or [])
|
|
)
|
|
if database is not None:
|
|
from custom_components.nestquest.const import DOMAIN
|
|
|
|
flow.hass.data = {
|
|
DOMAIN: {entry.entry_id: SimpleNamespace(database=database)}
|
|
}
|
|
return flow
|
|
|
|
|
|
def _user(user_id: str, name: str):
|
|
return SimpleNamespace(id=user_id, name=name)
|
|
|
|
|
|
def _open_allowlist_db(tmp_path, name):
|
|
"""A real migrated temp database for the flow to read and write."""
|
|
from unittest.mock import AsyncMock, MagicMock
|
|
|
|
from custom_components.nestquest.db import NestQuestDatabase
|
|
from custom_components.nestquest.migrations import apply_migrations
|
|
|
|
hass = MagicMock()
|
|
hass.async_add_executor_job = AsyncMock(side_effect=(lambda fn, *a: fn(*a)))
|
|
database = NestQuestDatabase(hass.async_add_executor_job)
|
|
_run(database.open(tmp_path / name))
|
|
_run(apply_migrations(database))
|
|
return database
|
|
|
|
|
|
def _schema_values(schema) -> dict:
|
|
"""Extract the per-field default values from a real voluptuous schema."""
|
|
return {
|
|
marker.schema: marker.default()
|
|
for marker in schema.schema
|
|
if hasattr(marker, "schema")
|
|
}
|
|
|
|
|
|
def test_options_flow_reachable_via_configure_button() -> None:
|
|
"""The config flow exposes async_get_options_flow returning the options flow."""
|
|
entry = _make_entry()
|
|
handler = config_flow.NestQuestConfigFlow.async_get_options_flow(entry)
|
|
assert isinstance(handler, options_flow.NestQuestOptionsFlow)
|
|
assert handler.config_entry is entry
|
|
|
|
|
|
def test_options_flow_get_options_flow_is_ha_callback() -> None:
|
|
"""async_get_options_flow is wrapped by the homeassistant.core.callback decorator."""
|
|
assert getattr(
|
|
config_flow.NestQuestConfigFlow.async_get_options_flow, "__ha_callback__", False
|
|
) is True
|
|
|
|
|
|
def test_options_flow_constructs_through_2024_6_surface() -> None:
|
|
"""The handler constructs against the mocked 2024.6 base and inherits its contract."""
|
|
entry = _make_entry(options={CONF_HORIZON_DAYS: 21})
|
|
flow = options_flow.NestQuestOptionsFlow(config_entry=entry)
|
|
assert isinstance(flow, _options_flow_base())
|
|
assert flow.config_entry is entry
|
|
assert flow.options == {CONF_HORIZON_DAYS: 21}
|
|
|
|
|
|
def test_options_flow_submits_through_2024_6_surface() -> None:
|
|
"""The handler can be submitted through the mocked 2024.6-shaped surface."""
|
|
entry = _make_entry()
|
|
handler = config_flow.NestQuestConfigFlow.async_get_options_flow(entry)
|
|
result = _run(handler.async_step_init(dict(VALID_INPUT)))
|
|
assert result["type"] == "create_entry"
|
|
assert result["title"] == "NestQuest"
|
|
assert result["data"] == FULL_INPUT
|
|
|
|
|
|
def test_options_flow_shows_form_initially() -> None:
|
|
"""The init step shows a form."""
|
|
flow = _make_flow(_make_entry())
|
|
result = _run(flow.async_step_init(None))
|
|
assert result["type"] == "form"
|
|
assert result["step_id"] == "init"
|
|
|
|
|
|
def test_options_flow_schema_is_voluptuous() -> None:
|
|
"""The form's data_schema is a real voluptuous Schema (no dict fallback)."""
|
|
flow = _make_flow(_make_entry())
|
|
result = _run(flow.async_step_init(None))
|
|
import voluptuous as vol
|
|
|
|
assert isinstance(result["data_schema"], vol.Schema)
|
|
|
|
|
|
def test_options_flow_prefills_from_defaults() -> None:
|
|
"""Without stored options, the form pre-fills with the constant defaults."""
|
|
flow = _make_flow(_make_entry())
|
|
result = _run(flow.async_step_init(None))
|
|
values = _schema_values(result["data_schema"])
|
|
assert values[CONF_HORIZON_DAYS] == DEFAULT_HORIZON_DAYS
|
|
assert values[CONF_DAY_ROLLOVER_TIME] == DEFAULT_DAY_ROLLOVER_TIME
|
|
assert values[CONF_PANEL_IDLE_TIMEOUT] == DEFAULT_PANEL_IDLE_TIMEOUT
|
|
|
|
|
|
def test_options_flow_prefills_from_entry_options() -> None:
|
|
"""Stored entry options win over entry data and defaults."""
|
|
entry = _make_entry(
|
|
options={
|
|
CONF_HORIZON_DAYS: 21,
|
|
CONF_DAY_ROLLOVER_TIME: "05:45",
|
|
CONF_PANEL_IDLE_TIMEOUT: 120,
|
|
},
|
|
data={
|
|
CONF_HORIZON_DAYS: 99,
|
|
CONF_DAY_ROLLOVER_TIME: "23:59",
|
|
CONF_PANEL_IDLE_TIMEOUT: 999,
|
|
},
|
|
)
|
|
result = _run(_make_flow(entry).async_step_init(None))
|
|
values = _schema_values(result["data_schema"])
|
|
assert values[CONF_HORIZON_DAYS] == 21
|
|
assert values[CONF_DAY_ROLLOVER_TIME] == "05:45"
|
|
assert values[CONF_PANEL_IDLE_TIMEOUT] == 120
|
|
|
|
|
|
def test_options_flow_falls_back_to_entry_data() -> None:
|
|
"""Pre-upgrade entries without options fall back to entry.data."""
|
|
entry = _make_entry(
|
|
data={
|
|
CONF_HORIZON_DAYS: 10,
|
|
CONF_DAY_ROLLOVER_TIME: "01:15",
|
|
CONF_PANEL_IDLE_TIMEOUT: 90,
|
|
}
|
|
)
|
|
result = _run(_make_flow(entry).async_step_init(None))
|
|
values = _schema_values(result["data_schema"])
|
|
assert values[CONF_HORIZON_DAYS] == 10
|
|
assert values[CONF_DAY_ROLLOVER_TIME] == "01:15"
|
|
assert values[CONF_PANEL_IDLE_TIMEOUT] == 90
|
|
|
|
|
|
def test_options_flow_schema_applies_stored_defaults() -> None:
|
|
"""A real voluptuous validate({}) fills every field from the stored defaults.
|
|
|
|
Without a runtime database (and no auth in the bare harness), the
|
|
admin picker's default is the empty selection and its choice set is
|
|
empty — the three legacy settings are unaffected.
|
|
"""
|
|
entry = _make_entry(
|
|
options={
|
|
CONF_HORIZON_DAYS: 21,
|
|
CONF_DAY_ROLLOVER_TIME: "05:45",
|
|
CONF_PANEL_IDLE_TIMEOUT: 120,
|
|
CONF_UPDATE_INTERVAL: 90,
|
|
CONF_NOTIFY_TARGET: "notify.mobile_app_test",
|
|
CONF_MORNING_SUMMARY_TIME: "07:00",
|
|
CONF_AFTERNOON_REMINDER_TIME: "14:00",
|
|
CONF_END_OF_DAY_REPORT_TIME: "21:30",
|
|
}
|
|
)
|
|
result = _run(_make_flow(entry).async_step_init(None))
|
|
assert result["data_schema"]({}) == {
|
|
CONF_HORIZON_DAYS: 21,
|
|
CONF_DAY_ROLLOVER_TIME: "05:45",
|
|
CONF_PANEL_IDLE_TIMEOUT: 120,
|
|
CONF_UPDATE_INTERVAL: 90,
|
|
CONF_NOTIFY_TARGET: "notify.mobile_app_test",
|
|
CONF_MORNING_SUMMARY_TIME: "07:00",
|
|
CONF_AFTERNOON_REMINDER_TIME: "14:00",
|
|
CONF_END_OF_DAY_REPORT_TIME: "21:30",
|
|
CONF_MORNING_SUMMARY_ENABLED: True,
|
|
CONF_AFTERNOON_REMINDER_ENABLED: True,
|
|
CONF_END_OF_DAY_REPORT_ENABLED: True,
|
|
CONF_CELEBRATION_ENABLED: True,
|
|
CONF_API_BASE_URL: DEFAULT_API_BASE_URL,
|
|
CONF_PANEL_TOKEN: DEFAULT_PANEL_TOKEN,
|
|
CONF_ADMIN_USER_IDS: [],
|
|
}
|
|
|
|
|
|
def test_options_flow_schema_coerces_field_types() -> None:
|
|
"""The schema validates the expected field types via voluptuous."""
|
|
flow = _make_flow(_make_entry())
|
|
result = _run(flow.async_step_init(None))
|
|
schema = result["data_schema"]
|
|
validators = {marker.schema: validator for marker, validator in schema.schema.items()}
|
|
assert validators[CONF_HORIZON_DAYS] is int
|
|
assert validators[CONF_DAY_ROLLOVER_TIME] is str
|
|
assert validators[CONF_PANEL_IDLE_TIMEOUT] is int
|
|
# The Feature 18 API connection fields are free-text strings.
|
|
assert validators[CONF_API_BASE_URL] is str
|
|
assert validators[CONF_PANEL_TOKEN] is str
|
|
|
|
|
|
@pytest.mark.parametrize("bad_horizon", [0, -5])
|
|
def test_options_flow_rejects_non_positive_horizon(bad_horizon: int) -> None:
|
|
"""Zero or negative horizon values yield a field-level error, no exception."""
|
|
flow = _make_flow(_make_entry())
|
|
result = _run(
|
|
flow.async_step_init({**VALID_INPUT, CONF_HORIZON_DAYS: bad_horizon})
|
|
)
|
|
assert result["type"] == "form"
|
|
assert result["errors"] == {CONF_HORIZON_DAYS: "invalid"}
|
|
|
|
|
|
def test_options_flow_rejects_malformed_time_not_a_time() -> None:
|
|
"""Non-HH:MM strings are rejected with invalid_time."""
|
|
flow = _make_flow(_make_entry())
|
|
result = _run(
|
|
flow.async_step_init({**VALID_INPUT, CONF_DAY_ROLLOVER_TIME: "noon"})
|
|
)
|
|
assert result["type"] == "form"
|
|
assert result["errors"] == {CONF_DAY_ROLLOVER_TIME: "invalid_time"}
|
|
|
|
|
|
def test_options_flow_rejects_malformed_time_out_of_range() -> None:
|
|
"""Out-of-range HH:MM values are rejected with invalid_time."""
|
|
flow = _make_flow(_make_entry())
|
|
result = _run(
|
|
flow.async_step_init({**VALID_INPUT, CONF_DAY_ROLLOVER_TIME: "25:99"})
|
|
)
|
|
assert result["type"] == "form"
|
|
assert result["errors"] == {CONF_DAY_ROLLOVER_TIME: "invalid_time"}
|
|
|
|
|
|
def test_options_flow_rejects_pasted_unicode_superscript_time() -> None:
|
|
"""Pasted Unicode digits like '²²:²²' yield a field-level error, no exception."""
|
|
flow = _make_flow(_make_entry())
|
|
result = _run(
|
|
flow.async_step_init({**VALID_INPUT, CONF_DAY_ROLLOVER_TIME: "²²:²²"})
|
|
)
|
|
assert result["type"] == "form"
|
|
assert result["errors"] == {CONF_DAY_ROLLOVER_TIME: "invalid_time"}
|
|
|
|
|
|
@pytest.mark.parametrize("bad_timeout", ["abc", 0, -10, 10])
|
|
def test_options_flow_rejects_invalid_idle_timeout(bad_timeout) -> None:
|
|
"""Non-numeric or too-low idle timeouts yield a field-level error."""
|
|
flow = _make_flow(_make_entry())
|
|
result = _run(
|
|
flow.async_step_init({**VALID_INPUT, CONF_PANEL_IDLE_TIMEOUT: bad_timeout})
|
|
)
|
|
assert result["type"] == "form"
|
|
assert result["errors"] == {CONF_PANEL_IDLE_TIMEOUT: "invalid"}
|
|
|
|
|
|
def test_options_flow_creates_entry_with_validated_keys() -> None:
|
|
"""A valid submit creates an entry with exactly the validated keys
|
|
(the legacy settings plus every later section's keys, filled from
|
|
defaults where the raw submit omitted them)."""
|
|
flow = _make_flow(_make_entry())
|
|
result = _run(flow.async_step_init(dict(VALID_INPUT)))
|
|
assert result["type"] == "create_entry"
|
|
assert result["title"] == "NestQuest"
|
|
assert set(result["data"]) == set(FULL_INPUT)
|
|
assert result["data"] == FULL_INPUT
|
|
|
|
|
|
def test_options_flow_create_entry_data_lands_in_entry_options_and_reloads_once() -> None:
|
|
"""The flow's created data is written to entry.options and reloads exactly once."""
|
|
hass, registry = _make_hass()
|
|
entry = _make_entry()
|
|
entry.add_update_listener = lambda listener: registry.add(entry.entry_id, listener)
|
|
|
|
# Simulate setup registering the merged lifecycle update listener.
|
|
from custom_components.nestquest import async_setup_entry
|
|
|
|
_run(async_setup_entry(hass, entry))
|
|
|
|
# Run the options flow and apply its created data to entry.options like HA does.
|
|
flow = _make_flow(entry)
|
|
result = _run(flow.async_step_init(dict(VALID_INPUT)))
|
|
assert result["type"] == "create_entry"
|
|
entry.options = dict(result["data"])
|
|
|
|
_run(registry.dispatch_options_update(entry))
|
|
assert hass.config_entries.async_reload.call_count == 1
|
|
assert registry.reloaded == [entry.entry_id]
|
|
assert entry.options == FULL_INPUT
|
|
|
|
|
|
def test_options_flow_validation_never_raises() -> None:
|
|
"""Validation returns errors instead of raising for arbitrary junk input."""
|
|
junk = {CONF_HORIZON_DAYS: None, CONF_DAY_ROLLOVER_TIME: None, CONF_PANEL_IDLE_TIMEOUT: None}
|
|
flow = _make_flow(_make_entry())
|
|
result = _run(flow.async_step_init(junk))
|
|
assert result["type"] == "form"
|
|
assert result["errors"] == {
|
|
CONF_HORIZON_DAYS: "invalid",
|
|
CONF_DAY_ROLLOVER_TIME: "invalid_time",
|
|
CONF_PANEL_IDLE_TIMEOUT: "invalid",
|
|
}
|
|
|
|
|
|
@pytest.fixture
|
|
def strings() -> dict:
|
|
"""Load and parse strings.json."""
|
|
assert STRINGS_PATH.exists(), f"strings.json does not exist at {STRINGS_PATH}"
|
|
with open(STRINGS_PATH, encoding="utf-8") as f:
|
|
return json.load(f)
|
|
|
|
|
|
@pytest.fixture
|
|
def translations_en() -> dict:
|
|
"""Load and parse translations/en.json."""
|
|
assert TRANSLATIONS_EN_PATH.exists(), (
|
|
f"translations/en.json does not exist at {TRANSLATIONS_EN_PATH}"
|
|
)
|
|
with open(TRANSLATIONS_EN_PATH, encoding="utf-8") as f:
|
|
return json.load(f)
|
|
|
|
|
|
def test_options_strings_present_and_matching(strings: dict, translations_en: dict) -> None:
|
|
"""Both string files carry identical options step/error strings."""
|
|
assert strings.get("options") == translations_en.get("options")
|
|
options = strings["options"]
|
|
init = options["step"]["init"]
|
|
assert init["title"]
|
|
assert init["description"]
|
|
data = init["data"]
|
|
for key in (CONF_HORIZON_DAYS, CONF_DAY_ROLLOVER_TIME, CONF_PANEL_IDLE_TIMEOUT):
|
|
assert data.get(key), f"Missing field name for '{key}'"
|
|
errors = options["errors"]
|
|
for key in ("invalid", "invalid_time"):
|
|
assert errors.get(key), f"Missing error string for '{key}'"
|
|
|
|
|
|
def test_options_strings_no_raw_keys(strings: dict, translations_en: dict) -> None:
|
|
"""No placeholder/raw keys appear in the options strings."""
|
|
|
|
def check(value) -> None:
|
|
if isinstance(value, dict):
|
|
for key, val in value.items():
|
|
assert val, f"Empty string value for key '{key}'"
|
|
check(val)
|
|
elif isinstance(value, str):
|
|
assert value.strip() != ""
|
|
|
|
check(strings["options"])
|
|
check(translations_en["options"])
|
|
|
|
|
|
def test_options_flow_no_plain_dict_schema_fallback() -> None:
|
|
"""No try/except ImportError fallback for voluptuous in the options flow module."""
|
|
py_file = Path(options_flow.__file__)
|
|
tree = ast.parse(py_file.read_text(encoding="utf-8"), filename=str(py_file))
|
|
for node in ast.walk(tree):
|
|
if isinstance(node, ast.Try):
|
|
handlers = node.handlers or []
|
|
has_import_error_handler = any(
|
|
getattr(handler.type, "id", getattr(handler.type, "attr", None))
|
|
== "ImportError"
|
|
for handler in handlers
|
|
)
|
|
assert not has_import_error_handler, (
|
|
f"Plain-dict schema fallback (try/except ImportError) found in {py_file.name}:{node.lineno}"
|
|
)
|
|
|
|
|
|
def test_options_flow_subclasses_options_flow_with_config_entry() -> None:
|
|
"""NestQuestOptionsFlow subclasses config_entries.OptionsFlowWithConfigEntry."""
|
|
base = _options_flow_base()
|
|
assert issubclass(options_flow.NestQuestOptionsFlow, base)
|
|
assert options_flow.NestQuestOptionsFlow.__bases__ == (base,)
|
|
|
|
|
|
def test_options_flow_no_hardcoded_domain_or_db_filename() -> None:
|
|
"""The options flow module does not hardcode the domain or db filename."""
|
|
py_file = Path(options_flow.__file__)
|
|
tree = ast.parse(py_file.read_text(encoding="utf-8"), filename=str(py_file))
|
|
for node in ast.walk(tree):
|
|
if isinstance(node, ast.Constant) and isinstance(node.value, str):
|
|
assert node.value != DOMAIN, (
|
|
f"Hard-coded domain string in {py_file.name}:{node.lineno}"
|
|
)
|
|
assert node.value != "nestquest.db", (
|
|
f"Hard-coded database filename in {py_file.name}:{node.lineno}"
|
|
)
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Admin allowlist picker (Feature 03)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_options_flow_picker_lists_existing_users(tmp_path) -> None:
|
|
"""The picker offers EXISTING HA users (id -> name), never free text."""
|
|
import voluptuous as vol
|
|
|
|
database = _open_allowlist_db(tmp_path, "picker-users.db")
|
|
try:
|
|
from custom_components.nestquest.admin_allowlist import set_admin_ids
|
|
|
|
_run(set_admin_ids(database, ["u1"]))
|
|
entry = _make_entry()
|
|
flow = _make_flow(
|
|
entry,
|
|
users=[_user("u1", "Joshua"), _user("u2", "Sam")],
|
|
database=database,
|
|
)
|
|
result = _run(flow.async_step_init(None))
|
|
schema = result["data_schema"]
|
|
marker = next(
|
|
m for m in schema.schema
|
|
if getattr(m, "schema", None) == CONF_ADMIN_USER_IDS
|
|
)
|
|
# Default is the CURRENT database allowlist.
|
|
values = _schema_values(schema)
|
|
assert values[CONF_ADMIN_USER_IDS] == ["u1"]
|
|
# Both users are selectable; an unknown id is rejected by the
|
|
# schema itself (this is what real HA submits go through).
|
|
validated = schema(
|
|
{
|
|
CONF_HORIZON_DAYS: 7,
|
|
CONF_DAY_ROLLOVER_TIME: "03:30",
|
|
CONF_PANEL_IDLE_TIMEOUT: 60,
|
|
CONF_ADMIN_USER_IDS: ["u1", "u2"],
|
|
}
|
|
)
|
|
assert validated[CONF_ADMIN_USER_IDS] == ["u1", "u2"]
|
|
with pytest.raises(vol.Invalid):
|
|
schema(
|
|
{
|
|
CONF_HORIZON_DAYS: 7,
|
|
CONF_DAY_ROLLOVER_TIME: "03:30",
|
|
CONF_PANEL_IDLE_TIMEOUT: 60,
|
|
CONF_ADMIN_USER_IDS: ["impostor"],
|
|
}
|
|
)
|
|
finally:
|
|
_run(database.close())
|
|
|
|
|
|
def test_options_flow_submit_applies_selection_to_database(tmp_path) -> None:
|
|
database = _open_allowlist_db(tmp_path, "picker-apply.db")
|
|
try:
|
|
from custom_components.nestquest.admin_allowlist import (
|
|
list_admin_ids,
|
|
set_admin_ids,
|
|
)
|
|
|
|
_run(set_admin_ids(database, ["u1"]))
|
|
entry = _make_entry()
|
|
flow = _make_flow(
|
|
entry,
|
|
users=[_user("u1", "Joshua"), _user("u2", "Sam")],
|
|
database=database,
|
|
)
|
|
result = _run(
|
|
flow.async_step_init({**VALID_INPUT, CONF_ADMIN_USER_IDS: ["u2"]})
|
|
)
|
|
assert result["type"] == "create_entry"
|
|
assert result["data"][CONF_ADMIN_USER_IDS] == ["u2"]
|
|
# The allowlist (and the entry's recovery copy) carry the pick.
|
|
assert await_gather(list_admin_ids(database)) == ["u2"]
|
|
finally:
|
|
_run(database.close())
|
|
|
|
|
|
def await_gather(coro):
|
|
return _run(coro)
|
|
|
|
|
|
def test_options_flow_empty_selection_refused(tmp_path) -> None:
|
|
"""Saving zero admins is refused: an empty allowlist fails closed."""
|
|
database = _open_allowlist_db(tmp_path, "picker-empty.db")
|
|
try:
|
|
from custom_components.nestquest.admin_allowlist import (
|
|
list_admin_ids,
|
|
set_admin_ids,
|
|
)
|
|
|
|
_run(set_admin_ids(database, ["u1"]))
|
|
entry = _make_entry()
|
|
flow = _make_flow(
|
|
entry,
|
|
users=[_user("u1", "Joshua")],
|
|
database=database,
|
|
)
|
|
result = _run(
|
|
flow.async_step_init({**VALID_INPUT, CONF_ADMIN_USER_IDS: []})
|
|
)
|
|
assert result["type"] == "form"
|
|
assert result["errors"] == {CONF_ADMIN_USER_IDS: "no_admins"}
|
|
assert await_gather(list_admin_ids(database)) == ["u1"]
|
|
finally:
|
|
_run(database.close())
|
|
|
|
|
|
def test_options_flow_unknown_selection_refused(tmp_path) -> None:
|
|
database = _open_allowlist_db(tmp_path, "picker-unknown.db")
|
|
try:
|
|
from custom_components.nestquest.admin_allowlist import (
|
|
list_admin_ids,
|
|
set_admin_ids,
|
|
)
|
|
|
|
_run(set_admin_ids(database, ["u1"]))
|
|
entry = _make_entry()
|
|
flow = _make_flow(
|
|
entry,
|
|
users=[_user("u1", "Joshua")],
|
|
database=database,
|
|
)
|
|
result = _run(
|
|
flow.async_step_init(
|
|
{**VALID_INPUT, CONF_ADMIN_USER_IDS: ["impostor"]}
|
|
)
|
|
)
|
|
assert result["type"] == "form"
|
|
assert result["errors"] == {CONF_ADMIN_USER_IDS: "invalid_admin"}
|
|
assert await_gather(list_admin_ids(database)) == ["u1"]
|
|
finally:
|
|
_run(database.close())
|
|
|
|
|
|
def test_options_flow_submit_without_admin_key_keeps_allowlist(tmp_path) -> None:
|
|
"""Raw callers omitting the key (pre-picker payloads) do not
|
|
touch the allowlist; the CURRENT allowlist is carried forward on
|
|
the options copy so no removed admin can be resurrected later."""
|
|
database = _open_allowlist_db(tmp_path, "picker-legacy.db")
|
|
try:
|
|
from custom_components.nestquest.admin_allowlist import (
|
|
list_admin_ids,
|
|
set_admin_ids,
|
|
)
|
|
|
|
_run(set_admin_ids(database, ["u1"]))
|
|
entry = _make_entry()
|
|
flow = _make_flow(
|
|
entry,
|
|
users=[_user("u1", "Joshua")],
|
|
database=database,
|
|
)
|
|
result = _run(flow.async_step_init(dict(VALID_INPUT)))
|
|
assert result["type"] == "create_entry"
|
|
assert result["data"][CONF_ADMIN_USER_IDS] == ["u1"]
|
|
assert await_gather(list_admin_ids(database)) == ["u1"]
|
|
finally:
|
|
_run(database.close())
|
|
|
|
|
|
def test_options_flow_duplicate_selection_refused(tmp_path) -> None:
|
|
"""Duplicate ids surface as a field error, not an unhandled
|
|
ValueError from the business layer."""
|
|
database = _open_allowlist_db(tmp_path, "picker-dupe.db")
|
|
try:
|
|
from custom_components.nestquest.admin_allowlist import (
|
|
list_admin_ids,
|
|
set_admin_ids,
|
|
)
|
|
|
|
_run(set_admin_ids(database, ["u1"]))
|
|
entry = _make_entry()
|
|
flow = _make_flow(
|
|
entry,
|
|
users=[_user("u1", "Joshua"), _user("u2", "Sam")],
|
|
database=database,
|
|
)
|
|
result = _run(
|
|
flow.async_step_init(
|
|
{**VALID_INPUT, CONF_ADMIN_USER_IDS: ["u2", "u2"]}
|
|
)
|
|
)
|
|
assert result["type"] == "form"
|
|
assert result["errors"] == {CONF_ADMIN_USER_IDS: "invalid_admin"}
|
|
assert await_gather(list_admin_ids(database)) == ["u1"]
|
|
finally:
|
|
_run(database.close())
|
|
|
|
|
|
def test_options_flow_omitted_picker_carries_allowlist_forward(
|
|
tmp_path,
|
|
) -> None:
|
|
"""An omitted picker field must not drop the allowlist copy from
|
|
the returned options: HA replaces entry.options wholesale, and a
|
|
later database loss must not resurrect removed admins from the
|
|
stale config-flow copy in entry.data."""
|
|
database = _open_allowlist_db(tmp_path, "picker-carry.db")
|
|
try:
|
|
from custom_components.nestquest.admin_allowlist import (
|
|
set_admin_ids,
|
|
)
|
|
|
|
_run(set_admin_ids(database, ["u2"]))
|
|
entry = _make_entry()
|
|
flow = _make_flow(
|
|
entry,
|
|
users=[_user("u1", "Joshua"), _user("u2", "Sam")],
|
|
database=database,
|
|
)
|
|
result = _run(flow.async_step_init(dict(VALID_INPUT)))
|
|
assert result["type"] == "create_entry"
|
|
# The current allowlist rides along on the legacy three-key path.
|
|
assert result["data"][CONF_ADMIN_USER_IDS] == ["u2"]
|
|
assert result["data"][CONF_HORIZON_DAYS] == VALID_INPUT[CONF_HORIZON_DAYS]
|
|
finally:
|
|
_run(database.close())
|
|
|
|
|
|
def test_options_flow_notification_section_defaults() -> None:
|
|
"""A bare submit fills the notification section from const defaults:
|
|
08:00 / 15:00 / 20:00 — the report deliberately BEFORE the midnight
|
|
rollover sweep — an empty notify target, and all toggles enabled."""
|
|
flow = _make_flow(_make_entry())
|
|
result = _run(flow.async_step_init(dict(VALID_INPUT)))
|
|
assert result["type"] == "create_entry"
|
|
assert result["data"][CONF_MORNING_SUMMARY_TIME] == "08:00"
|
|
assert result["data"][CONF_AFTERNOON_REMINDER_TIME] == "15:00"
|
|
assert result["data"][CONF_END_OF_DAY_REPORT_TIME] == "20:00"
|
|
assert result["data"][CONF_NOTIFY_TARGET] == ""
|
|
for toggle in (
|
|
CONF_MORNING_SUMMARY_ENABLED,
|
|
CONF_AFTERNOON_REMINDER_ENABLED,
|
|
CONF_END_OF_DAY_REPORT_ENABLED,
|
|
CONF_CELEBRATION_ENABLED,
|
|
):
|
|
assert result["data"][toggle] is True
|
|
|
|
|
|
def test_options_flow_notification_times_must_be_strict_hh_mm() -> None:
|
|
flow = _make_flow(_make_entry())
|
|
for bad_time in ("9:00", "25:00", "08:0", "08-00", 800):
|
|
result = _run(
|
|
flow.async_step_init(
|
|
{
|
|
**VALID_INPUT,
|
|
CONF_MORNING_SUMMARY_TIME: bad_time,
|
|
}
|
|
)
|
|
)
|
|
assert result["type"] == "form"
|
|
assert result["errors"] == {CONF_MORNING_SUMMARY_TIME: "invalid_time"}
|
|
|
|
|
|
def test_options_flow_notify_target_validation() -> None:
|
|
"""The empty target is VALID (it is the shipped default and must be
|
|
savable); only malformed non-empty values reject."""
|
|
flow = _make_flow(_make_entry())
|
|
# The empty default saves cleanly.
|
|
result = _run(
|
|
flow.async_step_init(
|
|
{**VALID_INPUT, CONF_NOTIFY_TARGET: ""}
|
|
)
|
|
)
|
|
assert result["type"] == "create_entry"
|
|
assert result["data"][CONF_NOTIFY_TARGET] == ""
|
|
|
|
# Malformed non-empty values reject.
|
|
for bad_target in (" notify.x ", " ", 42):
|
|
flow = _make_flow(_make_entry())
|
|
result = _run(
|
|
flow.async_step_init(
|
|
{
|
|
**VALID_INPUT,
|
|
CONF_NOTIFY_TARGET: bad_target,
|
|
}
|
|
)
|
|
)
|
|
assert result["type"] == "form"
|
|
assert result["errors"] == {CONF_NOTIFY_TARGET: "invalid"}
|
|
|
|
|
|
def test_options_flow_toggles_must_be_real_booleans() -> None:
|
|
flow = _make_flow(_make_entry())
|
|
result = _run(
|
|
flow.async_step_init(
|
|
{
|
|
**VALID_INPUT,
|
|
CONF_CELEBRATION_ENABLED: "on",
|
|
}
|
|
)
|
|
)
|
|
assert result["type"] == "form"
|
|
assert result["errors"] == {CONF_CELEBRATION_ENABLED: "invalid"}
|
|
|
|
|
|
def test_options_flow_notification_stored_values_round_trip() -> None:
|
|
"""Stored notification settings are the form's defaults and a raw
|
|
submit without them keeps the stored values."""
|
|
entry = _make_entry(
|
|
options={
|
|
**VALID_INPUT,
|
|
CONF_UPDATE_INTERVAL: 90,
|
|
CONF_NOTIFY_TARGET: "notify.mobile_app_dad",
|
|
CONF_MORNING_SUMMARY_TIME: "07:15",
|
|
CONF_AFTERNOON_REMINDER_TIME: "16:45",
|
|
CONF_END_OF_DAY_REPORT_TIME: "21:00",
|
|
CONF_MORNING_SUMMARY_ENABLED: False,
|
|
CONF_AFTERNOON_REMINDER_ENABLED: True,
|
|
CONF_END_OF_DAY_REPORT_ENABLED: False,
|
|
CONF_CELEBRATION_ENABLED: True,
|
|
}
|
|
)
|
|
flow = _make_flow(entry)
|
|
form = _run(flow.async_step_init(None))
|
|
schema_defaults = form["data_schema"]({})
|
|
assert schema_defaults[CONF_NOTIFY_TARGET] == "notify.mobile_app_dad"
|
|
assert schema_defaults[CONF_MORNING_SUMMARY_TIME] == "07:15"
|
|
assert schema_defaults[CONF_END_OF_DAY_REPORT_ENABLED] is False
|
|
|
|
# A raw submit carrying only the legacy keys keeps the stored
|
|
# notification settings.
|
|
result = _run(flow.async_step_init(dict(VALID_INPUT)))
|
|
assert result["type"] == "create_entry"
|
|
assert result["data"][CONF_NOTIFY_TARGET] == "notify.mobile_app_dad"
|
|
assert result["data"][CONF_MORNING_SUMMARY_TIME] == "07:15"
|
|
assert result["data"][CONF_MORNING_SUMMARY_ENABLED] is False
|
|
assert result["data"][CONF_CELEBRATION_ENABLED] is True
|
|
|
|
|
|
def test_options_flow_no_hard_coded_personal_target_in_source() -> None:
|
|
"""The notify target's DEFAULT is the empty string: the household
|
|
fills their own service name, and no personal target ships as a
|
|
value anywhere in the const/flow defaults."""
|
|
from custom_components.nestquest import const
|
|
|
|
assert getattr(const, "CONF_NOTIFY_TARGET", None) == "notify_target"
|
|
# No const default carries a concrete notify service.
|
|
assert const.__dict__.get("DEFAULT_NOTIFY_TARGET", "") == ""
|
|
form = _run(_make_flow(_make_entry()).async_step_init(None))
|
|
assert form["data_schema"]({})[CONF_NOTIFY_TARGET] == ""
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Panel-plane API connection section (Feature 18)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_options_flow_api_section_defaults_saved() -> None:
|
|
"""A bare submit saves the API connection defaults: the const base
|
|
URL and an EMPTY panel token (no token is ever invented)."""
|
|
flow = _make_flow(_make_entry())
|
|
result = _run(flow.async_step_init(dict(VALID_INPUT)))
|
|
assert result["type"] == "create_entry"
|
|
assert result["data"][CONF_API_BASE_URL] == DEFAULT_API_BASE_URL
|
|
assert result["data"][CONF_PANEL_TOKEN] == ""
|
|
|
|
|
|
def test_options_flow_api_base_url_must_be_full_http_url() -> None:
|
|
"""Scheme-less, host-less, empty, and non-string base URLs reject
|
|
with a field error — the same rule the client's constructor
|
|
enforces, so a malformed URL fails at the form, not at the first
|
|
request."""
|
|
flow = _make_flow(_make_entry())
|
|
for bad_url in ("api.lan:8000", "http://", "", " ", 42):
|
|
result = _run(
|
|
flow.async_step_init(
|
|
{**VALID_INPUT, CONF_API_BASE_URL: bad_url}
|
|
)
|
|
)
|
|
assert result["type"] == "form"
|
|
assert result["errors"] == {CONF_API_BASE_URL: "invalid_url"}
|
|
|
|
|
|
def test_options_flow_panel_token_validation() -> None:
|
|
"""The EMPTY token is valid (the API connection is simply not
|
|
configured yet); padded and non-string values reject — a padded
|
|
paste would fail the token check at request time instead."""
|
|
flow = _make_flow(_make_entry())
|
|
result = _run(
|
|
flow.async_step_init({**VALID_INPUT, CONF_PANEL_TOKEN: ""})
|
|
)
|
|
assert result["type"] == "create_entry"
|
|
assert result["data"][CONF_PANEL_TOKEN] == ""
|
|
|
|
for bad_token in (" secret ", " ", 42):
|
|
flow = _make_flow(_make_entry())
|
|
result = _run(
|
|
flow.async_step_init(
|
|
{**VALID_INPUT, CONF_PANEL_TOKEN: bad_token}
|
|
)
|
|
)
|
|
assert result["type"] == "form"
|
|
assert result["errors"] == {CONF_PANEL_TOKEN: "invalid"}
|
|
|
|
|
|
def test_options_flow_api_section_stored_values_round_trip() -> None:
|
|
"""Stored API settings are the form's defaults and a raw legacy
|
|
submit without them keeps the stored values."""
|
|
entry = _make_entry(
|
|
options={
|
|
**VALID_INPUT,
|
|
CONF_API_BASE_URL: "http://panel.lan:8443",
|
|
CONF_PANEL_TOKEN: "stored-token",
|
|
}
|
|
)
|
|
flow = _make_flow(entry)
|
|
form = _run(flow.async_step_init(None))
|
|
schema_defaults = form["data_schema"]({})
|
|
assert schema_defaults[CONF_API_BASE_URL] == "http://panel.lan:8443"
|
|
assert schema_defaults[CONF_PANEL_TOKEN] == "stored-token"
|
|
|
|
# A raw submit carrying only the legacy keys keeps the stored
|
|
# API connection settings.
|
|
result = _run(flow.async_step_init(dict(VALID_INPUT)))
|
|
assert result["type"] == "create_entry"
|
|
assert result["data"][CONF_API_BASE_URL] == "http://panel.lan:8443"
|
|
assert result["data"][CONF_PANEL_TOKEN] == "stored-token"
|
|
|
|
|
|
def test_options_flow_strings_cover_api_section(
|
|
strings: dict, translations_en: dict
|
|
) -> None:
|
|
"""Both string files name the API fields and the URL error."""
|
|
data = strings["options"]["step"]["init"]["data"]
|
|
for key in (CONF_API_BASE_URL, CONF_PANEL_TOKEN):
|
|
assert data.get(key), f"Missing field name for '{key}'"
|
|
assert strings["options"]["errors"]["invalid_url"]
|
|
assert translations_en["options"] == strings["options"]
|