forked from home-assistant/core
/
config_flow.py
162 lines (140 loc) · 5.3 KB
/
config_flow.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
"""Config flow for Opower integration."""
from __future__ import annotations
from collections.abc import Mapping
import logging
import socket
from typing import Any
from opower import (
CannotConnect,
InvalidAuth,
Opower,
get_supported_utility_names,
select_utility,
)
import voluptuous as vol
from homeassistant import config_entries
from homeassistant.const import CONF_PASSWORD, CONF_USERNAME
from homeassistant.core import HomeAssistant, callback
from homeassistant.data_entry_flow import FlowResult
from homeassistant.helpers.aiohttp_client import async_create_clientsession
from .const import CONF_TOTP_SECRET, CONF_UTILITY, DOMAIN
_LOGGER = logging.getLogger(__name__)
STEP_USER_DATA_SCHEMA = vol.Schema(
{
vol.Required(CONF_UTILITY): vol.In(get_supported_utility_names()),
vol.Required(CONF_USERNAME): str,
vol.Required(CONF_PASSWORD): str,
}
)
async def _validate_login(
hass: HomeAssistant, login_data: dict[str, str]
) -> dict[str, str]:
"""Validate login data and return any errors."""
api = Opower(
async_create_clientsession(hass, family=socket.AF_INET),
login_data[CONF_UTILITY],
login_data[CONF_USERNAME],
login_data[CONF_PASSWORD],
login_data.get(CONF_TOTP_SECRET),
)
errors: dict[str, str] = {}
try:
await api.async_login()
except InvalidAuth:
errors["base"] = "invalid_auth"
except CannotConnect:
errors["base"] = "cannot_connect"
return errors
class OpowerConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
"""Handle a config flow for Opower."""
VERSION = 1
def __init__(self) -> None:
"""Initialize a new OpowerConfigFlow."""
self.reauth_entry: config_entries.ConfigEntry | None = None
self.utility_info: dict[str, Any] | None = None
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Handle the initial step."""
errors: dict[str, str] = {}
if user_input is not None:
self._async_abort_entries_match(
{
CONF_UTILITY: user_input[CONF_UTILITY],
CONF_USERNAME: user_input[CONF_USERNAME],
}
)
if select_utility(user_input[CONF_UTILITY]).accepts_mfa():
self.utility_info = user_input
return await self.async_step_mfa()
errors = await _validate_login(self.hass, user_input)
if not errors:
return self._async_create_opower_entry(user_input)
return self.async_show_form(
step_id="user", data_schema=STEP_USER_DATA_SCHEMA, errors=errors
)
async def async_step_mfa(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Handle MFA step."""
assert self.utility_info is not None
errors: dict[str, str] = {}
if user_input is not None:
data = {**self.utility_info, **user_input}
errors = await _validate_login(self.hass, data)
if not errors:
return self._async_create_opower_entry(data)
if errors:
schema = {
vol.Required(
CONF_USERNAME, default=self.utility_info[CONF_USERNAME]
): str,
vol.Required(CONF_PASSWORD): str,
}
else:
schema = {}
schema[vol.Required(CONF_TOTP_SECRET)] = str
return self.async_show_form(
step_id="mfa",
data_schema=vol.Schema(schema),
errors=errors,
)
@callback
def _async_create_opower_entry(self, data: dict[str, Any]) -> FlowResult:
"""Create the config entry."""
return self.async_create_entry(
title=f"{data[CONF_UTILITY]} ({data[CONF_USERNAME]})",
data=data,
)
async def async_step_reauth(self, entry_data: Mapping[str, Any]) -> FlowResult:
"""Handle configuration by re-auth."""
self.reauth_entry = self.hass.config_entries.async_get_entry(
self.context["entry_id"]
)
return await self.async_step_reauth_confirm()
async def async_step_reauth_confirm(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Dialog that informs the user that reauth is required."""
assert self.reauth_entry
errors: dict[str, str] = {}
if user_input is not None:
data = {**self.reauth_entry.data, **user_input}
errors = await _validate_login(self.hass, data)
if not errors:
self.hass.config_entries.async_update_entry(
self.reauth_entry, data=data
)
await self.hass.config_entries.async_reload(self.reauth_entry.entry_id)
return self.async_abort(reason="reauth_successful")
schema = {
vol.Required(CONF_USERNAME): self.reauth_entry.data[CONF_USERNAME],
vol.Required(CONF_PASSWORD): str,
}
if select_utility(self.reauth_entry.data[CONF_UTILITY]).accepts_mfa():
schema[vol.Optional(CONF_TOTP_SECRET)] = str
return self.async_show_form(
step_id="reauth_confirm",
data_schema=vol.Schema(schema),
errors=errors,
)