|
1 | 1 | """Config flow for Liebherr Integration.""" |
2 | 2 |
|
3 | 3 | from collections.abc import Mapping |
| 4 | +import re |
4 | 5 | from typing import Any |
5 | 6 |
|
6 | 7 | from pyliebherr import LiebherrAPI, LiebherrDevice |
@@ -145,13 +146,16 @@ async def async_step_user( |
145 | 146 |
|
146 | 147 | if user_input is not None: |
147 | 148 | # check the API key |
148 | | - api: LiebherrAPI = LiebherrAPI(user_input[CONF_API_KEY]) |
149 | | - devices: list[LiebherrDevice] = [] |
150 | | - try: |
151 | | - devices = await api.async_get_appliances() |
152 | | - except LiebherrAuthException: |
153 | | - errors["base"] = "auth_error" |
154 | | - await api.async_close() |
| 149 | + if not re.fullmatch("^\\S.*\\S$", user_input[CONF_API_KEY]): |
| 150 | + errors[CONF_API_KEY] = "whitespace_api_key" |
| 151 | + else: |
| 152 | + api: LiebherrAPI = LiebherrAPI(user_input[CONF_API_KEY]) |
| 153 | + devices: list[LiebherrDevice] = [] |
| 154 | + try: |
| 155 | + devices = await api.async_get_appliances() |
| 156 | + except LiebherrAuthException: |
| 157 | + errors["base"] = "auth_error" |
| 158 | + await api.async_close() |
155 | 159 | if not errors: |
156 | 160 | await self.async_set_unique_id(f"{DOMAIN}_{user_input[CONF_API_KEY]}") |
157 | 161 | self._abort_if_unique_id_configured() |
@@ -181,9 +185,7 @@ async def async_step_user( |
181 | 185 | return self.async_abort(reason="invalid_source") |
182 | 186 |
|
183 | 187 | data_schema: vol.Schema = vol.Schema( |
184 | | - { |
185 | | - vol.Required(CONF_API_KEY): TextSelector(), |
186 | | - } |
| 188 | + {vol.Required(CONF_API_KEY): vol.All(TextSelector())} |
187 | 189 | ) |
188 | 190 | return self.async_show_form( |
189 | 191 | step_id="user", data_schema=data_schema, errors=errors |
|
0 commit comments