Add multi-partition support for TotalConnect (#55429)

This commit is contained in:
Austin Mroczek 2021-10-27 10:15:13 -07:00 committed by GitHub
parent 6cd83d1f66
commit baaaf3d2bc
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
11 changed files with 463 additions and 256 deletions

View file

@ -1,17 +1,25 @@
"""The totalconnect component.""" """The totalconnect component."""
from total_connect_client import TotalConnectClient
from datetime import timedelta
import logging
from total_connect_client.client import TotalConnectClient
from total_connect_client.exceptions import AuthenticationError, TotalConnectError
from homeassistant.config_entries import ConfigEntry from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_PASSWORD, CONF_USERNAME from homeassistant.const import CONF_PASSWORD, CONF_USERNAME
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryAuthFailed from homeassistant.exceptions import ConfigEntryAuthFailed
import homeassistant.helpers.config_validation as cv import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from .const import CONF_USERCODES, DOMAIN from .const import CONF_USERCODES, DOMAIN
PLATFORMS = ["alarm_control_panel", "binary_sensor"] PLATFORMS = ["alarm_control_panel", "binary_sensor"]
CONFIG_SCHEMA = cv.deprecated(DOMAIN) CONFIG_SCHEMA = cv.deprecated(DOMAIN)
SCAN_INTERVAL = timedelta(seconds=30)
_LOGGER = logging.getLogger(__name__)
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
@ -27,17 +35,18 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
temp_codes = conf[CONF_USERCODES] temp_codes = conf[CONF_USERCODES]
usercodes = {int(code): temp_codes[code] for code in temp_codes} usercodes = {int(code): temp_codes[code] for code in temp_codes}
client = await hass.async_add_executor_job( client = await hass.async_add_executor_job(
TotalConnectClient.TotalConnectClient, username, password, usercodes TotalConnectClient, username, password, usercodes
) )
if not client.is_valid_credentials(): if not client.is_valid_credentials():
raise ConfigEntryAuthFailed("TotalConnect authentication failed") raise ConfigEntryAuthFailed("TotalConnect authentication failed")
coordinator = TotalConnectDataUpdateCoordinator(hass, client)
await coordinator.async_config_entry_first_refresh()
hass.data.setdefault(DOMAIN, {}) hass.data.setdefault(DOMAIN, {})
hass.data[DOMAIN][entry.entry_id] = client hass.data[DOMAIN][entry.entry_id] = coordinator
hass.config_entries.async_setup_platforms(entry, PLATFORMS) hass.config_entries.async_setup_platforms(entry, PLATFORMS)
return True return True
@ -48,3 +57,36 @@ async def async_unload_entry(hass, entry: ConfigEntry):
hass.data[DOMAIN].pop(entry.entry_id) hass.data[DOMAIN].pop(entry.entry_id)
return unload_ok return unload_ok
class TotalConnectDataUpdateCoordinator(DataUpdateCoordinator):
"""Class to fetch data from TotalConnect."""
def __init__(self, hass: HomeAssistant, client):
"""Initialize."""
self.hass = hass
self.client = client
super().__init__(
hass, logger=_LOGGER, name=DOMAIN, update_interval=SCAN_INTERVAL
)
async def _async_update_data(self):
"""Update data."""
await self.hass.async_add_executor_job(self.sync_update_data)
def sync_update_data(self):
"""Fetch synchronous data from TotalConnect."""
try:
for location_id in self.client.locations:
self.client.locations[location_id].get_panel_meta_data()
except AuthenticationError as exception:
# should only encounter if password changes during operation
raise ConfigEntryAuthFailed(
"TotalConnect authentication failed"
) from exception
except TotalConnectError as exception:
raise UpdateFailed(exception) from exception
except ValueError as exception:
raise UpdateFailed("Unknown state from TotalConnect") from exception
return True

View file

@ -1,6 +1,9 @@
"""Interfaces with TotalConnect alarm control panels.""" """Interfaces with TotalConnect alarm control panels."""
import logging import logging
from total_connect_client import ArmingHelper
from total_connect_client.exceptions import BadResultCodeError
import homeassistant.components.alarm_control_panel as alarm import homeassistant.components.alarm_control_panel as alarm
from homeassistant.components.alarm_control_panel.const import ( from homeassistant.components.alarm_control_panel.const import (
SUPPORT_ALARM_ARM_AWAY, SUPPORT_ALARM_ARM_AWAY,
@ -18,35 +21,60 @@ from homeassistant.const import (
STATE_ALARM_TRIGGERED, STATE_ALARM_TRIGGERED,
) )
from homeassistant.exceptions import HomeAssistantError from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .const import DOMAIN from .const import DOMAIN
_LOGGER = logging.getLogger(__name__)
async def async_setup_entry(hass, entry, async_add_entities) -> None: async def async_setup_entry(hass, entry, async_add_entities) -> None:
"""Set up TotalConnect alarm panels based on a config entry.""" """Set up TotalConnect alarm panels based on a config entry."""
alarms = [] alarms = []
client = hass.data[DOMAIN][entry.entry_id] coordinator = hass.data[DOMAIN][entry.entry_id]
for location_id, location in client.locations.items(): for location_id, location in coordinator.client.locations.items():
location_name = location.location_name location_name = location.location_name
alarms.append(TotalConnectAlarm(location_name, location_id, client)) for partition_id in location.partitions:
alarms.append(
TotalConnectAlarm(
coordinator=coordinator,
name=location_name,
location_id=location_id,
partition_id=partition_id,
)
)
async_add_entities(alarms, True) async_add_entities(alarms, True)
class TotalConnectAlarm(alarm.AlarmControlPanelEntity): class TotalConnectAlarm(CoordinatorEntity, alarm.AlarmControlPanelEntity):
"""Represent an TotalConnect status.""" """Represent an TotalConnect status."""
def __init__(self, name, location_id, client): def __init__(self, coordinator, name, location_id, partition_id):
"""Initialize the TotalConnect status.""" """Initialize the TotalConnect status."""
self._name = name super().__init__(coordinator)
self._location_id = location_id self._location_id = location_id
self._unique_id = str(location_id) self._location = coordinator.client.locations[location_id]
self._client = client self._partition_id = partition_id
self._partition = self._location.partitions[partition_id]
self._device = self._location.devices[self._location.security_device_id]
self._state = None self._state = None
self._extra_state_attributes = {} self._extra_state_attributes = {}
"""
Set unique_id to location_id for partition 1 to avoid breaking change
for most users with new support for partitions.
Add _# for partition 2 and beyond.
"""
if partition_id == 1:
self._name = name
self._unique_id = f"{location_id}"
else:
self._name = f"{name} partition {partition_id}"
self._unique_id = f"{location_id}_{partition_id}"
@property @property
def name(self): def name(self):
"""Return the name of the device.""" """Return the name of the device."""
@ -57,9 +85,55 @@ class TotalConnectAlarm(alarm.AlarmControlPanelEntity):
"""Return the unique id.""" """Return the unique id."""
return self._unique_id return self._unique_id
@property
def device_info(self):
"""Return device info."""
return {
"identifiers": {(DOMAIN, self._device.serial_number)},
"name": self._device.name,
}
@property @property
def state(self): def state(self):
"""Return the state of the device.""" """Return the state of the device."""
attr = {
"location_name": self._name,
"location_id": self._location_id,
"partition": self._partition_id,
"ac_loss": self._location.ac_loss,
"low_battery": self._location.low_battery,
"cover_tampered": self._location.is_cover_tampered(),
"triggered_source": None,
"triggered_zone": None,
}
if self._partition.arming_state.is_disarmed():
state = STATE_ALARM_DISARMED
elif self._partition.arming_state.is_armed_night():
state = STATE_ALARM_ARMED_NIGHT
elif self._partition.arming_state.is_armed_home():
state = STATE_ALARM_ARMED_HOME
elif self._partition.arming_state.is_armed_away():
state = STATE_ALARM_ARMED_AWAY
elif self._partition.arming_state.is_armed_custom_bypass():
state = STATE_ALARM_ARMED_CUSTOM_BYPASS
elif self._partition.arming_state.is_arming():
state = STATE_ALARM_ARMING
elif self._partition.arming_state.is_disarming():
state = STATE_ALARM_DISARMING
elif self._partition.arming_state.is_triggered_police():
state = STATE_ALARM_TRIGGERED
attr["triggered_source"] = "Police/Medical"
elif self._partition.arming_state.is_triggered_fire():
state = STATE_ALARM_TRIGGERED
attr["triggered_source"] = "Fire/Smoke"
elif self._partition.arming_state.is_triggered_gas():
state = STATE_ALARM_TRIGGERED
attr["triggered_source"] = "Carbon Monoxide"
self._state = state
self._extra_state_attributes = attr
return self._state return self._state
@property @property
@ -72,67 +146,58 @@ class TotalConnectAlarm(alarm.AlarmControlPanelEntity):
"""Return the state attributes of the device.""" """Return the state attributes of the device."""
return self._extra_state_attributes return self._extra_state_attributes
def update(self): async def async_alarm_disarm(self, code=None):
"""Return the state of the device."""
self._client.get_armed_status(self._location_id)
attr = {
"location_name": self._name,
"location_id": self._location_id,
"ac_loss": self._client.locations[self._location_id].ac_loss,
"low_battery": self._client.locations[self._location_id].low_battery,
"cover_tampered": self._client.locations[
self._location_id
].is_cover_tampered(),
"triggered_source": None,
"triggered_zone": None,
}
if self._client.locations[self._location_id].is_disarmed():
state = STATE_ALARM_DISARMED
elif self._client.locations[self._location_id].is_armed_night():
state = STATE_ALARM_ARMED_NIGHT
elif self._client.locations[self._location_id].is_armed_home():
state = STATE_ALARM_ARMED_HOME
elif self._client.locations[self._location_id].is_armed_away():
state = STATE_ALARM_ARMED_AWAY
elif self._client.locations[self._location_id].is_armed_custom_bypass():
state = STATE_ALARM_ARMED_CUSTOM_BYPASS
elif self._client.locations[self._location_id].is_arming():
state = STATE_ALARM_ARMING
elif self._client.locations[self._location_id].is_disarming():
state = STATE_ALARM_DISARMING
elif self._client.locations[self._location_id].is_triggered_police():
state = STATE_ALARM_TRIGGERED
attr["triggered_source"] = "Police/Medical"
elif self._client.locations[self._location_id].is_triggered_fire():
state = STATE_ALARM_TRIGGERED
attr["triggered_source"] = "Fire/Smoke"
elif self._client.locations[self._location_id].is_triggered_gas():
state = STATE_ALARM_TRIGGERED
attr["triggered_source"] = "Carbon Monoxide"
else:
logging.info("Total Connect Client returned unknown status")
state = None
self._state = state
self._extra_state_attributes = attr
def alarm_disarm(self, code=None):
"""Send disarm command.""" """Send disarm command."""
if self._client.disarm(self._location_id) is not True: await self.hass.async_add_executor_job(self._disarm)
raise HomeAssistantError(f"TotalConnect failed to disarm {self._name}.") await self.coordinator.async_request_refresh()
def alarm_arm_home(self, code=None): def _disarm(self, code=None):
"""Disarm synchronous."""
try:
ArmingHelper(self._partition).disarm()
except BadResultCodeError as error:
raise HomeAssistantError(
f"TotalConnect failed to disarm {self._name}."
) from error
async def async_alarm_arm_home(self, code=None):
"""Send arm home command.""" """Send arm home command."""
if self._client.arm_stay(self._location_id) is not True: await self.hass.async_add_executor_job(self._arm_home)
raise HomeAssistantError(f"TotalConnect failed to arm home {self._name}.") await self.coordinator.async_request_refresh()
def alarm_arm_away(self, code=None): def _arm_home(self):
"""Arm home synchronous."""
try:
ArmingHelper(self._partition).arm_stay()
except BadResultCodeError as error:
raise HomeAssistantError(
f"TotalConnect failed to arm home {self._name}."
) from error
async def async_alarm_arm_away(self, code=None):
"""Send arm away command.""" """Send arm away command."""
if self._client.arm_away(self._location_id) is not True: await self.hass.async_add_executor_job(self._arm_away)
raise HomeAssistantError(f"TotalConnect failed to arm away {self._name}.") await self.coordinator.async_request_refresh()
def alarm_arm_night(self, code=None): def _arm_away(self, code=None):
"""Arm away synchronous."""
try:
ArmingHelper(self._partition).arm_away()
except BadResultCodeError as error:
raise HomeAssistantError(
f"TotalConnect failed to arm away {self._name}."
) from error
async def async_alarm_arm_night(self, code=None):
"""Send arm night command.""" """Send arm night command."""
if self._client.arm_stay_night(self._location_id) is not True: await self.hass.async_add_executor_job(self._arm_night)
raise HomeAssistantError(f"TotalConnect failed to arm night {self._name}.") await self.coordinator.async_request_refresh()
def _arm_night(self, code=None):
"""Arm night synchronous."""
try:
ArmingHelper(self._partition).arm_stay_night()
except BadResultCodeError as error:
raise HomeAssistantError(
f"TotalConnect failed to arm night {self._name}."
) from error

View file

@ -2,6 +2,8 @@
from homeassistant.components.binary_sensor import ( from homeassistant.components.binary_sensor import (
DEVICE_CLASS_DOOR, DEVICE_CLASS_DOOR,
DEVICE_CLASS_GAS, DEVICE_CLASS_GAS,
DEVICE_CLASS_MOTION,
DEVICE_CLASS_SAFETY,
DEVICE_CLASS_SMOKE, DEVICE_CLASS_SMOKE,
BinarySensorEntity, BinarySensorEntity,
) )
@ -13,7 +15,7 @@ async def async_setup_entry(hass, entry, async_add_entities) -> None:
"""Set up TotalConnect device sensors based on a config entry.""" """Set up TotalConnect device sensors based on a config entry."""
sensors = [] sensors = []
client_locations = hass.data[DOMAIN][entry.entry_id].locations client_locations = hass.data[DOMAIN][entry.entry_id].client.locations
for location_id, location in client_locations.items(): for location_id, location in client_locations.items():
for zone_id, zone in location.zones.items(): for zone_id, zone in location.zones.items():
@ -70,6 +72,10 @@ class TotalConnectBinarySensor(BinarySensorEntity):
return DEVICE_CLASS_SMOKE return DEVICE_CLASS_SMOKE
if self._zone.is_type_carbon_monoxide(): if self._zone.is_type_carbon_monoxide():
return DEVICE_CLASS_GAS return DEVICE_CLASS_GAS
if self._zone.is_type_motion():
return DEVICE_CLASS_MOTION
if self._zone.is_type_medical():
return DEVICE_CLASS_SAFETY
return None return None
@property @property
@ -80,5 +86,6 @@ class TotalConnectBinarySensor(BinarySensorEntity):
"location_id": self._location_id, "location_id": self._location_id,
"low_battery": self._is_low_battery, "low_battery": self._is_low_battery,
"tampered": self._is_tampered, "tampered": self._is_tampered,
"partition": self._zone.partition,
} }
return attributes return attributes

View file

@ -1,5 +1,5 @@
"""Config flow for the Total Connect component.""" """Config flow for the Total Connect component."""
from total_connect_client import TotalConnectClient from total_connect_client.client import TotalConnectClient
import voluptuous as vol import voluptuous as vol
from homeassistant import config_entries from homeassistant import config_entries
@ -37,7 +37,7 @@ class TotalConnectConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
self._abort_if_unique_id_configured() self._abort_if_unique_id_configured()
client = await self.hass.async_add_executor_job( client = await self.hass.async_add_executor_job(
TotalConnectClient.TotalConnectClient, username, password, None TotalConnectClient, username, password, None
) )
if client.is_valid_credentials(): if client.is_valid_credentials():
@ -130,7 +130,7 @@ class TotalConnectConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
) )
client = await self.hass.async_add_executor_job( client = await self.hass.async_add_executor_job(
TotalConnectClient.TotalConnectClient, TotalConnectClient,
self.username, self.username,
user_input[CONF_PASSWORD], user_input[CONF_PASSWORD],
self.usercodes, self.usercodes,

View file

@ -2,7 +2,7 @@
"domain": "totalconnect", "domain": "totalconnect",
"name": "Total Connect", "name": "Total Connect",
"documentation": "https://www.home-assistant.io/integrations/totalconnect", "documentation": "https://www.home-assistant.io/integrations/totalconnect",
"requirements": ["total_connect_client==0.57"], "requirements": ["total_connect_client==2021.8.3"],
"dependencies": [], "dependencies": [],
"codeowners": ["@austinmroczek"], "codeowners": ["@austinmroczek"],
"config_flow": true, "config_flow": true,

View file

@ -2314,7 +2314,7 @@ todoist-python==8.0.0
toonapi==0.2.1 toonapi==0.2.1
# homeassistant.components.totalconnect # homeassistant.components.totalconnect
total_connect_client==0.57 total_connect_client==2021.8.3
# homeassistant.components.tplink_lte # homeassistant.components.tplink_lte
tp-connected==0.0.4 tp-connected==0.0.4

View file

@ -1330,7 +1330,7 @@ tesla-powerwall==0.3.12
toonapi==0.2.1 toonapi==0.2.1
# homeassistant.components.totalconnect # homeassistant.components.totalconnect
total_connect_client==0.57 total_connect_client==2021.8.3
# homeassistant.components.transmission # homeassistant.components.transmission
transmissionrpc==0.11 transmissionrpc==0.11

View file

@ -1,7 +1,9 @@
"""Common methods used across tests for TotalConnect.""" """Common methods used across tests for TotalConnect."""
from unittest.mock import patch from unittest.mock import patch
from total_connect_client import TotalConnectClient from total_connect_client.client import TotalConnectClient
from total_connect_client.const import ArmingState
from total_connect_client.zone import ZoneStatus, ZoneType
from homeassistant.components.totalconnect.const import CONF_USERCODES, DOMAIN from homeassistant.components.totalconnect.const import CONF_USERCODES, DOMAIN
from homeassistant.const import CONF_PASSWORD, CONF_USERNAME from homeassistant.const import CONF_PASSWORD, CONF_USERNAME
@ -11,13 +13,24 @@ from tests.common import MockConfigEntry
LOCATION_ID = "123456" LOCATION_ID = "123456"
DEVICE_INFO_BASIC_1 = {
"DeviceID": "987654",
"DeviceName": "test",
"DeviceClassID": 1,
"DeviceSerialNumber": "987654321ABC",
"DeviceFlags": "PromptForUserCode=0,PromptForInstallerCode=0,PromptForImportSecuritySettings=0,AllowUserSlotEditing=0,CalCapable=1,CanBeSentToPanel=0,CanArmNightStay=0,CanSupportMultiPartition=0,PartitionCount=0,MaxPartitionCount=0,OnBoardingSupport=0,PartitionAdded=0,DuplicateUserSyncStatus=0,PanelType=8,PanelVariant=1,BLEDisarmCapable=0,ArmHomeSupported=0,DuplicateUserCodeCheck=1,CanSupportRapid=0,IsKeypadSupported=1,WifiEnrollmentSupported=0,IsConnectedPanel=0,ArmNightInSceneSupported=0,BuiltInCameraSettingsSupported=0,ZWaveThermostatScheduleDisabled=0,MultipleAuthorityLevelSupported=0,VideoOnPanelSupported=0,EnableBLEMode=0,IsPanelWiFiResetSupported=0,IsCompetitorClearBypass=0,IsNotReadyStateSupported=0,isArmStatusWithoutExitDelayNotSupported=0",
"SecurityPanelTypeID": None,
"DeviceSerialText": None,
}
DEVICE_LIST = [DEVICE_INFO_BASIC_1]
LOCATION_INFO_BASIC_NORMAL = { LOCATION_INFO_BASIC_NORMAL = {
"LocationID": LOCATION_ID, "LocationID": LOCATION_ID,
"LocationName": "test", "LocationName": "test",
"SecurityDeviceID": "987654", "SecurityDeviceID": "987654",
"PhotoURL": "http://www.example.com/some/path/to/file.jpg", "PhotoURL": "http://www.example.com/some/path/to/file.jpg",
"LocationModuleFlags": "Security=1,Video=0,Automation=0,GPS=0,VideoPIR=0", "LocationModuleFlags": "Security=1,Video=0,Automation=0,GPS=0,VideoPIR=0",
"DeviceList": None, "DeviceList": {"DeviceInfoBasic": DEVICE_LIST},
} }
LOCATIONS = {"LocationInfoBasic": [LOCATION_INFO_BASIC_NORMAL]} LOCATIONS = {"LocationInfoBasic": [LOCATION_INFO_BASIC_NORMAL]}
@ -31,7 +44,7 @@ USER = {
} }
RESPONSE_AUTHENTICATE = { RESPONSE_AUTHENTICATE = {
"ResultCode": TotalConnectClient.TotalConnectClient.SUCCESS, "ResultCode": TotalConnectClient.SUCCESS,
"SessionID": 1, "SessionID": 1,
"Locations": LOCATIONS, "Locations": LOCATIONS,
"ModuleFlags": MODULE_FLAGS, "ModuleFlags": MODULE_FLAGS,
@ -39,58 +52,68 @@ RESPONSE_AUTHENTICATE = {
} }
RESPONSE_AUTHENTICATE_FAILED = { RESPONSE_AUTHENTICATE_FAILED = {
"ResultCode": TotalConnectClient.TotalConnectClient.BAD_USER_OR_PASSWORD, "ResultCode": TotalConnectClient.BAD_USER_OR_PASSWORD,
"ResultData": "test bad authentication", "ResultData": "test bad authentication",
} }
PARTITION_DISARMED = { PARTITION_DISARMED = {
"PartitionID": "1", "PartitionID": "1",
"ArmingState": TotalConnectClient.TotalConnectLocation.DISARMED, "ArmingState": ArmingState.DISARMED,
}
PARTITION_DISARMED2 = {
"PartitionID": "2",
"ArmingState": ArmingState.DISARMED,
} }
PARTITION_ARMED_STAY = { PARTITION_ARMED_STAY = {
"PartitionID": "1", "PartitionID": "1",
"ArmingState": TotalConnectClient.TotalConnectLocation.ARMED_STAY, "ArmingState": ArmingState.ARMED_STAY,
}
PARTITION_ARMED_STAY2 = {
"PartitionID": "2",
"ArmingState": ArmingState.DISARMED,
} }
PARTITION_ARMED_AWAY = { PARTITION_ARMED_AWAY = {
"PartitionID": "1", "PartitionID": "1",
"ArmingState": TotalConnectClient.TotalConnectLocation.ARMED_AWAY, "ArmingState": ArmingState.ARMED_AWAY,
} }
PARTITION_ARMED_CUSTOM = { PARTITION_ARMED_CUSTOM = {
"PartitionID": "1", "PartitionID": "1",
"ArmingState": TotalConnectClient.TotalConnectLocation.ARMED_CUSTOM_BYPASS, "ArmingState": ArmingState.ARMED_CUSTOM_BYPASS,
} }
PARTITION_ARMED_NIGHT = { PARTITION_ARMED_NIGHT = {
"PartitionID": "1", "PartitionID": "1",
"ArmingState": TotalConnectClient.TotalConnectLocation.ARMED_STAY_NIGHT, "ArmingState": ArmingState.ARMED_STAY_NIGHT,
} }
PARTITION_ARMING = { PARTITION_ARMING = {
"PartitionID": "1", "PartitionID": "1",
"ArmingState": TotalConnectClient.TotalConnectLocation.ARMING, "ArmingState": ArmingState.ARMING,
} }
PARTITION_DISARMING = { PARTITION_DISARMING = {
"PartitionID": "1", "PartitionID": "1",
"ArmingState": TotalConnectClient.TotalConnectLocation.DISARMING, "ArmingState": ArmingState.DISARMING,
} }
PARTITION_TRIGGERED_POLICE = { PARTITION_TRIGGERED_POLICE = {
"PartitionID": "1", "PartitionID": "1",
"ArmingState": TotalConnectClient.TotalConnectLocation.ALARMING, "ArmingState": ArmingState.ALARMING,
} }
PARTITION_TRIGGERED_FIRE = { PARTITION_TRIGGERED_FIRE = {
"PartitionID": "1", "PartitionID": "1",
"ArmingState": TotalConnectClient.TotalConnectLocation.ALARMING_FIRE_SMOKE, "ArmingState": ArmingState.ALARMING_FIRE_SMOKE,
} }
PARTITION_TRIGGERED_CARBON_MONOXIDE = { PARTITION_TRIGGERED_CARBON_MONOXIDE = {
"PartitionID": "1", "PartitionID": "1",
"ArmingState": TotalConnectClient.TotalConnectLocation.ALARMING_CARBON_MONOXIDE, "ArmingState": ArmingState.ALARMING_CARBON_MONOXIDE,
} }
PARTITION_UNKNOWN = { PARTITION_UNKNOWN = {
@ -99,17 +122,17 @@ PARTITION_UNKNOWN = {
} }
PARTITION_INFO_DISARMED = {0: PARTITION_DISARMED} PARTITION_INFO_DISARMED = [PARTITION_DISARMED, PARTITION_DISARMED2]
PARTITION_INFO_ARMED_STAY = {0: PARTITION_ARMED_STAY} PARTITION_INFO_ARMED_STAY = [PARTITION_ARMED_STAY, PARTITION_ARMED_STAY2]
PARTITION_INFO_ARMED_AWAY = {0: PARTITION_ARMED_AWAY} PARTITION_INFO_ARMED_AWAY = [PARTITION_ARMED_AWAY]
PARTITION_INFO_ARMED_CUSTOM = {0: PARTITION_ARMED_CUSTOM} PARTITION_INFO_ARMED_CUSTOM = [PARTITION_ARMED_CUSTOM]
PARTITION_INFO_ARMED_NIGHT = {0: PARTITION_ARMED_NIGHT} PARTITION_INFO_ARMED_NIGHT = [PARTITION_ARMED_NIGHT]
PARTITION_INFO_ARMING = {0: PARTITION_ARMING} PARTITION_INFO_ARMING = [PARTITION_ARMING]
PARTITION_INFO_DISARMING = {0: PARTITION_DISARMING} PARTITION_INFO_DISARMING = [PARTITION_DISARMING]
PARTITION_INFO_TRIGGERED_POLICE = {0: PARTITION_TRIGGERED_POLICE} PARTITION_INFO_TRIGGERED_POLICE = [PARTITION_TRIGGERED_POLICE]
PARTITION_INFO_TRIGGERED_FIRE = {0: PARTITION_TRIGGERED_FIRE} PARTITION_INFO_TRIGGERED_FIRE = [PARTITION_TRIGGERED_FIRE]
PARTITION_INFO_TRIGGERED_CARBON_MONOXIDE = {0: PARTITION_TRIGGERED_CARBON_MONOXIDE} PARTITION_INFO_TRIGGERED_CARBON_MONOXIDE = [PARTITION_TRIGGERED_CARBON_MONOXIDE]
PARTITION_INFO_UNKNOWN = {0: PARTITION_UNKNOWN} PARTITION_INFO_UNKNOWN = [PARTITION_UNKNOWN]
PARTITIONS_DISARMED = {"PartitionInfo": PARTITION_INFO_DISARMED} PARTITIONS_DISARMED = {"PartitionInfo": PARTITION_INFO_DISARMED}
PARTITIONS_ARMED_STAY = {"PartitionInfo": PARTITION_INFO_ARMED_STAY} PARTITIONS_ARMED_STAY = {"PartitionInfo": PARTITION_INFO_ARMED_STAY}
@ -128,7 +151,7 @@ PARTITIONS_UNKNOWN = {"PartitionInfo": PARTITION_INFO_UNKNOWN}
ZONE_NORMAL = { ZONE_NORMAL = {
"ZoneID": "1", "ZoneID": "1",
"ZoneDescription": "Normal", "ZoneDescription": "Normal",
"ZoneStatus": TotalConnectClient.ZONE_STATUS_NORMAL, "ZoneStatus": ZoneStatus.NORMAL,
"PartitionId": "1", "PartitionId": "1",
} }
@ -176,46 +199,74 @@ METADATA_TRIGGERED_CARBON_MONOXIDE["Partitions"] = PARTITIONS_TRIGGERED_CARBON_M
METADATA_UNKNOWN = METADATA_DISARMED.copy() METADATA_UNKNOWN = METADATA_DISARMED.copy()
METADATA_UNKNOWN["Partitions"] = PARTITIONS_UNKNOWN METADATA_UNKNOWN["Partitions"] = PARTITIONS_UNKNOWN
RESPONSE_DISARMED = {"ResultCode": 0, "PanelMetadataAndStatus": METADATA_DISARMED} RESPONSE_DISARMED = {
RESPONSE_ARMED_STAY = {"ResultCode": 0, "PanelMetadataAndStatus": METADATA_ARMED_STAY} "ResultCode": 0,
RESPONSE_ARMED_AWAY = {"ResultCode": 0, "PanelMetadataAndStatus": METADATA_ARMED_AWAY} "PanelMetadataAndStatus": METADATA_DISARMED,
"ArmingState": ArmingState.DISARMED,
}
RESPONSE_ARMED_STAY = {
"ResultCode": 0,
"PanelMetadataAndStatus": METADATA_ARMED_STAY,
"ArmingState": ArmingState.ARMED_STAY,
}
RESPONSE_ARMED_AWAY = {
"ResultCode": 0,
"PanelMetadataAndStatus": METADATA_ARMED_AWAY,
"ArmingState": ArmingState.ARMED_AWAY,
}
RESPONSE_ARMED_CUSTOM = { RESPONSE_ARMED_CUSTOM = {
"ResultCode": 0, "ResultCode": 0,
"PanelMetadataAndStatus": METADATA_ARMED_CUSTOM, "PanelMetadataAndStatus": METADATA_ARMED_CUSTOM,
"ArmingState": ArmingState.ARMED_CUSTOM_BYPASS,
}
RESPONSE_ARMED_NIGHT = {
"ResultCode": 0,
"PanelMetadataAndStatus": METADATA_ARMED_NIGHT,
"ArmingState": ArmingState.ARMED_STAY_NIGHT,
}
RESPONSE_ARMING = {
"ResultCode": 0,
"PanelMetadataAndStatus": METADATA_ARMING,
"ArmingState": ArmingState.ARMING,
}
RESPONSE_DISARMING = {
"ResultCode": 0,
"PanelMetadataAndStatus": METADATA_DISARMING,
"ArmingState": ArmingState.DISARMING,
} }
RESPONSE_ARMED_NIGHT = {"ResultCode": 0, "PanelMetadataAndStatus": METADATA_ARMED_NIGHT}
RESPONSE_ARMING = {"ResultCode": 0, "PanelMetadataAndStatus": METADATA_ARMING}
RESPONSE_DISARMING = {"ResultCode": 0, "PanelMetadataAndStatus": METADATA_DISARMING}
RESPONSE_TRIGGERED_POLICE = { RESPONSE_TRIGGERED_POLICE = {
"ResultCode": 0, "ResultCode": 0,
"PanelMetadataAndStatus": METADATA_TRIGGERED_POLICE, "PanelMetadataAndStatus": METADATA_TRIGGERED_POLICE,
"ArmingState": ArmingState.ALARMING,
} }
RESPONSE_TRIGGERED_FIRE = { RESPONSE_TRIGGERED_FIRE = {
"ResultCode": 0, "ResultCode": 0,
"PanelMetadataAndStatus": METADATA_TRIGGERED_FIRE, "PanelMetadataAndStatus": METADATA_TRIGGERED_FIRE,
"ArmingState": ArmingState.ALARMING_FIRE_SMOKE,
} }
RESPONSE_TRIGGERED_CARBON_MONOXIDE = { RESPONSE_TRIGGERED_CARBON_MONOXIDE = {
"ResultCode": 0, "ResultCode": 0,
"PanelMetadataAndStatus": METADATA_TRIGGERED_CARBON_MONOXIDE, "PanelMetadataAndStatus": METADATA_TRIGGERED_CARBON_MONOXIDE,
"ArmingState": ArmingState.ALARMING_CARBON_MONOXIDE,
}
RESPONSE_UNKNOWN = {
"ResultCode": 0,
"PanelMetadataAndStatus": METADATA_UNKNOWN,
"ArmingState": ArmingState.DISARMED,
} }
RESPONSE_UNKNOWN = {"ResultCode": 0, "PanelMetadataAndStatus": METADATA_UNKNOWN}
RESPONSE_ARM_SUCCESS = {"ResultCode": TotalConnectClient.TotalConnectClient.ARM_SUCCESS} RESPONSE_ARM_SUCCESS = {"ResultCode": TotalConnectClient.ARM_SUCCESS}
RESPONSE_ARM_FAILURE = { RESPONSE_ARM_FAILURE = {"ResultCode": TotalConnectClient.COMMAND_FAILED}
"ResultCode": TotalConnectClient.TotalConnectClient.COMMAND_FAILED RESPONSE_DISARM_SUCCESS = {"ResultCode": TotalConnectClient.DISARM_SUCCESS}
}
RESPONSE_DISARM_SUCCESS = {
"ResultCode": TotalConnectClient.TotalConnectClient.DISARM_SUCCESS
}
RESPONSE_DISARM_FAILURE = { RESPONSE_DISARM_FAILURE = {
"ResultCode": TotalConnectClient.TotalConnectClient.COMMAND_FAILED, "ResultCode": TotalConnectClient.COMMAND_FAILED,
"ResultData": "Command Failed", "ResultData": "Command Failed",
} }
RESPONSE_USER_CODE_INVALID = { RESPONSE_USER_CODE_INVALID = {
"ResultCode": TotalConnectClient.TotalConnectClient.USER_CODE_INVALID, "ResultCode": TotalConnectClient.USER_CODE_INVALID,
"ResultData": "testing user code invalid", "ResultData": "testing user code invalid",
} }
RESPONSE_SUCCESS = {"ResultCode": TotalConnectClient.TotalConnectClient.SUCCESS} RESPONSE_SUCCESS = {"ResultCode": TotalConnectClient.SUCCESS}
USERNAME = "username@me.com" USERNAME = "username@me.com"
PASSWORD = "password" PASSWORD = "password"
@ -227,40 +278,72 @@ CONFIG_DATA = {
} }
CONFIG_DATA_NO_USERCODES = {CONF_USERNAME: USERNAME, CONF_PASSWORD: PASSWORD} CONFIG_DATA_NO_USERCODES = {CONF_USERNAME: USERNAME, CONF_PASSWORD: PASSWORD}
PARTITION_DETAILS_1 = {
USERNAME = "username@me.com" "PartitionID": 1,
PASSWORD = "password" "ArmingState": ArmingState.DISARMED.value,
USERCODES = {123456: "7890"} "PartitionName": "Test1",
CONFIG_DATA = {
CONF_USERNAME: USERNAME,
CONF_PASSWORD: PASSWORD,
CONF_USERCODES: USERCODES,
} }
CONFIG_DATA_NO_USERCODES = {CONF_USERNAME: USERNAME, CONF_PASSWORD: PASSWORD}
PARTITION_DETAILS_2 = {
"PartitionID": 2,
"ArmingState": ArmingState.DISARMED.value,
"PartitionName": "Test2",
}
PARTITION_DETAILS = {"PartitionDetails": [PARTITION_DETAILS_1, PARTITION_DETAILS_2]}
RESPONSE_PARTITION_DETAILS = {
"ResultCode": TotalConnectClient.SUCCESS,
"ResultData": "testing partition details",
"PartitionsInfoList": PARTITION_DETAILS,
}
ZONE_DETAILS_NORMAL = {
"PartitionId": "1",
"Batterylevel": "-1",
"Signalstrength": "-1",
"zoneAdditionalInfo": None,
"ZoneID": "1",
"ZoneStatus": ZoneStatus.NORMAL,
"ZoneTypeId": ZoneType.SECURITY,
"CanBeBypassed": 1,
"ZoneFlags": None,
}
ZONE_STATUS_INFO = [ZONE_DETAILS_NORMAL]
ZONE_DETAILS = {"ZoneStatusInfoWithPartitionId": ZONE_STATUS_INFO}
ZONE_DETAIL_STATUS = {"Zones": ZONE_DETAILS}
RESPONSE_GET_ZONE_DETAILS_SUCCESS = {
"ResultCode": 0,
"ResultData": "Success",
"ZoneStatus": ZONE_DETAIL_STATUS,
}
TOTALCONNECT_REQUEST = (
"homeassistant.components.totalconnect.TotalConnectClient.request"
)
async def setup_platform(hass, platform): async def setup_platform(hass, platform):
"""Set up the TotalConnect platform.""" """Set up the TotalConnect platform."""
# first set up a config entry and add it to hass # first set up a config entry and add it to hass
mock_entry = MockConfigEntry( mock_entry = MockConfigEntry(domain=DOMAIN, data=CONFIG_DATA)
domain=DOMAIN,
data=CONFIG_DATA,
)
mock_entry.add_to_hass(hass) mock_entry.add_to_hass(hass)
responses = [RESPONSE_AUTHENTICATE, RESPONSE_DISARMED] responses = [
RESPONSE_AUTHENTICATE,
RESPONSE_PARTITION_DETAILS,
RESPONSE_GET_ZONE_DETAILS_SUCCESS,
RESPONSE_DISARMED,
RESPONSE_DISARMED,
]
with patch("homeassistant.components.totalconnect.PLATFORMS", [platform]), patch( with patch("homeassistant.components.totalconnect.PLATFORMS", [platform]), patch(
"zeep.Client", autospec=True TOTALCONNECT_REQUEST,
), patch(
"homeassistant.components.totalconnect.TotalConnectClient.TotalConnectClient.request",
side_effect=responses, side_effect=responses,
) as mock_request, patch( ) as mock_request:
"homeassistant.components.totalconnect.TotalConnectClient.TotalConnectClient.get_zone_details",
return_value=True,
):
assert await async_setup_component(hass, DOMAIN, {}) assert await async_setup_component(hass, DOMAIN, {})
assert mock_request.call_count == 2 assert mock_request.call_count == 5
await hass.async_block_till_done() await hass.async_block_till_done()
return mock_entry return mock_entry

View file

@ -1,4 +1,5 @@
"""Tests for the TotalConnect alarm control panel device.""" """Tests for the TotalConnect alarm control panel device."""
from datetime import timedelta
from unittest.mock import patch from unittest.mock import patch
import pytest import pytest
@ -19,8 +20,11 @@ from homeassistant.const import (
STATE_ALARM_DISARMED, STATE_ALARM_DISARMED,
STATE_ALARM_DISARMING, STATE_ALARM_DISARMING,
STATE_ALARM_TRIGGERED, STATE_ALARM_TRIGGERED,
STATE_UNAVAILABLE,
) )
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError from homeassistant.exceptions import HomeAssistantError
from homeassistant.util import dt
from .common import ( from .common import (
LOCATION_ID, LOCATION_ID,
@ -41,18 +45,23 @@ from .common import (
RESPONSE_TRIGGERED_POLICE, RESPONSE_TRIGGERED_POLICE,
RESPONSE_UNKNOWN, RESPONSE_UNKNOWN,
RESPONSE_USER_CODE_INVALID, RESPONSE_USER_CODE_INVALID,
TOTALCONNECT_REQUEST,
setup_platform, setup_platform,
) )
from tests.common import async_fire_time_changed
ENTITY_ID = "alarm_control_panel.test" ENTITY_ID = "alarm_control_panel.test"
ENTITY_ID_2 = "alarm_control_panel.test_partition_2"
CODE = "-1" CODE = "-1"
DATA = {ATTR_ENTITY_ID: ENTITY_ID} DATA = {ATTR_ENTITY_ID: ENTITY_ID}
DELAY = timedelta(seconds=10)
async def test_attributes(hass): async def test_attributes(hass: HomeAssistant) -> None:
"""Test the alarm control panel attributes are correct.""" """Test the alarm control panel attributes are correct."""
with patch( with patch(
"homeassistant.components.totalconnect.TotalConnectClient.TotalConnectClient.request", "homeassistant.components.totalconnect.TotalConnectClient.request",
return_value=RESPONSE_DISARMED, return_value=RESPONSE_DISARMED,
) as mock_request: ) as mock_request:
await setup_platform(hass, ALARM_DOMAIN) await setup_platform(hass, ALARM_DOMAIN)
@ -63,37 +72,44 @@ async def test_attributes(hass):
entity_registry = await hass.helpers.entity_registry.async_get_registry() entity_registry = await hass.helpers.entity_registry.async_get_registry()
entry = entity_registry.async_get(ENTITY_ID) entry = entity_registry.async_get(ENTITY_ID)
# TotalConnect alarm device unique_id is the location_id # TotalConnect partition #1 alarm device unique_id is the location_id
assert entry.unique_id == LOCATION_ID assert entry.unique_id == LOCATION_ID
entry2 = entity_registry.async_get(ENTITY_ID_2)
# TotalConnect partition #2 unique_id is the location_id + "_{partition_number}"
assert entry2.unique_id == LOCATION_ID + "_2"
assert mock_request.call_count == 1
async def test_arm_home_success(hass):
async def test_arm_home_success(hass: HomeAssistant) -> None:
"""Test arm home method success.""" """Test arm home method success."""
responses = [RESPONSE_DISARMED, RESPONSE_ARM_SUCCESS, RESPONSE_ARMED_STAY] responses = [RESPONSE_DISARMED, RESPONSE_ARM_SUCCESS, RESPONSE_ARMED_STAY]
with patch( with patch(TOTALCONNECT_REQUEST, side_effect=responses) as mock_request:
"homeassistant.components.totalconnect.TotalConnectClient.TotalConnectClient.request",
side_effect=responses,
):
await setup_platform(hass, ALARM_DOMAIN) await setup_platform(hass, ALARM_DOMAIN)
assert hass.states.get(ENTITY_ID).state == STATE_ALARM_DISARMED assert hass.states.get(ENTITY_ID).state == STATE_ALARM_DISARMED
assert hass.states.get(ENTITY_ID_2).state == STATE_ALARM_DISARMED
assert mock_request.call_count == 1
await hass.services.async_call( await hass.services.async_call(
ALARM_DOMAIN, SERVICE_ALARM_ARM_HOME, DATA, blocking=True ALARM_DOMAIN, SERVICE_ALARM_ARM_HOME, DATA, blocking=True
) )
assert mock_request.call_count == 2
async_fire_time_changed(hass, dt.utcnow() + DELAY)
await hass.async_block_till_done() await hass.async_block_till_done()
assert mock_request.call_count == 3
assert hass.states.get(ENTITY_ID).state == STATE_ALARM_ARMED_HOME assert hass.states.get(ENTITY_ID).state == STATE_ALARM_ARMED_HOME
# second partition should not be armed
assert hass.states.get(ENTITY_ID_2).state == STATE_ALARM_DISARMED
async def test_arm_home_failure(hass): async def test_arm_home_failure(hass: HomeAssistant) -> None:
"""Test arm home method failure.""" """Test arm home method failure."""
responses = [RESPONSE_DISARMED, RESPONSE_ARM_FAILURE, RESPONSE_DISARMED] responses = [RESPONSE_DISARMED, RESPONSE_ARM_FAILURE]
with patch( with patch(TOTALCONNECT_REQUEST, side_effect=responses) as mock_request:
"homeassistant.components.totalconnect.TotalConnectClient.TotalConnectClient.request",
side_effect=responses,
):
await setup_platform(hass, ALARM_DOMAIN) await setup_platform(hass, ALARM_DOMAIN)
assert hass.states.get(ENTITY_ID).state == STATE_ALARM_DISARMED assert hass.states.get(ENTITY_ID).state == STATE_ALARM_DISARMED
assert mock_request.call_count == 1
with pytest.raises(HomeAssistantError) as err: with pytest.raises(HomeAssistantError) as err:
await hass.services.async_call( await hass.services.async_call(
@ -102,17 +118,16 @@ async def test_arm_home_failure(hass):
await hass.async_block_till_done() await hass.async_block_till_done()
assert f"{err.value}" == "TotalConnect failed to arm home test." assert f"{err.value}" == "TotalConnect failed to arm home test."
assert hass.states.get(ENTITY_ID).state == STATE_ALARM_DISARMED assert hass.states.get(ENTITY_ID).state == STATE_ALARM_DISARMED
assert mock_request.call_count == 2
async def test_arm_home_invalid_usercode(hass): async def test_arm_home_invalid_usercode(hass: HomeAssistant) -> None:
"""Test arm home method with invalid usercode.""" """Test arm home method with invalid usercode."""
responses = [RESPONSE_DISARMED, RESPONSE_USER_CODE_INVALID, RESPONSE_DISARMED] responses = [RESPONSE_DISARMED, RESPONSE_USER_CODE_INVALID]
with patch( with patch(TOTALCONNECT_REQUEST, side_effect=responses) as mock_request:
"homeassistant.components.totalconnect.TotalConnectClient.TotalConnectClient.request",
side_effect=responses,
):
await setup_platform(hass, ALARM_DOMAIN) await setup_platform(hass, ALARM_DOMAIN)
assert hass.states.get(ENTITY_ID).state == STATE_ALARM_DISARMED assert hass.states.get(ENTITY_ID).state == STATE_ALARM_DISARMED
assert mock_request.call_count == 1
with pytest.raises(HomeAssistantError) as err: with pytest.raises(HomeAssistantError) as err:
await hass.services.async_call( await hass.services.async_call(
@ -121,34 +136,35 @@ async def test_arm_home_invalid_usercode(hass):
await hass.async_block_till_done() await hass.async_block_till_done()
assert f"{err.value}" == "TotalConnect failed to arm home test." assert f"{err.value}" == "TotalConnect failed to arm home test."
assert hass.states.get(ENTITY_ID).state == STATE_ALARM_DISARMED assert hass.states.get(ENTITY_ID).state == STATE_ALARM_DISARMED
assert mock_request.call_count == 2
async def test_arm_away_success(hass): async def test_arm_away_success(hass: HomeAssistant) -> None:
"""Test arm away method success.""" """Test arm away method success."""
responses = [RESPONSE_DISARMED, RESPONSE_ARM_SUCCESS, RESPONSE_ARMED_AWAY] responses = [RESPONSE_DISARMED, RESPONSE_ARM_SUCCESS, RESPONSE_ARMED_AWAY]
with patch( with patch(TOTALCONNECT_REQUEST, side_effect=responses) as mock_request:
"homeassistant.components.totalconnect.TotalConnectClient.TotalConnectClient.request",
side_effect=responses,
):
await setup_platform(hass, ALARM_DOMAIN) await setup_platform(hass, ALARM_DOMAIN)
assert hass.states.get(ENTITY_ID).state == STATE_ALARM_DISARMED assert hass.states.get(ENTITY_ID).state == STATE_ALARM_DISARMED
assert mock_request.call_count == 1
await hass.services.async_call( await hass.services.async_call(
ALARM_DOMAIN, SERVICE_ALARM_ARM_AWAY, DATA, blocking=True ALARM_DOMAIN, SERVICE_ALARM_ARM_AWAY, DATA, blocking=True
) )
assert mock_request.call_count == 2
async_fire_time_changed(hass, dt.utcnow() + DELAY)
await hass.async_block_till_done() await hass.async_block_till_done()
assert mock_request.call_count == 3
assert hass.states.get(ENTITY_ID).state == STATE_ALARM_ARMED_AWAY assert hass.states.get(ENTITY_ID).state == STATE_ALARM_ARMED_AWAY
async def test_arm_away_failure(hass): async def test_arm_away_failure(hass: HomeAssistant) -> None:
"""Test arm away method failure.""" """Test arm away method failure."""
responses = [RESPONSE_DISARMED, RESPONSE_ARM_FAILURE, RESPONSE_DISARMED] responses = [RESPONSE_DISARMED, RESPONSE_ARM_FAILURE]
with patch( with patch(TOTALCONNECT_REQUEST, side_effect=responses) as mock_request:
"homeassistant.components.totalconnect.TotalConnectClient.TotalConnectClient.request",
side_effect=responses,
):
await setup_platform(hass, ALARM_DOMAIN) await setup_platform(hass, ALARM_DOMAIN)
assert hass.states.get(ENTITY_ID).state == STATE_ALARM_DISARMED assert hass.states.get(ENTITY_ID).state == STATE_ALARM_DISARMED
assert mock_request.call_count == 1
with pytest.raises(HomeAssistantError) as err: with pytest.raises(HomeAssistantError) as err:
await hass.services.async_call( await hass.services.async_call(
@ -157,34 +173,35 @@ async def test_arm_away_failure(hass):
await hass.async_block_till_done() await hass.async_block_till_done()
assert f"{err.value}" == "TotalConnect failed to arm away test." assert f"{err.value}" == "TotalConnect failed to arm away test."
assert hass.states.get(ENTITY_ID).state == STATE_ALARM_DISARMED assert hass.states.get(ENTITY_ID).state == STATE_ALARM_DISARMED
assert mock_request.call_count == 2
async def test_disarm_success(hass): async def test_disarm_success(hass: HomeAssistant) -> None:
"""Test disarm method success.""" """Test disarm method success."""
responses = [RESPONSE_ARMED_AWAY, RESPONSE_DISARM_SUCCESS, RESPONSE_DISARMED] responses = [RESPONSE_ARMED_AWAY, RESPONSE_DISARM_SUCCESS, RESPONSE_DISARMED]
with patch( with patch(TOTALCONNECT_REQUEST, side_effect=responses) as mock_request:
"homeassistant.components.totalconnect.TotalConnectClient.TotalConnectClient.request",
side_effect=responses,
):
await setup_platform(hass, ALARM_DOMAIN) await setup_platform(hass, ALARM_DOMAIN)
assert hass.states.get(ENTITY_ID).state == STATE_ALARM_ARMED_AWAY assert hass.states.get(ENTITY_ID).state == STATE_ALARM_ARMED_AWAY
assert mock_request.call_count == 1
await hass.services.async_call( await hass.services.async_call(
ALARM_DOMAIN, SERVICE_ALARM_DISARM, DATA, blocking=True ALARM_DOMAIN, SERVICE_ALARM_DISARM, DATA, blocking=True
) )
assert mock_request.call_count == 2
async_fire_time_changed(hass, dt.utcnow() + DELAY)
await hass.async_block_till_done() await hass.async_block_till_done()
assert mock_request.call_count == 3
assert hass.states.get(ENTITY_ID).state == STATE_ALARM_DISARMED assert hass.states.get(ENTITY_ID).state == STATE_ALARM_DISARMED
async def test_disarm_failure(hass): async def test_disarm_failure(hass: HomeAssistant) -> None:
"""Test disarm method failure.""" """Test disarm method failure."""
responses = [RESPONSE_ARMED_AWAY, RESPONSE_DISARM_FAILURE, RESPONSE_ARMED_AWAY] responses = [RESPONSE_ARMED_AWAY, RESPONSE_DISARM_FAILURE]
with patch( with patch(TOTALCONNECT_REQUEST, side_effect=responses) as mock_request:
"homeassistant.components.totalconnect.TotalConnectClient.TotalConnectClient.request",
side_effect=responses,
):
await setup_platform(hass, ALARM_DOMAIN) await setup_platform(hass, ALARM_DOMAIN)
assert hass.states.get(ENTITY_ID).state == STATE_ALARM_ARMED_AWAY assert hass.states.get(ENTITY_ID).state == STATE_ALARM_ARMED_AWAY
assert mock_request.call_count == 1
with pytest.raises(HomeAssistantError) as err: with pytest.raises(HomeAssistantError) as err:
await hass.services.async_call( await hass.services.async_call(
@ -193,17 +210,16 @@ async def test_disarm_failure(hass):
await hass.async_block_till_done() await hass.async_block_till_done()
assert f"{err.value}" == "TotalConnect failed to disarm test." assert f"{err.value}" == "TotalConnect failed to disarm test."
assert hass.states.get(ENTITY_ID).state == STATE_ALARM_ARMED_AWAY assert hass.states.get(ENTITY_ID).state == STATE_ALARM_ARMED_AWAY
assert mock_request.call_count == 2
async def test_disarm_invalid_usercode(hass): async def test_disarm_invalid_usercode(hass: HomeAssistant) -> None:
"""Test disarm method failure.""" """Test disarm method failure."""
responses = [RESPONSE_ARMED_AWAY, RESPONSE_USER_CODE_INVALID, RESPONSE_ARMED_AWAY] responses = [RESPONSE_ARMED_AWAY, RESPONSE_USER_CODE_INVALID]
with patch( with patch(TOTALCONNECT_REQUEST, side_effect=responses) as mock_request:
"homeassistant.components.totalconnect.TotalConnectClient.TotalConnectClient.request",
side_effect=responses,
):
await setup_platform(hass, ALARM_DOMAIN) await setup_platform(hass, ALARM_DOMAIN)
assert hass.states.get(ENTITY_ID).state == STATE_ALARM_ARMED_AWAY assert hass.states.get(ENTITY_ID).state == STATE_ALARM_ARMED_AWAY
assert mock_request.call_count == 1
with pytest.raises(HomeAssistantError) as err: with pytest.raises(HomeAssistantError) as err:
await hass.services.async_call( await hass.services.async_call(
@ -212,35 +228,35 @@ async def test_disarm_invalid_usercode(hass):
await hass.async_block_till_done() await hass.async_block_till_done()
assert f"{err.value}" == "TotalConnect failed to disarm test." assert f"{err.value}" == "TotalConnect failed to disarm test."
assert hass.states.get(ENTITY_ID).state == STATE_ALARM_ARMED_AWAY assert hass.states.get(ENTITY_ID).state == STATE_ALARM_ARMED_AWAY
assert mock_request.call_count == 2
async def test_arm_night_success(hass): async def test_arm_night_success(hass: HomeAssistant) -> None:
"""Test arm night method success.""" """Test arm night method success."""
responses = [RESPONSE_DISARMED, RESPONSE_ARM_SUCCESS, RESPONSE_ARMED_NIGHT] responses = [RESPONSE_DISARMED, RESPONSE_ARM_SUCCESS, RESPONSE_ARMED_NIGHT]
with patch( with patch(TOTALCONNECT_REQUEST, side_effect=responses) as mock_request:
"homeassistant.components.totalconnect.TotalConnectClient.TotalConnectClient.request",
side_effect=responses,
):
await setup_platform(hass, ALARM_DOMAIN) await setup_platform(hass, ALARM_DOMAIN)
assert hass.states.get(ENTITY_ID).state == STATE_ALARM_DISARMED assert hass.states.get(ENTITY_ID).state == STATE_ALARM_DISARMED
assert mock_request.call_count == 1
await hass.services.async_call( await hass.services.async_call(
ALARM_DOMAIN, SERVICE_ALARM_ARM_NIGHT, DATA, blocking=True ALARM_DOMAIN, SERVICE_ALARM_ARM_NIGHT, DATA, blocking=True
) )
assert mock_request.call_count == 2
async_fire_time_changed(hass, dt.utcnow() + DELAY)
await hass.async_block_till_done() await hass.async_block_till_done()
assert mock_request.call_count == 3
assert hass.states.get(ENTITY_ID).state == STATE_ALARM_ARMED_NIGHT assert hass.states.get(ENTITY_ID).state == STATE_ALARM_ARMED_NIGHT
async def test_arm_night_failure(hass): async def test_arm_night_failure(hass: HomeAssistant) -> None:
"""Test arm night method failure.""" """Test arm night method failure."""
responses = [RESPONSE_DISARMED, RESPONSE_ARM_FAILURE, RESPONSE_DISARMED] responses = [RESPONSE_DISARMED, RESPONSE_ARM_FAILURE]
with patch( with patch(TOTALCONNECT_REQUEST, side_effect=responses) as mock_request:
"homeassistant.components.totalconnect.TotalConnectClient.TotalConnectClient.request",
side_effect=responses,
):
await setup_platform(hass, ALARM_DOMAIN) await setup_platform(hass, ALARM_DOMAIN)
assert hass.states.get(ENTITY_ID).state == STATE_ALARM_DISARMED assert hass.states.get(ENTITY_ID).state == STATE_ALARM_DISARMED
assert mock_request.call_count == 1
with pytest.raises(HomeAssistantError) as err: with pytest.raises(HomeAssistantError) as err:
await hass.services.async_call( await hass.services.async_call(
@ -249,98 +265,93 @@ async def test_arm_night_failure(hass):
await hass.async_block_till_done() await hass.async_block_till_done()
assert f"{err.value}" == "TotalConnect failed to arm night test." assert f"{err.value}" == "TotalConnect failed to arm night test."
assert hass.states.get(ENTITY_ID).state == STATE_ALARM_DISARMED assert hass.states.get(ENTITY_ID).state == STATE_ALARM_DISARMED
assert mock_request.call_count == 2
async def test_arming(hass): async def test_arming(hass: HomeAssistant) -> None:
"""Test arming.""" """Test arming."""
responses = [RESPONSE_DISARMED, RESPONSE_SUCCESS, RESPONSE_ARMING] responses = [RESPONSE_DISARMED, RESPONSE_SUCCESS, RESPONSE_ARMING]
with patch( with patch(TOTALCONNECT_REQUEST, side_effect=responses) as mock_request:
"homeassistant.components.totalconnect.TotalConnectClient.TotalConnectClient.request",
side_effect=responses,
):
await setup_platform(hass, ALARM_DOMAIN) await setup_platform(hass, ALARM_DOMAIN)
assert hass.states.get(ENTITY_ID).state == STATE_ALARM_DISARMED assert hass.states.get(ENTITY_ID).state == STATE_ALARM_DISARMED
assert mock_request.call_count == 1
await hass.services.async_call( await hass.services.async_call(
ALARM_DOMAIN, SERVICE_ALARM_ARM_NIGHT, DATA, blocking=True ALARM_DOMAIN, SERVICE_ALARM_ARM_NIGHT, DATA, blocking=True
) )
assert mock_request.call_count == 2
async_fire_time_changed(hass, dt.utcnow() + DELAY)
await hass.async_block_till_done()
assert mock_request.call_count == 3
assert hass.states.get(ENTITY_ID).state == STATE_ALARM_ARMING assert hass.states.get(ENTITY_ID).state == STATE_ALARM_ARMING
async def test_disarming(hass): async def test_disarming(hass: HomeAssistant) -> None:
"""Test disarming.""" """Test disarming."""
responses = [RESPONSE_ARMED_AWAY, RESPONSE_SUCCESS, RESPONSE_DISARMING] responses = [RESPONSE_ARMED_AWAY, RESPONSE_SUCCESS, RESPONSE_DISARMING]
with patch( with patch(TOTALCONNECT_REQUEST, side_effect=responses) as mock_request:
"homeassistant.components.totalconnect.TotalConnectClient.TotalConnectClient.request",
side_effect=responses,
):
await setup_platform(hass, ALARM_DOMAIN) await setup_platform(hass, ALARM_DOMAIN)
assert hass.states.get(ENTITY_ID).state == STATE_ALARM_ARMED_AWAY assert hass.states.get(ENTITY_ID).state == STATE_ALARM_ARMED_AWAY
assert mock_request.call_count == 1
await hass.services.async_call( await hass.services.async_call(
ALARM_DOMAIN, SERVICE_ALARM_DISARM, DATA, blocking=True ALARM_DOMAIN, SERVICE_ALARM_DISARM, DATA, blocking=True
) )
assert mock_request.call_count == 2
async_fire_time_changed(hass, dt.utcnow() + DELAY)
await hass.async_block_till_done()
assert mock_request.call_count == 3
assert hass.states.get(ENTITY_ID).state == STATE_ALARM_DISARMING assert hass.states.get(ENTITY_ID).state == STATE_ALARM_DISARMING
async def test_triggered_fire(hass): async def test_triggered_fire(hass: HomeAssistant) -> None:
"""Test triggered by fire.""" """Test triggered by fire."""
responses = [RESPONSE_TRIGGERED_FIRE] responses = [RESPONSE_TRIGGERED_FIRE]
with patch( with patch(TOTALCONNECT_REQUEST, side_effect=responses) as mock_request:
"homeassistant.components.totalconnect.TotalConnectClient.TotalConnectClient.request",
side_effect=responses,
):
await setup_platform(hass, ALARM_DOMAIN) await setup_platform(hass, ALARM_DOMAIN)
state = hass.states.get(ENTITY_ID) state = hass.states.get(ENTITY_ID)
assert state.state == STATE_ALARM_TRIGGERED assert state.state == STATE_ALARM_TRIGGERED
assert state.attributes.get("triggered_source") == "Fire/Smoke" assert state.attributes.get("triggered_source") == "Fire/Smoke"
assert mock_request.call_count == 1
async def test_triggered_police(hass): async def test_triggered_police(hass: HomeAssistant) -> None:
"""Test triggered by police.""" """Test triggered by police."""
responses = [RESPONSE_TRIGGERED_POLICE] responses = [RESPONSE_TRIGGERED_POLICE]
with patch( with patch(TOTALCONNECT_REQUEST, side_effect=responses) as mock_request:
"homeassistant.components.totalconnect.TotalConnectClient.TotalConnectClient.request",
side_effect=responses,
):
await setup_platform(hass, ALARM_DOMAIN) await setup_platform(hass, ALARM_DOMAIN)
state = hass.states.get(ENTITY_ID) state = hass.states.get(ENTITY_ID)
assert state.state == STATE_ALARM_TRIGGERED assert state.state == STATE_ALARM_TRIGGERED
assert state.attributes.get("triggered_source") == "Police/Medical" assert state.attributes.get("triggered_source") == "Police/Medical"
assert mock_request.call_count == 1
async def test_triggered_carbon_monoxide(hass): async def test_triggered_carbon_monoxide(hass: HomeAssistant) -> None:
"""Test triggered by carbon monoxide.""" """Test triggered by carbon monoxide."""
responses = [RESPONSE_TRIGGERED_CARBON_MONOXIDE] responses = [RESPONSE_TRIGGERED_CARBON_MONOXIDE]
with patch( with patch(TOTALCONNECT_REQUEST, side_effect=responses) as mock_request:
"homeassistant.components.totalconnect.TotalConnectClient.TotalConnectClient.request",
side_effect=responses,
):
await setup_platform(hass, ALARM_DOMAIN) await setup_platform(hass, ALARM_DOMAIN)
state = hass.states.get(ENTITY_ID) state = hass.states.get(ENTITY_ID)
assert state.state == STATE_ALARM_TRIGGERED assert state.state == STATE_ALARM_TRIGGERED
assert state.attributes.get("triggered_source") == "Carbon Monoxide" assert state.attributes.get("triggered_source") == "Carbon Monoxide"
assert mock_request.call_count == 1
async def test_armed_custom(hass): async def test_armed_custom(hass: HomeAssistant) -> None:
"""Test armed custom.""" """Test armed custom."""
responses = [RESPONSE_ARMED_CUSTOM] responses = [RESPONSE_ARMED_CUSTOM]
with patch( with patch(TOTALCONNECT_REQUEST, side_effect=responses) as mock_request:
"homeassistant.components.totalconnect.TotalConnectClient.TotalConnectClient.request",
side_effect=responses,
):
await setup_platform(hass, ALARM_DOMAIN) await setup_platform(hass, ALARM_DOMAIN)
state = hass.states.get(ENTITY_ID) assert hass.states.get(ENTITY_ID).state == STATE_ALARM_ARMED_CUSTOM_BYPASS
assert state.state == STATE_ALARM_ARMED_CUSTOM_BYPASS assert mock_request.call_count == 1
async def test_unknown(hass): async def test_unknown(hass: HomeAssistant) -> None:
"""Test unknown arm status.""" """Test unknown arm status."""
responses = [RESPONSE_UNKNOWN] responses = [RESPONSE_UNKNOWN]
with patch( with patch(TOTALCONNECT_REQUEST, side_effect=responses) as mock_request:
"homeassistant.components.totalconnect.TotalConnectClient.TotalConnectClient.request",
side_effect=responses,
):
await setup_platform(hass, ALARM_DOMAIN) await setup_platform(hass, ALARM_DOMAIN)
state = hass.states.get(ENTITY_ID) assert hass.states.get(ENTITY_ID).state == STATE_UNAVAILABLE
assert state.state == "unknown" assert mock_request.call_count == 1

View file

@ -11,8 +11,11 @@ from .common import (
CONFIG_DATA_NO_USERCODES, CONFIG_DATA_NO_USERCODES,
RESPONSE_AUTHENTICATE, RESPONSE_AUTHENTICATE,
RESPONSE_DISARMED, RESPONSE_DISARMED,
RESPONSE_GET_ZONE_DETAILS_SUCCESS,
RESPONSE_PARTITION_DETAILS,
RESPONSE_SUCCESS, RESPONSE_SUCCESS,
RESPONSE_USER_CODE_INVALID, RESPONSE_USER_CODE_INVALID,
TOTALCONNECT_REQUEST,
USERNAME, USERNAME,
) )
@ -37,18 +40,14 @@ async def test_user_show_locations(hass):
# user/pass provided, so check if valid then ask for usercodes on locations form # user/pass provided, so check if valid then ask for usercodes on locations form
responses = [ responses = [
RESPONSE_AUTHENTICATE, RESPONSE_AUTHENTICATE,
RESPONSE_PARTITION_DETAILS,
RESPONSE_GET_ZONE_DETAILS_SUCCESS,
RESPONSE_DISARMED, RESPONSE_DISARMED,
RESPONSE_USER_CODE_INVALID, RESPONSE_USER_CODE_INVALID,
RESPONSE_SUCCESS, RESPONSE_SUCCESS,
] ]
with patch("zeep.Client", autospec=True), patch( with patch(TOTALCONNECT_REQUEST, side_effect=responses,) as mock_request, patch(
"homeassistant.components.totalconnect.TotalConnectClient.TotalConnectClient.request",
side_effect=responses,
) as mock_request, patch(
"homeassistant.components.totalconnect.TotalConnectClient.TotalConnectClient.get_zone_details",
return_value=True,
), patch(
"homeassistant.components.totalconnect.async_setup_entry", return_value=True "homeassistant.components.totalconnect.async_setup_entry", return_value=True
): ):
@ -61,8 +60,8 @@ async def test_user_show_locations(hass):
# first it should show the locations form # first it should show the locations form
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "locations" assert result["step_id"] == "locations"
# client should have sent two requests, authenticate and get status # client should have sent four requests for init
assert mock_request.call_count == 2 assert mock_request.call_count == 4
# user enters an invalid usercode # user enters an invalid usercode
result2 = await hass.config_entries.flow.async_configure( result2 = await hass.config_entries.flow.async_configure(
@ -71,8 +70,8 @@ async def test_user_show_locations(hass):
) )
assert result2["type"] == data_entry_flow.RESULT_TYPE_FORM assert result2["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result2["step_id"] == "locations" assert result2["step_id"] == "locations"
# client should have sent 3rd request to validate usercode # client should have sent 5th request to validate usercode
assert mock_request.call_count == 3 assert mock_request.call_count == 5
# user enters a valid usercode # user enters a valid usercode
result3 = await hass.config_entries.flow.async_configure( result3 = await hass.config_entries.flow.async_configure(
@ -81,7 +80,7 @@ async def test_user_show_locations(hass):
) )
assert result3["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY assert result3["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
# client should have sent another request to validate usercode # client should have sent another request to validate usercode
assert mock_request.call_count == 4 assert mock_request.call_count == 6
async def test_abort_if_already_setup(hass): async def test_abort_if_already_setup(hass):
@ -94,7 +93,7 @@ async def test_abort_if_already_setup(hass):
# Should fail, same USERNAME (flow) # Should fail, same USERNAME (flow)
with patch( with patch(
"homeassistant.components.totalconnect.config_flow.TotalConnectClient.TotalConnectClient" "homeassistant.components.totalconnect.config_flow.TotalConnectClient"
) as client_mock: ) as client_mock:
client_mock.return_value.is_valid_credentials.return_value = True client_mock.return_value.is_valid_credentials.return_value = True
result = await hass.config_entries.flow.async_init( result = await hass.config_entries.flow.async_init(
@ -110,7 +109,7 @@ async def test_abort_if_already_setup(hass):
async def test_login_failed(hass): async def test_login_failed(hass):
"""Test when we have errors during login.""" """Test when we have errors during login."""
with patch( with patch(
"homeassistant.components.totalconnect.config_flow.TotalConnectClient.TotalConnectClient" "homeassistant.components.totalconnect.config_flow.TotalConnectClient"
) as client_mock: ) as client_mock:
client_mock.return_value.is_valid_credentials.return_value = False client_mock.return_value.is_valid_credentials.return_value = False
result = await hass.config_entries.flow.async_init( result = await hass.config_entries.flow.async_init(
@ -139,7 +138,7 @@ async def test_reauth(hass):
assert result["step_id"] == "reauth_confirm" assert result["step_id"] == "reauth_confirm"
with patch( with patch(
"homeassistant.components.totalconnect.config_flow.TotalConnectClient.TotalConnectClient" "homeassistant.components.totalconnect.config_flow.TotalConnectClient"
) as client_mock, patch( ) as client_mock, patch(
"homeassistant.components.totalconnect.async_setup_entry", return_value=True "homeassistant.components.totalconnect.async_setup_entry", return_value=True
): ):

View file

@ -19,7 +19,7 @@ async def test_reauth_started(hass):
mock_entry.add_to_hass(hass) mock_entry.add_to_hass(hass)
with patch( with patch(
"homeassistant.components.totalconnect.TotalConnectClient.TotalConnectClient", "homeassistant.components.totalconnect.TotalConnectClient",
autospec=True, autospec=True,
) as mock_client: ) as mock_client:
mock_client.return_value.is_valid_credentials.return_value = False mock_client.return_value.is_valid_credentials.return_value = False