23368799114

Committed 21 Mar 2026 01:12AM UTC coverage: 59.232% (+3.1%) from 56.166%

Build # 23368799114

Build Type

push

github

Committed by

web-flow

Commit Message

Repair issue if telegram_bot configuration become missing (#30)

* Repair issue if telegram_bot configuration become missing

* Apply suggestions from code review

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

* Apply suggestions from code review

---------

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

Coverage Stats

60 of 83 new or added lines in 3 files covered. (72.29%)

5 existing lines in 3 files now uncovered.

478 of 807 relevant lines covered (59.23%)

1.78 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

72.48

/custom_components/telegram_bot_conversation/recursive_data_flow.py

"""Recursive config flow and options flow."""

from __future__ import annotations

from collections.abc import Generator, Iterable, Mapping
from functools import partial
from types import MappingProxyType
from typing import Any

import voluptuous as vol

from homeassistant.config_entries import (
    HANDLERS,
    SOURCE_RECONFIGURE,
    ConfigEntry,
    ConfigEntryState,
    ConfigError,
    ConfigFlow,
    ConfigFlowContext,
    ConfigFlowResult,
    ConfigSubentryData,
    ConfigSubentryFlow,
    OptionsFlow,
    SubentryFlowContext,
    SubentryFlowResult,
)
from homeassistant.core import HomeAssistant, callback
from homeassistant.data_entry_flow import AbortFlow, section
from homeassistant.exceptions import HomeAssistantError


class AbortRecursiveFlow(AbortFlow):
    """Error in recursive config flow."""


class RecursiveBaseFlow:
    """Overwrite methods in this class with integration-specific config."""

    VERSION = 1
    MINOR_VERSION = 1

    async def async_validate_input(
        self, step_id: str, user_input: dict[str, Any]
    ) -> dict[str, str]:
        """Validate step data."""
        return {}

    def step_enabled(self, step_id: str) -> bool:
        """Check if the current data flow step is enabled."""
        return True

    @property
    def title(self) -> str:
        """Return config flow title."""
        return self.domain

    @property
    def subentry_title(self) -> str:
        """Return config subentry flow title."""
        return self._subentry_type

    async def get_data_schema(self) -> vol.Schema:
        """Get data schema."""
        if hasattr(self, "data_schema") and self.data_schema is not None:
            return self.data_schema
        raise NotImplementedError

    async def get_options_schema(self) -> vol.Schema:
        """Get options schema."""
        if hasattr(self, "options_schema") and self.options_schema is not None:
            return self.options_schema
        raise NotImplementedError

    async def get_default_subentries(self) -> Iterable[ConfigSubentryData] | None:
        """Get default subentries."""
        return None

    async def get_subentry_schema(self, subentry_type: str) -> vol.Schema:
        """Get subentry schema."""
        if (
            hasattr(self, "subentries_schema")
            and self.subentries_schema is not None
            and subentry_type in self.subentries_schema
        ):
            return self.subentries_schema[subentry_type]
        raise NotImplementedError

    @classmethod
    @callback
    def get_subentries(cls, config_entry: ConfigEntry) -> Iterable[str]:
        """Get subentries list."""
        raise NotImplementedError


class RecursiveDataFlow(RecursiveBaseFlow):
    """Handle both config and option flow."""

    data_schema: vol.Schema | None = None
    options_schema: vol.Schema | None = None
    domain: str | None = None

    def __init_subclass__(
        cls,
        *,
        data_schema: vol.Schema | None = None,
        options_schema: vol.Schema | None = None,
        **kwargs: Any,
    ) -> None:
        """Set config and options schema if provided."""
        super().__init_subclass__(**kwargs)
        cls.data_schema = data_schema
        cls.options_schema = options_schema
        cls.domain = kwargs.get("domain")

    def __init__(self) -> None:
        """Initialize the flow."""
        self.data: Mapping[str, Any] | None = None
        self.options: Mapping[str, Any] | None = None
        self.config_step = None
        self.current_step_schema = None
        self.current_step_id = None
        self.current_step_data = None

    def config_step_generator(
        self,
    ) -> Generator[tuple[str, vol.Schema, dict, bool]]:
        """Return a generator of the next step config."""

        def traverse_config(
            name: str, schema: vol.Schema, data: dict, last_config: bool = False
        ) -> tuple[str, vol.Schema, dict, bool]:
            current_schema = {}
            recursive_schema = {}
            for var, val in schema.schema.items():
                if isinstance(val, vol.Schema):
                    recursive_schema[var] = val
                elif isinstance(val, dict):
                    recursive_schema[var] = vol.Schema(val)
                else:
                    current_schema[var] = val

            yield (
                name,
                vol.Schema(current_schema),
                data,
                last_config and not recursive_schema,
            )
            for index, (var, val) in enumerate(recursive_schema.items(), start=1):
                if self.step_enabled(str(var)):
                    data[str(var)] = data.get(str(var), {}).copy()
                    yield from traverse_config(
                        str(var),
                        val,
                        data[str(var)],
                        last_config and index == len(recursive_schema),
                    )

        if not isinstance(self, (OptionsFlow, ConfigSubentryFlow)) and self.data_schema:
            yield from traverse_config(
                "user", self.data_schema, self.data, not self.options_schema
            )
        if self.options_schema:
            yield from traverse_config("init", self.options_schema, self.options, True)

    async def async_step(
        self, step_id: str, user_input: dict[str, Any] | None = None
    ) -> ConfigFlowResult:
        """Handle the step."""
        if self.config_step is None:
            self.config_step = self.config_step_generator()
            (
                self.current_step_id,
                self.current_step_schema,
                self.current_step_data,
                self.last_step,
            ) = next(self.config_step)
        if self.current_step_id != step_id:
            raise ConfigError("Unexpected step id")

        try:
            errors = {}
            if user_input is not None:
                for name, var in user_input.items():
                    self.current_step_data[name] = var
                errors = await self.async_validate_input(
                    step_id=step_id,
                    user_input=user_input,
                )
                if not errors:
                    for name in list(self.current_step_data.keys()):
                        if name not in user_input:
                            for key in self.current_step_schema.schema:
                                if key == name and isinstance(key, vol.Optional):
                                    self.current_step_data.pop(name)
                                    break
                    try:
                        (
                            self.current_step_id,
                            self.current_step_schema,
                            self.current_step_data,
                            self.last_step,
                        ) = next(self.config_step)
                        return await self.async_step(self.current_step_id)
                    except StopIteration:
                        return self.async_create_entry(
                            title=self.title,
                            data=self.data,
                            options=self.options,
                            subentries=(
                                await self.get_default_subentries()
                                if not isinstance(
                                    self, (OptionsFlow, ConfigSubentryFlow)
                                )
                                else None
                            ),
                        )
        except AbortRecursiveFlow as err:
            return self.async_abort(reason=str(err))

        schema = self.add_suggested_values_to_schema(
            self.current_step_schema, self.current_step_data
        )

        return self.async_show_form(
            step_id=self.current_step_id,
            data_schema=schema,
            errors=errors,
            last_step=self.last_step,
        )

    def __getattr__(self, attr: str) -> Any:
        """Get step method."""
        if attr.startswith("async_step_"):
            return partial(self.async_step, attr[11:])
        if hasattr(super(), "__getattr__"):
            return super().__getattr__(attr)
        raise AttributeError

    def suggested_values_from_default(
        self, data_schema: vol.Schema | Mapping[str, Any] | section
    ) -> Mapping[str, Any]:
        """Generate suggested values from schema markers."""
        if isinstance(data_schema, section):
            data_schema = data_schema.schema
        if isinstance(data_schema, vol.Schema):
            data_schema = data_schema.schema

        suggested_values = {}
        for key, value in data_schema.items():
            if isinstance(key, vol.Marker) and not isinstance(
                key.default, vol.Undefined
            ):
                suggested_values[str(key)] = key.default()
            if isinstance(value, (vol.Schema, dict, section)):
                value = self.suggested_values_from_default(value)
                if value:
                    suggested_values[str(key)] = value
        return suggested_values


class RecursiveOptionsFlow(RecursiveDataFlow, OptionsFlow):
    """Handle an options flow."""

    def __init__(self, config_entry: ConfigEntry) -> None:
        """Initialize options flow."""
        super().__init__()
        self.data = config_entry.data
        self.options = config_entry.options.copy()

    async def async_step_init(
        self, user_input: dict[str, Any] | None = None
    ) -> ConfigFlowResult:
        """Options flow entry point."""
        try:
            if self.options_schema is None:
                self.options_schema = await self.get_options_schema()
            return await self.async_step("init", user_input)
        except AbortRecursiveFlow as err:
            return self.async_abort(reason=str(err))

    @callback
    def async_create_entry(
        self,
        *,
        data: Mapping[str, Any],
        options: Mapping[str, Any] | None = None,
        subentries: Iterable[ConfigSubentryData] | None = None,
        **kwargs,
    ) -> ConfigFlowResult:
        """Return result entry for option flow."""
        return super().async_create_entry(data=options, **kwargs)


class RecursiveSubentryFlow(RecursiveDataFlow, ConfigSubentryFlow):
    """Handle a config subentry flow."""

    async def async_step_user(
        self, user_input: dict[str, Any] | None = None
    ) -> SubentryFlowResult:
        """Add a subentry."""
        return await self.async_step_init()

    async def async_step_reconfigure(
        self, user_input: dict[str, Any] | None = None
    ) -> SubentryFlowResult:
        """Handle reconfiguration of a subentry."""
        self.options = self._get_reconfigure_subentry().data.copy()
        return await self.async_step_init()

    async def async_step_init(
        self, user_input: dict[str, Any] | None = None
    ) -> SubentryFlowResult:
        """Set initial options."""
        if self._get_entry().state != ConfigEntryState.LOADED:
            return self.async_abort(reason="entry_not_loaded")
        try:
            if self.data is None:
                self.data = self._get_entry().data
            if self.options_schema is None:
                self.options_schema = await self.get_subentry_schema(
                    self._subentry_type
                )
            if self.options is None:
                self.options = self.suggested_values_from_default(self.options_schema)
            return await self.async_step("init", user_input)
        except AbortRecursiveFlow as err:
            return self.async_abort(reason=str(err))

    @property
    def title(self) -> str:
        """Return config subentry flow title."""
        return self.subentry_title

    @callback
    def async_create_entry(
        self,
        *,
        title: str,
        data: Mapping[str, Any],
        options: Mapping[str, Any] | None = None,
        subentries: Iterable[ConfigSubentryData] | None = None,
        **kwargs,
    ) -> ConfigFlowResult:
        """Return result entry for subentry flow."""
        if self.source == "user":
            return super().async_create_entry(
                title=title,
                data=self.options,
            )
        return self.async_update_and_abort(
            self._get_entry(),
            self._get_reconfigure_subentry(),
            data=self.options,
        )


class RecursiveConfigFlow(RecursiveDataFlow, ConfigFlow):
    """Handle a config flow."""

    subentries_schema: dict[str, vol.Schema] | None = None

    def __init_subclass__(
        cls,
        *,
        subentries_schema: dict[str, vol.Schema] | None = None,
        **kwargs: Any,
    ) -> None:
        """Set config and options schema if provided."""
        super().__init_subclass__(**kwargs)
        cls.subentries_schema = subentries_schema

    async def async_step_user(
        self, user_input: dict[str, Any] | None = None
    ) -> ConfigFlowResult:
        """Config flow entry point."""
        try:
            if self.data_schema is None:
                self.data_schema = await self.get_data_schema()
            if self.options_schema is None:
                self.options_schema = await self.get_options_schema()
            if self.data is None:
                self.data = self.suggested_values_from_default(self.data_schema)
            if self.options is None:
                self.options = self.suggested_values_from_default(self.options_schema)
            return await self.async_step("user", user_input)
        except AbortRecursiveFlow as err:
            return self.async_abort(reason=str(err))

    @classmethod
    @callback
    def async_get_options_flow(cls, config_entry: ConfigEntry) -> OptionsFlow:
        """Create the options flow."""

        class MyOptionsFlow(
            RecursiveOptionsFlow,
            cls,
            data_schema=cls.data_schema,
            options_schema=cls.options_schema,
        ):
            pass

        return MyOptionsFlow(config_entry)

    @classmethod
    @callback
    def async_supports_options_flow(cls, config_entry: ConfigEntry) -> bool:
        """Return options flow support for this handler."""

        def _func(obj: Any) -> Any:
            return getattr(obj, "__func__", obj)

        return bool(
            _func(cls.get_options_schema)
            is not _func(RecursiveBaseFlow.get_options_schema)
            or (cls.options_schema is not None and cls.options_schema.schema)
        )

    @classmethod
    @callback
    def async_get_supported_subentry_types(
        cls, config_entry: ConfigEntry
    ) -> dict[str, type[ConfigSubentryFlow]]:
        """Return subentries supported by this integration."""

        def _func(obj: Any) -> Any:
            return getattr(obj, "__func__", obj)

        if cls.subentries_schema is not None:
            subentries_schema = cls.subentries_schema
        elif _func(cls.get_subentries) is not _func(RecursiveBaseFlow.get_subentries):
            subentries_schema = dict.fromkeys(cls.get_subentries(config_entry))
        else:
            subentries_schema = {}

        def subentry_factory(schema: vol.Schema) -> type[ConfigSubentryFlow]:
            class MySubentryFlow(RecursiveSubentryFlow, cls, options_schema=schema):
                pass

            return MySubentryFlow

        return {
            key: subentry_factory(schema) for key, schema in subentries_schema.items()
        }


async def validate_data(
    hass: HomeAssistant, config_entry: ConfigEntry
) -> MappingProxyType[str, Any]:
    """Validate config data."""
    handler = HANDLERS.get(config_entry.domain)
    if handler is None or not issubclass(handler, RecursiveBaseFlow):
        raise NotImplementedError(
            f"Handler for domain {config_entry.domain} is not a RecursiveBaseFlow"
        )
    flow = handler()
    flow.hass = hass
    flow.handler = config_entry.entry_id
    flow.context = ConfigFlowContext(
        source=SOURCE_RECONFIGURE,
        show_advanced_options=True,
        entry_id=config_entry.entry_id,
    )
    flow.data = config_entry.data
    flow.options = config_entry.options
    try:
        schema = await flow.get_data_schema()
        return MappingProxyType(schema(config_entry.data.copy()))
    except (AbortRecursiveFlow, vol.MultipleInvalid) as err:
        raise HomeAssistantError(str(err)) from err


async def validate_options(
    hass: HomeAssistant, config_entry: ConfigEntry
) -> MappingProxyType[str, Any]:
    """Validate options."""
    handler = HANDLERS.get(config_entry.domain)
    if handler is None or not issubclass(handler, RecursiveBaseFlow):
        raise NotImplementedError(
            f"Handler for domain {config_entry.domain} is not a RecursiveBaseFlow"
        )
    flow = handler.async_get_options_flow(config_entry)
    flow.hass = hass
    flow.handler = config_entry.entry_id
    flow.context = ConfigFlowContext(
        source=SOURCE_RECONFIGURE,
        show_advanced_options=True,
        entry_id=config_entry.entry_id,
    )
    flow.data = config_entry.data
    flow.options = config_entry.options
    try:
        schema = await flow.get_options_schema()
        return MappingProxyType(schema(config_entry.options.copy()))
    except (AbortRecursiveFlow, vol.MultipleInvalid) as err:
        raise HomeAssistantError(str(err)) from err


async def validate_subentry_data(
    hass: HomeAssistant, config_entry: ConfigEntry, subentry_id: str
) -> MappingProxyType[str, Any]:
    """Validate subentry config."""
    handler = HANDLERS.get(config_entry.domain)
    if handler is None or not issubclass(handler, RecursiveBaseFlow):
        raise NotImplementedError(
            f"Handler for domain {config_entry.domain} is not a RecursiveBaseFlow"
        )
    subentry = config_entry.subentries[subentry_id]
    flow = handler.async_get_supported_subentry_types(config_entry)[
        subentry.subentry_type
    ]()
    flow.hass = hass
    flow.handler = config_entry.entry_id, subentry.subentry_type
    flow.context = SubentryFlowContext(
        source=SOURCE_RECONFIGURE,
        show_advanced_options=True,
        entry_id=config_entry.entry_id,
        subentry_id=subentry.subentry_id,
    )
    flow.data = config_entry.data
    flow.options = subentry.data
    try:
        schema = await flow.get_subentry_schema(subentry.subentry_type)
        return MappingProxyType(schema(subentry.data.copy()))
    except (AbortRecursiveFlow, vol.MultipleInvalid) as err:
        raise HomeAssistantError(str(err)) from err

1	"""Recursive config flow and options flow."""
2
3	from __future__ import annotations	3✔
4
5	from collections.abc import Generator, Iterable, Mapping	3✔
6	from functools import partial	3✔
7	from types import MappingProxyType	3✔
8	from typing import Any	3✔
9
10	import voluptuous as vol	3✔
11
12	from homeassistant.config_entries import (	3✔
13	HANDLERS,
14	SOURCE_RECONFIGURE,
15	ConfigEntry,
16	ConfigEntryState,
17	ConfigError,
18	ConfigFlow,
19	ConfigFlowContext,
20	ConfigFlowResult,
21	ConfigSubentryData,
22	ConfigSubentryFlow,
23	OptionsFlow,
24	SubentryFlowContext,
25	SubentryFlowResult,
26	)
27	from homeassistant.core import HomeAssistant, callback	3✔
28	from homeassistant.data_entry_flow import AbortFlow, section	3✔
29	from homeassistant.exceptions import HomeAssistantError	3✔
30
31
32	class AbortRecursiveFlow(AbortFlow):	3✔
33	"""Error in recursive config flow."""
34
35
36	class RecursiveBaseFlow:	3✔
37	"""Overwrite methods in this class with integration-specific config."""
38
39	VERSION = 1	3✔
40	MINOR_VERSION = 1	3✔
41
42	async def async_validate_input(	3✔
43	self, step_id: str, user_input: dict[str, Any]
44	) -> dict[str, str]:
45	"""Validate step data."""
46	return {}	×
47
48	def step_enabled(self, step_id: str) -> bool:	3✔
49	"""Check if the current data flow step is enabled."""
50	return True	×
51
52	@property	3✔
53	def title(self) -> str:	3✔
54	"""Return config flow title."""
55	return self.domain	×
56
57	@property	3✔
58	def subentry_title(self) -> str:	3✔
59	"""Return config subentry flow title."""
60	return self._subentry_type	×
61
62	async def get_data_schema(self) -> vol.Schema:	3✔
63	"""Get data schema."""
NEW 64	if hasattr(self, "data_schema") and self.data_schema is not None:	×
NEW 65	return self.data_schema	×
UNCOV 66	raise NotImplementedError	×
67
68	async def get_options_schema(self) -> vol.Schema:	3✔
69	"""Get options schema."""
NEW 70	if hasattr(self, "options_schema") and self.options_schema is not None:	×
NEW 71	return self.options_schema	×
UNCOV 72	raise NotImplementedError	×
73
74	async def get_default_subentries(self) -> Iterable[ConfigSubentryData] \| None:	3✔
75	"""Get default subentries."""
76	return None	×
77
78	async def get_subentry_schema(self, subentry_type: str) -> vol.Schema:	3✔
79	"""Get subentry schema."""
NEW 80	if (	×
81	hasattr(self, "subentries_schema")
82	and self.subentries_schema is not None
83	and subentry_type in self.subentries_schema
84	):
NEW 85	return self.subentries_schema[subentry_type]	×
UNCOV 86	raise NotImplementedError	×
87
88	@classmethod	3✔
89	@callback	3✔
90	def get_subentries(cls, config_entry: ConfigEntry) -> Iterable[str]:	3✔
91	"""Get subentries list."""
92	raise NotImplementedError	×
93
94
95	class RecursiveDataFlow(RecursiveBaseFlow):	3✔
96	"""Handle both config and option flow."""
97
98	data_schema: vol.Schema \| None = None	3✔
99	options_schema: vol.Schema \| None = None	3✔
100	domain: str \| None = None	3✔
101
102	def __init_subclass__(	3✔
103	cls,
104	*,
105	data_schema: vol.Schema \| None = None,
106	options_schema: vol.Schema \| None = None,
107	**kwargs: Any,
108	) -> None:
109	"""Set config and options schema if provided."""
110	super().__init_subclass__(**kwargs)	3✔
111	cls.data_schema = data_schema	3✔
112	cls.options_schema = options_schema	3✔
113	cls.domain = kwargs.get("domain")	3✔
114
115	def __init__(self) -> None:	3✔
116	"""Initialize the flow."""
117	self.data: Mapping[str, Any] \| None = None	3✔
118	self.options: Mapping[str, Any] \| None = None	3✔
119	self.config_step = None	3✔
120	self.current_step_schema = None	3✔
121	self.current_step_id = None	3✔
122	self.current_step_data = None	3✔
123
124	def config_step_generator(	3✔
125	self,
126	) -> Generator[tuple[str, vol.Schema, dict, bool]]:
127	"""Return a generator of the next step config."""
128
129	def traverse_config(	3✔
130	name: str, schema: vol.Schema, data: dict, last_config: bool = False
131	) -> tuple[str, vol.Schema, dict, bool]:
132	current_schema = {}	3✔
133	recursive_schema = {}	3✔
134	for var, val in schema.schema.items():	3✔
135	if isinstance(val, vol.Schema):	3✔
136	recursive_schema[var] = val	×
137	elif isinstance(val, dict):	3✔
138	recursive_schema[var] = vol.Schema(val)	×
139	else:
140	current_schema[var] = val	3✔
141
142	yield (	3✔
143	name,
144	vol.Schema(current_schema),
145	data,
146	last_config and not recursive_schema,
147	)
148	for index, (var, val) in enumerate(recursive_schema.items(), start=1):	3✔
149	if self.step_enabled(str(var)):	×
150	data[str(var)] = data.get(str(var), {}).copy()	×
151	yield from traverse_config(	×
152	str(var),
153	val,
154	data[str(var)],
155	last_config and index == len(recursive_schema),
156	)
157
158	if not isinstance(self, (OptionsFlow, ConfigSubentryFlow)) and self.data_schema:	3✔
159	yield from traverse_config(	3✔
160	"user", self.data_schema, self.data, not self.options_schema
161	)
162	if self.options_schema:	3✔
163	yield from traverse_config("init", self.options_schema, self.options, True)	3✔
164
165	async def async_step(	3✔
166	self, step_id: str, user_input: dict[str, Any] \| None = None
167	) -> ConfigFlowResult:
168	"""Handle the step."""
169	if self.config_step is None:	3✔
170	self.config_step = self.config_step_generator()	3✔
171	(	3✔
172	self.current_step_id,
173	self.current_step_schema,
174	self.current_step_data,
175	self.last_step,
176	) = next(self.config_step)
177	if self.current_step_id != step_id:	3✔
178	raise ConfigError("Unexpected step id")	×
179
180	try:	3✔
181	errors = {}	3✔
182	if user_input is not None:	3✔
183	for name, var in user_input.items():	3✔
184	self.current_step_data[name] = var	3✔
185	errors = await self.async_validate_input(	3✔
186	step_id=step_id,
187	user_input=user_input,
188	)
189	if not errors:	3✔
190	for name in list(self.current_step_data.keys()):	3✔
191	if name not in user_input:	3✔
192	for key in self.current_step_schema.schema:	×
193	if key == name and isinstance(key, vol.Optional):	×
194	self.current_step_data.pop(name)	×
195	break	×
196	try:	3✔
197	(	3✔
198	self.current_step_id,
199	self.current_step_schema,
200	self.current_step_data,
201	self.last_step,
202	) = next(self.config_step)
203	return await self.async_step(self.current_step_id)	3✔
204	except StopIteration:	3✔
205	return self.async_create_entry(	3✔
206	title=self.title,
207	data=self.data,
208	options=self.options,
209	subentries=(
210	await self.get_default_subentries()
211	if not isinstance(
212	self, (OptionsFlow, ConfigSubentryFlow)
213	)
214	else None
215	),
216	)
217	except AbortRecursiveFlow as err:	×
218	return self.async_abort(reason=str(err))	×
219
220	schema = self.add_suggested_values_to_schema(	3✔
221	self.current_step_schema, self.current_step_data
222	)
223
224	return self.async_show_form(	3✔
225	step_id=self.current_step_id,
226	data_schema=schema,
227	errors=errors,
228	last_step=self.last_step,
229	)
230
231	def __getattr__(self, attr: str) -> Any:	3✔
232	"""Get step method."""
233	if attr.startswith("async_step_"):	3✔
234	return partial(self.async_step, attr[11:])	3✔
235	if hasattr(super(), "__getattr__"):	×
236	return super().__getattr__(attr)	×
237	raise AttributeError	×
238
239	def suggested_values_from_default(	3✔
240	self, data_schema: vol.Schema \| Mapping[str, Any] \| section
241	) -> Mapping[str, Any]:
242	"""Generate suggested values from schema markers."""
243	if isinstance(data_schema, section):	3✔
244	data_schema = data_schema.schema	×
245	if isinstance(data_schema, vol.Schema):	3✔
246	data_schema = data_schema.schema	3✔
247
248	suggested_values = {}	3✔
249	for key, value in data_schema.items():	3✔
250	if isinstance(key, vol.Marker) and not isinstance(	3✔
251	key.default, vol.Undefined
252	):
253	suggested_values[str(key)] = key.default()	3✔
254	if isinstance(value, (vol.Schema, dict, section)):	3✔
255	value = self.suggested_values_from_default(value)	×
256	if value:	×
257	suggested_values[str(key)] = value	×
258	return suggested_values	3✔
259
260
261	class RecursiveOptionsFlow(RecursiveDataFlow, OptionsFlow):	3✔
262	"""Handle an options flow."""
263
264	def __init__(self, config_entry: ConfigEntry) -> None:	3✔
265	"""Initialize options flow."""
266	super().__init__()	3✔
267	self.data = config_entry.data	3✔
268	self.options = config_entry.options.copy()	3✔
269
270	async def async_step_init(	3✔
271	self, user_input: dict[str, Any] \| None = None
272	) -> ConfigFlowResult:
273	"""Options flow entry point."""
274	try:	3✔
275	if self.options_schema is None:	3✔
276	self.options_schema = await self.get_options_schema()	3✔
277	return await self.async_step("init", user_input)	3✔
278	except AbortRecursiveFlow as err:	×
279	return self.async_abort(reason=str(err))	×
280
281	@callback	3✔
282	def async_create_entry(	3✔
283	self,
284	*,
285	data: Mapping[str, Any],
286	options: Mapping[str, Any] \| None = None,
287	subentries: Iterable[ConfigSubentryData] \| None = None,
288	**kwargs,
289	) -> ConfigFlowResult:
290	"""Return result entry for option flow."""
291	return super().async_create_entry(data=options, **kwargs)	3✔
292
293
294	class RecursiveSubentryFlow(RecursiveDataFlow, ConfigSubentryFlow):	3✔
295	"""Handle a config subentry flow."""
296
297	async def async_step_user(	3✔
298	self, user_input: dict[str, Any] \| None = None
299	) -> SubentryFlowResult:
300	"""Add a subentry."""
301	return await self.async_step_init()	×
302
303	async def async_step_reconfigure(	3✔
304	self, user_input: dict[str, Any] \| None = None
305	) -> SubentryFlowResult:
306	"""Handle reconfiguration of a subentry."""
307	self.options = self._get_reconfigure_subentry().data.copy()	×
308	return await self.async_step_init()	×
309
310	async def async_step_init(	3✔
311	self, user_input: dict[str, Any] \| None = None
312	) -> SubentryFlowResult:
313	"""Set initial options."""
314	if self._get_entry().state != ConfigEntryState.LOADED:	×
315	return self.async_abort(reason="entry_not_loaded")	×
316	try:	×
317	if self.data is None:	×
318	self.data = self._get_entry().data	×
319	if self.options_schema is None:	×
320	self.options_schema = await self.get_subentry_schema(	×
321	self._subentry_type
322	)
323	if self.options is None:	×
324	self.options = self.suggested_values_from_default(self.options_schema)	×
325	return await self.async_step("init", user_input)	×
326	except AbortRecursiveFlow as err:	×
327	return self.async_abort(reason=str(err))	×
328
329	@property	3✔
330	def title(self) -> str:	3✔
331	"""Return config subentry flow title."""
332	return self.subentry_title	×
333
334	@callback	3✔
335	def async_create_entry(	3✔
336	self,
337	*,
338	title: str,
339	data: Mapping[str, Any],
340	options: Mapping[str, Any] \| None = None,
341	subentries: Iterable[ConfigSubentryData] \| None = None,
342	**kwargs,
343	) -> ConfigFlowResult:
344	"""Return result entry for subentry flow."""
345	if self.source == "user":	×
346	return super().async_create_entry(	×
347	title=title,
348	data=self.options,
349	)
350	return self.async_update_and_abort(	×
351	self._get_entry(),
352	self._get_reconfigure_subentry(),
353	data=self.options,
354	)
355
356
357	class RecursiveConfigFlow(RecursiveDataFlow, ConfigFlow):	3✔
358	"""Handle a config flow."""
359
360	subentries_schema: dict[str, vol.Schema] \| None = None	3✔
361
362	def __init_subclass__(	3✔
363	cls,
364	*,
365	subentries_schema: dict[str, vol.Schema] \| None = None,
366	**kwargs: Any,
367	) -> None:
368	"""Set config and options schema if provided."""
369	super().__init_subclass__(**kwargs)	3✔
370	cls.subentries_schema = subentries_schema	3✔
371
372	async def async_step_user(	3✔
373	self, user_input: dict[str, Any] \| None = None
374	) -> ConfigFlowResult:
375	"""Config flow entry point."""
376	try:	3✔
377	if self.data_schema is None:	3✔
378	self.data_schema = await self.get_data_schema()	3✔
379	if self.options_schema is None:	3✔
380	self.options_schema = await self.get_options_schema()	3✔
381	if self.data is None:	3✔
382	self.data = self.suggested_values_from_default(self.data_schema)	3✔
383	if self.options is None:	3✔
384	self.options = self.suggested_values_from_default(self.options_schema)	3✔
385	return await self.async_step("user", user_input)	3✔
386	except AbortRecursiveFlow as err:	×
387	return self.async_abort(reason=str(err))	×
388
389	@classmethod	3✔
390	@callback	3✔
391	def async_get_options_flow(cls, config_entry: ConfigEntry) -> OptionsFlow:	3✔
392	"""Create the options flow."""
393
394	class MyOptionsFlow(	3✔
395	RecursiveOptionsFlow,
396	cls,
397	data_schema=cls.data_schema,
398	options_schema=cls.options_schema,
399	):
400	pass	3✔
401
402	return MyOptionsFlow(config_entry)	3✔
403
404	@classmethod	3✔
405	@callback	3✔
406	def async_supports_options_flow(cls, config_entry: ConfigEntry) -> bool:	3✔
407	"""Return options flow support for this handler."""
408
409	def _func(obj: Any) -> Any:	×
410	return getattr(obj, "__func__", obj)	×
411
412	return bool(	×
413	_func(cls.get_options_schema)
414	is not _func(RecursiveBaseFlow.get_options_schema)
415	or (cls.options_schema is not None and cls.options_schema.schema)
416	)
417
418	@classmethod	3✔
419	@callback	3✔
420	def async_get_supported_subentry_types(	3✔
421	cls, config_entry: ConfigEntry
422	) -> dict[str, type[ConfigSubentryFlow]]:
423	"""Return subentries supported by this integration."""
424
425	def _func(obj: Any) -> Any:	3✔
426	return getattr(obj, "__func__", obj)	3✔
427
428	if cls.subentries_schema is not None:	3✔
NEW 429	subentries_schema = cls.subentries_schema	×
430	elif _func(cls.get_subentries) is not _func(RecursiveBaseFlow.get_subentries):	3✔
431	subentries_schema = dict.fromkeys(cls.get_subentries(config_entry))	3✔
432	else:
433	subentries_schema = {}	×
434
435	def subentry_factory(schema: vol.Schema) -> type[ConfigSubentryFlow]:	3✔
436	class MySubentryFlow(RecursiveSubentryFlow, cls, options_schema=schema):	3✔
437	pass	3✔
438
439	return MySubentryFlow	3✔
440
441	return {	3✔
442	key: subentry_factory(schema) for key, schema in subentries_schema.items()
443	}
444
445
446	async def validate_data(	3✔
447	hass: HomeAssistant, config_entry: ConfigEntry
448	) -> MappingProxyType[str, Any]:
449	"""Validate config data."""
450	handler = HANDLERS.get(config_entry.domain)	3✔
451	if handler is None or not issubclass(handler, RecursiveBaseFlow):	3✔
NEW 452	raise NotImplementedError(	×
453	f"Handler for domain {config_entry.domain} is not a RecursiveBaseFlow"
454	)
455	flow = handler()	3✔
456	flow.hass = hass	3✔
457	flow.handler = config_entry.entry_id	3✔
458	flow.context = ConfigFlowContext(	3✔
459	source=SOURCE_RECONFIGURE,
460	show_advanced_options=True,
461	entry_id=config_entry.entry_id,
462	)
463	flow.data = config_entry.data	3✔
464	flow.options = config_entry.options	3✔
465	try:	3✔
466	schema = await flow.get_data_schema()	3✔
467	return MappingProxyType(schema(config_entry.data.copy()))	3✔
NEW 468	except (AbortRecursiveFlow, vol.MultipleInvalid) as err:	×
NEW 469	raise HomeAssistantError(str(err)) from err	×
470
471
472	async def validate_options(	3✔
473	hass: HomeAssistant, config_entry: ConfigEntry
474	) -> MappingProxyType[str, Any]:
475	"""Validate options."""
476	handler = HANDLERS.get(config_entry.domain)	3✔
477	if handler is None or not issubclass(handler, RecursiveBaseFlow):	3✔
NEW 478	raise NotImplementedError(	×
479	f"Handler for domain {config_entry.domain} is not a RecursiveBaseFlow"
480	)
481	flow = handler.async_get_options_flow(config_entry)	3✔
482	flow.hass = hass	3✔
483	flow.handler = config_entry.entry_id	3✔
484	flow.context = ConfigFlowContext(	3✔
485	source=SOURCE_RECONFIGURE,
486	show_advanced_options=True,
487	entry_id=config_entry.entry_id,
488	)
489	flow.data = config_entry.data	3✔
490	flow.options = config_entry.options	3✔
491	try:	3✔
492	schema = await flow.get_options_schema()	3✔
493	return MappingProxyType(schema(config_entry.options.copy()))	3✔
NEW 494	except (AbortRecursiveFlow, vol.MultipleInvalid) as err:	×
NEW 495	raise HomeAssistantError(str(err)) from err	×
496
497
498	async def validate_subentry_data(	3✔
499	hass: HomeAssistant, config_entry: ConfigEntry, subentry_id: str
500	) -> MappingProxyType[str, Any]:
501	"""Validate subentry config."""
502	handler = HANDLERS.get(config_entry.domain)	3✔
503	if handler is None or not issubclass(handler, RecursiveBaseFlow):	3✔
NEW 504	raise NotImplementedError(	×
505	f"Handler for domain {config_entry.domain} is not a RecursiveBaseFlow"
506	)
507	subentry = config_entry.subentries[subentry_id]	3✔
508	flow = handler.async_get_supported_subentry_types(config_entry)[	3✔
509	subentry.subentry_type
510	]()
511	flow.hass = hass	3✔
512	flow.handler = config_entry.entry_id, subentry.subentry_type	3✔
513	flow.context = SubentryFlowContext(	3✔
514	source=SOURCE_RECONFIGURE,
515	show_advanced_options=True,
516	entry_id=config_entry.entry_id,
517	subentry_id=subentry.subentry_id,
518	)
519	flow.data = config_entry.data	3✔
520	flow.options = subentry.data	3✔
521	try:	3✔
522	schema = await flow.get_subentry_schema(subentry.subentry_type)	3✔
523	return MappingProxyType(schema(subentry.data.copy()))	3✔
NEW 524	except (AbortRecursiveFlow, vol.MultipleInvalid) as err:	×
NEW 525	raise HomeAssistantError(str(err)) from err	×

Shulyaka / telegram_bot_conversation / 23368799114

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous