Home-Assistant
/
core
mirror of https://github.com/home-assistant/core.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Remove the vulcan integration (#151504)

pull/151544/head
Antoni Czaplicki 2025-09-02 01:10:07 +02:00 committed by GitHub
parent e9dcde1bb5
commit 180f898bfa
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
20 changed files with 0 additions and 1757 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
CODEOWNERS generated
View File

@ -1720,8 +1720,6 @@ build.json @home-assistant/supervisor
/tests/components/volvo/ @thomasddn
/homeassistant/components/volvooncall/ @molobrakos
/tests/components/volvooncall/ @molobrakos
/homeassistant/components/vulcan/ @Antoni-Czaplicki
/tests/components/vulcan/ @Antoni-Czaplicki
/homeassistant/components/wake_on_lan/ @ntilley905
/tests/components/wake_on_lan/ @ntilley905
/homeassistant/components/wake_word/ @home-assistant/core @synesthesiam

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

@ -1,48 +0,0 @@
"""The Vulcan component."""
from aiohttp import ClientConnectorError
from vulcan import Account, Keystore, UnauthorizedCertificateException, Vulcan
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import Platform
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryAuthFailed, ConfigEntryNotReady
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from .const import DOMAIN
PLATFORMS = [Platform.CALENDAR]
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up Uonet+ Vulcan integration."""
hass.data.setdefault(DOMAIN, {})
try:
keystore = Keystore.load(entry.data["keystore"])
account = Account.load(entry.data["account"])
client = Vulcan(keystore, account, async_get_clientsession(hass))
await client.select_student()
students = await client.get_students()
for student in students:
if str(student.pupil.id) == str(entry.data["student_id"]):
client.student = student
break
except UnauthorizedCertificateException as err:
raise ConfigEntryAuthFailed("The certificate is not authorized.") from err
except ClientConnectorError as err:
raise ConfigEntryNotReady(
f"Connection error - please check your internet connection: {err}"
) from err
hass.data[DOMAIN][entry.entry_id] = client
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
return True
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Unload a config entry."""
if unload_ok := await hass.config_entries.async_unload_platforms(entry, PLATFORMS):
hass.data[DOMAIN].pop(entry.entry_id)
return unload_ok

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

@ -1,176 +0,0 @@
"""Support for Vulcan Calendar platform."""
from __future__ import annotations
from datetime import date, datetime, timedelta
import logging
from typing import cast
from zoneinfo import ZoneInfo
from aiohttp import ClientConnectorError
from vulcan import UnauthorizedCertificateException
from homeassistant.components.calendar import (
ENTITY_ID_FORMAT,
CalendarEntity,
CalendarEvent,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryAuthFailed
from homeassistant.helpers.device_registry import DeviceEntryType, DeviceInfo
from homeassistant.helpers.entity import generate_entity_id
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from . import DOMAIN
from .fetch_data import get_lessons, get_student_info
_LOGGER = logging.getLogger(__name__)
async def async_setup_entry(
hass: HomeAssistant,
config_entry: ConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up the calendar platform for entity."""
client = hass.data[DOMAIN][config_entry.entry_id]
data = {
"student_info": await get_student_info(
client, config_entry.data.get("student_id")
),
}
async_add_entities(
[
VulcanCalendarEntity(
client,
data,
generate_entity_id(
ENTITY_ID_FORMAT,
f"vulcan_calendar_{data['student_info']['full_name']}",
hass=hass,
),
)
],
)
class VulcanCalendarEntity(CalendarEntity):
"""A calendar entity."""
_attr_has_entity_name = True
_attr_translation_key = "calendar"
def __init__(self, client, data, entity_id) -> None:
"""Create the Calendar entity."""
self._event: CalendarEvent | None = None
self.client = client
self.entity_id = entity_id
student_info = data["student_info"]
self._attr_unique_id = f"vulcan_calendar_{student_info['id']}"
self._attr_device_info = DeviceInfo(
identifiers={(DOMAIN, f"calendar_{student_info['id']}")},
entry_type=DeviceEntryType.SERVICE,
name=cast(str, student_info["full_name"]),
model=(
f"{student_info['full_name']} -"
f" {student_info['class']} {student_info['school']}"
),
manufacturer="Uonet +",
configuration_url=(
f"https://uonetplus.vulcan.net.pl/{student_info['symbol']}"
),
)
@property
def event(self) -> CalendarEvent | None:
"""Return the next upcoming event."""
return self._event
async def async_get_events(
self, hass: HomeAssistant, start_date: datetime, end_date: datetime
) -> list[CalendarEvent]:
"""Get all events in a specific time frame."""
try:
events = await get_lessons(
self.client,
date_from=start_date,
date_to=end_date,
)
except UnauthorizedCertificateException as err:
raise ConfigEntryAuthFailed(
"The certificate is not authorized, please authorize integration again"
) from err
except ClientConnectorError as err:
if self.available:
_LOGGER.warning(
"Connection error - please check your internet connection: %s", err
)
events = []
event_list = []
for item in events:
event = CalendarEvent(
start=datetime.combine(
item["date"], item["time"].from_, ZoneInfo("Europe/Warsaw")
),
end=datetime.combine(
item["date"], item["time"].to, ZoneInfo("Europe/Warsaw")
),
summary=item["lesson"],
location=item["room"],
description=item["teacher"],
)
event_list.append(event)
return event_list
async def async_update(self) -> None:
"""Get the latest data."""
try:
events = await get_lessons(self.client)
if not self.available:
_LOGGER.warning("Restored connection with API")
self._attr_available = True
if events == []:
events = await get_lessons(
self.client,
date_to=date.today() + timedelta(days=7),
)
if events == []:
self._event = None
return
except UnauthorizedCertificateException as err:
raise ConfigEntryAuthFailed(
"The certificate is not authorized, please authorize integration again"
) from err
except ClientConnectorError as err:
if self.available:
_LOGGER.warning(
"Connection error - please check your internet connection: %s", err
)
self._attr_available = False
return
new_event = min(
events,
key=lambda d: (
datetime.combine(d["date"], d["time"].to) < datetime.now(),
abs(datetime.combine(d["date"], d["time"].to) - datetime.now()),
),
)
self._event = CalendarEvent(
start=datetime.combine(
new_event["date"], new_event["time"].from_, ZoneInfo("Europe/Warsaw")
),
end=datetime.combine(
new_event["date"], new_event["time"].to, ZoneInfo("Europe/Warsaw")
),
summary=new_event["lesson"],
location=new_event["room"],
description=new_event["teacher"],
)

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

@ -1,327 +0,0 @@
"""Adds config flow for Vulcan."""
from collections.abc import Mapping
import logging
from typing import TYPE_CHECKING, Any
from aiohttp import ClientConnectionError
import voluptuous as vol
from vulcan import (
Account,
ExpiredTokenException,
InvalidPINException,
InvalidSymbolException,
InvalidTokenException,
Keystore,
UnauthorizedCertificateException,
Vulcan,
)
from vulcan.model import Student
from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
from homeassistant.const import CONF_PIN, CONF_REGION, CONF_TOKEN
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from . import DOMAIN
from .register import register
_LOGGER = logging.getLogger(__name__)
LOGIN_SCHEMA = {
vol.Required(CONF_TOKEN): str,
vol.Required(CONF_REGION): str,
vol.Required(CONF_PIN): str,
}
class VulcanFlowHandler(ConfigFlow, domain=DOMAIN):
"""Handle a Uonet+ Vulcan config flow."""
VERSION = 1
account: Account
keystore: Keystore
def __init__(self) -> None:
"""Initialize config flow."""
self.students: list[Student] | None = None
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Handle config flow."""
if self._async_current_entries():
return await self.async_step_add_next_config_entry()
return await self.async_step_auth()
async def async_step_auth(
self,
user_input: dict[str, str] | None = None,
errors: dict[str, str] | None = None,
) -> ConfigFlowResult:
"""Authorize integration."""
if user_input is not None:
try:
credentials = await register(
user_input[CONF_TOKEN],
user_input[CONF_REGION],
user_input[CONF_PIN],
)
except InvalidSymbolException:
errors = {"base": "invalid_symbol"}
except InvalidTokenException:
errors = {"base": "invalid_token"}
except InvalidPINException:
errors = {"base": "invalid_pin"}
except ExpiredTokenException:
errors = {"base": "expired_token"}
except ClientConnectionError as err:
errors = {"base": "cannot_connect"}
_LOGGER.error("Connection error: %s", err)
except Exception:
_LOGGER.exception("Unexpected exception")
errors = {"base": "unknown"}
if not errors:
account = credentials["account"]
keystore = credentials["keystore"]
client = Vulcan(keystore, account, async_get_clientsession(self.hass))
students = await client.get_students()
if len(students) > 1:
self.account = account
self.keystore = keystore
self.students = students
return await self.async_step_select_student()
student = students[0]
await self.async_set_unique_id(str(student.pupil.id))
self._abort_if_unique_id_configured()
return self.async_create_entry(
title=f"{student.pupil.first_name} {student.pupil.last_name}",
data={
"student_id": str(student.pupil.id),
"keystore": keystore.as_dict,
"account": account.as_dict,
},
)
return self.async_show_form(
step_id="auth",
data_schema=vol.Schema(LOGIN_SCHEMA),
errors=errors,
)
async def async_step_select_student(
self, user_input: dict[str, str] | None = None
) -> ConfigFlowResult:
"""Allow user to select student."""
errors: dict[str, str] = {}
students: dict[str, str] = {}
if self.students is not None:
for student in self.students:
students[str(student.pupil.id)] = (
f"{student.pupil.first_name} {student.pupil.last_name}"
)
if user_input is not None:
if TYPE_CHECKING:
assert self.keystore is not None
student_id = user_input["student"]
await self.async_set_unique_id(str(student_id))
self._abort_if_unique_id_configured()
return self.async_create_entry(
title=students[student_id],
data={
"student_id": str(student_id),
"keystore": self.keystore.as_dict,
"account": self.account.as_dict,
},
)
return self.async_show_form(
step_id="select_student",
data_schema=vol.Schema({vol.Required("student"): vol.In(students)}),
errors=errors,
)
async def async_step_select_saved_credentials(
self,
user_input: dict[str, str] | None = None,
errors: dict[str, str] | None = None,
) -> ConfigFlowResult:
"""Allow user to select saved credentials."""
credentials: dict[str, Any] = {}
for entry in self.hass.config_entries.async_entries(DOMAIN):
credentials[entry.entry_id] = entry.data["account"]["UserName"]
if user_input is not None:
existing_entry = self.hass.config_entries.async_get_entry(
user_input["credentials"]
)
if TYPE_CHECKING:
assert existing_entry is not None
keystore = Keystore.load(existing_entry.data["keystore"])
account = Account.load(existing_entry.data["account"])
client = Vulcan(keystore, account, async_get_clientsession(self.hass))
try:
students = await client.get_students()
except UnauthorizedCertificateException:
return await self.async_step_auth(
errors={"base": "expired_credentials"}
)
except ClientConnectionError as err:
_LOGGER.error("Connection error: %s", err)
return await self.async_step_select_saved_credentials(
errors={"base": "cannot_connect"}
)
except Exception:
_LOGGER.exception("Unexpected exception")
return await self.async_step_auth(errors={"base": "unknown"})
if len(students) == 1:
student = students[0]
await self.async_set_unique_id(str(student.pupil.id))
self._abort_if_unique_id_configured()
return self.async_create_entry(
title=f"{student.pupil.first_name} {student.pupil.last_name}",
data={
"student_id": str(student.pupil.id),
"keystore": keystore.as_dict,
"account": account.as_dict,
},
)
self.account = account
self.keystore = keystore
self.students = students
return await self.async_step_select_student()
data_schema = {
vol.Required(
"credentials",
): vol.In(credentials),
}
return self.async_show_form(
step_id="select_saved_credentials",
data_schema=vol.Schema(data_schema),
errors=errors,
)
async def async_step_add_next_config_entry(
self, user_input: dict[str, bool] | None = None
) -> ConfigFlowResult:
"""Flow initialized when user is adding next entry of that integration."""
existing_entries = self.hass.config_entries.async_entries(DOMAIN)
errors: dict[str, str] = {}
if user_input is not None:
if not user_input["use_saved_credentials"]:
return await self.async_step_auth()
if len(existing_entries) > 1:
return await self.async_step_select_saved_credentials()
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()
existing_entry_ids = [
entry.data["student_id"] for entry in existing_entries
]
new_students = [
student
for student in students
if str(student.pupil.id) not in existing_entry_ids
]
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 = {
vol.Required("use_saved_credentials", default=True): bool,
}
return self.async_show_form(
step_id="add_next_config_entry",
data_schema=vol.Schema(data_schema),
errors=errors,
)
async def async_step_reauth(
self, entry_data: Mapping[str, Any]
) -> ConfigFlowResult:
"""Perform reauth upon an API authentication error."""
return await self.async_step_reauth_confirm()
async def async_step_reauth_confirm(
self, user_input: dict[str, str] | None = None
) -> ConfigFlowResult:
"""Reauthorize integration."""
errors = {}
if user_input is not None:
try:
credentials = await register(
user_input[CONF_TOKEN],
user_input[CONF_REGION],
user_input[CONF_PIN],
)
except InvalidSymbolException:
errors = {"base": "invalid_symbol"}
except InvalidTokenException:
errors = {"base": "invalid_token"}
except InvalidPINException:
errors = {"base": "invalid_pin"}
except ExpiredTokenException:
errors = {"base": "expired_token"}
except ClientConnectionError as err:
errors["base"] = "cannot_connect"
_LOGGER.error("Connection error: %s", err)
except Exception:
_LOGGER.exception("Unexpected exception")
errors["base"] = "unknown"
if not errors:
account = credentials["account"]
keystore = credentials["keystore"]
client = Vulcan(keystore, account, async_get_clientsession(self.hass))
students = await client.get_students()
existing_entries = self.hass.config_entries.async_entries(DOMAIN)
matching_entries = False
for student in students:
for entry in existing_entries:
if str(student.pupil.id) == str(entry.data["student_id"]):
self.hass.config_entries.async_update_entry(
entry,
title=(
f"{student.pupil.first_name} {student.pupil.last_name}"
),
data={
"student_id": str(student.pupil.id),
"keystore": keystore.as_dict,
"account": account.as_dict,
},
)
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_show_form(
step_id="reauth_confirm",
data_schema=vol.Schema(LOGIN_SCHEMA),
errors=errors,
)

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

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

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

@ -1,98 +0,0 @@
"""Support for fetching Vulcan data."""
async def get_lessons(client, date_from=None, date_to=None):
"""Support for fetching Vulcan lessons."""
changes = {}
list_ans = []
async for lesson in await client.data.get_changed_lessons(
date_from=date_from, date_to=date_to
):
temp_dict = {}
_id = str(lesson.id)
temp_dict["id"] = lesson.id
temp_dict["number"] = lesson.time.position if lesson.time is not None else None
temp_dict["lesson"] = (
lesson.subject.name if lesson.subject is not None else None
)
temp_dict["room"] = lesson.room.code if lesson.room is not None else None
temp_dict["changes"] = lesson.changes
temp_dict["note"] = lesson.note
temp_dict["reason"] = lesson.reason
temp_dict["event"] = lesson.event
temp_dict["group"] = lesson.group
temp_dict["teacher"] = (
lesson.teacher.display_name if lesson.teacher is not None else None
)
temp_dict["from_to"] = (
lesson.time.displayed_time if lesson.time is not None else None
)
changes[str(_id)] = temp_dict
async for lesson in await client.data.get_lessons(
date_from=date_from, date_to=date_to
):
temp_dict = {}
temp_dict["id"] = lesson.id
temp_dict["number"] = lesson.time.position
temp_dict["time"] = lesson.time
temp_dict["date"] = lesson.date.date
temp_dict["lesson"] = (
lesson.subject.name if lesson.subject is not None else None
)
if lesson.room is not None:
temp_dict["room"] = lesson.room.code
else:
temp_dict["room"] = "-"
temp_dict["visible"] = lesson.visible
temp_dict["changes"] = lesson.changes
temp_dict["group"] = lesson.group
temp_dict["reason"] = None
temp_dict["teacher"] = (
lesson.teacher.display_name if lesson.teacher is not None else None
)
temp_dict["from_to"] = (
lesson.time.displayed_time if lesson.time is not None else None
)
if temp_dict["changes"] is None:
temp_dict["changes"] = ""
elif temp_dict["changes"].type == 1:
temp_dict["lesson"] = f"Lekcja odwołana ({temp_dict['lesson']})"
temp_dict["changes_info"] = f"Lekcja odwołana ({temp_dict['lesson']})"
if str(temp_dict["changes"].id) in changes:
temp_dict["reason"] = changes[str(temp_dict["changes"].id)]["reason"]
elif temp_dict["changes"].type == 2:
temp_dict["lesson"] = f"{temp_dict['lesson']} (Zastępstwo)"
temp_dict["teacher"] = changes[str(temp_dict["changes"].id)]["teacher"]
if str(temp_dict["changes"].id) in changes:
temp_dict["teacher"] = changes[str(temp_dict["changes"].id)]["teacher"]
temp_dict["reason"] = changes[str(temp_dict["changes"].id)]["reason"]
elif temp_dict["changes"].type == 4:
temp_dict["lesson"] = f"Lekcja odwołana ({temp_dict['lesson']})"
if str(temp_dict["changes"].id) in changes:
temp_dict["reason"] = changes[str(temp_dict["changes"].id)]["reason"]
if temp_dict["visible"]:
list_ans.append(temp_dict)
return list_ans
async def get_student_info(client, student_id):
"""Support for fetching Student info by student id."""
student_info = {}
for student in await client.get_students():
if str(student.pupil.id) == str(student_id):
student_info["first_name"] = student.pupil.first_name
if student.pupil.second_name:
student_info["second_name"] = student.pupil.second_name
student_info["last_name"] = student.pupil.last_name
student_info["full_name"] = (
f"{student.pupil.first_name} {student.pupil.last_name}"
)
student_info["id"] = student.pupil.id
student_info["class"] = student.class_
student_info["school"] = student.school.name
student_info["symbol"] = student.symbol
break
return student_info

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

@ -1,9 +0,0 @@
{
"domain": "vulcan",
"name": "Uonet+ Vulcan",
"codeowners": ["@Antoni-Czaplicki"],
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/vulcan",
"iot_class": "cloud_polling",
"requirements": ["vulcan-api==2.4.2"]
}

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

@ -1,12 +0,0 @@
"""Support for register Vulcan account."""
from typing import Any
from vulcan import Account, Keystore
async def register(token: str, symbol: str, pin: str) -> dict[str, Any]:
"""Register integration and save credentials."""
keystore = await Keystore.create(device_model="Home Assistant")
account = await Account.register(keystore, token, symbol, pin)
return {"account": account, "keystore": keystore}

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

@ -1,62 +0,0 @@
{
"config": {
"abort": {
"already_configured": "That student has already been added.",
"all_student_already_configured": "All students have already been added.",
"reauth_successful": "Reauth successful",
"no_matching_entries": "No matching entries found, please use different account or remove outdated student integration."
},
"error": {
"unknown": "[%key:common::config_flow::error::unknown%]",
"invalid_token": "[%key:common::config_flow::error::invalid_access_token%]",
"expired_token": "Expired token - please generate a new token",
"invalid_pin": "Invalid PIN",
"invalid_symbol": "Invalid symbol",
"expired_credentials": "Expired credentials - please create new on Vulcan mobile app registration page",
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]"
},
"step": {
"auth": {
"description": "Log in to your Vulcan Account using mobile app registration page.",
"data": {
"token": "Token",
"region": "Symbol",
"pin": "[%key:common::config_flow::data::pin%]"
}
},
"reauth_confirm": {
"description": "[%key:component::vulcan::config::step::auth::description%]",
"data": {
"token": "Token",
"region": "[%key:component::vulcan::config::step::auth::data::region%]",
"pin": "[%key:common::config_flow::data::pin%]"
}
},
"select_student": {
"description": "Select student, you can add more students by adding integration again.",
"data": {
"student_name": "Select student"
}
},
"select_saved_credentials": {
"description": "Select saved credentials.",
"data": {
"credentials": "Login"
}
},
"add_next_config_entry": {
"description": "Add another student.",
"data": {
"use_saved_credentials": "Use saved credentials"
}
}
}
},
"entity": {
"calendar": {
"calendar": {
"name": "[%key:component::calendar::title%]"
}
}
}
}

1
homeassistant/generated/config_flows.py generated
View File

@ -709,7 +709,6 @@ FLOWS = {
"volumio",
"volvo",
"volvooncall",
"vulcan",
"wake_on_lan",
"wallbox",
"waqi",

6
homeassistant/generated/integrations.json
View File

@ -7312,12 +7312,6 @@
"config_flow": true,
"iot_class": "cloud_polling"
},
"vulcan": {
"name": "Uonet+ Vulcan",
"integration_type": "hub",
"config_flow": true,
"iot_class": "cloud_polling"
},
"vultr": {
"name": "Vultr",
"integration_type": "hub",

3
requirements_all.txt generated
View File

@ -3078,9 +3078,6 @@ vsure==2.6.7
# homeassistant.components.vasttrafik
vtjp==0.2.1
# homeassistant.components.vulcan
vulcan-api==2.4.2
# homeassistant.components.vultr
vultr==0.1.2

3
requirements_test_all.txt generated
View File

@ -2540,9 +2540,6 @@ volvooncall==0.10.3
# homeassistant.components.verisure
vsure==2.6.7
# homeassistant.components.vulcan
vulcan-api==2.4.2
# homeassistant.components.vultr
vultr==0.1.2

2
script/hassfest/quality_scale.py
View File

@ -1069,7 +1069,6 @@ INTEGRATIONS_WITHOUT_QUALITY_SCALE_FILE = [
"volkszaehler",
"volumio",
"volvooncall",
"vulcan",
"vultr",
"w800rf32",
"wake_on_lan",
@ -2124,7 +2123,6 @@ INTEGRATIONS_WITHOUT_SCALE = [
"volkszaehler",
"volumio",
"volvooncall",
"vulcan",
"vultr",
"w800rf32",
"wake_on_lan",

1
tests/components/analytics_insights/fixtures/current_data.json
View File

@ -799,7 +799,6 @@
"geofency": 313,
"hvv_departures": 70,
"devolo_home_control": 65,
"vulcan": 24,
"laundrify": 151,
"openhome": 730,
"rainmachine": 381,

1
tests/components/vulcan/__init__.py
View File

@ -1 +0,0 @@
"""Tests for the Uonet+ Vulcan integration."""

16
tests/components/vulcan/fixtures/fake_config_entry_data.json
View File

@ -1,16 +0,0 @@
{
"student_id": "123",
"keystore": {
"Certificate": "certificate",
"DeviceModel": "Home Assistant",
"Fingerprint": "fingerprint",
"FirebaseToken": "firebase_token",
"PrivateKey": "private_key"
},
"account": {
"LoginId": 0,
"RestURL": "",
"UserLogin": "example@example.com",
"UserName": "example@example.com"
}
}

35
tests/components/vulcan/fixtures/fake_student_1.json
View File

@ -1,35 +0,0 @@
{
"TopLevelPartition": "",
"Partition": "",
"ClassDisplay": "",
"Unit": {
"Id": 1,
"Symbol": "",
"Short": "",
"RestURL": "",
"Name": "",
"DisplayName": ""
},
"ConstituentUnit": {
"Id": 1,
"Short": "",
"Name": "",
"Address": ""
},
"Pupil": {
"Id": 0,
"LoginId": 0,
"LoginValue": "",
"FirstName": "Jan",
"SecondName": "Maciej",
"Surname": "Kowalski",
"Sex": true
},
"Periods": [],
"State": 0,
"MessageBox": {
"Id": 1,
"GlobalKey": "00000000-0000-0000-0000-000000000000",
"Name": "Test"
}
}

35
tests/components/vulcan/fixtures/fake_student_2.json
View File

@ -1,35 +0,0 @@
{
"TopLevelPartition": "",
"Partition": "",
"ClassDisplay": "",
"Unit": {
"Id": 1,
"Symbol": "",
"Short": "",
"RestURL": "",
"Name": "",
"DisplayName": ""
},
"ConstituentUnit": {
"Id": 1,
"Short": "",
"Name": "",
"Address": ""
},
"Pupil": {
"Id": 1,
"LoginId": 1,
"LoginValue": "",
"FirstName": "Magda",
"SecondName": "",
"Surname": "Kowalska",
"Sex": false
},
"Periods": [],
"State": 0,
"MessageBox": {
"Id": 1,
"GlobalKey": "00000000-0000-0000-0000-000000000000",
"Name": "Test"
}
}

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

@ -1,917 +0,0 @@
"""Test the Uonet+ Vulcan config flow."""
import json
from unittest import mock
from unittest.mock import patch
from vulcan import (
Account,
ExpiredTokenException,
InvalidPINException,
InvalidSymbolException,
InvalidTokenException,
UnauthorizedCertificateException,
)
from vulcan.model import Student
from homeassistant import config_entries
from homeassistant.components.vulcan import config_flow, register
from homeassistant.components.vulcan.config_flow import ClientConnectionError, Keystore
from homeassistant.components.vulcan.const import DOMAIN
from homeassistant.const import CONF_PIN, CONF_REGION, CONF_TOKEN
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from tests.common import MockConfigEntry, async_load_fixture
fake_keystore = Keystore("", "", "", "", "")
fake_account = Account(
login_id=1,
user_login="example@example.com",
user_name="example@example.com",
rest_url="rest_url",
)
async def test_show_form(hass: HomeAssistant) -> None:
"""Test that the form is served with no input."""
flow = config_flow.VulcanFlowHandler()
flow.hass = hass
result = await flow.async_step_user(user_input=None)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "auth"
@mock.patch("homeassistant.components.vulcan.config_flow.Vulcan.get_students")
@mock.patch("homeassistant.components.vulcan.config_flow.Account.register")
@mock.patch("homeassistant.components.vulcan.config_flow.Keystore.create")
async def test_config_flow_auth_success(
mock_keystore, mock_account, mock_student, hass: HomeAssistant
) -> None:
"""Test a successful config flow initialized by the user."""
mock_keystore.return_value = fake_keystore
mock_account.return_value = fake_account
mock_student.return_value = [
Student.load(await async_load_fixture(hass, "fake_student_1.json", DOMAIN))
]
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "auth"
assert result["errors"] is None
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"},
)
await hass.async_block_till_done()
assert result["type"] is FlowResultType.CREATE_ENTRY
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.Account.register")
@mock.patch("homeassistant.components.vulcan.config_flow.Keystore.create")
async def test_config_flow_auth_success_with_multiple_students(
mock_keystore, mock_account, mock_student, hass: HomeAssistant
) -> None:
"""Test a successful config flow with multiple students."""
mock_keystore.return_value = fake_keystore
mock_account.return_value = fake_account
mock_student.return_value = [
Student.load(student)
for student in (
await async_load_fixture(hass, "fake_student_1.json", DOMAIN),
await async_load_fixture(hass, "fake_student_2.json", DOMAIN),
)
]
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "auth"
assert result["errors"] is None
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{CONF_TOKEN: "token", CONF_REGION: "region", CONF_PIN: "000000"},
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "select_student"
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"],
{"student": "0"},
)
assert result["type"] is FlowResultType.CREATE_ENTRY
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.Keystore.create")
@mock.patch("homeassistant.components.vulcan.config_flow.Account.register")
async def test_config_flow_reauth_success(
mock_account, mock_keystore, mock_student, hass: HomeAssistant
) -> None:
"""Test a successful config flow reauth."""
mock_keystore.return_value = fake_keystore
mock_account.return_value = fake_account
mock_student.return_value = [
Student.load(await async_load_fixture(hass, "fake_student_1.json", DOMAIN))
]
entry = MockConfigEntry(
domain=DOMAIN,
unique_id="0",
data={"student_id": "0"},
)
entry.add_to_hass(hass)
result = await entry.start_reauth_flow(hass)
assert result["type"] is FlowResultType.FORM
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"] is FlowResultType.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: HomeAssistant
) -> None:
"""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(await async_load_fixture(hass, "fake_student_1.json", DOMAIN))
]
entry = MockConfigEntry(
domain=DOMAIN,
unique_id="0",
data={"student_id": "1"},
)
entry.add_to_hass(hass)
result = await entry.start_reauth_flow(hass)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "reauth_confirm"
assert result["errors"] == {}
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{CONF_TOKEN: "token", CONF_REGION: "region", CONF_PIN: "000000"},
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "no_matching_entries"
@mock.patch("homeassistant.components.vulcan.config_flow.Keystore.create")
@mock.patch("homeassistant.components.vulcan.config_flow.Account.register")
async def test_config_flow_reauth_with_errors(
mock_account, mock_keystore, hass: HomeAssistant
) -> None:
"""Test reauth config flow with errors."""
mock_keystore.return_value = fake_keystore
mock_account.return_value = fake_account
entry = MockConfigEntry(
domain=DOMAIN,
unique_id="0",
data={"student_id": "0"},
)
entry.add_to_hass(hass)
result = await entry.start_reauth_flow(hass)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "reauth_confirm"
assert result["errors"] == {}
with patch(
"homeassistant.components.vulcan.config_flow.Account.register",
side_effect=InvalidTokenException,
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{CONF_TOKEN: "token", CONF_REGION: "region", CONF_PIN: "000000"},
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "reauth_confirm"
assert result["errors"] == {"base": "invalid_token"}
with patch(
"homeassistant.components.vulcan.config_flow.Account.register",
side_effect=ExpiredTokenException,
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{CONF_TOKEN: "token", CONF_REGION: "region", CONF_PIN: "000000"},
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "reauth_confirm"
assert result["errors"] == {"base": "expired_token"}
with patch(
"homeassistant.components.vulcan.config_flow.Account.register",
side_effect=InvalidPINException,
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{CONF_TOKEN: "token", CONF_REGION: "region", CONF_PIN: "000000"},
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "reauth_confirm"
assert result["errors"] == {"base": "invalid_pin"}
with patch(
"homeassistant.components.vulcan.config_flow.Account.register",
side_effect=InvalidSymbolException,
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{CONF_TOKEN: "token", CONF_REGION: "region", CONF_PIN: "000000"},
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "reauth_confirm"
assert result["errors"] == {"base": "invalid_symbol"}
with patch(
"homeassistant.components.vulcan.config_flow.Account.register",
side_effect=ClientConnectionError,
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{CONF_TOKEN: "token", CONF_REGION: "region", CONF_PIN: "000000"},
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "reauth_confirm"
assert result["errors"] == {"base": "cannot_connect"}
with patch(
"homeassistant.components.vulcan.config_flow.Account.register",
side_effect=Exception,
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{CONF_TOKEN: "token", CONF_REGION: "region", CONF_PIN: "000000"},
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "reauth_confirm"
assert result["errors"] == {"base": "unknown"}
@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_multiple_config_entries(
mock_account, mock_keystore, mock_student, hass: HomeAssistant
) -> None:
"""Test a successful config flow for multiple config entries."""
mock_keystore.return_value = fake_keystore
mock_account.return_value = fake_account
mock_student.return_value = [
Student.load(await async_load_fixture(hass, "fake_student_1.json", DOMAIN))
]
MockConfigEntry(
domain=DOMAIN,
unique_id="123456",
data=json.loads(
await async_load_fixture(hass, "fake_config_entry_data.json", DOMAIN)
),
).add_to_hass(hass)
await register.register("token", "region", "000000")
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] is FlowResultType.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": False},
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "auth"
assert result["errors"] is None
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"] is FlowResultType.CREATE_ENTRY
assert result["title"] == "Jan Kowalski"
assert len(mock_setup_entry.mock_calls) == 2
@mock.patch("homeassistant.components.vulcan.config_flow.Vulcan.get_students")
async def test_multiple_config_entries_using_saved_credentials(
mock_student, hass: HomeAssistant
) -> None:
"""Test a successful config flow for multiple config entries using saved credentials."""
mock_student.return_value = [
Student.load(await async_load_fixture(hass, "fake_student_1.json", DOMAIN))
]
MockConfigEntry(
domain=DOMAIN,
unique_id="123456",
data=json.loads(
await async_load_fixture(hass, "fake_config_entry_data.json", DOMAIN)
),
).add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "add_next_config_entry"
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"],
{"use_saved_credentials": True},
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == "Jan Kowalski"
assert len(mock_setup_entry.mock_calls) == 2
@mock.patch("homeassistant.components.vulcan.config_flow.Vulcan.get_students")
async def test_multiple_config_entries_using_saved_credentials_2(
mock_student, hass: HomeAssistant
) -> None:
"""Test a successful config flow for multiple config entries using saved credentials (different situation)."""
mock_student.return_value = [
Student.load(await async_load_fixture(hass, "fake_student_1.json", DOMAIN)),
Student.load(await async_load_fixture(hass, "fake_student_2.json", DOMAIN)),
]
MockConfigEntry(
domain=DOMAIN,
unique_id="123456",
data=json.loads(
await async_load_fixture(hass, "fake_config_entry_data.json", DOMAIN)
),
).add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] is FlowResultType.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},
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "select_student"
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"],
{"student": "0"},
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == "Jan Kowalski"
assert len(mock_setup_entry.mock_calls) == 2
@mock.patch("homeassistant.components.vulcan.config_flow.Vulcan.get_students")
async def test_multiple_config_entries_using_saved_credentials_3(
mock_student, hass: HomeAssistant
) -> None:
"""Test a successful config flow for multiple config entries using saved credentials."""
mock_student.return_value = [
Student.load(await async_load_fixture(hass, "fake_student_1.json", DOMAIN))
]
MockConfigEntry(
entry_id="456",
domain=DOMAIN,
unique_id="234567",
data=json.loads(
await async_load_fixture(hass, "fake_config_entry_data.json", DOMAIN)
)
| {"student_id": "456"},
).add_to_hass(hass)
MockConfigEntry(
entry_id="123",
domain=DOMAIN,
unique_id="123456",
data=json.loads(
await async_load_fixture(hass, "fake_config_entry_data.json", DOMAIN)
),
).add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] is FlowResultType.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},
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "select_saved_credentials"
assert result["errors"] is None
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"],
{"credentials": "123"},
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == "Jan Kowalski"
assert len(mock_setup_entry.mock_calls) == 3
@mock.patch("homeassistant.components.vulcan.config_flow.Vulcan.get_students")
async def test_multiple_config_entries_using_saved_credentials_4(
mock_student, hass: HomeAssistant
) -> None:
"""Test a successful config flow for multiple config entries using saved credentials (different situation)."""
mock_student.return_value = [
Student.load(await async_load_fixture(hass, "fake_student_1.json", DOMAIN)),
Student.load(await async_load_fixture(hass, "fake_student_2.json", DOMAIN)),
]
MockConfigEntry(
entry_id="456",
domain=DOMAIN,
unique_id="234567",
data=json.loads(
await async_load_fixture(hass, "fake_config_entry_data.json", DOMAIN)
)
| {"student_id": "456"},
).add_to_hass(hass)
MockConfigEntry(
entry_id="123",
domain=DOMAIN,
unique_id="123456",
data=json.loads(
await async_load_fixture(hass, "fake_config_entry_data.json", DOMAIN)
),
).add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] is FlowResultType.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},
)
assert result["type"] is FlowResultType.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"] is FlowResultType.FORM
assert result["step_id"] == "select_student"
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"],
{"student": "0"},
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == "Jan Kowalski"
assert len(mock_setup_entry.mock_calls) == 3
async def test_multiple_config_entries_without_valid_saved_credentials(
hass: HomeAssistant,
) -> None:
"""Test a unsuccessful config flow for multiple config entries without valid saved credentials."""
MockConfigEntry(
entry_id="456",
domain=DOMAIN,
unique_id="234567",
data=json.loads(
await async_load_fixture(hass, "fake_config_entry_data.json", DOMAIN)
)
| {"student_id": "456"},
).add_to_hass(hass)
MockConfigEntry(
entry_id="123",
domain=DOMAIN,
unique_id="123456",
data=json.loads(
await async_load_fixture(hass, "fake_config_entry_data.json", DOMAIN)
),
).add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] is FlowResultType.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=UnauthorizedCertificateException,
):
assert result["type"] is FlowResultType.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"] is FlowResultType.FORM
assert result["step_id"] == "auth"
assert result["errors"] == {"base": "expired_credentials"}
async def test_multiple_config_entries_using_saved_credentials_with_connections_issues(
hass: HomeAssistant,
) -> None:
"""Test a unsuccessful config flow for multiple config entries without valid saved credentials."""
MockConfigEntry(
entry_id="456",
domain=DOMAIN,
unique_id="234567",
data=json.loads(
await async_load_fixture(hass, "fake_config_entry_data.json", DOMAIN)
)
| {"student_id": "456"},
).add_to_hass(hass)
MockConfigEntry(
entry_id="123",
domain=DOMAIN,
unique_id="123456",
data=json.loads(
await async_load_fixture(hass, "fake_config_entry_data.json", DOMAIN)
),
).add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] is FlowResultType.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=ClientConnectionError,
):
assert result["type"] is FlowResultType.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"] is FlowResultType.FORM
assert result["step_id"] == "select_saved_credentials"
assert result["errors"] == {"base": "cannot_connect"}
async def test_multiple_config_entries_using_saved_credentials_with_unknown_error(
hass: HomeAssistant,
) -> None:
"""Test a unsuccessful config flow for multiple config entries without valid saved credentials."""
MockConfigEntry(
entry_id="456",
domain=DOMAIN,
unique_id="234567",
data=json.loads(
await async_load_fixture(hass, "fake_config_entry_data.json", DOMAIN)
)
| {"student_id": "456"},
).add_to_hass(hass)
MockConfigEntry(
entry_id="123",
domain=DOMAIN,
unique_id="123456",
data=json.loads(
await async_load_fixture(hass, "fake_config_entry_data.json", DOMAIN)
),
).add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] is FlowResultType.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=Exception,
):
assert result["type"] is FlowResultType.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"] is FlowResultType.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.Keystore.create")
@mock.patch("homeassistant.components.vulcan.config_flow.Account.register")
async def test_student_already_exists(
mock_account, mock_keystore, mock_student, hass: HomeAssistant
) -> None:
"""Test config entry when student's entry already exists."""
mock_keystore.return_value = fake_keystore
mock_account.return_value = fake_account
mock_student.return_value = [
Student.load(await async_load_fixture(hass, "fake_student_1.json", DOMAIN))
]
MockConfigEntry(
domain=DOMAIN,
unique_id="0",
data=json.loads(
await async_load_fixture(hass, "fake_config_entry_data.json", DOMAIN)
)
| {"student_id": "0"},
).add_to_hass(hass)
await register.register("token", "region", "000000")
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] is FlowResultType.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},
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "all_student_already_configured"
@mock.patch("homeassistant.components.vulcan.config_flow.Keystore.create")
async def test_config_flow_auth_invalid_token(
mock_keystore, hass: HomeAssistant
) -> None:
"""Test a config flow initialized by the user using invalid token."""
mock_keystore.return_value = fake_keystore
with patch(
"homeassistant.components.vulcan.config_flow.Account.register",
side_effect=InvalidTokenException,
):
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "auth"
assert result["errors"] is None
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{CONF_TOKEN: "3S20000", CONF_REGION: "region", CONF_PIN: "000000"},
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "auth"
assert result["errors"] == {"base": "invalid_token"}
@mock.patch("homeassistant.components.vulcan.config_flow.Keystore.create")
async def test_config_flow_auth_invalid_region(
mock_keystore, hass: HomeAssistant
) -> None:
"""Test a config flow initialized by the user using invalid region."""
mock_keystore.return_value = fake_keystore
with patch(
"homeassistant.components.vulcan.config_flow.Account.register",
side_effect=InvalidSymbolException,
):
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] is FlowResultType.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: "invalid_region", CONF_PIN: "000000"},
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "auth"
assert result["errors"] == {"base": "invalid_symbol"}
@mock.patch("homeassistant.components.vulcan.config_flow.Keystore.create")
async def test_config_flow_auth_invalid_pin(mock_keystore, hass: HomeAssistant) -> None:
"""Test a config flow initialized by the with invalid pin."""
mock_keystore.return_value = fake_keystore
with patch(
"homeassistant.components.vulcan.config_flow.Account.register",
side_effect=InvalidPINException,
):
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] is FlowResultType.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"] is FlowResultType.FORM
assert result["step_id"] == "auth"
assert result["errors"] == {"base": "invalid_pin"}
@mock.patch("homeassistant.components.vulcan.config_flow.Keystore.create")
async def test_config_flow_auth_expired_token(
mock_keystore, hass: HomeAssistant
) -> None:
"""Test a config flow initialized by the with expired token."""
mock_keystore.return_value = fake_keystore
with patch(
"homeassistant.components.vulcan.config_flow.Account.register",
side_effect=ExpiredTokenException,
):
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] is FlowResultType.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"] is FlowResultType.FORM
assert result["step_id"] == "auth"
assert result["errors"] == {"base": "expired_token"}
@mock.patch("homeassistant.components.vulcan.config_flow.Keystore.create")
async def test_config_flow_auth_connection_error(
mock_keystore, hass: HomeAssistant
) -> None:
"""Test a config flow with connection error."""
mock_keystore.return_value = fake_keystore
with patch(
"homeassistant.components.vulcan.config_flow.Account.register",
side_effect=ClientConnectionError,
):
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] is FlowResultType.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"] is FlowResultType.FORM
assert result["step_id"] == "auth"
assert result["errors"] == {"base": "cannot_connect"}
@mock.patch("homeassistant.components.vulcan.config_flow.Keystore.create")
async def test_config_flow_auth_unknown_error(
mock_keystore, hass: HomeAssistant
) -> None:
"""Test a config flow with unknown error."""
mock_keystore.return_value = fake_keystore
with patch(
"homeassistant.components.vulcan.config_flow.Account.register",
side_effect=Exception,
):
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] is FlowResultType.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: "invalid_region", CONF_PIN: "000000"},
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "auth"
assert result["errors"] == {"base": "unknown"}