mirror of https://github.com/home-assistant/core
Add Prosegur Alarms (#44679)
Co-authored-by: Franck Nijhof <frenck@frenck.dev> Co-authored-by: Franck Nijhof <git@frenck.dev>
This commit is contained in:
Diogo Gomes
GitHub
parent
f92ba75791
commit
7ad7cdad3d
|
|
@ -390,6 +390,7 @@ homeassistant/components/powerwall/* @bdraco @jrester
|
|||
homeassistant/components/profiler/* @bdraco
|
||||
homeassistant/components/progettihwsw/* @ardaseremet
|
||||
homeassistant/components/prometheus/* @knyar
|
||||
homeassistant/components/prosegur/* @dgomes
|
||||
homeassistant/components/proxmoxve/* @k4ds3 @jhollowe @Corbeno
|
||||
homeassistant/components/ps4/* @ktnrg45
|
||||
homeassistant/components/push/* @dgomes
|
||||
|
|
|
|||
|
|
@ -0,0 +1,57 @@
|
|||
"""The Prosegur Alarm integration."""
|
||||
import logging
|
||||
|
||||
from pyprosegur.auth import Auth
|
||||
|
||||
from homeassistant.config_entries import SOURCE_REAUTH, ConfigEntry
|
||||
from homeassistant.const import CONF_PASSWORD, CONF_USERNAME
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.exceptions import ConfigEntryNotReady
|
||||
from homeassistant.helpers import aiohttp_client
|
||||
|
||||
from .const import CONF_COUNTRY, DOMAIN
|
||||
|
||||
PLATFORMS = ["alarm_control_panel"]
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
||||
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
|
||||
"""Set up Prosegur Alarm from a config entry."""
|
||||
try:
|
||||
session = aiohttp_client.async_get_clientsession(hass)
|
||||
hass.data.setdefault(DOMAIN, {})
|
||||
hass.data[DOMAIN][entry.entry_id] = Auth(
|
||||
session,
|
||||
entry.data[CONF_USERNAME],
|
||||
entry.data[CONF_PASSWORD],
|
||||
entry.data[CONF_COUNTRY],
|
||||
)
|
||||
await hass.data[DOMAIN][entry.entry_id].login()
|
||||
|
||||
except ConnectionRefusedError as error:
|
||||
_LOGGER.error("Configured credential are invalid, %s", error)
|
||||
|
||||
hass.async_create_task(
|
||||
hass.config_entries.flow.async_init(
|
||||
DOMAIN,
|
||||
context={"source": SOURCE_REAUTH, "entry_id": entry.data["entry_id"]},
|
||||
)
|
||||
)
|
||||
|
||||
except ConnectionError as error:
|
||||
_LOGGER.error("Could not connect with Prosegur backend: %s", error)
|
||||
raise ConfigEntryNotReady from error
|
||||
|
||||
hass.config_entries.async_setup_platforms(entry, PLATFORMS)
|
||||
|
||||
return True
|
||||
|
||||
|
||||
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
|
||||
"""Unload a config entry."""
|
||||
unload_ok = await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
|
||||
if unload_ok:
|
||||
hass.data[DOMAIN].pop(entry.entry_id)
|
||||
|
||||
return unload_ok
|
||||
|
|
@ -0,0 +1,76 @@
|
|||
"""Support for Prosegur alarm control panels."""
|
||||
import logging
|
||||
|
||||
from pyprosegur.auth import Auth
|
||||
from pyprosegur.installation import Installation, Status
|
||||
|
||||
import homeassistant.components.alarm_control_panel as alarm
|
||||
from homeassistant.components.alarm_control_panel import (
|
||||
SUPPORT_ALARM_ARM_AWAY,
|
||||
SUPPORT_ALARM_ARM_HOME,
|
||||
)
|
||||
from homeassistant.const import (
|
||||
STATE_ALARM_ARMED_AWAY,
|
||||
STATE_ALARM_ARMED_HOME,
|
||||
STATE_ALARM_DISARMED,
|
||||
)
|
||||
|
||||
from . import DOMAIN
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
STATE_MAPPING = {
|
||||
Status.DISARMED: STATE_ALARM_DISARMED,
|
||||
Status.ARMED: STATE_ALARM_ARMED_AWAY,
|
||||
Status.PARTIALLY: STATE_ALARM_ARMED_HOME,
|
||||
Status.ERROR_PARTIALLY: STATE_ALARM_ARMED_HOME,
|
||||
}
|
||||
|
||||
|
||||
async def async_setup_entry(hass, entry, async_add_entities):
|
||||
"""Set up the Prosegur alarm control panel platform."""
|
||||
async_add_entities(
|
||||
[ProsegurAlarm(entry.data["contract"], hass.data[DOMAIN][entry.entry_id])],
|
||||
update_before_add=True,
|
||||
)
|
||||
|
||||
|
||||
class ProsegurAlarm(alarm.AlarmControlPanelEntity):
|
||||
"""Representation of a Prosegur alarm status."""
|
||||
|
||||
def __init__(self, contract: str, auth: Auth) -> None:
|
||||
"""Initialize the Prosegur alarm panel."""
|
||||
self._changed_by = None
|
||||
|
||||
self._installation = None
|
||||
self.contract = contract
|
||||
self._auth = auth
|
||||
|
||||
self._attr_name = f"contract {self.contract}"
|
||||
self._attr_unique_id = self.contract
|
||||
self._attr_supported_features = SUPPORT_ALARM_ARM_AWAY | SUPPORT_ALARM_ARM_HOME
|
||||
|
||||
async def async_update(self):
|
||||
"""Update alarm status."""
|
||||
|
||||
try:
|
||||
self._installation = await Installation.retrieve(self._auth)
|
||||
except ConnectionError as err:
|
||||
_LOGGER.error(err)
|
||||
self._attr_available = False
|
||||
return
|
||||
|
||||
self._attr_state = STATE_MAPPING.get(self._installation.status)
|
||||
self._attr_available = True
|
||||
|
||||
async def async_alarm_disarm(self, code=None):
|
||||
"""Send disarm command."""
|
||||
await self._installation.disarm(self._auth)
|
||||
|
||||
async def async_alarm_arm_home(self, code=None):
|
||||
"""Send arm away command."""
|
||||
await self._installation.arm_partially(self._auth)
|
||||
|
||||
async def async_alarm_arm_away(self, code=None):
|
||||
"""Send arm away command."""
|
||||
await self._installation.arm(self._auth)
|
||||
|
|
@ -0,0 +1,135 @@
|
|||
"""Config flow for Prosegur Alarm integration."""
|
||||
import logging
|
||||
|
||||
from pyprosegur.auth import COUNTRY, Auth
|
||||
from pyprosegur.installation import Installation
|
||||
import voluptuous as vol
|
||||
|
||||
from homeassistant import config_entries, core, exceptions
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
from homeassistant.const import CONF_PASSWORD, CONF_USERNAME
|
||||
from homeassistant.helpers import aiohttp_client
|
||||
|
||||
from .const import CONF_COUNTRY, DOMAIN
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
STEP_USER_DATA_SCHEMA = vol.Schema(
|
||||
{
|
||||
vol.Required(CONF_USERNAME): str,
|
||||
vol.Required(CONF_PASSWORD): str,
|
||||
vol.Required(CONF_COUNTRY): vol.In(COUNTRY.keys()),
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
async def validate_input(hass: core.HomeAssistant, data):
|
||||
"""Validate the user input allows us to connect."""
|
||||
try:
|
||||
session = aiohttp_client.async_get_clientsession(hass)
|
||||
auth = Auth(
|
||||
session, data[CONF_USERNAME], data[CONF_PASSWORD], data[CONF_COUNTRY]
|
||||
)
|
||||
install = await Installation.retrieve(auth)
|
||||
except ConnectionRefusedError:
|
||||
raise InvalidAuth from ConnectionRefusedError
|
||||
except ConnectionError:
|
||||
raise CannotConnect from ConnectionError
|
||||
|
||||
# Info to store in the config entry.
|
||||
return {
|
||||
"title": f"Contract {install.contract}",
|
||||
"contract": install.contract,
|
||||
"username": data[CONF_USERNAME],
|
||||
"password": data[CONF_PASSWORD],
|
||||
"country": data[CONF_COUNTRY],
|
||||
}
|
||||
|
||||
|
||||
class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
|
||||
"""Handle a config flow for Prosegur Alarm."""
|
||||
|
||||
VERSION = 1
|
||||
entry: ConfigEntry
|
||||
|
||||
async def async_step_user(self, user_input=None):
|
||||
"""Handle the initial step."""
|
||||
errors = {}
|
||||
|
||||
if user_input:
|
||||
try:
|
||||
info = await validate_input(self.hass, user_input)
|
||||
except CannotConnect:
|
||||
errors["base"] = "cannot_connect"
|
||||
except InvalidAuth:
|
||||
errors["base"] = "invalid_auth"
|
||||
except Exception as exception: # pylint: disable=broad-except
|
||||
_LOGGER.exception(exception)
|
||||
errors["base"] = "unknown"
|
||||
else:
|
||||
await self.async_set_unique_id(info["contract"])
|
||||
self._abort_if_unique_id_configured()
|
||||
|
||||
user_input["contract"] = info["contract"]
|
||||
return self.async_create_entry(title=info["title"], data=user_input)
|
||||
|
||||
return self.async_show_form(
|
||||
step_id="user", data_schema=STEP_USER_DATA_SCHEMA, errors=errors
|
||||
)
|
||||
|
||||
async def async_step_reauth(self, data):
|
||||
"""Handle initiation of re-authentication with Prosegur."""
|
||||
self.entry = self.hass.config_entries.async_get_entry(self.context["entry_id"])
|
||||
return await self.async_step_reauth_confirm()
|
||||
|
||||
async def async_step_reauth_confirm(self, user_input=None):
|
||||
"""Handle re-authentication with Prosegur."""
|
||||
errors = {}
|
||||
|
||||
if user_input:
|
||||
try:
|
||||
user_input[CONF_COUNTRY] = self.entry.data[CONF_COUNTRY]
|
||||
await validate_input(self.hass, user_input)
|
||||
|
||||
except CannotConnect:
|
||||
errors["base"] = "cannot_connect"
|
||||
except InvalidAuth:
|
||||
errors["base"] = "invalid_auth"
|
||||
except Exception as exception: # pylint: disable=broad-except
|
||||
_LOGGER.exception(exception)
|
||||
errors["base"] = "unknown"
|
||||
else:
|
||||
data = self.entry.data.copy()
|
||||
self.hass.config_entries.async_update_entry(
|
||||
self.entry,
|
||||
data={
|
||||
**data,
|
||||
CONF_USERNAME: user_input[CONF_USERNAME],
|
||||
CONF_PASSWORD: user_input[CONF_PASSWORD],
|
||||
},
|
||||
)
|
||||
self.hass.async_create_task(
|
||||
self.hass.config_entries.async_reload(self.entry.entry_id)
|
||||
)
|
||||
return self.async_abort(reason="reauth_successful")
|
||||
|
||||
return self.async_show_form(
|
||||
step_id="reauth_confirm",
|
||||
data_schema=vol.Schema(
|
||||
{
|
||||
vol.Required(
|
||||
CONF_USERNAME, default=self.entry.data[CONF_USERNAME]
|
||||
): str,
|
||||
vol.Required(CONF_PASSWORD): str,
|
||||
}
|
||||
),
|
||||
errors=errors,
|
||||
)
|
||||
|
||||
|
||||
class CannotConnect(exceptions.HomeAssistantError):
|
||||
"""Error to indicate we cannot connect."""
|
||||
|
||||
|
||||
class InvalidAuth(exceptions.HomeAssistantError):
|
||||
"""Error to indicate there is invalid auth."""
|
||||
|
|
@ -0,0 +1,5 @@
|
|||
"""Constants for the Prosegur Alarm integration."""
|
||||
|
||||
DOMAIN = "prosegur"
|
||||
|
||||
CONF_COUNTRY = "country"
|
||||
|
|
@ -0,0 +1,13 @@
|
|||
{
|
||||
"domain": "prosegur",
|
||||
"name": "Prosegur Alarm",
|
||||
"config_flow": true,
|
||||
"documentation": "https://www.home-assistant.io/integrations/prosegur",
|
||||
"requirements": [
|
||||
"pyprosegur==0.0.5"
|
||||
],
|
||||
"codeowners": [
|
||||
"@dgomes"
|
||||
],
|
||||
"iot_class": "cloud_polling"
|
||||
}
|
||||
|
|
@ -0,0 +1,29 @@
|
|||
{
|
||||
"config": {
|
||||
"step": {
|
||||
"user": {
|
||||
"data": {
|
||||
"username": "[%key:common::config_flow::data::username%]",
|
||||
"password": "[%key:common::config_flow::data::password%]",
|
||||
"country": "Country"
|
||||
}
|
||||
},
|
||||
"reauth_confirm": {
|
||||
"data": {
|
||||
"description": "Re-authenticate with Prosegur account.",
|
||||
"username": "[%key:common::config_flow::data::username%]",
|
||||
"password": "[%key:common::config_flow::data::password%]"
|
||||
}
|
||||
}
|
||||
},
|
||||
"error": {
|
||||
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
|
||||
"invalid_auth": "[%key:common::config_flow::error::invalid_auth%]",
|
||||
"unknown": "[%key:common::config_flow::error::unknown%]"
|
||||
},
|
||||
"abort": {
|
||||
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]",
|
||||
"reauth_successful": "[%key:common::config_flow::abort::reauth_successful%]"
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,29 @@
|
|||
{
|
||||
"config": {
|
||||
"abort": {
|
||||
"already_configured": "Device is already configured",
|
||||
"reauth_successful": "Re-authentication was successful"
|
||||
},
|
||||
"error": {
|
||||
"cannot_connect": "Failed to connect",
|
||||
"invalid_auth": "Invalid authentication",
|
||||
"unknown": "Unexpected error"
|
||||
},
|
||||
"step": {
|
||||
"reauth_confirm": {
|
||||
"data": {
|
||||
"description": "Re-authenticate with Prosegur account.",
|
||||
"password": "Password",
|
||||
"username": "Username"
|
||||
}
|
||||
},
|
||||
"user": {
|
||||
"data": {
|
||||
"country": "Country",
|
||||
"password": "Password",
|
||||
"username": "Username"
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -210,6 +210,7 @@ FLOWS = [
|
|||
"powerwall",
|
||||
"profiler",
|
||||
"progettihwsw",
|
||||
"prosegur",
|
||||
"ps4",
|
||||
"pvpc_hourly_pricing",
|
||||
"rachio",
|
||||
|
|
|
|||
|
|
@ -1691,6 +1691,9 @@ pypoint==2.1.0
|
|||
# homeassistant.components.profiler
|
||||
pyprof2calltree==1.4.5
|
||||
|
||||
# homeassistant.components.prosegur
|
||||
pyprosegur==0.0.5
|
||||
|
||||
# homeassistant.components.ps4
|
||||
pyps4-2ndscreen==1.2.0
|
||||
|
||||
|
|
|
|||
|
|
@ -972,6 +972,9 @@ pypoint==2.1.0
|
|||
# homeassistant.components.profiler
|
||||
pyprof2calltree==1.4.5
|
||||
|
||||
# homeassistant.components.prosegur
|
||||
pyprosegur==0.0.5
|
||||
|
||||
# homeassistant.components.ps4
|
||||
pyps4-2ndscreen==1.2.0
|
||||
|
||||
|
|
|
|||
|
|
@ -0,0 +1 @@
|
|||
"""Tests for the Prosegur Alarm integration."""
|
||||
|
|
@ -0,0 +1,27 @@
|
|||
"""Common methods used across tests for Prosegur."""
|
||||
from homeassistant.components.prosegur import DOMAIN as PROSEGUR_DOMAIN
|
||||
from homeassistant.const import CONF_PASSWORD, CONF_USERNAME
|
||||
from homeassistant.setup import async_setup_component
|
||||
|
||||
from tests.common import MockConfigEntry
|
||||
|
||||
CONTRACT = "1234abcd"
|
||||
|
||||
|
||||
async def setup_platform(hass, platform):
|
||||
"""Set up the Prosegur platform."""
|
||||
mock_entry = MockConfigEntry(
|
||||
domain=PROSEGUR_DOMAIN,
|
||||
data={
|
||||
"contract": "1234abcd",
|
||||
CONF_USERNAME: "user@email.com",
|
||||
CONF_PASSWORD: "password",
|
||||
"country": "PT",
|
||||
},
|
||||
)
|
||||
mock_entry.add_to_hass(hass)
|
||||
|
||||
assert await async_setup_component(hass, PROSEGUR_DOMAIN, {})
|
||||
await hass.async_block_till_done()
|
||||
|
||||
return mock_entry
|
||||
|
|
@ -0,0 +1,120 @@
|
|||
"""Tests for the Prosegur alarm control panel device."""
|
||||
|
||||
from unittest.mock import AsyncMock, patch
|
||||
|
||||
from pyprosegur.installation import Status
|
||||
from pytest import fixture, mark
|
||||
|
||||
from homeassistant.components.alarm_control_panel import DOMAIN as ALARM_DOMAIN
|
||||
from homeassistant.const import (
|
||||
ATTR_ENTITY_ID,
|
||||
ATTR_FRIENDLY_NAME,
|
||||
ATTR_SUPPORTED_FEATURES,
|
||||
SERVICE_ALARM_ARM_AWAY,
|
||||
SERVICE_ALARM_ARM_HOME,
|
||||
SERVICE_ALARM_DISARM,
|
||||
STATE_ALARM_ARMED_AWAY,
|
||||
STATE_ALARM_ARMED_HOME,
|
||||
STATE_ALARM_DISARMED,
|
||||
STATE_UNAVAILABLE,
|
||||
)
|
||||
from homeassistant.helpers import entity_component
|
||||
|
||||
from .common import CONTRACT, setup_platform
|
||||
|
||||
PROSEGUR_ALARM_ENTITY = f"alarm_control_panel.contract_{CONTRACT}"
|
||||
|
||||
|
||||
@fixture
|
||||
def mock_auth():
|
||||
"""Setups authentication."""
|
||||
|
||||
with patch("pyprosegur.auth.Auth.login", return_value=True):
|
||||
yield
|
||||
|
||||
|
||||
@fixture(params=list(Status))
|
||||
def mock_status(request):
|
||||
"""Mock the status of the alarm."""
|
||||
|
||||
install = AsyncMock()
|
||||
install.contract = "123"
|
||||
install.installationId = "1234abcd"
|
||||
install.status = request.param
|
||||
|
||||
with patch("pyprosegur.installation.Installation.retrieve", return_value=install):
|
||||
yield
|
||||
|
||||
|
||||
async def test_entity_registry(hass, mock_auth, mock_status):
|
||||
"""Tests that the devices are registered in the entity registry."""
|
||||
await setup_platform(hass, ALARM_DOMAIN)
|
||||
entity_registry = await hass.helpers.entity_registry.async_get_registry()
|
||||
|
||||
entry = entity_registry.async_get(PROSEGUR_ALARM_ENTITY)
|
||||
# Prosegur alarm device unique_id is the contract id associated to the alarm account
|
||||
assert entry.unique_id == CONTRACT
|
||||
|
||||
await hass.async_block_till_done()
|
||||
|
||||
state = hass.states.get(PROSEGUR_ALARM_ENTITY)
|
||||
|
||||
assert state.attributes.get(ATTR_FRIENDLY_NAME) == "contract 1234abcd"
|
||||
assert state.attributes.get(ATTR_SUPPORTED_FEATURES) == 3
|
||||
|
||||
|
||||
async def test_connection_error(hass, mock_auth):
|
||||
"""Test the alarm control panel when connection can't be made to the cloud service."""
|
||||
|
||||
install = AsyncMock()
|
||||
install.arm = AsyncMock(return_value=False)
|
||||
install.arm_partially = AsyncMock(return_value=True)
|
||||
install.disarm = AsyncMock(return_value=True)
|
||||
install.status = Status.ARMED
|
||||
|
||||
with patch("pyprosegur.installation.Installation.retrieve", return_value=install):
|
||||
|
||||
await setup_platform(hass, ALARM_DOMAIN)
|
||||
|
||||
await hass.async_block_till_done()
|
||||
|
||||
with patch(
|
||||
"pyprosegur.installation.Installation.retrieve", side_effect=ConnectionError
|
||||
):
|
||||
|
||||
await entity_component.async_update_entity(hass, PROSEGUR_ALARM_ENTITY)
|
||||
|
||||
state = hass.states.get(PROSEGUR_ALARM_ENTITY)
|
||||
assert state.state == STATE_UNAVAILABLE
|
||||
|
||||
|
||||
@mark.parametrize(
|
||||
"code, alarm_service, alarm_state",
|
||||
[
|
||||
(Status.ARMED, SERVICE_ALARM_ARM_AWAY, STATE_ALARM_ARMED_AWAY),
|
||||
(Status.PARTIALLY, SERVICE_ALARM_ARM_HOME, STATE_ALARM_ARMED_HOME),
|
||||
(Status.DISARMED, SERVICE_ALARM_DISARM, STATE_ALARM_DISARMED),
|
||||
],
|
||||
)
|
||||
async def test_arm(hass, mock_auth, code, alarm_service, alarm_state):
|
||||
"""Test the alarm control panel can be set to away."""
|
||||
|
||||
install = AsyncMock()
|
||||
install.arm = AsyncMock(return_value=False)
|
||||
install.arm_partially = AsyncMock(return_value=True)
|
||||
install.disarm = AsyncMock(return_value=True)
|
||||
install.status = code
|
||||
|
||||
with patch("pyprosegur.installation.Installation.retrieve", return_value=install):
|
||||
await setup_platform(hass, ALARM_DOMAIN)
|
||||
|
||||
await hass.services.async_call(
|
||||
ALARM_DOMAIN,
|
||||
alarm_service,
|
||||
{ATTR_ENTITY_ID: PROSEGUR_ALARM_ENTITY},
|
||||
blocking=True,
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
state = hass.states.get(PROSEGUR_ALARM_ENTITY)
|
||||
assert state.state == alarm_state
|
||||
|
|
@ -0,0 +1,247 @@
|
|||
"""Test the Prosegur Alarm config flow."""
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
from pytest import mark
|
||||
|
||||
from homeassistant import config_entries, setup
|
||||
from homeassistant.components.prosegur.config_flow import CannotConnect, InvalidAuth
|
||||
from homeassistant.components.prosegur.const import DOMAIN
|
||||
from homeassistant.data_entry_flow import RESULT_TYPE_ABORT, RESULT_TYPE_FORM
|
||||
|
||||
from tests.common import MockConfigEntry
|
||||
|
||||
|
||||
async def test_form(hass):
|
||||
"""Test we get the form."""
|
||||
await setup.async_setup_component(hass, "persistent_notification", {})
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": config_entries.SOURCE_USER}
|
||||
)
|
||||
assert result["type"] == "form"
|
||||
assert result["errors"] == {}
|
||||
|
||||
install = MagicMock()
|
||||
install.contract = "123"
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.prosegur.config_flow.Installation.retrieve",
|
||||
return_value=install,
|
||||
), patch(
|
||||
"homeassistant.components.prosegur.async_setup_entry",
|
||||
return_value=True,
|
||||
) as mock_setup_entry:
|
||||
result2 = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
{
|
||||
"username": "test-username",
|
||||
"password": "test-password",
|
||||
"country": "PT",
|
||||
},
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert result2["type"] == "create_entry"
|
||||
assert result2["title"] == "Contract 123"
|
||||
assert result2["data"] == {
|
||||
"contract": "123",
|
||||
"username": "test-username",
|
||||
"password": "test-password",
|
||||
"country": "PT",
|
||||
}
|
||||
assert len(mock_setup_entry.mock_calls) == 1
|
||||
|
||||
|
||||
async def test_form_invalid_auth(hass):
|
||||
"""Test we handle invalid auth."""
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": config_entries.SOURCE_USER}
|
||||
)
|
||||
|
||||
with patch(
|
||||
"pyprosegur.auth.Auth",
|
||||
side_effect=ConnectionRefusedError,
|
||||
):
|
||||
result2 = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
{
|
||||
"username": "test-username",
|
||||
"password": "test-password",
|
||||
"country": "PT",
|
||||
},
|
||||
)
|
||||
|
||||
assert result2["type"] == "form"
|
||||
assert result2["errors"] == {"base": "invalid_auth"}
|
||||
|
||||
|
||||
async def test_form_cannot_connect(hass):
|
||||
"""Test we handle cannot connect error."""
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": config_entries.SOURCE_USER}
|
||||
)
|
||||
|
||||
with patch(
|
||||
"pyprosegur.installation.Installation",
|
||||
side_effect=ConnectionError,
|
||||
):
|
||||
result2 = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
{
|
||||
"username": "test-username",
|
||||
"password": "test-password",
|
||||
"country": "PT",
|
||||
},
|
||||
)
|
||||
|
||||
assert result2["type"] == "form"
|
||||
assert result2["errors"] == {"base": "cannot_connect"}
|
||||
|
||||
|
||||
async def test_form_unknown_exception(hass):
|
||||
"""Test we handle unknown exceptions."""
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": config_entries.SOURCE_USER}
|
||||
)
|
||||
|
||||
with patch(
|
||||
"pyprosegur.installation.Installation",
|
||||
side_effect=ValueError,
|
||||
):
|
||||
result2 = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
{
|
||||
"username": "test-username",
|
||||
"password": "test-password",
|
||||
"country": "PT",
|
||||
},
|
||||
)
|
||||
|
||||
assert result2["type"] == "form"
|
||||
assert result2["errors"] == {"base": "unknown"}
|
||||
|
||||
|
||||
async def test_form_validate_input(hass):
|
||||
"""Test we retrieve data from Installation."""
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": config_entries.SOURCE_USER}
|
||||
)
|
||||
|
||||
with patch(
|
||||
"pyprosegur.installation.Installation.retrieve",
|
||||
return_value=MagicMock,
|
||||
) as mock_retrieve:
|
||||
await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
{
|
||||
"username": "test-username",
|
||||
"password": "test-password",
|
||||
"country": "PT",
|
||||
},
|
||||
)
|
||||
|
||||
assert len(mock_retrieve.mock_calls) == 1
|
||||
|
||||
|
||||
async def test_reauth_flow(hass):
|
||||
"""Test a reauthentication flow."""
|
||||
entry = MockConfigEntry(
|
||||
domain=DOMAIN,
|
||||
unique_id="12345",
|
||||
data={
|
||||
"username": "test-username",
|
||||
"password": "test-password",
|
||||
"country": "PT",
|
||||
},
|
||||
)
|
||||
entry.add_to_hass(hass)
|
||||
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN,
|
||||
context={
|
||||
"source": config_entries.SOURCE_REAUTH,
|
||||
"unique_id": entry.unique_id,
|
||||
"entry_id": entry.entry_id,
|
||||
},
|
||||
data=entry.data,
|
||||
)
|
||||
assert result["step_id"] == "reauth_confirm"
|
||||
assert result["type"] == RESULT_TYPE_FORM
|
||||
assert result["errors"] == {}
|
||||
|
||||
install = MagicMock()
|
||||
install.contract = "123"
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.prosegur.config_flow.Installation.retrieve",
|
||||
return_value=install,
|
||||
) as mock_installation, patch(
|
||||
"homeassistant.components.prosegur.async_setup_entry",
|
||||
return_value=True,
|
||||
) as mock_setup_entry:
|
||||
result2 = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
{
|
||||
"username": "test-username",
|
||||
"password": "new_password",
|
||||
},
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert result2["type"] == RESULT_TYPE_ABORT
|
||||
assert result2["reason"] == "reauth_successful"
|
||||
assert entry.data == {
|
||||
"country": "PT",
|
||||
"username": "test-username",
|
||||
"password": "new_password",
|
||||
}
|
||||
|
||||
assert len(mock_installation.mock_calls) == 1
|
||||
assert len(mock_setup_entry.mock_calls) == 1
|
||||
|
||||
|
||||
@mark.parametrize(
|
||||
"exception, base_error",
|
||||
[
|
||||
(CannotConnect, "cannot_connect"),
|
||||
(InvalidAuth, "invalid_auth"),
|
||||
(Exception, "unknown"),
|
||||
],
|
||||
)
|
||||
async def test_reauth_flow_error(hass, exception, base_error):
|
||||
"""Test a reauthentication flow with errors."""
|
||||
entry = MockConfigEntry(
|
||||
domain=DOMAIN,
|
||||
unique_id="12345",
|
||||
data={
|
||||
"username": "test-username",
|
||||
"password": "test-password",
|
||||
"country": "PT",
|
||||
},
|
||||
)
|
||||
entry.add_to_hass(hass)
|
||||
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN,
|
||||
context={
|
||||
"source": config_entries.SOURCE_REAUTH,
|
||||
"unique_id": entry.unique_id,
|
||||
"entry_id": entry.entry_id,
|
||||
},
|
||||
data=entry.data,
|
||||
)
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.prosegur.config_flow.Installation.retrieve",
|
||||
side_effect=exception,
|
||||
):
|
||||
result2 = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
{
|
||||
"username": "test-username",
|
||||
"password": "new_password",
|
||||
},
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert result2["type"] == RESULT_TYPE_FORM
|
||||
assert result2["errors"]["base"] == base_error
|
||||
|
|
@ -0,0 +1,74 @@
|
|||
"""Tests prosegur setup."""
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
from pytest import mark
|
||||
|
||||
from homeassistant.components.prosegur import DOMAIN
|
||||
|
||||
from tests.common import MockConfigEntry
|
||||
|
||||
|
||||
@mark.parametrize(
|
||||
"error",
|
||||
[
|
||||
ConnectionRefusedError,
|
||||
ConnectionError,
|
||||
],
|
||||
)
|
||||
async def test_setup_entry_fail_retrieve(hass, error):
|
||||
"""Test loading the Prosegur entry."""
|
||||
|
||||
hass.config.components.add(DOMAIN)
|
||||
config_entry = MockConfigEntry(
|
||||
domain=DOMAIN,
|
||||
data={
|
||||
"username": "test-username",
|
||||
"password": "test-password",
|
||||
"country": "PT",
|
||||
"contract": "xpto",
|
||||
},
|
||||
)
|
||||
config_entry.add_to_hass(hass)
|
||||
|
||||
with patch(
|
||||
"pyprosegur.auth.Auth.login",
|
||||
side_effect=error,
|
||||
):
|
||||
assert not await hass.config_entries.async_setup(config_entry.entry_id)
|
||||
|
||||
await hass.async_block_till_done()
|
||||
|
||||
|
||||
async def test_unload_entry(hass, aioclient_mock):
|
||||
"""Test unloading the Prosegur entry."""
|
||||
|
||||
aioclient_mock.post(
|
||||
"https://smart.prosegur.com/smart-server/ws/access/login",
|
||||
json={"data": {"token": "123456789"}},
|
||||
)
|
||||
|
||||
hass.config.components.add(DOMAIN)
|
||||
config_entry = MockConfigEntry(
|
||||
domain=DOMAIN,
|
||||
data={
|
||||
"username": "test-username",
|
||||
"password": "test-password",
|
||||
"country": "PT",
|
||||
"contract": "xpto",
|
||||
},
|
||||
)
|
||||
config_entry.add_to_hass(hass)
|
||||
|
||||
install = MagicMock()
|
||||
install.contract = "123"
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.prosegur.config_flow.Installation.retrieve",
|
||||
return_value=install,
|
||||
):
|
||||
|
||||
assert await hass.config_entries.async_setup(config_entry.entry_id)
|
||||
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert await hass.config_entries.async_unload(config_entry.entry_id)
|
||||
Loading…
Reference in New Issue