jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-25 14:17:45 +00:00
Code Issues Packages Projects Releases Wiki Activity

Refactor vulcan integration (#71175)

Browse Source
This commit is contained in:
Antoni Czaplicki 2022-05-05 14:32:36 +02:00 committed by GitHub
parent 248f01f41f
commit 191230f535
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
11 changed files with 293 additions and 529 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

52
homeassistant/components/vulcan/__init__.py
View File

@ -1,18 +1,15 @@
"""The Vulcan component.""" """The Vulcan component."""
import logging
from aiohttp import ClientConnectorError from aiohttp import ClientConnectorError
from vulcan import Account, Keystore, Vulcan from vulcan import Account, Keystore, UnauthorizedCertificateException, Vulcan
from vulcan._utils import VulcanAPIException
from homeassistant.config_entries import ConfigEntry from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryAuthFailed, ConfigEntryNotReady from homeassistant.exceptions import ConfigEntryAuthFailed, ConfigEntryNotReady
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from .const import DOMAIN from .const import DOMAIN
_LOGGER = logging.getLogger(__name__)
PLATFORMS = ["calendar"] PLATFORMS = ["calendar"]
@ -22,54 +19,29 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
try: try:
keystore = Keystore.load(entry.data["keystore"]) keystore = Keystore.load(entry.data["keystore"])
account = Account.load(entry.data["account"]) account = Account.load(entry.data["account"])
client = Vulcan(keystore, account) client = Vulcan(keystore, account, async_get_clientsession(hass))
await client.select_student() await client.select_student()
students = await client.get_students() students = await client.get_students()
for student in students: for student in students:
if str(student.pupil.id) == str(entry.data["student_id"]): if str(student.pupil.id) == str(entry.data["student_id"]):
client.student = student client.student = student
break break
except VulcanAPIException as err: except UnauthorizedCertificateException as err:
if str(err) == "The certificate is not authorized.": raise ConfigEntryAuthFailed("The certificate is not authorized.") from err
_LOGGER.error(
"The certificate is not authorized, please authorize integration again"
)
raise ConfigEntryAuthFailed from err
_LOGGER.error("Vulcan API error: %s", err)
return False
except ClientConnectorError as err: except ClientConnectorError as err:
if "connection_error" not in hass.data[DOMAIN]: raise ConfigEntryNotReady(
_LOGGER.error( f"Connection error - please check your internet connection: {err}"
"Connection error - please check your internet connection: %s", err ) from err
)
hass.data[DOMAIN]["connection_error"] = True
await client.close()
raise ConfigEntryNotReady from err
hass.data[DOMAIN]["students_number"] = len(
hass.config_entries.async_entries(DOMAIN)
)
hass.data[DOMAIN][entry.entry_id] = client hass.data[DOMAIN][entry.entry_id] = client
if not entry.update_listeners: hass.config_entries.async_setup_platforms(entry, PLATFORMS)
entry.add_update_listener(_async_update_options)
for platform in PLATFORMS:
hass.async_create_task(
hass.config_entries.async_forward_entry_setup(entry, platform)
)
return True return True
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Unload a config entry.""" """Unload a config entry."""
await hass.data[DOMAIN][entry.entry_id].close() if unload_ok := await hass.config_entries.async_unload_platforms(entry, PLATFORMS):
for platform in PLATFORMS: hass.data[DOMAIN].pop(entry.entry_id)
await hass.config_entries.async_forward_entry_unload(entry, platform)
return True return unload_ok
async def _async_update_options(hass, entry):
"""Update options."""
await hass.config_entries.async_reload(entry.entry_id)

167
homeassistant/components/vulcan/calendar.py
View File

@ -1,24 +1,25 @@
"""Support for Vulcan Calendar platform.""" """Support for Vulcan Calendar platform."""
import copy from __future__ import annotations
from datetime import date, datetime, timedelta from datetime import date, datetime, timedelta
import logging import logging
from aiohttp import ClientConnectorError from aiohttp import ClientConnectorError
from vulcan._utils import VulcanAPIException from vulcan import UnauthorizedCertificateException
from homeassistant.components.calendar import ENTITY_ID_FORMAT, CalendarEventDevice from homeassistant.components.calendar import (
ENTITY_ID_FORMAT,
CalendarEntity,
CalendarEvent,
)
from homeassistant.config_entries import ConfigEntry from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_SCAN_INTERVAL
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryAuthFailed from homeassistant.exceptions import ConfigEntryAuthFailed
from homeassistant.helpers.device_registry import DeviceEntryType from homeassistant.helpers.device_registry import DeviceEntryType
from homeassistant.helpers.entity import generate_entity_id from homeassistant.helpers.entity import generate_entity_id
from homeassistant.helpers.entity_platform import AddEntitiesCallback from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.template import DATE_STR_FORMAT
from homeassistant.util import Throttle, dt
from . import DOMAIN from . import DOMAIN
from .const import DEFAULT_SCAN_INTERVAL
from .fetch_data import get_lessons, get_student_info from .fetch_data import get_lessons, get_student_info
_LOGGER = logging.getLogger(__name__) _LOGGER = logging.getLogger(__name__)
@ -29,20 +30,16 @@ async def async_setup_entry(
config_entry: ConfigEntry, config_entry: ConfigEntry,
async_add_entities: AddEntitiesCallback, async_add_entities: AddEntitiesCallback,
) -> None: ) -> None:
"""Set up the calendar platform for event devices.""" """Set up the calendar platform for entity."""
VulcanCalendarData.MIN_TIME_BETWEEN_UPDATES = timedelta(
minutes=config_entry.options.get(CONF_SCAN_INTERVAL, DEFAULT_SCAN_INTERVAL)
)
client = hass.data[DOMAIN][config_entry.entry_id] client = hass.data[DOMAIN][config_entry.entry_id]
data = { data = {
"student_info": await get_student_info( "student_info": await get_student_info(
client, config_entry.data.get("student_id") client, config_entry.data.get("student_id")
), ),
"students_number": hass.data[DOMAIN]["students_number"],
} }
async_add_entities( async_add_entities(
[ [
VulcanCalendarEventDevice( VulcanCalendarEntity(
client, client,
data, data,
generate_entity_id( generate_entity_id(
@ -55,80 +52,33 @@ async def async_setup_entry(
) )
class VulcanCalendarEventDevice(CalendarEventDevice): class VulcanCalendarEntity(CalendarEntity):
"""A calendar event device.""" """A calendar entity."""
def __init__(self, client, data, entity_id): def __init__(self, client, data, entity_id) -> None:
"""Create the Calendar event device.""" """Create the Calendar entity."""
self.student_info = data["student_info"] self.student_info = data["student_info"]
self.data = VulcanCalendarData( self._event: CalendarEvent | None = None
client, self.client = client
self.student_info,
self.hass,
)
self._event = None
self.entity_id = entity_id self.entity_id = entity_id
self._unique_id = f"vulcan_calendar_{self.student_info['id']}" self._unique_id = f"vulcan_calendar_{self.student_info['id']}"
self._attr_name = f"Vulcan calendar - {self.student_info['full_name']}"
if data["students_number"] == 1:
self._attr_name = "Vulcan calendar"
self.device_name = "Calendar"
else:
self._attr_name = f"Vulcan calendar - {self.student_info['full_name']}"
self.device_name = f"{self.student_info['full_name']}: Calendar"
self._attr_unique_id = f"vulcan_calendar_{self.student_info['id']}" self._attr_unique_id = f"vulcan_calendar_{self.student_info['id']}"
self._attr_device_info = { self._attr_device_info = {
"identifiers": {(DOMAIN, f"calendar_{self.student_info['id']}")}, "identifiers": {(DOMAIN, f"calendar_{self.student_info['id']}")},
"entry_type": DeviceEntryType.SERVICE, "entry_type": DeviceEntryType.SERVICE,
"name": self.device_name, "name": f"{self.student_info['full_name']}: Calendar",
"model": f"{self.student_info['full_name']} - {self.student_info['class']} {self.student_info['school']}", "model": f"{self.student_info['full_name']} - {self.student_info['class']} {self.student_info['school']}",
"manufacturer": "Uonet +", "manufacturer": "Uonet +",
"configuration_url": f"https://uonetplus.vulcan.net.pl/{self.student_info['symbol']}", "configuration_url": f"https://uonetplus.vulcan.net.pl/{self.student_info['symbol']}",
} }
@property @property
def event(self): def event(self) -> CalendarEvent | None:
"""Return the next upcoming event.""" """Return the next upcoming event."""
return self._event return self._event
async def async_get_events(self, hass, start_date, end_date): async def async_get_events(self, hass, start_date, end_date) -> list[CalendarEvent]:
"""Get all events in a specific time frame."""
return await self.data.async_get_events(hass, start_date, end_date)
async def async_update(self):
"""Update event data."""
await self.data.async_update()
event = copy.deepcopy(self.data.event)
if event is None:
self._event = event
return
event["start"] = {
"dateTime": datetime.combine(event["date"], event["time"].from_)
.astimezone(dt.DEFAULT_TIME_ZONE)
.isoformat()
}
event["end"] = {
"dateTime": datetime.combine(event["date"], event["time"].to)
.astimezone(dt.DEFAULT_TIME_ZONE)
.isoformat()
}
self._event = event
class VulcanCalendarData:
"""Class to utilize calendar service object to get next event."""
MIN_TIME_BETWEEN_UPDATES = timedelta(minutes=DEFAULT_SCAN_INTERVAL)
def __init__(self, client, student_info, hass):
"""Set up how we are going to search the Vulcan calendar."""
self.client = client
self.event = None
self.hass = hass
self.student_info = student_info
self._available = True
async def async_get_events(self, hass, start_date, end_date):
"""Get all events in a specific time frame.""" """Get all events in a specific time frame."""
try: try:
events = await get_lessons( events = await get_lessons(
@ -136,16 +86,12 @@ class VulcanCalendarData:
date_from=start_date, date_from=start_date,
date_to=end_date, date_to=end_date,
) )
except VulcanAPIException as err: except UnauthorizedCertificateException as err:
if str(err) == "The certificate is not authorized.": raise ConfigEntryAuthFailed(
_LOGGER.error( "The certificate is not authorized, please authorize integration again"
"The certificate is not authorized, please authorize integration again" ) from err
)
raise ConfigEntryAuthFailed from err
_LOGGER.error("An API error has occurred: %s", err)
events = []
except ClientConnectorError as err: except ClientConnectorError as err:
if self._available: if self.available:
_LOGGER.warning( _LOGGER.warning(
"Connection error - please check your internet connection: %s", err "Connection error - please check your internet connection: %s", err
) )
@ -153,37 +99,27 @@ class VulcanCalendarData:
event_list = [] event_list = []
for item in events: for item in events:
event = { event = CalendarEvent(
"uid": item["id"], start=datetime.combine(item["date"], item["time"].from_),
"start": { end=datetime.combine(item["date"], item["time"].to),
"dateTime": datetime.combine( summary=item["lesson"],
item["date"], item["time"].from_ location=item["room"],
).strftime(DATE_STR_FORMAT) description=item["teacher"],
}, )
"end": {
"dateTime": datetime.combine(
item["date"], item["time"].to
).strftime(DATE_STR_FORMAT)
},
"summary": item["lesson"],
"location": item["room"],
"description": item["teacher"],
}
event_list.append(event) event_list.append(event)
return event_list return event_list
@Throttle(MIN_TIME_BETWEEN_UPDATES) async def async_update(self) -> None:
async def async_update(self):
"""Get the latest data.""" """Get the latest data."""
try: try:
events = await get_lessons(self.client) events = await get_lessons(self.client)
if not self._available: if not self.available:
_LOGGER.info("Restored connection with API") _LOGGER.info("Restored connection with API")
self._available = True self._attr_available = True
if events == []: if events == []:
events = await get_lessons( events = await get_lessons(
@ -191,22 +127,18 @@ class VulcanCalendarData:
date_to=date.today() + timedelta(days=7), date_to=date.today() + timedelta(days=7),
) )
if events == []: if events == []:
self.event = None self._event = None
return return
except VulcanAPIException as err: except UnauthorizedCertificateException as err:
if str(err) == "The certificate is not authorized.": raise ConfigEntryAuthFailed(
_LOGGER.error( "The certificate is not authorized, please authorize integration again"
"The certificate is not authorized, please authorize integration again" ) from err
)
raise ConfigEntryAuthFailed from err
_LOGGER.error("An API error has occurred: %s", err)
return
except ClientConnectorError as err: except ClientConnectorError as err:
if self._available: if self.available:
_LOGGER.warning( _LOGGER.warning(
"Connection error - please check your internet connection: %s", err "Connection error - please check your internet connection: %s", err
) )
self._available = False self._attr_available = False
return return
new_event = min( new_event = min(
@ -216,11 +148,10 @@ class VulcanCalendarData:
abs(datetime.combine(d["date"], d["time"].to) - datetime.now()), abs(datetime.combine(d["date"], d["time"].to) - datetime.now()),
), ),
) )
self.event = { self._event = CalendarEvent(
"uid": new_event["id"], start=datetime.combine(new_event["date"], new_event["time"].from_),
"date": new_event["date"], end=datetime.combine(new_event["date"], new_event["time"].to),
"time": new_event["time"], summary=new_event["lesson"],
"summary": new_event["lesson"], location=new_event["room"],
"location": new_event["room"], description=new_event["teacher"],
"description": new_event["teacher"], )
}

223
homeassistant/components/vulcan/config_flow.py
View File

@ -3,16 +3,22 @@ import logging
from aiohttp import ClientConnectionError from aiohttp import ClientConnectionError
import voluptuous as vol import voluptuous as vol
from vulcan import Account, Keystore, Vulcan from vulcan import (
from vulcan._utils import VulcanAPIException Account,
ExpiredTokenException,
InvalidPINException,
InvalidSymbolException,
InvalidTokenException,
Keystore,
UnauthorizedCertificateException,
Vulcan,
)
from homeassistant import config_entries from homeassistant import config_entries
from homeassistant.const import CONF_PIN, CONF_REGION, CONF_SCAN_INTERVAL, CONF_TOKEN from homeassistant.const import CONF_PIN, CONF_REGION, CONF_TOKEN
from homeassistant.core import callback from homeassistant.helpers.aiohttp_client import async_get_clientsession
import homeassistant.helpers.config_validation as cv
from . import DOMAIN from . import DOMAIN
from .const import DEFAULT_SCAN_INTERVAL
from .register import register from .register import register
_LOGGER = logging.getLogger(__name__) _LOGGER = logging.getLogger(__name__)
@ -29,11 +35,11 @@ class VulcanFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
VERSION = 1 VERSION = 1
@staticmethod def __init__(self):
@callback """Initialize config flow."""
def async_get_options_flow(config_entry): self.account = None
"""Get the options flow for this handler.""" self.keystore = None
return VulcanOptionsFlowHandler(config_entry) self.students = None
async def async_step_user(self, user_input=None): async def async_step_user(self, user_input=None):
"""Handle config flow.""" """Handle config flow."""
@ -53,22 +59,14 @@ class VulcanFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
user_input[CONF_REGION], user_input[CONF_REGION],
user_input[CONF_PIN], user_input[CONF_PIN],
) )
except VulcanAPIException as err: except InvalidSymbolException:
if str(err) == "Invalid token!" or str(err) == "Invalid token.": errors = {"base": "invalid_symbol"}
errors = {"base": "invalid_token"} except InvalidTokenException:
elif str(err) == "Expired token.": errors = {"base": "invalid_token"}
errors = {"base": "expired_token"} except InvalidPINException:
elif str(err) == "Invalid PIN.": errors = {"base": "invalid_pin"}
errors = {"base": "invalid_pin"} except ExpiredTokenException:
else: errors = {"base": "expired_token"}
errors = {"base": "unknown"}
_LOGGER.error(err)
except RuntimeError as err:
if str(err) == "Internal Server Error (ArgumentException)":
errors = {"base": "invalid_symbol"}
else:
errors = {"base": "unknown"}
_LOGGER.error(err)
except ClientConnectionError as err: except ClientConnectionError as err:
errors = {"base": "cannot_connect"} errors = {"base": "cannot_connect"}
_LOGGER.error("Connection error: %s", err) _LOGGER.error("Connection error: %s", err)
@ -78,12 +76,10 @@ class VulcanFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
if not errors: if not errors:
account = credentials["account"] account = credentials["account"]
keystore = credentials["keystore"] keystore = credentials["keystore"]
client = Vulcan(keystore, account) client = Vulcan(keystore, account, async_get_clientsession(self.hass))
students = await client.get_students() students = await client.get_students()
await client.close()
if len(students) > 1: if len(students) > 1:
# pylint:disable=attribute-defined-outside-init
self.account = account self.account = account
self.keystore = keystore self.keystore = keystore
self.students = students self.students = students
@ -109,10 +105,10 @@ class VulcanFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
async def async_step_select_student(self, user_input=None): async def async_step_select_student(self, user_input=None):
"""Allow user to select student.""" """Allow user to select student."""
errors = {} errors = {}
students_list = {} students = {}
if self.students is not None: if self.students is not None:
for student in self.students: for student in self.students:
students_list[ students[
str(student.pupil.id) str(student.pupil.id)
] = f"{student.pupil.first_name} {student.pupil.last_name}" ] = f"{student.pupil.first_name} {student.pupil.last_name}"
if user_input is not None: if user_input is not None:
@ -120,7 +116,7 @@ class VulcanFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
await self.async_set_unique_id(str(student_id)) await self.async_set_unique_id(str(student_id))
self._abort_if_unique_id_configured() self._abort_if_unique_id_configured()
return self.async_create_entry( return self.async_create_entry(
title=students_list[student_id], title=students[student_id],
data={ data={
"student_id": str(student_id), "student_id": str(student_id),
"keystore": self.keystore.as_dict, "keystore": self.keystore.as_dict,
@ -128,37 +124,30 @@ class VulcanFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
}, },
) )
data_schema = {
vol.Required(
"student",
): vol.In(students_list),
}
return self.async_show_form( return self.async_show_form(
step_id="select_student", step_id="select_student",
data_schema=vol.Schema(data_schema), data_schema=vol.Schema({vol.Required("student"): vol.In(students)}),
errors=errors, errors=errors,
) )
async def async_step_select_saved_credentials(self, user_input=None, errors=None): async def async_step_select_saved_credentials(self, user_input=None, errors=None):
"""Allow user to select saved credentials.""" """Allow user to select saved credentials."""
credentials_list = {}
credentials = {}
for entry in self.hass.config_entries.async_entries(DOMAIN): for entry in self.hass.config_entries.async_entries(DOMAIN):
credentials_list[entry.entry_id] = entry.data["account"]["UserName"] credentials[entry.entry_id] = entry.data["account"]["UserName"]
if user_input is not None: if user_input is not None:
entry = self.hass.config_entries.async_get_entry(user_input["credentials"]) entry = self.hass.config_entries.async_get_entry(user_input["credentials"])
keystore = Keystore.load(entry.data["keystore"]) keystore = Keystore.load(entry.data["keystore"])
account = Account.load(entry.data["account"]) account = Account.load(entry.data["account"])
client = Vulcan(keystore, account) client = Vulcan(keystore, account, async_get_clientsession(self.hass))
try: try:
students = await client.get_students() students = await client.get_students()
except VulcanAPIException as err: except UnauthorizedCertificateException:
if str(err) == "The certificate is not authorized.": return await self.async_step_auth(
return await self.async_step_auth( errors={"base": "expired_credentials"}
errors={"base": "expired_credentials"} )
)
_LOGGER.error(err)
return await self.async_step_auth(errors={"base": "unknown"})
except ClientConnectionError as err: except ClientConnectionError as err:
_LOGGER.error("Connection error: %s", err) _LOGGER.error("Connection error: %s", err)
return await self.async_step_select_saved_credentials( return await self.async_step_select_saved_credentials(
@ -167,8 +156,6 @@ class VulcanFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
except Exception: # pylint: disable=broad-except except Exception: # pylint: disable=broad-except
_LOGGER.exception("Unexpected exception") _LOGGER.exception("Unexpected exception")
return await self.async_step_auth(errors={"base": "unknown"}) return await self.async_step_auth(errors={"base": "unknown"})
finally:
await client.close()
if len(students) == 1: if len(students) == 1:
student = students[0] student = students[0]
await self.async_set_unique_id(str(student.pupil.id)) await self.async_set_unique_id(str(student.pupil.id))
@ -181,7 +168,6 @@ class VulcanFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
"account": account.as_dict, "account": account.as_dict,
}, },
) )
# pylint:disable=attribute-defined-outside-init
self.account = account self.account = account
self.keystore = keystore self.keystore = keystore
self.students = students self.students = students
@ -190,7 +176,7 @@ class VulcanFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
data_schema = { data_schema = {
vol.Required( vol.Required(
"credentials", "credentials",
): vol.In(credentials_list), ): vol.In(credentials),
} }
return self.async_show_form( return self.async_show_form(
step_id="select_saved_credentials", step_id="select_saved_credentials",
@ -200,46 +186,46 @@ class VulcanFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
async def async_step_add_next_config_entry(self, user_input=None): async def async_step_add_next_config_entry(self, user_input=None):
"""Flow initialized when user is adding next entry of that integration.""" """Flow initialized when user is adding next entry of that integration."""
existing_entries = [] existing_entries = []
for entry in self.hass.config_entries.async_entries(DOMAIN): for entry in self.hass.config_entries.async_entries(DOMAIN):
existing_entries.append(entry) existing_entries.append(entry)
errors = {} errors = {}
if user_input is not None: if user_input is not None:
if user_input["use_saved_credentials"]: if not user_input["use_saved_credentials"]:
if len(existing_entries) == 1: return await self.async_step_auth()
keystore = Keystore.load(existing_entries[0].data["keystore"]) if len(existing_entries) > 1:
account = Account.load(existing_entries[0].data["account"])
client = Vulcan(keystore, account)
students = await client.get_students()
await client.close()
new_students = []
existing_entry_ids = []
for entry in self.hass.config_entries.async_entries(DOMAIN):
existing_entry_ids.append(entry.data["student_id"])
for student in students:
if str(student.pupil.id) not in existing_entry_ids:
new_students.append(student)
if not new_students:
return self.async_abort(reason="all_student_already_configured")
if len(new_students) == 1:
await self.async_set_unique_id(str(new_students[0].pupil.id))
self._abort_if_unique_id_configured()
return self.async_create_entry(
title=f"{new_students[0].pupil.first_name} {new_students[0].pupil.last_name}",
data={
"student_id": str(new_students[0].pupil.id),
"keystore": keystore.as_dict,
"account": account.as_dict,
},
)
# pylint:disable=attribute-defined-outside-init
self.account = account
self.keystore = keystore
self.students = new_students
return await self.async_step_select_student()
return await self.async_step_select_saved_credentials() return await self.async_step_select_saved_credentials()
return await self.async_step_auth() keystore = Keystore.load(existing_entries[0].data["keystore"])
account = Account.load(existing_entries[0].data["account"])
client = Vulcan(keystore, account, async_get_clientsession(self.hass))
students = await client.get_students()
new_students = []
existing_entry_ids = []
for entry in self.hass.config_entries.async_entries(DOMAIN):
existing_entry_ids.append(entry.data["student_id"])
for student in students:
if str(student.pupil.id) not in existing_entry_ids:
new_students.append(student)
if not new_students:
return self.async_abort(reason="all_student_already_configured")
if len(new_students) == 1:
await self.async_set_unique_id(str(new_students[0].pupil.id))
self._abort_if_unique_id_configured()
return self.async_create_entry(
title=f"{new_students[0].pupil.first_name} {new_students[0].pupil.last_name}",
data={
"student_id": str(new_students[0].pupil.id),
"keystore": keystore.as_dict,
"account": account.as_dict,
},
)
self.account = account
self.keystore = keystore
self.students = new_students
return await self.async_step_select_student()
data_schema = { data_schema = {
vol.Required("use_saved_credentials", default=True): bool, vol.Required("use_saved_credentials", default=True): bool,
@ -251,6 +237,10 @@ class VulcanFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
) )
async def async_step_reauth(self, user_input=None): async def async_step_reauth(self, user_input=None):
"""Perform reauth upon an API authentication error."""
return await self.async_step_reauth_confirm()
async def async_step_reauth_confirm(self, user_input=None):
"""Reauthorize integration.""" """Reauthorize integration."""
errors = {} errors = {}
if user_input is not None: if user_input is not None:
@ -261,22 +251,14 @@ class VulcanFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
user_input[CONF_REGION], user_input[CONF_REGION],
user_input[CONF_PIN], user_input[CONF_PIN],
) )
except VulcanAPIException as err: except InvalidSymbolException:
if str(err) == "Invalid token!" or str(err) == "Invalid token.": errors = {"base": "invalid_symbol"}
errors["base"] = "invalid_token" except InvalidTokenException:
elif str(err) == "Expired token.": errors = {"base": "invalid_token"}
errors["base"] = "expired_token" except InvalidPINException:
elif str(err) == "Invalid PIN.": errors = {"base": "invalid_pin"}
errors["base"] = "invalid_pin" except ExpiredTokenException:
else: errors = {"base": "expired_token"}
errors["base"] = "unknown"
_LOGGER.error(err)
except RuntimeError as err:
if str(err) == "Internal Server Error (ArgumentException)":
errors["base"] = "invalid_symbol"
else:
errors["base"] = "unknown"
_LOGGER.error(err)
except ClientConnectionError as err: except ClientConnectionError as err:
errors["base"] = "cannot_connect" errors["base"] = "cannot_connect"
_LOGGER.error("Connection error: %s", err) _LOGGER.error("Connection error: %s", err)
@ -286,12 +268,12 @@ class VulcanFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
if not errors: if not errors:
account = credentials["account"] account = credentials["account"]
keystore = credentials["keystore"] keystore = credentials["keystore"]
client = Vulcan(keystore, account) client = Vulcan(keystore, account, async_get_clientsession(self.hass))
students = await client.get_students() students = await client.get_students()
await client.close()
existing_entries = [] existing_entries = []
for entry in self.hass.config_entries.async_entries(DOMAIN): for entry in self.hass.config_entries.async_entries(DOMAIN):
existing_entries.append(entry) existing_entries.append(entry)
matching_entries = False
for student in students: for student in students:
for entry in existing_entries: for entry in existing_entries:
if str(student.pupil.id) == str(entry.data["student_id"]): if str(student.pupil.id) == str(entry.data["student_id"]):
@ -305,38 +287,13 @@ class VulcanFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
}, },
) )
await self.hass.config_entries.async_reload(entry.entry_id) await self.hass.config_entries.async_reload(entry.entry_id)
matching_entries = True
if not matching_entries:
return self.async_abort(reason="no_matching_entries")
return self.async_abort(reason="reauth_successful") return self.async_abort(reason="reauth_successful")
return self.async_show_form( return self.async_show_form(
step_id="reauth", step_id="reauth_confirm",
data_schema=vol.Schema(LOGIN_SCHEMA), data_schema=vol.Schema(LOGIN_SCHEMA),
errors=errors, errors=errors,
) )
class VulcanOptionsFlowHandler(config_entries.OptionsFlow):
"""Config flow options for Uonet+ Vulcan."""
def __init__(self, config_entry):
"""Initialize options flow."""
self.config_entry = config_entry
async def async_step_init(self, user_input=None):
"""Manage the options."""
errors = {}
if user_input is not None:
return self.async_create_entry(title="", data=user_input)
options = {
vol.Optional(
CONF_SCAN_INTERVAL,
default=self.config_entry.options.get(
CONF_SCAN_INTERVAL, DEFAULT_SCAN_INTERVAL
),
): cv.positive_int,
}
return self.async_show_form(
step_id="init", data_schema=vol.Schema(options), errors=errors
)

1
homeassistant/components/vulcan/const.py
View File

@ -1,4 +1,3 @@
"""Constants for the Vulcan integration.""" """Constants for the Vulcan integration."""
DOMAIN = "vulcan" DOMAIN = "vulcan"
DEFAULT_SCAN_INTERVAL = 5

1
homeassistant/components/vulcan/fetch_data.py
View File

@ -94,4 +94,5 @@ async def get_student_info(client, student_id):
student_info["class"] = student.class_ student_info["class"] = student.class_
student_info["school"] = student.school.name student_info["school"] = student.school.name
student_info["symbol"] = student.symbol student_info["symbol"] = student.symbol
break
return student_info return student_info

5
homeassistant/components/vulcan/manifest.json
View File

@ -3,9 +3,8 @@
"name": "Uonet+ Vulcan", "name": "Uonet+ Vulcan",
"config_flow": true, "config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/vulcan", "documentation": "https://www.home-assistant.io/integrations/vulcan",
"requirements": ["vulcan-api==2.0.3"], "requirements": ["vulcan-api==2.1.1"],
"dependencies": [],
"codeowners": ["@Antoni-Czaplicki"], "codeowners": ["@Antoni-Czaplicki"],
"iot_class": "cloud_polling", "iot_class": "cloud_polling",
"quality_scale": "platinum" "quality_scale": "silver"
} }

5
homeassistant/components/vulcan/register.py
View File

@ -1,13 +1,10 @@
"""Support for register Vulcan account.""" """Support for register Vulcan account."""
from functools import partial
from vulcan import Account, Keystore from vulcan import Account, Keystore
async def register(hass, token, symbol, pin): async def register(hass, token, symbol, pin):
"""Register integration and save credentials.""" """Register integration and save credentials."""
keystore = await hass.async_add_executor_job( keystore = await Keystore.create(device_model="Home Assistant")
partial(Keystore.create, device_model="Home Assistant")
)
account = await Account.register(keystore, token, symbol, pin) account = await Account.register(keystore, token, symbol, pin)
return {"account": account, "keystore": keystore} return {"account": account, "keystore": keystore}

20
homeassistant/components/vulcan/strings.json
View File

@ -3,7 +3,8 @@
"abort": { "abort": {
"already_configured": "That student has already been added.", "already_configured": "That student has already been added.",
"all_student_already_configured": "All students have already been added.", "all_student_already_configured": "All students have already been added.",
"reauth_successful": "Reauth successful" "reauth_successful": "Reauth successful",
"no_matching_entries": "No matching entries found, please use different account or remove integration with outdated student.."
}, },
"error": { "error": {
"unknown": "Unknown error occurred", "unknown": "Unknown error occurred",
@ -23,7 +24,7 @@
"pin": "Pin" "pin": "Pin"
} }
}, },
"reauth": { "reauth_confirm": {
"description": "Login to your Vulcan Account using mobile app registration page.", "description": "Login to your Vulcan Account using mobile app registration page.",
"data": { "data": {
"token": "Token", "token": "Token",
@ -50,20 +51,5 @@
} }
} }
} }
},
"options": {
"error": {
"error": "Error occurred"
},
"step": {
"init": {
"data": {
"message_notify": "Show notifications when new message received",
"attendance_notify": "Show notifications about the latest attendance entries",
"grade_notify": "Show notifications about the latest grades",
"scan_interval": "Update interval (in minutes)"
}
}
}
} }
} }

2
requirements_all.txt
View File

@ -2396,7 +2396,7 @@ vsure==1.7.3
vtjp==0.1.14 vtjp==0.1.14
# homeassistant.components.vulcan # homeassistant.components.vulcan
vulcan-api==2.0.3 vulcan-api==2.1.1
# homeassistant.components.vultr # homeassistant.components.vultr
vultr==0.1.2 vultr==0.1.2

2
requirements_test_all.txt
View File

@ -1563,7 +1563,7 @@ vilfo-api-client==0.3.2
vsure==1.7.3 vsure==1.7.3
# homeassistant.components.vulcan # homeassistant.components.vulcan
vulcan-api==2.0.3 vulcan-api==2.1.1
# homeassistant.components.vultr # homeassistant.components.vultr
vultr==0.1.2 vultr==0.1.2

344
tests/components/vulcan/test_config_flow.py
View File

@ -3,17 +3,20 @@ import json
from unittest import mock from unittest import mock
from unittest.mock import patch from unittest.mock import patch
from vulcan import Account from vulcan import (
Account,
ExpiredTokenException,
InvalidPINException,
InvalidSymbolException,
InvalidTokenException,
UnauthorizedCertificateException,
)
from vulcan.model import Student from vulcan.model import Student
from homeassistant import config_entries, data_entry_flow from homeassistant import config_entries, data_entry_flow
from homeassistant.components.vulcan import config_flow, const, register from homeassistant.components.vulcan import config_flow, const, register
from homeassistant.components.vulcan.config_flow import ( from homeassistant.components.vulcan.config_flow import ClientConnectionError, Keystore
ClientConnectionError, from homeassistant.const import CONF_PIN, CONF_REGION, CONF_TOKEN
Keystore,
VulcanAPIException,
)
from homeassistant.const import CONF_PIN, CONF_REGION, CONF_SCAN_INTERVAL, CONF_TOKEN
from tests.common import MockConfigEntry, load_fixture from tests.common import MockConfigEntry, load_fixture
@ -56,13 +59,20 @@ async def test_config_flow_auth_success(
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "auth" assert result["step_id"] == "auth"
assert result["errors"] is None assert result["errors"] is None
result = await hass.config_entries.flow.async_configure(
result["flow_id"], with patch(
{CONF_TOKEN: "token", CONF_REGION: "region", CONF_PIN: "000000"}, "homeassistant.components.vulcan.async_setup_entry",
) return_value=True,
) as mock_setup_entry:
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{CONF_TOKEN: "token", CONF_REGION: "region", CONF_PIN: "000000"},
)
await hass.async_block_till_done()
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert result["title"] == "Jan Kowalski" assert result["title"] == "Jan Kowalski"
assert len(mock_setup_entry.mock_calls) == 1
@mock.patch("homeassistant.components.vulcan.config_flow.Vulcan.get_students") @mock.patch("homeassistant.components.vulcan.config_flow.Vulcan.get_students")
@ -96,13 +106,18 @@ async def test_config_flow_auth_success_with_multiple_students(
assert result["step_id"] == "select_student" assert result["step_id"] == "select_student"
assert result["errors"] == {} assert result["errors"] == {}
result = await hass.config_entries.flow.async_configure( with patch(
result["flow_id"], "homeassistant.components.vulcan.async_setup_entry",
{"student": "0"}, return_value=True,
) ) as mock_setup_entry:
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{"student": "0"},
)
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert result["title"] == "Jan Kowalski" assert result["title"] == "Jan Kowalski"
assert len(mock_setup_entry.mock_calls) == 1
@mock.patch("homeassistant.components.vulcan.config_flow.Vulcan.get_students") @mock.patch("homeassistant.components.vulcan.config_flow.Vulcan.get_students")
@ -120,14 +135,53 @@ async def test_config_flow_reauth_success(
MockConfigEntry( MockConfigEntry(
domain=const.DOMAIN, domain=const.DOMAIN,
unique_id="0", unique_id="0",
data={"student_id": "0", "login": "example@example.com"}, data={"student_id": "0"},
).add_to_hass(hass) ).add_to_hass(hass)
result = await hass.config_entries.flow.async_init( result = await hass.config_entries.flow.async_init(
const.DOMAIN, context={"source": config_entries.SOURCE_REAUTH} const.DOMAIN, context={"source": config_entries.SOURCE_REAUTH}
) )
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "reauth" assert result["step_id"] == "reauth_confirm"
assert result["errors"] == {}
with patch(
"homeassistant.components.vulcan.async_setup_entry",
return_value=True,
) as mock_setup_entry:
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{CONF_TOKEN: "token", CONF_REGION: "region", CONF_PIN: "000000"},
)
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
assert result["reason"] == "reauth_successful"
assert len(mock_setup_entry.mock_calls) == 1
@mock.patch("homeassistant.components.vulcan.config_flow.Vulcan.get_students")
@mock.patch("homeassistant.components.vulcan.config_flow.Keystore.create")
@mock.patch("homeassistant.components.vulcan.config_flow.Account.register")
async def test_config_flow_reauth_without_matching_entries(
mock_account, mock_keystore, mock_student, hass
):
"""Test a aborted config flow reauth caused by leak of matching entries."""
mock_keystore.return_value = fake_keystore
mock_account.return_value = fake_account
mock_student.return_value = [
Student.load(load_fixture("fake_student_1.json", "vulcan"))
]
MockConfigEntry(
domain=const.DOMAIN,
unique_id="0",
data={"student_id": "1"},
).add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
const.DOMAIN, context={"source": config_entries.SOURCE_REAUTH}
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "reauth_confirm"
assert result["errors"] == {} assert result["errors"] == {}
result = await hass.config_entries.flow.async_configure( result = await hass.config_entries.flow.async_configure(
@ -136,7 +190,7 @@ async def test_config_flow_reauth_success(
) )
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
assert result["reason"] == "reauth_successful" assert result["reason"] == "no_matching_entries"
@mock.patch("homeassistant.components.vulcan.config_flow.Keystore.create") @mock.patch("homeassistant.components.vulcan.config_flow.Keystore.create")
@ -149,11 +203,11 @@ async def test_config_flow_reauth_with_errors(mock_account, mock_keystore, hass)
const.DOMAIN, context={"source": config_entries.SOURCE_REAUTH} const.DOMAIN, context={"source": config_entries.SOURCE_REAUTH}
) )
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "reauth" assert result["step_id"] == "reauth_confirm"
assert result["errors"] == {} assert result["errors"] == {}
with patch( with patch(
"homeassistant.components.vulcan.config_flow.Account.register", "homeassistant.components.vulcan.config_flow.Account.register",
side_effect=VulcanAPIException("Invalid token."), side_effect=InvalidTokenException,
): ):
result = await hass.config_entries.flow.async_configure( result = await hass.config_entries.flow.async_configure(
@ -162,12 +216,12 @@ async def test_config_flow_reauth_with_errors(mock_account, mock_keystore, hass)
) )
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "reauth" assert result["step_id"] == "reauth_confirm"
assert result["errors"] == {"base": "invalid_token"} assert result["errors"] == {"base": "invalid_token"}
with patch( with patch(
"homeassistant.components.vulcan.config_flow.Account.register", "homeassistant.components.vulcan.config_flow.Account.register",
side_effect=VulcanAPIException("Expired token."), side_effect=ExpiredTokenException,
): ):
result = await hass.config_entries.flow.async_configure( result = await hass.config_entries.flow.async_configure(
@ -176,12 +230,12 @@ async def test_config_flow_reauth_with_errors(mock_account, mock_keystore, hass)
) )
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "reauth" assert result["step_id"] == "reauth_confirm"
assert result["errors"] == {"base": "expired_token"} assert result["errors"] == {"base": "expired_token"}
with patch( with patch(
"homeassistant.components.vulcan.config_flow.Account.register", "homeassistant.components.vulcan.config_flow.Account.register",
side_effect=VulcanAPIException("Invalid PIN."), side_effect=InvalidPINException,
): ):
result = await hass.config_entries.flow.async_configure( result = await hass.config_entries.flow.async_configure(
@ -190,12 +244,12 @@ async def test_config_flow_reauth_with_errors(mock_account, mock_keystore, hass)
) )
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "reauth" assert result["step_id"] == "reauth_confirm"
assert result["errors"] == {"base": "invalid_pin"} assert result["errors"] == {"base": "invalid_pin"}
with patch( with patch(
"homeassistant.components.vulcan.config_flow.Account.register", "homeassistant.components.vulcan.config_flow.Account.register",
side_effect=VulcanAPIException("Unknown error"), side_effect=InvalidSymbolException,
): ):
result = await hass.config_entries.flow.async_configure( result = await hass.config_entries.flow.async_configure(
@ -204,37 +258,9 @@ async def test_config_flow_reauth_with_errors(mock_account, mock_keystore, hass)
) )
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "reauth" assert result["step_id"] == "reauth_confirm"
assert result["errors"] == {"base": "unknown"}
with patch(
"homeassistant.components.vulcan.config_flow.Account.register",
side_effect=RuntimeError("Internal Server Error (ArgumentException)"),
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{CONF_TOKEN: "token", CONF_REGION: "region", CONF_PIN: "000000"},
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "reauth"
assert result["errors"] == {"base": "invalid_symbol"} assert result["errors"] == {"base": "invalid_symbol"}
with patch(
"homeassistant.components.vulcan.config_flow.Account.register",
side_effect=RuntimeError("Unknown error"),
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{CONF_TOKEN: "token", CONF_REGION: "region", CONF_PIN: "000000"},
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "reauth"
assert result["errors"] == {"base": "unknown"}
with patch( with patch(
"homeassistant.components.vulcan.config_flow.Account.register", "homeassistant.components.vulcan.config_flow.Account.register",
side_effect=ClientConnectionError, side_effect=ClientConnectionError,
@ -246,7 +272,7 @@ async def test_config_flow_reauth_with_errors(mock_account, mock_keystore, hass)
) )
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "reauth" assert result["step_id"] == "reauth_confirm"
assert result["errors"] == {"base": "cannot_connect"} assert result["errors"] == {"base": "cannot_connect"}
with patch( with patch(
@ -260,7 +286,7 @@ async def test_config_flow_reauth_with_errors(mock_account, mock_keystore, hass)
) )
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "reauth" assert result["step_id"] == "reauth_confirm"
assert result["errors"] == {"base": "unknown"} assert result["errors"] == {"base": "unknown"}
@ -297,13 +323,18 @@ async def test_multiple_config_entries(mock_account, mock_keystore, mock_student
assert result["step_id"] == "auth" assert result["step_id"] == "auth"
assert result["errors"] is None assert result["errors"] is None
result = await hass.config_entries.flow.async_configure( with patch(
result["flow_id"], "homeassistant.components.vulcan.async_setup_entry",
{CONF_TOKEN: "token", CONF_REGION: "region", CONF_PIN: "000000"}, return_value=True,
) ) as mock_setup_entry:
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{CONF_TOKEN: "token", CONF_REGION: "region", CONF_PIN: "000000"},
)
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert result["title"] == "Jan Kowalski" assert result["title"] == "Jan Kowalski"
assert len(mock_setup_entry.mock_calls) == 2
@mock.patch("homeassistant.components.vulcan.config_flow.Vulcan.get_students") @mock.patch("homeassistant.components.vulcan.config_flow.Vulcan.get_students")
@ -326,13 +357,18 @@ async def test_multiple_config_entries_using_saved_credentials(mock_student, has
assert result["step_id"] == "add_next_config_entry" assert result["step_id"] == "add_next_config_entry"
assert result["errors"] == {} assert result["errors"] == {}
result = await hass.config_entries.flow.async_configure( with patch(
result["flow_id"], "homeassistant.components.vulcan.async_setup_entry",
{"use_saved_credentials": True}, return_value=True,
) ) as mock_setup_entry:
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{"use_saved_credentials": True},
)
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert result["title"] == "Jan Kowalski" assert result["title"] == "Jan Kowalski"
assert len(mock_setup_entry.mock_calls) == 2
@mock.patch("homeassistant.components.vulcan.config_flow.Vulcan.get_students") @mock.patch("homeassistant.components.vulcan.config_flow.Vulcan.get_students")
@ -364,13 +400,18 @@ async def test_multiple_config_entries_using_saved_credentials_2(mock_student, h
assert result["step_id"] == "select_student" assert result["step_id"] == "select_student"
assert result["errors"] == {} assert result["errors"] == {}
result = await hass.config_entries.flow.async_configure( with patch(
result["flow_id"], "homeassistant.components.vulcan.async_setup_entry",
{"student": "0"}, return_value=True,
) ) as mock_setup_entry:
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{"student": "0"},
)
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert result["title"] == "Jan Kowalski" assert result["title"] == "Jan Kowalski"
assert len(mock_setup_entry.mock_calls) == 2
@mock.patch("homeassistant.components.vulcan.config_flow.Vulcan.get_students") @mock.patch("homeassistant.components.vulcan.config_flow.Vulcan.get_students")
@ -410,13 +451,18 @@ async def test_multiple_config_entries_using_saved_credentials_3(mock_student, h
assert result["step_id"] == "select_saved_credentials" assert result["step_id"] == "select_saved_credentials"
assert result["errors"] is None assert result["errors"] is None
result = await hass.config_entries.flow.async_configure( with patch(
result["flow_id"], "homeassistant.components.vulcan.async_setup_entry",
{"credentials": "123"}, return_value=True,
) ) as mock_setup_entry:
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{"credentials": "123"},
)
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert result["title"] == "Jan Kowalski" assert result["title"] == "Jan Kowalski"
assert len(mock_setup_entry.mock_calls) == 3
@mock.patch("homeassistant.components.vulcan.config_flow.Vulcan.get_students") @mock.patch("homeassistant.components.vulcan.config_flow.Vulcan.get_students")
@ -465,13 +511,18 @@ async def test_multiple_config_entries_using_saved_credentials_4(mock_student, h
assert result["step_id"] == "select_student" assert result["step_id"] == "select_student"
assert result["errors"] == {} assert result["errors"] == {}
result = await hass.config_entries.flow.async_configure( with patch(
result["flow_id"], "homeassistant.components.vulcan.async_setup_entry",
{"student": "0"}, return_value=True,
) ) as mock_setup_entry:
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{"student": "0"},
)
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert result["title"] == "Jan Kowalski" assert result["title"] == "Jan Kowalski"
assert len(mock_setup_entry.mock_calls) == 3
async def test_multiple_config_entries_without_valid_saved_credentials(hass): async def test_multiple_config_entries_without_valid_saved_credentials(hass):
@ -504,7 +555,7 @@ async def test_multiple_config_entries_without_valid_saved_credentials(hass):
) )
with patch( with patch(
"homeassistant.components.vulcan.config_flow.Vulcan.get_students", "homeassistant.components.vulcan.config_flow.Vulcan.get_students",
side_effect=VulcanAPIException("The certificate is not authorized."), side_effect=UnauthorizedCertificateException,
): ):
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "select_saved_credentials" assert result["step_id"] == "select_saved_credentials"
@ -614,54 +665,6 @@ async def test_multiple_config_entries_using_saved_credentials_with_unknown_erro
assert result["errors"] == {"base": "unknown"} assert result["errors"] == {"base": "unknown"}
async def test_multiple_config_entries_using_saved_credentials_with_unknown_api_error(
hass,
):
"""Test a unsuccessful config flow for multiple config entries without valid saved credentials."""
MockConfigEntry(
entry_id="456",
domain=const.DOMAIN,
unique_id="234567",
data=json.loads(load_fixture("fake_config_entry_data.json", "vulcan"))
| {"student_id": "456"},
).add_to_hass(hass)
MockConfigEntry(
entry_id="123",
domain=const.DOMAIN,
unique_id="123456",
data=json.loads(load_fixture("fake_config_entry_data.json", "vulcan")),
).add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
const.DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "add_next_config_entry"
assert result["errors"] == {}
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{"use_saved_credentials": True},
)
with patch(
"homeassistant.components.vulcan.config_flow.Vulcan.get_students",
side_effect=VulcanAPIException("Unknown error"),
):
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "select_saved_credentials"
assert result["errors"] is None
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{"credentials": "123"},
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "auth"
assert result["errors"] == {"base": "unknown"}
@mock.patch("homeassistant.components.vulcan.config_flow.Vulcan.get_students") @mock.patch("homeassistant.components.vulcan.config_flow.Vulcan.get_students")
@mock.patch("homeassistant.components.vulcan.config_flow.Keystore.create") @mock.patch("homeassistant.components.vulcan.config_flow.Keystore.create")
@mock.patch("homeassistant.components.vulcan.config_flow.Account.register") @mock.patch("homeassistant.components.vulcan.config_flow.Account.register")
@ -704,7 +707,7 @@ async def test_config_flow_auth_invalid_token(mock_keystore, hass):
mock_keystore.return_value = fake_keystore mock_keystore.return_value = fake_keystore
with patch( with patch(
"homeassistant.components.vulcan.config_flow.Account.register", "homeassistant.components.vulcan.config_flow.Account.register",
side_effect=VulcanAPIException("Invalid token."), side_effect=InvalidTokenException,
): ):
result = await hass.config_entries.flow.async_init( result = await hass.config_entries.flow.async_init(
const.DOMAIN, context={"source": config_entries.SOURCE_USER} const.DOMAIN, context={"source": config_entries.SOURCE_USER}
@ -730,7 +733,7 @@ async def test_config_flow_auth_invalid_region(mock_keystore, hass):
mock_keystore.return_value = fake_keystore mock_keystore.return_value = fake_keystore
with patch( with patch(
"homeassistant.components.vulcan.config_flow.Account.register", "homeassistant.components.vulcan.config_flow.Account.register",
side_effect=RuntimeError("Internal Server Error (ArgumentException)"), side_effect=InvalidSymbolException,
): ):
result = await hass.config_entries.flow.async_init( result = await hass.config_entries.flow.async_init(
const.DOMAIN, context={"source": config_entries.SOURCE_USER} const.DOMAIN, context={"source": config_entries.SOURCE_USER}
@ -756,7 +759,7 @@ async def test_config_flow_auth_invalid_pin(mock_keystore, hass):
mock_keystore.return_value = fake_keystore mock_keystore.return_value = fake_keystore
with patch( with patch(
"homeassistant.components.vulcan.config_flow.Account.register", "homeassistant.components.vulcan.config_flow.Account.register",
side_effect=VulcanAPIException("Invalid PIN."), side_effect=InvalidPINException,
): ):
result = await hass.config_entries.flow.async_init( result = await hass.config_entries.flow.async_init(
const.DOMAIN, context={"source": config_entries.SOURCE_USER} const.DOMAIN, context={"source": config_entries.SOURCE_USER}
@ -782,7 +785,7 @@ async def test_config_flow_auth_expired_token(mock_keystore, hass):
mock_keystore.return_value = fake_keystore mock_keystore.return_value = fake_keystore
with patch( with patch(
"homeassistant.components.vulcan.config_flow.Account.register", "homeassistant.components.vulcan.config_flow.Account.register",
side_effect=VulcanAPIException("Expired token."), side_effect=ExpiredTokenException,
): ):
result = await hass.config_entries.flow.async_init( result = await hass.config_entries.flow.async_init(
const.DOMAIN, context={"source": config_entries.SOURCE_USER} const.DOMAIN, context={"source": config_entries.SOURCE_USER}
@ -802,58 +805,6 @@ async def test_config_flow_auth_expired_token(mock_keystore, hass):
assert result["errors"] == {"base": "expired_token"} assert result["errors"] == {"base": "expired_token"}
@mock.patch("homeassistant.components.vulcan.config_flow.Keystore.create")
async def test_config_flow_auth_api_unknown_error(mock_keystore, hass):
"""Test a config flow with unknown API error."""
mock_keystore.return_value = fake_keystore
with patch(
"homeassistant.components.vulcan.config_flow.Account.register",
side_effect=VulcanAPIException("Unknown error"),
):
result = await hass.config_entries.flow.async_init(
const.DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "auth"
assert result["errors"] is None
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{CONF_TOKEN: "3S10000", CONF_REGION: "region", CONF_PIN: "000000"},
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "auth"
assert result["errors"] == {"base": "unknown"}
@mock.patch("homeassistant.components.vulcan.config_flow.Keystore.create")
async def test_config_flow_auth_api_unknown_runtime_error(mock_keystore, hass):
"""Test a config flow with runtime error."""
mock_keystore.return_value = fake_keystore
with patch(
"homeassistant.components.vulcan.config_flow.Account.register",
side_effect=RuntimeError("Unknown error"),
):
result = await hass.config_entries.flow.async_init(
const.DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "auth"
assert result["errors"] is None
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{CONF_TOKEN: "3S10000", CONF_REGION: "region", CONF_PIN: "000000"},
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "auth"
assert result["errors"] == {"base": "unknown"}
@mock.patch("homeassistant.components.vulcan.config_flow.Keystore.create") @mock.patch("homeassistant.components.vulcan.config_flow.Keystore.create")
async def test_config_flow_auth_connection_error(mock_keystore, hass): async def test_config_flow_auth_connection_error(mock_keystore, hass):
"""Test a config flow with connection error.""" """Test a config flow with connection error."""
@ -904,32 +855,3 @@ async def test_config_flow_auth_unknown_error(mock_keystore, hass):
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "auth" assert result["step_id"] == "auth"
assert result["errors"] == {"base": "unknown"} assert result["errors"] == {"base": "unknown"}
@mock.patch("homeassistant.components.vulcan.Vulcan.get_students")
async def test_options_flow(mock_student, hass):
"""Test config flow options."""
mock_student.return_value = [
Student.load(load_fixture("fake_student_1.json", "vulcan"))
]
config_entry = MockConfigEntry(
domain=const.DOMAIN,
unique_id="0",
data=json.loads(load_fixture("fake_config_entry_data.json", "vulcan")),
)
config_entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
result = await hass.config_entries.options.async_init(config_entry.entry_id)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "init"
result = await hass.config_entries.options.async_configure(
result["flow_id"], user_input={CONF_SCAN_INTERVAL: 2137}
)
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert config_entry.options == {CONF_SCAN_INTERVAL: 2137}