Coverage for custom_components/supernotify/__init__.py: 97%

1"""The Supernotify integration""" 

2 

3from __future__ import annotations 

4 

5import logging 

6from typing import TYPE_CHECKING 

7 

8import voluptuous as vol 

9from homeassistant.const import CONF_NAME, SERVICE_RELOAD, Platform 

10from homeassistant.exceptions import ConfigEntryNotReady 

11from homeassistant.helpers.reload import async_integration_yaml_config 

12from homeassistant.helpers.service import async_register_admin_service 

13from homeassistant.loader import async_get_integration 

14from homeassistant.util import slugify 

15 

16if TYPE_CHECKING: 

17 from homeassistant.config_entries import ConfigEntry 

18 from homeassistant.core import HomeAssistant, ServiceCall 

19 from homeassistant.helpers.typing import ConfigType 

20 

21 from .notify import SupernotifyAction 

22 

23 type SupernotifyConfigEntry = ConfigEntry[SupernotifyAction] 

24 

25DOMAIN = "supernotify" 

26 

27TEMPLATE_DIR: str = "supernotify/templates" 

28MEDIA_DIR: str = "supernotify/media" 

29ARCHIVE_DIR: str = "supernotify/archive" 

30 

31_LOGGER = logging.getLogger(__name__) 

32 

33NOTIFY_SERVICE_NAME = "supernotify" 

34 

35# Key under hass.data[DOMAIN] holding the validated top-level `supernotify:` YAML section 

36# (delivery/transports/scenarios/recipients/cameras/action_groups/links/snooze - the 8 keys not 

37# yet configurable via ConfigFlow). Populated by async_setup, read by _entry_full_config. 

38KEY_YAML_CONFIG = "yaml_config" 

39 

40# Deferred import: schema.py imports ARCHIVE_DIR/MEDIA_DIR/TEMPLATE_DIR back from this module, so it can 

41# only be imported here once those (and DOMAIN) are already defined above. 

42from .schema import SUPERNOTIFY_YAML_SCHEMA # noqa: E402, RUF100, I001 

43 

44CONFIG_SCHEMA: vol.Schema = vol.Schema( 

45 {vol.Optional(DOMAIN, default=dict): SUPERNOTIFY_YAML_SCHEMA}, 

46 extra=vol.ALLOW_EXTRA, 

47) 

48 

49 

50async def async_reload_yaml_config_and_entries(hass: HomeAssistant) -> None: 

51 """Re-read the top-level `supernotify:` YAML section from disk and reload every entry. 

52 

53 Shared by the supernotify.reload service below and repairs.py's migration flow (which needs 

54 the freshly-migrated `supernotify.yaml` picked up immediately, without a restart). 

55 """ 

56 new_config = await async_integration_yaml_config(hass, DOMAIN) 

57 hass.data.setdefault(DOMAIN, {})[KEY_YAML_CONFIG] = (new_config or {}).get(DOMAIN, {}) 

58 for entry in hass.config_entries.async_entries(DOMAIN): 

59 await hass.config_entries.async_reload(entry.entry_id) 

60 

61 

62async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool: 

63 """Stash the top-level `supernotify:` YAML section and wire up supernotify.reload. 

64 

65 This is the only place that YAML section is ever read for a running instance - editing it 

66 and calling supernotify.reload (or the repairs.py migration flow) re-reads it and reloads 

67 the config entry, which is the sole, unconditional owner of notify.supernotify. 

68 """ 

69 hass.data.setdefault(DOMAIN, {})[KEY_YAML_CONFIG] = config.get(DOMAIN, {}) 

70 

71 async def _async_reload(_call: ServiceCall) -> None: 

72 await async_reload_yaml_config_and_entries(hass) 

73 

74 async_register_admin_service(hass, DOMAIN, SERVICE_RELOAD, _async_reload) 

75 

76 if not hass.config_entries.async_entries(DOMAIN): 

77 from homeassistant.config_entries import SOURCE_IMPORT 

78 

79 hass.async_create_task(hass.config_entries.flow.async_init(DOMAIN, context={"source": SOURCE_IMPORT}, data={})) 

80 

81 return True 

82 

83 

84def _entry_full_config(hass: HomeAssistant, entry: SupernotifyConfigEntry) -> ConfigType: 

85 """Fold a config entry's data/options and the stashed top-level YAML section into a full 

86 FULL_CONFIG_SCHEMA config dict. 

87 

88 entry.data holds the flat "user" step fields; entry.options holds the nested 

89 archive/dupe_check/housekeeping sections from the options flow - both validated here via 

90 CONFIG_ENTRY_SCHEMA. hass.data[DOMAIN] holds the delivery/transports/scenarios/recipients/ 

91 cameras/action_groups/links/snooze YAML section (see async_setup) - already validated once 

92 by CONFIG_SCHEMA when that YAML was first loaded, so it's merged in as-is rather than 

93 re-validated (a second pass would reject values schema validation already coerced, e.g. 

94 cv.template's Template objects - see CONFIG_ENTRY_SCHEMA's docstring in schema.py). 

95 """ 

96 from .schema import CONFIG_ENTRY_SCHEMA 

97 

98 yaml_config = hass.data.get(DOMAIN, {}).get(KEY_YAML_CONFIG, {}) 

99 return {**CONFIG_ENTRY_SCHEMA({**entry.data, **entry.options}), **yaml_config} 

100 

101 

102async def async_setup_entry(hass: HomeAssistant, entry: SupernotifyConfigEntry) -> bool: 

103 from .notification import set_version 

104 from .notify import async_register_supplemental_services, build_supernotify_action 

105 

106 integration = await async_get_integration(hass, DOMAIN) 

107 set_version(str(integration.version) if integration.version else "unknown") 

108 

109 # Matches the slugify(conf_name or SERVICE_NOTIFY) logic the legacy notify platform loader 

110 # used to apply to a YAML `name:` field, so an existing custom notify.<name> action (e.g. 

111 # `name: SuperNotifier` -> notify.supernotifier) keeps working once this entry becomes the 

112 # sole owner of registering it - see repairs.py, which carries a migrated entry's name over. 

113 service_name = slugify(entry.data.get(CONF_NAME) or NOTIFY_SERVICE_NAME) 

114 

115 if hass.services.has_service("notify", service_name): 

116 _LOGGER.warning( 

117 "SUPERNOTIFY notify.%s is already registered - not registering it again from this config entry", 

118 service_name, 

119 ) 

120 return True 

121 

122 full_config = _entry_full_config(hass, entry) 

123 service: SupernotifyAction = build_supernotify_action(hass, full_config) 

124 try: 

125 await service.initialize() 

126 except Exception as err: 

127 _LOGGER.exception("SUPERNOTIFY Failed to initialize, will retry") 

128 raise ConfigEntryNotReady(f"SUPERNOTIFY Failed to initialize: {err}") from err 

129 await service.async_setup(hass, service_name, service_name) 

130 await service.async_register_services() 

131 async_register_supplemental_services(hass, service, full_config) 

132 entry.runtime_data = service 

133 entry.async_on_unload(entry.add_update_listener(_async_update_listener)) 

134 await hass.config_entries.async_forward_entry_setups(entry, [Platform.NOTIFY]) 

135 return True 

136 

137 

138async def _async_update_listener(hass: HomeAssistant, entry: SupernotifyConfigEntry) -> None: 

139 """Reload the entry so archive/dupe_check/housekeeping options apply immediately.""" 

140 await hass.config_entries.async_reload(entry.entry_id) 

141 

142 

143async def async_unload_entry(hass: HomeAssistant, entry: SupernotifyConfigEntry) -> bool: 

144 from .notify import async_unregister_supplemental_services 

145 

146 unload_ok = await hass.config_entries.async_unload_platforms(entry, [Platform.NOTIFY]) 

147 service: SupernotifyAction | None = getattr(entry, "runtime_data", None) 

148 if service is not None: 

149 await service.async_unregister_services() 

150 async_unregister_supplemental_services(hass) 

151 return unload_ok