Coverage for custom_components/supernotify/config_flow.py: 98%
135 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-09-01 18:25 +0000
1"""Config flow for the Supernotify integration.
3A UI setup path that reproduces examples/minimal.yaml (everything auto-discovered, no required
4fields), plus options pages for the archive, dupe_check and housekeeping settings. Delivery,
5transports, scenarios, recipients, cameras, action_groups, links and snooze stay YAML-only, now
6under a top-level `supernotify:` key (see CONFIG_SCHEMA/async_setup in __init__.py) rather than
7the legacy `notify: - platform: supernotify` block - this config entry is the sole,
8unconditional owner of registering notify.supernotify in every case.
9"""
11from __future__ import annotations
13import logging
14from typing import Any
16import voluptuous as vol
17from anyio import Path
18from homeassistant.config_entries import ConfigEntry, ConfigFlow, ConfigFlowResult, OptionsFlow
19from homeassistant.const import CONF_ENABLED, CONF_NAME
20from homeassistant.data_entry_flow import section
21from homeassistant.helpers import config_validation as cv
22from homeassistant.helpers.selector import SelectSelector, SelectSelectorConfig
24from . import ARCHIVE_DIR, DOMAIN, MEDIA_DIR, TEMPLATE_DIR
25from .const import (
26 ATTR_DUPE_POLICY_MT,
27 ATTR_DUPE_POLICY_MTSLP,
28 ATTR_DUPE_POLICY_NONE,
29 CONF_ARCHIVE,
30 CONF_ARCHIVE_DAYS,
31 CONF_ARCHIVE_DIAGNOSTICS,
32 CONF_ARCHIVE_EVENT_NAME,
33 CONF_ARCHIVE_EVENT_SELECTION,
34 CONF_ARCHIVE_MQTT_QOS,
35 CONF_ARCHIVE_MQTT_RETAIN,
36 CONF_ARCHIVE_MQTT_TOPIC,
37 CONF_ARCHIVE_PATH,
38 CONF_ARCHIVE_PURGE_INTERVAL,
39 CONF_DUPE_CHECK,
40 CONF_DUPE_POLICY,
41 CONF_HOUSEKEEPING,
42 CONF_HOUSEKEEPING_TIME,
43 CONF_MEDIA_PATH,
44 CONF_MEDIA_STORAGE_DAYS,
45 CONF_MEDIA_URL_PREFIX,
46 CONF_MOBILE_DISCOVERY,
47 CONF_RECIPIENTS_DISCOVERY,
48 CONF_SIZE,
49 CONF_TEMPLATE_PATH,
50 CONF_TTL,
51)
52from .schema import OutcomeSelection
54_LOGGER = logging.getLogger(__name__)
56_DUPE_POLICIES = [ATTR_DUPE_POLICY_MTSLP, ATTR_DUPE_POLICY_MT, ATTR_DUPE_POLICY_NONE]
59def _event_policy_str(value: Any) -> str:
60 """Render an OutcomeSelection as the pipe-separated name string parse_event_policy
61 expects (e.g. "ERROR|DUPE"), whatever form it currently happens to be in.
63 A YAML-imported archive config already went through ARCHIVE_SCHEMA, which turns
64 event_selection/diagnostics into OutcomeSelection (IntFlag) instances - stored as a raw
65 int once round-tripped through config-entry storage. Left unconverted, that raw int would
66 show up as a bare number in this form instead of the "ERROR"-style text it's meant to be.
67 """
68 if isinstance(value, str):
69 return value
70 if isinstance(value, int):
71 try:
72 return OutcomeSelection(value).name or "NONE"
73 except ValueError:
74 return "NONE"
75 return str(value)
78# NONE isn't a real, independently selectable outcome - it's the empty bitmask, and an
79# always-false no-op check in archive.py (outcome_policy & OutcomeSelection.NONE is always
80# 0). "No outcomes ticked" already means NONE, so it's excluded from the checkbox list.
81# Option values are lowercased since HA selector translation keys must match [a-z0-9-_]+ -
82# the stored/parsed policy strings stay uppercase (OutcomeSelection member names).
83_OUTCOME_OPTIONS = [flag.name.lower() for flag in OutcomeSelection if flag != OutcomeSelection.NONE and flag.name]
86def _event_policy_to_list(value: Any) -> list[str]:
87 """Turn a stored OutcomeSelection value into the list of names a multi-select needs."""
88 policy_str = _event_policy_str(value)
89 return [] if policy_str in ("", "NONE") else [part.lower() for part in policy_str.split("|")]
92def _event_policy_from_list(values: list[str]) -> str:
93 """Turn a submitted multi-select list back into the pipe-separated string
94 ARCHIVE_SCHEMA's parse_event_policy expects."""
95 return "|".join(value.upper() for value in values) if values else "NONE"
98def extract_legacy_options(import_data: dict[str, Any]) -> dict[str, Any]:
99 """Pull the archive/dupe_check/housekeeping option blocks out of a legacy YAML config dict,
100 normalizing archive's event_selection/diagnostics the same way a fresh entry would.
102 Shared by async_step_import (fresh entry bootstrap) and repairs.py's migration flow, which
103 also needs this when merging legacy config into an entry that already exists - e.g. one
104 auto-bootstrapped blank by async_setup before the interactive repair ever runs (see
105 repairs.py's async_sync_entry_from_legacy_config).
106 """
107 archive: dict[str, Any] = dict(import_data.get(CONF_ARCHIVE) or {})
108 if CONF_ARCHIVE_EVENT_SELECTION in archive:
109 archive[CONF_ARCHIVE_EVENT_SELECTION] = _event_policy_str(archive[CONF_ARCHIVE_EVENT_SELECTION])
110 if CONF_ARCHIVE_DIAGNOSTICS in archive:
111 archive[CONF_ARCHIVE_DIAGNOSTICS] = _event_policy_str(archive[CONF_ARCHIVE_DIAGNOSTICS])
113 housekeeping: dict[str, Any] = dict(import_data.get(CONF_HOUSEKEEPING) or {})
114 housekeeping_time = housekeeping.get(CONF_HOUSEKEEPING_TIME)
115 if housekeeping_time is not None and not isinstance(housekeeping_time, str):
116 # cv.time on the YAML side already coerced this into a datetime.time
117 housekeeping[CONF_HOUSEKEEPING_TIME] = housekeeping_time.isoformat()
119 options: dict[str, Any] = {}
120 if archive:
121 options[CONF_ARCHIVE] = archive
122 if import_data.get(CONF_DUPE_CHECK):
123 options[CONF_DUPE_CHECK] = import_data[CONF_DUPE_CHECK]
124 if housekeeping:
125 options[CONF_HOUSEKEEPING] = housekeeping
126 return options
129def extract_legacy_data(import_data: dict[str, Any]) -> dict[str, Any]:
130 """Pull the template_path/media_path/media_url_prefix/mobile_discovery/recipients_discovery
131 settings out of a legacy YAML config dict, defaulting anything not set - the same defaults a
132 fresh entry would get. Deliberately excludes `name`, which repairs.py's
133 async_sync_entry_from_legacy_config merges in separately (it isn't defaulted the same way).
135 Shared by async_step_import (fresh entry bootstrap) and repairs.py's migration flow, which
136 also needs this when merging legacy config into an entry that already exists - e.g. one
137 auto-bootstrapped blank by async_setup before the interactive repair ever runs (see
138 repairs.py's async_sync_entry_from_legacy_config). Without this, a "simple" install with
139 nothing that needs a repair (no delivery/transports/scenarios/etc to move into
140 supernotify.yaml) would silently keep running on defaults forever, ignoring a customized
141 template_path/media_path/etc in the legacy block.
142 """
143 return {
144 CONF_TEMPLATE_PATH: import_data.get(CONF_TEMPLATE_PATH, TEMPLATE_DIR),
145 CONF_MEDIA_PATH: import_data.get(CONF_MEDIA_PATH, MEDIA_DIR),
146 CONF_MEDIA_URL_PREFIX: import_data.get(CONF_MEDIA_URL_PREFIX, "/supernotify/media"),
147 CONF_MOBILE_DISCOVERY: import_data.get(CONF_MOBILE_DISCOVERY, True),
148 CONF_RECIPIENTS_DISCOVERY: import_data.get(CONF_RECIPIENTS_DISCOVERY, True),
149 }
152def _user_schema(defaults: dict[str, Any] | None = None) -> vol.Schema:
153 # cv.string, not cv.path: cv.path fails voluptuous-serialize schema conversion used by
154 # the config flow frontend ("Unable to convert schema" / HTTP 500).
155 defaults = defaults or {}
156 return vol.Schema({
157 # Determines the registered notify.<name> action (slugified) - matches the legacy
158 # notify platform's `name:` YAML field, which this replaces as the sole source of truth.
159 vol.Optional(CONF_NAME, default=defaults.get(CONF_NAME, DOMAIN)): cv.string,
160 vol.Optional(CONF_TEMPLATE_PATH, default=defaults.get(CONF_TEMPLATE_PATH, TEMPLATE_DIR)): cv.string,
161 vol.Optional(CONF_MEDIA_PATH, default=defaults.get(CONF_MEDIA_PATH, MEDIA_DIR)): cv.string,
162 vol.Optional(CONF_MEDIA_URL_PREFIX, default=defaults.get(CONF_MEDIA_URL_PREFIX, "/supernotify/media")): cv.string,
163 vol.Optional(CONF_MOBILE_DISCOVERY, default=defaults.get(CONF_MOBILE_DISCOVERY, True)): cv.boolean,
164 vol.Optional(CONF_RECIPIENTS_DISCOVERY, default=defaults.get(CONF_RECIPIENTS_DISCOVERY, True)): cv.boolean,
165 })
168async def _ensure_directory_exists(path_str: str) -> str | None:
169 """Create a directory if it doesn't exist yet, mirroring MediaStorage.initialize()'s own
170 tolerant runtime behavior (media_grab.py). Returns an error description on a genuine
171 failure to create/access it, None on success."""
172 try:
173 path = Path(path_str)
174 if not path.is_absolute():
175 path = await path.absolute()
176 if not await path.exists():
177 await path.mkdir(parents=True, exist_ok=True)
178 except (OSError, ValueError) as err:
179 return str(err)
180 return None
183async def _validate_user_input(user_input: dict[str, Any]) -> dict[str, str]:
184 """Validate template_path/media_path at config-flow submission time, creating each
185 directory if it doesn't exist yet. Catching a genuine failure here surfaces a typo'd or
186 unwritable path immediately in the wizard, instead of leaving the user to discover it
187 later via a repair issue. A blank value is left untouched - both fields are optional.
188 """
189 errors: dict[str, str] = {}
191 template_path = user_input.get(CONF_TEMPLATE_PATH)
192 if template_path:
193 error = await _ensure_directory_exists(template_path)
194 if error:
195 errors[CONF_TEMPLATE_PATH] = "template_path_invalid"
196 _LOGGER.debug("SUPERNOTIFY Invalid template_path %s: %s", template_path, error)
198 media_path = user_input.get(CONF_MEDIA_PATH)
199 if media_path:
200 error = await _ensure_directory_exists(media_path)
201 if error:
202 errors[CONF_MEDIA_PATH] = "media_path_invalid"
203 _LOGGER.debug("SUPERNOTIFY Invalid media_path %s: %s", media_path, error)
205 return errors
208class SupernotifyConfigFlow(ConfigFlow, domain=DOMAIN):
209 """Handle a Supernotify config flow."""
211 VERSION = 1
213 async def async_step_user(self, user_input: dict[str, Any] | None = None) -> ConfigFlowResult:
214 errors: dict[str, str] = {}
215 if user_input is not None:
216 errors = await _validate_user_input(user_input)
217 if not errors:
218 return self.async_create_entry(title="Supernotify", data=user_input)
219 return self.async_show_form(step_id="user", data_schema=_user_schema(user_input), errors=errors)
221 async def async_step_reconfigure(self, user_input: dict[str, Any] | None = None) -> ConfigFlowResult:
222 """Let the user change the global settings of an existing entry.
224 These fields are config-entry *data* (set at initial setup), not options, so HA's
225 guidance is a reconfigure step rather than the options flow - which instead covers
226 archive/dupe_check/housekeeping, genuine runtime preferences.
227 """
228 entry = self._get_reconfigure_entry()
229 errors: dict[str, str] = {}
230 if user_input is not None:
231 errors = await _validate_user_input(user_input)
232 if not errors:
233 # async_update_and_abort, not async_update_reload_and_abort: async_setup_entry
234 # registers an update listener, which async_update_entry already fires (and
235 # reloads via) whenever entry.data changes - calling the "_reload" variant too
236 # would reload twice and log an HA deprecation warning about the redundancy.
237 return self.async_update_and_abort(entry, data_updates=user_input)
238 defaults = user_input if user_input is not None else dict(entry.data)
239 return self.async_show_form(step_id="reconfigure", data_schema=_user_schema(defaults), errors=errors)
241 async def async_step_import(self, import_data: dict[str, Any]) -> ConfigFlowResult:
242 """Bootstrap a config entry, optionally seeded from a legacy YAML config dict.
244 Two callers: __init__.py's async_setup (a from-scratch install with no entry yet) and
245 async_reload_yaml_config_and_entries's bootstrap, and repairs.py's migration flow (a
246 leftover legacy `notify: - platform: supernotify` block, possibly with real
247 archive/housekeeping/dupe_check/name/template_path/etc already set) - both pass
248 `data={}` for a genuinely fresh install, or the raw legacy config dict to preserve an
249 existing installation's settings (notably `name`, which determines the registered
250 notify.<name> action - see __init__.py's async_setup_entry).
252 Duplicate-entry protection is the same single_config_entry manifest flag used by the
253 user step (it covers SOURCE_IMPORT too), so no unique_id bookkeeping is needed here.
254 """
255 data: dict[str, Any] = {
256 CONF_NAME: import_data.get(CONF_NAME, DOMAIN),
257 **extract_legacy_data(import_data),
258 }
260 options = extract_legacy_options(import_data)
262 _LOGGER.info(
263 "SUPERNOTIFY Config entry bootstrapped (data=%s,options=%s)",
264 ";".join(data.keys()),
265 ";".join(options.keys()),
266 )
267 return self.async_create_entry(title="Supernotify", data=data, options=options)
269 @staticmethod
270 def async_get_options_flow(config_entry: ConfigEntry) -> SupernotifyOptionsFlow:
271 return SupernotifyOptionsFlow()
274class SupernotifyOptionsFlow(OptionsFlow):
275 """Options pages for archive, dupe_check and housekeeping settings."""
277 async def async_step_init(self, user_input: dict[str, Any] | None = None) -> ConfigFlowResult:
278 return self.async_show_menu(step_id="init", menu_options=["archive", "dupe_check", "housekeeping"])
280 async def async_step_archive(self, user_input: dict[str, Any] | None = None) -> ConfigFlowResult:
281 current: dict[str, Any] = self.config_entry.options.get(CONF_ARCHIVE, {})
282 if user_input is not None:
283 file_section = user_input["file"]
284 mqtt_section = user_input["mqtt"]
285 event_section = user_input["event"]
286 processed = {
287 CONF_ENABLED: user_input[CONF_ENABLED],
288 CONF_ARCHIVE_PATH: file_section[CONF_ARCHIVE_PATH],
289 CONF_ARCHIVE_DAYS: file_section[CONF_ARCHIVE_DAYS],
290 CONF_ARCHIVE_PURGE_INTERVAL: file_section[CONF_ARCHIVE_PURGE_INTERVAL],
291 CONF_ARCHIVE_MQTT_TOPIC: mqtt_section[CONF_ARCHIVE_MQTT_TOPIC],
292 CONF_ARCHIVE_MQTT_QOS: mqtt_section[CONF_ARCHIVE_MQTT_QOS],
293 CONF_ARCHIVE_MQTT_RETAIN: mqtt_section[CONF_ARCHIVE_MQTT_RETAIN],
294 CONF_ARCHIVE_EVENT_NAME: event_section[CONF_ARCHIVE_EVENT_NAME],
295 CONF_ARCHIVE_EVENT_SELECTION: _event_policy_from_list(event_section[CONF_ARCHIVE_EVENT_SELECTION]),
296 CONF_ARCHIVE_DIAGNOSTICS: _event_policy_from_list(event_section[CONF_ARCHIVE_DIAGNOSTICS]),
297 }
298 return self.async_create_entry(title="", data={**self.config_entry.options, CONF_ARCHIVE: processed})
299 outcome_selector = SelectSelector(
300 SelectSelectorConfig(options=_OUTCOME_OPTIONS, multiple=True, translation_key="outcome_selection")
301 )
302 schema = vol.Schema({
303 vol.Optional(CONF_ENABLED, default=False): cv.boolean,
304 # section wrapper keys must be vol.Required, not vol.Optional: with Optional, the
305 # frontend silently ignores both default and suggested_value for everything inside
306 # (and for the rest of the form too) - https://github.com/home-assistant/frontend/issues/22419
307 # No default= here - matches the working pattern in this repo's other integrations
308 # (autoarm, remote_logger).
309 vol.Required("file"): section(
310 vol.Schema({
311 # cv.string, not cv.path: cv.path fails voluptuous-serialize schema
312 # conversion used by the config flow frontend ("Unable to convert
313 # schema" / HTTP 500).
314 vol.Optional(CONF_ARCHIVE_PATH, default=ARCHIVE_DIR): cv.string,
315 vol.Optional(CONF_ARCHIVE_DAYS, default=3): cv.positive_int,
316 vol.Optional(CONF_ARCHIVE_PURGE_INTERVAL, default=60): cv.positive_int,
317 }),
318 {"collapsed": False},
319 ),
320 vol.Required("mqtt"): section(
321 vol.Schema({
322 vol.Optional(CONF_ARCHIVE_MQTT_TOPIC, default=""): cv.string,
323 vol.Optional(CONF_ARCHIVE_MQTT_QOS, default=0): cv.positive_int,
324 vol.Optional(CONF_ARCHIVE_MQTT_RETAIN, default=True): cv.boolean,
325 }),
326 {"collapsed": True},
327 ),
328 vol.Required("event"): section(
329 vol.Schema({
330 vol.Optional(CONF_ARCHIVE_EVENT_NAME, default="supernotification"): cv.string,
331 vol.Optional(CONF_ARCHIVE_EVENT_SELECTION, default=[]): outcome_selector,
332 vol.Optional(CONF_ARCHIVE_DIAGNOSTICS, default=[]): outcome_selector,
333 }),
334 {"collapsed": True},
335 ),
336 })
337 # A plain vol.Optional(default=...) only sets the server-side validation fallback -
338 # once a section is present, the frontend needs description.suggested_value (for
339 # every field, not just the section-nested ones) to pre-fill an existing entry's
340 # current values.
341 suggested_values = {
342 CONF_ENABLED: current.get(CONF_ENABLED, False),
343 "file": {
344 CONF_ARCHIVE_PATH: current.get(CONF_ARCHIVE_PATH, ARCHIVE_DIR),
345 CONF_ARCHIVE_DAYS: current.get(CONF_ARCHIVE_DAYS, 3),
346 CONF_ARCHIVE_PURGE_INTERVAL: current.get(CONF_ARCHIVE_PURGE_INTERVAL, 60),
347 },
348 "mqtt": {
349 CONF_ARCHIVE_MQTT_TOPIC: current.get(CONF_ARCHIVE_MQTT_TOPIC, ""),
350 CONF_ARCHIVE_MQTT_QOS: current.get(CONF_ARCHIVE_MQTT_QOS, 0),
351 CONF_ARCHIVE_MQTT_RETAIN: current.get(CONF_ARCHIVE_MQTT_RETAIN, True),
352 },
353 "event": {
354 CONF_ARCHIVE_EVENT_NAME: current.get(CONF_ARCHIVE_EVENT_NAME, "supernotification"),
355 CONF_ARCHIVE_EVENT_SELECTION: _event_policy_to_list(current.get(CONF_ARCHIVE_EVENT_SELECTION, "NONE")),
356 CONF_ARCHIVE_DIAGNOSTICS: _event_policy_to_list(current.get(CONF_ARCHIVE_DIAGNOSTICS, "ERROR")),
357 },
358 }
359 schema = self.add_suggested_values_to_schema(schema, suggested_values)
360 return self.async_show_form(step_id="archive", data_schema=schema)
362 async def async_step_dupe_check(self, user_input: dict[str, Any] | None = None) -> ConfigFlowResult:
363 current: dict[str, Any] = self.config_entry.options.get(CONF_DUPE_CHECK, {})
364 if user_input is not None:
365 return self.async_create_entry(title="", data={**self.config_entry.options, CONF_DUPE_CHECK: user_input})
366 schema = vol.Schema({
367 vol.Optional(CONF_TTL, default=current.get(CONF_TTL, 120)): cv.positive_int,
368 vol.Optional(CONF_SIZE, default=current.get(CONF_SIZE, 100)): cv.positive_int,
369 vol.Optional(CONF_DUPE_POLICY, default=current.get(CONF_DUPE_POLICY, ATTR_DUPE_POLICY_MTSLP)): SelectSelector(
370 SelectSelectorConfig(options=_DUPE_POLICIES, translation_key="dupe_policy")
371 ),
372 })
373 return self.async_show_form(step_id="dupe_check", data_schema=schema)
375 async def async_step_housekeeping(self, user_input: dict[str, Any] | None = None) -> ConfigFlowResult:
376 current: dict[str, Any] = self.config_entry.options.get(CONF_HOUSEKEEPING, {})
377 if user_input is not None:
378 return self.async_create_entry(title="", data={**self.config_entry.options, CONF_HOUSEKEEPING: user_input})
379 housekeeping_time = current.get(CONF_HOUSEKEEPING_TIME, "00:00:01")
380 if not isinstance(housekeeping_time, str):
381 # a YAML-imported entry may still have a raw datetime.time from before this was fixed
382 housekeeping_time = housekeeping_time.isoformat()
383 schema = vol.Schema({
384 vol.Optional(CONF_HOUSEKEEPING_TIME, default=housekeeping_time): cv.string,
385 vol.Optional(CONF_MEDIA_STORAGE_DAYS, default=current.get(CONF_MEDIA_STORAGE_DAYS, 7)): cv.positive_int,
386 })
387 return self.async_show_form(step_id="housekeeping", data_schema=schema)