Maniacs_Mansion/NestQuest
11
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Feature 18: HA Integration as API Client (#173)

Browse Source 
Co-authored-by: dev_agent <17+dev_agent@code.cubecraftcreations.com>
Co-committed-by: dev_agent <17+dev_agent@code.cubecraftcreations.com>
This commit was merged in pull request #173.
This commit is contained in:
dev_agent
2026-09-23 09:19:48 -04:00
committed by approver_agent
parent d78ef196d0
commit 6059a67122
79 changed files with 4874 additions and 7457 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_options_flow.py
+212 -452
View File
@@ -14,27 +14,16 @@ from conftest import make_config_entry, make_hass, options_flow_base as _options
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_DAY_ROLLOVER_TIME,
CONF_HORIZON_DAYS,
CONF_API_BASE_URL,
CONF_PANEL_IDLE_TIMEOUT,
CONF_PANEL_TOKEN,
CONF_SNAPSHOT_STALENESS,
CONF_UPDATE_INTERVAL,
CONF_AFTERNOON_REMINDER_ENABLED,
CONF_AFTERNOON_REMINDER_TIME,
CONF_CELEBRATION_ENABLED,
CONF_END_OF_DAY_REPORT_ENABLED,
CONF_END_OF_DAY_REPORT_TIME,
CONF_MORNING_SUMMARY_ENABLED,
CONF_MORNING_SUMMARY_TIME,
CONF_NOTIFY_TARGET,
DEFAULT_AFTERNOON_REMINDER_TIME,
DEFAULT_AUTOMATION_ENABLED,
DEFAULT_END_OF_DAY_REPORT_TIME,
DEFAULT_MORNING_SUMMARY_TIME,
DEFAULT_UPDATE_INTERVAL,
DEFAULT_DAY_ROLLOVER_TIME,
DEFAULT_HORIZON_DAYS,
DEFAULT_API_BASE_URL,
DEFAULT_PANEL_IDLE_TIMEOUT,
DEFAULT_PANEL_TOKEN,
DEFAULT_SNAPSHOT_STALENESS,
DEFAULT_UPDATE_INTERVAL,
DOMAIN,
)
@@ -43,25 +32,19 @@ 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 and the Feature 11
#: notification section fill from their defaults.
#: fresh entry: the Feature 10 refresh interval, the Feature 18
#: staleness threshold, 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_SNAPSHOT_STALENESS: DEFAULT_SNAPSHOT_STALENESS,
CONF_API_BASE_URL: DEFAULT_API_BASE_URL,
CONF_PANEL_TOKEN: DEFAULT_PANEL_TOKEN,
}
@@ -77,48 +60,13 @@ def _make_entry(entry_id: str = "test_entry", options: dict | None = None, data:
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
def _make_flow(entry) -> options_flow.NestQuestOptionsFlow:
"""Create an options flow instance bound to a stubbed config entry."""
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 {
@@ -145,11 +93,11 @@ def test_options_flow_get_options_flow_is_ha_callback() -> None:
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})
entry = _make_entry(options={CONF_UPDATE_INTERVAL: 90})
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}
assert flow.options == {CONF_UPDATE_INTERVAL: 90}
def test_options_flow_submits_through_2024_6_surface() -> None:
@@ -184,82 +132,66 @@ def test_options_flow_prefills_from_defaults() -> None:
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
assert values[CONF_UPDATE_INTERVAL] == DEFAULT_UPDATE_INTERVAL
assert values[CONF_SNAPSHOT_STALENESS] == DEFAULT_SNAPSHOT_STALENESS
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,
CONF_UPDATE_INTERVAL: 90,
CONF_SNAPSHOT_STALENESS: 450,
},
data={
CONF_HORIZON_DAYS: 99,
CONF_DAY_ROLLOVER_TIME: "23:59",
CONF_PANEL_IDLE_TIMEOUT: 999,
CONF_UPDATE_INTERVAL: 999,
CONF_SNAPSHOT_STALENESS: 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
assert values[CONF_UPDATE_INTERVAL] == 90
assert values[CONF_SNAPSHOT_STALENESS] == 450
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,
CONF_UPDATE_INTERVAL: 60,
CONF_SNAPSHOT_STALENESS: 240,
}
)
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
assert values[CONF_UPDATE_INTERVAL] == 60
assert values[CONF_SNAPSHOT_STALENESS] == 240
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.
"""
"""A real voluptuous validate({}) fills every field from the stored defaults."""
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",
CONF_SNAPSHOT_STALENESS: 450,
CONF_API_BASE_URL: "http://panel.lan:8443",
CONF_PANEL_TOKEN: "stored-token",
}
)
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_ADMIN_USER_IDS: [],
CONF_SNAPSHOT_STALENESS: 450,
CONF_API_BASE_URL: "http://panel.lan:8443",
CONF_PANEL_TOKEN: "stored-token",
}
@@ -269,50 +201,12 @@ def test_options_flow_schema_coerces_field_types() -> None:
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
@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"}
assert validators[CONF_UPDATE_INTERVAL] is int
assert validators[CONF_SNAPSHOT_STALENESS] 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_timeout", ["abc", 0, -10, 10])
@@ -326,8 +220,21 @@ def test_options_flow_rejects_invalid_idle_timeout(bad_timeout) -> None:
assert result["errors"] == {CONF_PANEL_IDLE_TIMEOUT: "invalid"}
def test_options_flow_creates_entry_with_four_keys() -> None:
"""A valid submit creates an entry with exactly the four validated keys."""
@pytest.mark.parametrize("bad_interval", ["abc", True, 0, -10, 29])
def test_options_flow_rejects_invalid_update_interval(bad_interval) -> None:
"""Non-numeric, boolean, and below-floor update intervals reject."""
flow = _make_flow(_make_entry())
result = _run(
flow.async_step_init({**VALID_INPUT, CONF_UPDATE_INTERVAL: bad_interval})
)
assert result["type"] == "form"
assert result["errors"] == {CONF_UPDATE_INTERVAL: "invalid"}
def test_options_flow_creates_entry_with_validated_keys() -> None:
"""A valid submit creates an entry with exactly the validated keys
(the client-facing settings, 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"
@@ -342,7 +249,7 @@ def test_options_flow_create_entry_data_lands_in_entry_options_and_reloads_once(
entry = _make_entry()
entry.add_update_listener = lambda listener: registry.add(entry.entry_id, listener)
# Simulate setup registering the merged lifecycle update listener.
# Simulate setup registering the lifecycle update listener.
from custom_components.nestquest import async_setup_entry
_run(async_setup_entry(hass, entry))
@@ -361,14 +268,13 @@ def test_options_flow_create_entry_data_lands_in_entry_options_and_reloads_once(
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}
junk = {CONF_PANEL_IDLE_TIMEOUT: None, CONF_UPDATE_INTERVAL: "abc"}
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",
CONF_UPDATE_INTERVAL: "invalid",
}
@@ -398,10 +304,10 @@ def test_options_strings_present_and_matching(strings: dict, translations_en: di
assert init["title"]
assert init["description"]
data = init["data"]
for key in (CONF_HORIZON_DAYS, CONF_DAY_ROLLOVER_TIME, CONF_PANEL_IDLE_TIMEOUT):
for key in (CONF_PANEL_IDLE_TIMEOUT,):
assert data.get(key), f"Missing field name for '{key}'"
errors = options["errors"]
for key in ("invalid", "invalid_time"):
for key in ("invalid",):
assert errors.get(key), f"Missing error string for '{key}'"
@@ -457,345 +363,199 @@ def test_options_flow_no_hardcoded_domain_or_db_filename() -> None:
f"Hard-coded database filename in {py_file.name}:{node.lineno}"
)
# ---------------------------------------------------------------------------
# Admin allowlist picker (Feature 03)
# Panel-plane API connection section (Feature 18)
# ---------------------------------------------------------------------------
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."""
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_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
assert result["data"][CONF_API_BASE_URL] == DEFAULT_API_BASE_URL
assert result["data"][CONF_PANEL_TOKEN] == ""
def test_options_flow_notification_times_must_be_strict_hh_mm() -> None:
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_time in ("9:00", "25:00", "08:0", "08-00", 800):
for bad_url in ("api.lan:8000", "http://", "", " ", 42):
result = _run(
flow.async_step_init(
{
**VALID_INPUT,
CONF_MORNING_SUMMARY_TIME: bad_time,
}
{**VALID_INPUT, CONF_API_BASE_URL: bad_url}
)
)
assert result["type"] == "form"
assert result["errors"] == {CONF_MORNING_SUMMARY_TIME: "invalid_time"}
assert result["errors"] == {CONF_API_BASE_URL: "invalid_url"}
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."""
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())
# The empty default saves cleanly.
result = _run(
flow.async_step_init(
{**VALID_INPUT, CONF_NOTIFY_TARGET: ""}
)
flow.async_step_init({**VALID_INPUT, CONF_PANEL_TOKEN: ""})
)
assert result["type"] == "create_entry"
assert result["data"][CONF_NOTIFY_TARGET] == ""
assert result["data"][CONF_PANEL_TOKEN] == ""
# Malformed non-empty values reject.
for bad_target in (" notify.x ", " ", 42):
for bad_token in (" secret ", " ", 42):
flow = _make_flow(_make_entry())
result = _run(
flow.async_step_init(
{
**VALID_INPUT,
CONF_NOTIFY_TARGET: bad_target,
}
{**VALID_INPUT, CONF_PANEL_TOKEN: bad_token}
)
)
assert result["type"] == "form"
assert result["errors"] == {CONF_NOTIFY_TARGET: "invalid"}
assert result["errors"] == {CONF_PANEL_TOKEN: "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
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_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,
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_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
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
# notification settings.
# A raw submit carrying only the panel-idle key keeps the stored
# API connection 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
assert result["data"][CONF_API_BASE_URL] == "http://panel.lan:8443"
assert result["data"][CONF_PANEL_TOKEN] == "stored-token"
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
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"]
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] == ""
# ---------------------------------------------------------------------------
# Last-good snapshot staleness section (Feature 18)
# ---------------------------------------------------------------------------
def test_options_flow_staleness_below_update_interval_refused() -> None:
"""A staleness threshold shorter than the submit's update interval
is refused with the dedicated error: such a threshold could never
serve even one cached snapshot."""
flow = _make_flow(_make_entry())
result = _run(
flow.async_step_init(
{
**VALID_INPUT,
CONF_UPDATE_INTERVAL: 300,
CONF_SNAPSHOT_STALENESS: 60,
}
)
)
assert result["type"] == "form"
assert result["errors"] == {CONF_SNAPSHOT_STALENESS: "invalid_staleness"}
def test_options_flow_staleness_below_stored_interval_refused() -> None:
"""A raw submit setting the staleness but omitting the update
interval is checked against the STORED interval, not the default."""
entry = _make_entry(options={CONF_UPDATE_INTERVAL: 300})
flow = _make_flow(entry)
result = _run(
flow.async_step_init({**VALID_INPUT, CONF_SNAPSHOT_STALENESS: 60})
)
assert result["type"] == "form"
assert result["errors"] == {CONF_SNAPSHOT_STALENESS: "invalid_staleness"}
def test_options_flow_staleness_must_be_positive_int() -> None:
"""Non-int, boolean, and non-positive staleness values reject with
the plain invalid error."""
flow = _make_flow(_make_entry())
for bad_staleness in ("900", True, 0, -10):
result = _run(
flow.async_step_init(
{**VALID_INPUT, CONF_SNAPSHOT_STALENESS: bad_staleness}
)
)
assert result["type"] == "form"
assert result["errors"] == {CONF_SNAPSHOT_STALENESS: "invalid"}
def test_options_flow_staleness_equal_to_interval_saves() -> None:
"""A threshold exactly as long as the update interval is valid (the
boundary is inclusive)."""
flow = _make_flow(_make_entry())
result = _run(
flow.async_step_init(
{
**VALID_INPUT,
CONF_UPDATE_INTERVAL: 300,
CONF_SNAPSHOT_STALENESS: 300,
}
)
)
assert result["type"] == "create_entry"
assert result["data"][CONF_SNAPSHOT_STALENESS] == 300
def test_options_flow_staleness_defaults_saved() -> None:
"""A bare submit saves the staleness default (three update
intervals) alongside the other defaults."""
flow = _make_flow(_make_entry())
result = _run(flow.async_step_init(dict(VALID_INPUT)))
assert result["type"] == "create_entry"
assert result["data"][CONF_SNAPSHOT_STALENESS] == DEFAULT_SNAPSHOT_STALENESS
def test_options_flow_staleness_stored_values_round_trip() -> None:
"""A stored staleness is the form's default and a raw legacy submit
without it keeps the stored value."""
entry = _make_entry(
options={
**VALID_INPUT,
CONF_UPDATE_INTERVAL: 300,
CONF_SNAPSHOT_STALENESS: 1200,
}
)
flow = _make_flow(entry)
form = _run(flow.async_step_init(None))
schema_defaults = form["data_schema"]({})
assert schema_defaults[CONF_SNAPSHOT_STALENESS] == 1200
# A raw submit carrying only the panel-idle key keeps the stored
# staleness (and never trips the interval cross-check).
result = _run(flow.async_step_init(dict(VALID_INPUT)))
assert result["type"] == "create_entry"
assert result["data"][CONF_SNAPSHOT_STALENESS] == 1200
def test_options_flow_strings_cover_staleness_section(
strings: dict, translations_en: dict
) -> None:
"""Both string files name the staleness field and its error."""
data = strings["options"]["step"]["init"]["data"]
assert data.get(CONF_SNAPSHOT_STALENESS), (
f"Missing field name for '{CONF_SNAPSHOT_STALENESS}'"
)
assert strings["options"]["errors"]["invalid_staleness"]
assert translations_en["options"] == strings["options"]