From baaaf3d2bc4fe11533f02fd14a349033e1dbc5f4 Mon Sep 17 00:00:00 2001 From: Austin Mroczek <austin@mroczek.org> Date: Wed, 27 Oct 2021 10:15:13 -0700 Subject: [PATCH] Add multi-partition support for TotalConnect (#55429) --- .../components/totalconnect/__init__.py | 52 ++++- .../totalconnect/alarm_control_panel.py | 175 ++++++++++----- .../components/totalconnect/binary_sensor.py | 9 +- .../components/totalconnect/config_flow.py | 6 +- .../components/totalconnect/manifest.json | 2 +- requirements_all.txt | 2 +- requirements_test_all.txt | 2 +- tests/components/totalconnect/common.py | 211 ++++++++++++------ .../totalconnect/test_alarm_control_panel.py | 207 +++++++++-------- .../totalconnect/test_config_flow.py | 29 ++- tests/components/totalconnect/test_init.py | 2 +- 11 files changed, 452 insertions(+), 245 deletions(-) diff --git a/homeassistant/components/totalconnect/__init__.py b/homeassistant/components/totalconnect/__init__.py index 2183448eed7..bd1f693fd07 100644 --- a/homeassistant/components/totalconnect/__init__.py +++ b/homeassistant/components/totalconnect/__init__.py @@ -1,17 +1,25 @@ """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.const import CONF_PASSWORD, CONF_USERNAME from homeassistant.core import HomeAssistant from homeassistant.exceptions import ConfigEntryAuthFailed import homeassistant.helpers.config_validation as cv +from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed from .const import CONF_USERCODES, DOMAIN PLATFORMS = ["alarm_control_panel", "binary_sensor"] CONFIG_SCHEMA = cv.deprecated(DOMAIN) +SCAN_INTERVAL = timedelta(seconds=30) +_LOGGER = logging.getLogger(__name__) 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] usercodes = {int(code): temp_codes[code] for code in temp_codes} client = await hass.async_add_executor_job( - TotalConnectClient.TotalConnectClient, username, password, usercodes + TotalConnectClient, username, password, usercodes ) if not client.is_valid_credentials(): raise ConfigEntryAuthFailed("TotalConnect authentication failed") - hass.data.setdefault(DOMAIN, {}) - hass.data[DOMAIN][entry.entry_id] = client + coordinator = TotalConnectDataUpdateCoordinator(hass, client) + await coordinator.async_config_entry_first_refresh() + hass.data.setdefault(DOMAIN, {}) + hass.data[DOMAIN][entry.entry_id] = coordinator hass.config_entries.async_setup_platforms(entry, PLATFORMS) - return True @@ -48,3 +57,36 @@ async def async_unload_entry(hass, entry: ConfigEntry): hass.data[DOMAIN].pop(entry.entry_id) 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 diff --git a/homeassistant/components/totalconnect/alarm_control_panel.py b/homeassistant/components/totalconnect/alarm_control_panel.py index 7e88322eca1..36f7e5297f2 100644 --- a/homeassistant/components/totalconnect/alarm_control_panel.py +++ b/homeassistant/components/totalconnect/alarm_control_panel.py @@ -1,6 +1,9 @@ """Interfaces with TotalConnect alarm control panels.""" import logging +from total_connect_client import ArmingHelper +from total_connect_client.exceptions import BadResultCodeError + import homeassistant.components.alarm_control_panel as alarm from homeassistant.components.alarm_control_panel.const import ( SUPPORT_ALARM_ARM_AWAY, @@ -18,35 +21,60 @@ from homeassistant.const import ( STATE_ALARM_TRIGGERED, ) from homeassistant.exceptions import HomeAssistantError +from homeassistant.helpers.update_coordinator import CoordinatorEntity from .const import DOMAIN +_LOGGER = logging.getLogger(__name__) + async def async_setup_entry(hass, entry, async_add_entities) -> None: """Set up TotalConnect alarm panels based on a config entry.""" 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 - 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) -class TotalConnectAlarm(alarm.AlarmControlPanelEntity): +class TotalConnectAlarm(CoordinatorEntity, alarm.AlarmControlPanelEntity): """Represent an TotalConnect status.""" - def __init__(self, name, location_id, client): + def __init__(self, coordinator, name, location_id, partition_id): """Initialize the TotalConnect status.""" - self._name = name + super().__init__(coordinator) self._location_id = location_id - self._unique_id = str(location_id) - self._client = client + self._location = coordinator.client.locations[location_id] + 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._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 def name(self): """Return the name of the device.""" @@ -58,81 +86,118 @@ class TotalConnectAlarm(alarm.AlarmControlPanelEntity): return self._unique_id @property - def state(self): - """Return the state of the device.""" - return self._state - - @property - def supported_features(self) -> int: - """Return the list of supported features.""" - return SUPPORT_ALARM_ARM_HOME | SUPPORT_ALARM_ARM_AWAY | SUPPORT_ALARM_ARM_NIGHT + def device_info(self): + """Return device info.""" + return { + "identifiers": {(DOMAIN, self._device.serial_number)}, + "name": self._device.name, + } @property - def extra_state_attributes(self): - """Return the state attributes of the device.""" - return self._extra_state_attributes - - def update(self): + def state(self): """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(), + "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._client.locations[self._location_id].is_disarmed(): + if self._partition.arming_state.is_disarmed(): state = STATE_ALARM_DISARMED - elif self._client.locations[self._location_id].is_armed_night(): + elif self._partition.arming_state.is_armed_night(): state = STATE_ALARM_ARMED_NIGHT - elif self._client.locations[self._location_id].is_armed_home(): + elif self._partition.arming_state.is_armed_home(): state = STATE_ALARM_ARMED_HOME - elif self._client.locations[self._location_id].is_armed_away(): + elif self._partition.arming_state.is_armed_away(): state = STATE_ALARM_ARMED_AWAY - elif self._client.locations[self._location_id].is_armed_custom_bypass(): + elif self._partition.arming_state.is_armed_custom_bypass(): state = STATE_ALARM_ARMED_CUSTOM_BYPASS - elif self._client.locations[self._location_id].is_arming(): + elif self._partition.arming_state.is_arming(): state = STATE_ALARM_ARMING - elif self._client.locations[self._location_id].is_disarming(): + elif self._partition.arming_state.is_disarming(): state = STATE_ALARM_DISARMING - elif self._client.locations[self._location_id].is_triggered_police(): + elif self._partition.arming_state.is_triggered_police(): state = STATE_ALARM_TRIGGERED attr["triggered_source"] = "Police/Medical" - elif self._client.locations[self._location_id].is_triggered_fire(): + elif self._partition.arming_state.is_triggered_fire(): state = STATE_ALARM_TRIGGERED attr["triggered_source"] = "Fire/Smoke" - elif self._client.locations[self._location_id].is_triggered_gas(): + elif self._partition.arming_state.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.""" - if self._client.disarm(self._location_id) is not True: - raise HomeAssistantError(f"TotalConnect failed to disarm {self._name}.") + return self._state - def alarm_arm_home(self, code=None): - """Send arm home command.""" - if self._client.arm_stay(self._location_id) is not True: - raise HomeAssistantError(f"TotalConnect failed to arm home {self._name}.") + @property + def supported_features(self) -> int: + """Return the list of supported features.""" + return SUPPORT_ALARM_ARM_HOME | SUPPORT_ALARM_ARM_AWAY | SUPPORT_ALARM_ARM_NIGHT - def alarm_arm_away(self, code=None): - """Send arm away command.""" - if self._client.arm_away(self._location_id) is not True: - raise HomeAssistantError(f"TotalConnect failed to arm away {self._name}.") + @property + def extra_state_attributes(self): + """Return the state attributes of the device.""" + return self._extra_state_attributes - def alarm_arm_night(self, code=None): + async def async_alarm_disarm(self, code=None): + """Send disarm command.""" + await self.hass.async_add_executor_job(self._disarm) + await self.coordinator.async_request_refresh() + + 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.""" + await self.hass.async_add_executor_job(self._arm_home) + await self.coordinator.async_request_refresh() + + 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.""" + await self.hass.async_add_executor_job(self._arm_away) + await self.coordinator.async_request_refresh() + + 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.""" - if self._client.arm_stay_night(self._location_id) is not True: - raise HomeAssistantError(f"TotalConnect failed to arm night {self._name}.") + await self.hass.async_add_executor_job(self._arm_night) + 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 diff --git a/homeassistant/components/totalconnect/binary_sensor.py b/homeassistant/components/totalconnect/binary_sensor.py index ef02c5d1fd3..e37482cf1e1 100644 --- a/homeassistant/components/totalconnect/binary_sensor.py +++ b/homeassistant/components/totalconnect/binary_sensor.py @@ -2,6 +2,8 @@ from homeassistant.components.binary_sensor import ( DEVICE_CLASS_DOOR, DEVICE_CLASS_GAS, + DEVICE_CLASS_MOTION, + DEVICE_CLASS_SAFETY, DEVICE_CLASS_SMOKE, 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.""" 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 zone_id, zone in location.zones.items(): @@ -70,6 +72,10 @@ class TotalConnectBinarySensor(BinarySensorEntity): return DEVICE_CLASS_SMOKE if self._zone.is_type_carbon_monoxide(): 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 @property @@ -80,5 +86,6 @@ class TotalConnectBinarySensor(BinarySensorEntity): "location_id": self._location_id, "low_battery": self._is_low_battery, "tampered": self._is_tampered, + "partition": self._zone.partition, } return attributes diff --git a/homeassistant/components/totalconnect/config_flow.py b/homeassistant/components/totalconnect/config_flow.py index 8f39346f47e..7ba96f11a1e 100644 --- a/homeassistant/components/totalconnect/config_flow.py +++ b/homeassistant/components/totalconnect/config_flow.py @@ -1,5 +1,5 @@ """Config flow for the Total Connect component.""" -from total_connect_client import TotalConnectClient +from total_connect_client.client import TotalConnectClient import voluptuous as vol from homeassistant import config_entries @@ -37,7 +37,7 @@ class TotalConnectConfigFlow(config_entries.ConfigFlow, domain=DOMAIN): self._abort_if_unique_id_configured() client = await self.hass.async_add_executor_job( - TotalConnectClient.TotalConnectClient, username, password, None + TotalConnectClient, username, password, None ) if client.is_valid_credentials(): @@ -130,7 +130,7 @@ class TotalConnectConfigFlow(config_entries.ConfigFlow, domain=DOMAIN): ) client = await self.hass.async_add_executor_job( - TotalConnectClient.TotalConnectClient, + TotalConnectClient, self.username, user_input[CONF_PASSWORD], self.usercodes, diff --git a/homeassistant/components/totalconnect/manifest.json b/homeassistant/components/totalconnect/manifest.json index 3bfba56f92c..25fb2fd2c75 100644 --- a/homeassistant/components/totalconnect/manifest.json +++ b/homeassistant/components/totalconnect/manifest.json @@ -2,7 +2,7 @@ "domain": "totalconnect", "name": "Total Connect", "documentation": "https://www.home-assistant.io/integrations/totalconnect", - "requirements": ["total_connect_client==0.57"], + "requirements": ["total_connect_client==2021.8.3"], "dependencies": [], "codeowners": ["@austinmroczek"], "config_flow": true, diff --git a/requirements_all.txt b/requirements_all.txt index 1c9178b070c..11789c4f06b 100644 --- a/requirements_all.txt +++ b/requirements_all.txt @@ -2314,7 +2314,7 @@ todoist-python==8.0.0 toonapi==0.2.1 # homeassistant.components.totalconnect -total_connect_client==0.57 +total_connect_client==2021.8.3 # homeassistant.components.tplink_lte tp-connected==0.0.4 diff --git a/requirements_test_all.txt b/requirements_test_all.txt index 411cd336cab..112605b1f76 100644 --- a/requirements_test_all.txt +++ b/requirements_test_all.txt @@ -1330,7 +1330,7 @@ tesla-powerwall==0.3.12 toonapi==0.2.1 # homeassistant.components.totalconnect -total_connect_client==0.57 +total_connect_client==2021.8.3 # homeassistant.components.transmission transmissionrpc==0.11 diff --git a/tests/components/totalconnect/common.py b/tests/components/totalconnect/common.py index b092a028c0b..29278d33273 100644 --- a/tests/components/totalconnect/common.py +++ b/tests/components/totalconnect/common.py @@ -1,7 +1,9 @@ """Common methods used across tests for TotalConnect.""" 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.const import CONF_PASSWORD, CONF_USERNAME @@ -11,13 +13,24 @@ from tests.common import MockConfigEntry 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 = { "LocationID": LOCATION_ID, "LocationName": "test", "SecurityDeviceID": "987654", "PhotoURL": "http://www.example.com/some/path/to/file.jpg", "LocationModuleFlags": "Security=1,Video=0,Automation=0,GPS=0,VideoPIR=0", - "DeviceList": None, + "DeviceList": {"DeviceInfoBasic": DEVICE_LIST}, } LOCATIONS = {"LocationInfoBasic": [LOCATION_INFO_BASIC_NORMAL]} @@ -31,7 +44,7 @@ USER = { } RESPONSE_AUTHENTICATE = { - "ResultCode": TotalConnectClient.TotalConnectClient.SUCCESS, + "ResultCode": TotalConnectClient.SUCCESS, "SessionID": 1, "Locations": LOCATIONS, "ModuleFlags": MODULE_FLAGS, @@ -39,58 +52,68 @@ RESPONSE_AUTHENTICATE = { } RESPONSE_AUTHENTICATE_FAILED = { - "ResultCode": TotalConnectClient.TotalConnectClient.BAD_USER_OR_PASSWORD, + "ResultCode": TotalConnectClient.BAD_USER_OR_PASSWORD, "ResultData": "test bad authentication", } PARTITION_DISARMED = { "PartitionID": "1", - "ArmingState": TotalConnectClient.TotalConnectLocation.DISARMED, + "ArmingState": ArmingState.DISARMED, +} + +PARTITION_DISARMED2 = { + "PartitionID": "2", + "ArmingState": ArmingState.DISARMED, } PARTITION_ARMED_STAY = { "PartitionID": "1", - "ArmingState": TotalConnectClient.TotalConnectLocation.ARMED_STAY, + "ArmingState": ArmingState.ARMED_STAY, +} + +PARTITION_ARMED_STAY2 = { + "PartitionID": "2", + "ArmingState": ArmingState.DISARMED, } PARTITION_ARMED_AWAY = { "PartitionID": "1", - "ArmingState": TotalConnectClient.TotalConnectLocation.ARMED_AWAY, + "ArmingState": ArmingState.ARMED_AWAY, } PARTITION_ARMED_CUSTOM = { "PartitionID": "1", - "ArmingState": TotalConnectClient.TotalConnectLocation.ARMED_CUSTOM_BYPASS, + "ArmingState": ArmingState.ARMED_CUSTOM_BYPASS, } PARTITION_ARMED_NIGHT = { "PartitionID": "1", - "ArmingState": TotalConnectClient.TotalConnectLocation.ARMED_STAY_NIGHT, + "ArmingState": ArmingState.ARMED_STAY_NIGHT, } PARTITION_ARMING = { "PartitionID": "1", - "ArmingState": TotalConnectClient.TotalConnectLocation.ARMING, + "ArmingState": ArmingState.ARMING, } PARTITION_DISARMING = { "PartitionID": "1", - "ArmingState": TotalConnectClient.TotalConnectLocation.DISARMING, + "ArmingState": ArmingState.DISARMING, } PARTITION_TRIGGERED_POLICE = { "PartitionID": "1", - "ArmingState": TotalConnectClient.TotalConnectLocation.ALARMING, + "ArmingState": ArmingState.ALARMING, } PARTITION_TRIGGERED_FIRE = { "PartitionID": "1", - "ArmingState": TotalConnectClient.TotalConnectLocation.ALARMING_FIRE_SMOKE, + "ArmingState": ArmingState.ALARMING_FIRE_SMOKE, } PARTITION_TRIGGERED_CARBON_MONOXIDE = { "PartitionID": "1", - "ArmingState": TotalConnectClient.TotalConnectLocation.ALARMING_CARBON_MONOXIDE, + "ArmingState": ArmingState.ALARMING_CARBON_MONOXIDE, } PARTITION_UNKNOWN = { @@ -99,17 +122,17 @@ PARTITION_UNKNOWN = { } -PARTITION_INFO_DISARMED = {0: PARTITION_DISARMED} -PARTITION_INFO_ARMED_STAY = {0: PARTITION_ARMED_STAY} -PARTITION_INFO_ARMED_AWAY = {0: PARTITION_ARMED_AWAY} -PARTITION_INFO_ARMED_CUSTOM = {0: PARTITION_ARMED_CUSTOM} -PARTITION_INFO_ARMED_NIGHT = {0: PARTITION_ARMED_NIGHT} -PARTITION_INFO_ARMING = {0: PARTITION_ARMING} -PARTITION_INFO_DISARMING = {0: PARTITION_DISARMING} -PARTITION_INFO_TRIGGERED_POLICE = {0: PARTITION_TRIGGERED_POLICE} -PARTITION_INFO_TRIGGERED_FIRE = {0: PARTITION_TRIGGERED_FIRE} -PARTITION_INFO_TRIGGERED_CARBON_MONOXIDE = {0: PARTITION_TRIGGERED_CARBON_MONOXIDE} -PARTITION_INFO_UNKNOWN = {0: PARTITION_UNKNOWN} +PARTITION_INFO_DISARMED = [PARTITION_DISARMED, PARTITION_DISARMED2] +PARTITION_INFO_ARMED_STAY = [PARTITION_ARMED_STAY, PARTITION_ARMED_STAY2] +PARTITION_INFO_ARMED_AWAY = [PARTITION_ARMED_AWAY] +PARTITION_INFO_ARMED_CUSTOM = [PARTITION_ARMED_CUSTOM] +PARTITION_INFO_ARMED_NIGHT = [PARTITION_ARMED_NIGHT] +PARTITION_INFO_ARMING = [PARTITION_ARMING] +PARTITION_INFO_DISARMING = [PARTITION_DISARMING] +PARTITION_INFO_TRIGGERED_POLICE = [PARTITION_TRIGGERED_POLICE] +PARTITION_INFO_TRIGGERED_FIRE = [PARTITION_TRIGGERED_FIRE] +PARTITION_INFO_TRIGGERED_CARBON_MONOXIDE = [PARTITION_TRIGGERED_CARBON_MONOXIDE] +PARTITION_INFO_UNKNOWN = [PARTITION_UNKNOWN] PARTITIONS_DISARMED = {"PartitionInfo": PARTITION_INFO_DISARMED} PARTITIONS_ARMED_STAY = {"PartitionInfo": PARTITION_INFO_ARMED_STAY} @@ -128,7 +151,7 @@ PARTITIONS_UNKNOWN = {"PartitionInfo": PARTITION_INFO_UNKNOWN} ZONE_NORMAL = { "ZoneID": "1", "ZoneDescription": "Normal", - "ZoneStatus": TotalConnectClient.ZONE_STATUS_NORMAL, + "ZoneStatus": ZoneStatus.NORMAL, "PartitionId": "1", } @@ -176,46 +199,74 @@ METADATA_TRIGGERED_CARBON_MONOXIDE["Partitions"] = PARTITIONS_TRIGGERED_CARBON_M METADATA_UNKNOWN = METADATA_DISARMED.copy() METADATA_UNKNOWN["Partitions"] = PARTITIONS_UNKNOWN -RESPONSE_DISARMED = {"ResultCode": 0, "PanelMetadataAndStatus": METADATA_DISARMED} -RESPONSE_ARMED_STAY = {"ResultCode": 0, "PanelMetadataAndStatus": METADATA_ARMED_STAY} -RESPONSE_ARMED_AWAY = {"ResultCode": 0, "PanelMetadataAndStatus": METADATA_ARMED_AWAY} +RESPONSE_DISARMED = { + "ResultCode": 0, + "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 = { "ResultCode": 0, "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 = { "ResultCode": 0, "PanelMetadataAndStatus": METADATA_TRIGGERED_POLICE, + "ArmingState": ArmingState.ALARMING, } RESPONSE_TRIGGERED_FIRE = { "ResultCode": 0, "PanelMetadataAndStatus": METADATA_TRIGGERED_FIRE, + "ArmingState": ArmingState.ALARMING_FIRE_SMOKE, } RESPONSE_TRIGGERED_CARBON_MONOXIDE = { "ResultCode": 0, "PanelMetadataAndStatus": METADATA_TRIGGERED_CARBON_MONOXIDE, + "ArmingState": ArmingState.ALARMING_CARBON_MONOXIDE, } -RESPONSE_UNKNOWN = {"ResultCode": 0, "PanelMetadataAndStatus": METADATA_UNKNOWN} - -RESPONSE_ARM_SUCCESS = {"ResultCode": TotalConnectClient.TotalConnectClient.ARM_SUCCESS} -RESPONSE_ARM_FAILURE = { - "ResultCode": TotalConnectClient.TotalConnectClient.COMMAND_FAILED -} -RESPONSE_DISARM_SUCCESS = { - "ResultCode": TotalConnectClient.TotalConnectClient.DISARM_SUCCESS +RESPONSE_UNKNOWN = { + "ResultCode": 0, + "PanelMetadataAndStatus": METADATA_UNKNOWN, + "ArmingState": ArmingState.DISARMED, } + +RESPONSE_ARM_SUCCESS = {"ResultCode": TotalConnectClient.ARM_SUCCESS} +RESPONSE_ARM_FAILURE = {"ResultCode": TotalConnectClient.COMMAND_FAILED} +RESPONSE_DISARM_SUCCESS = {"ResultCode": TotalConnectClient.DISARM_SUCCESS} RESPONSE_DISARM_FAILURE = { - "ResultCode": TotalConnectClient.TotalConnectClient.COMMAND_FAILED, + "ResultCode": TotalConnectClient.COMMAND_FAILED, "ResultData": "Command Failed", } RESPONSE_USER_CODE_INVALID = { - "ResultCode": TotalConnectClient.TotalConnectClient.USER_CODE_INVALID, + "ResultCode": TotalConnectClient.USER_CODE_INVALID, "ResultData": "testing user code invalid", } -RESPONSE_SUCCESS = {"ResultCode": TotalConnectClient.TotalConnectClient.SUCCESS} +RESPONSE_SUCCESS = {"ResultCode": TotalConnectClient.SUCCESS} USERNAME = "username@me.com" PASSWORD = "password" @@ -227,40 +278,72 @@ CONFIG_DATA = { } CONFIG_DATA_NO_USERCODES = {CONF_USERNAME: USERNAME, CONF_PASSWORD: PASSWORD} +PARTITION_DETAILS_1 = { + "PartitionID": 1, + "ArmingState": ArmingState.DISARMED.value, + "PartitionName": "Test1", +} -USERNAME = "username@me.com" -PASSWORD = "password" -USERCODES = {123456: "7890"} -CONFIG_DATA = { - CONF_USERNAME: USERNAME, - CONF_PASSWORD: PASSWORD, - CONF_USERCODES: USERCODES, +PARTITION_DETAILS_2 = { + "PartitionID": 2, + "ArmingState": ArmingState.DISARMED.value, + "PartitionName": "Test2", } -CONFIG_DATA_NO_USERCODES = {CONF_USERNAME: USERNAME, CONF_PASSWORD: PASSWORD} + +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): """Set up the TotalConnect platform.""" # first set up a config entry and add it to hass - mock_entry = MockConfigEntry( - domain=DOMAIN, - data=CONFIG_DATA, - ) + mock_entry = MockConfigEntry(domain=DOMAIN, data=CONFIG_DATA) 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( - "zeep.Client", autospec=True - ), patch( - "homeassistant.components.totalconnect.TotalConnectClient.TotalConnectClient.request", + TOTALCONNECT_REQUEST, side_effect=responses, - ) as mock_request, patch( - "homeassistant.components.totalconnect.TotalConnectClient.TotalConnectClient.get_zone_details", - return_value=True, - ): + ) as mock_request: 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() return mock_entry diff --git a/tests/components/totalconnect/test_alarm_control_panel.py b/tests/components/totalconnect/test_alarm_control_panel.py index a77adea5e27..368ebf93a04 100644 --- a/tests/components/totalconnect/test_alarm_control_panel.py +++ b/tests/components/totalconnect/test_alarm_control_panel.py @@ -1,4 +1,5 @@ """Tests for the TotalConnect alarm control panel device.""" +from datetime import timedelta from unittest.mock import patch import pytest @@ -19,8 +20,11 @@ from homeassistant.const import ( STATE_ALARM_DISARMED, STATE_ALARM_DISARMING, STATE_ALARM_TRIGGERED, + STATE_UNAVAILABLE, ) +from homeassistant.core import HomeAssistant from homeassistant.exceptions import HomeAssistantError +from homeassistant.util import dt from .common import ( LOCATION_ID, @@ -41,18 +45,23 @@ from .common import ( RESPONSE_TRIGGERED_POLICE, RESPONSE_UNKNOWN, RESPONSE_USER_CODE_INVALID, + TOTALCONNECT_REQUEST, setup_platform, ) +from tests.common import async_fire_time_changed + ENTITY_ID = "alarm_control_panel.test" +ENTITY_ID_2 = "alarm_control_panel.test_partition_2" CODE = "-1" 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.""" with patch( - "homeassistant.components.totalconnect.TotalConnectClient.TotalConnectClient.request", + "homeassistant.components.totalconnect.TotalConnectClient.request", return_value=RESPONSE_DISARMED, ) as mock_request: 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() 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 + 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.""" responses = [RESPONSE_DISARMED, RESPONSE_ARM_SUCCESS, RESPONSE_ARMED_STAY] - with patch( - "homeassistant.components.totalconnect.TotalConnectClient.TotalConnectClient.request", - side_effect=responses, - ): + with patch(TOTALCONNECT_REQUEST, side_effect=responses) as mock_request: await setup_platform(hass, ALARM_DOMAIN) 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( 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() + assert mock_request.call_count == 3 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.""" - responses = [RESPONSE_DISARMED, RESPONSE_ARM_FAILURE, RESPONSE_DISARMED] - with patch( - "homeassistant.components.totalconnect.TotalConnectClient.TotalConnectClient.request", - side_effect=responses, - ): + responses = [RESPONSE_DISARMED, RESPONSE_ARM_FAILURE] + with patch(TOTALCONNECT_REQUEST, side_effect=responses) as mock_request: await setup_platform(hass, ALARM_DOMAIN) assert hass.states.get(ENTITY_ID).state == STATE_ALARM_DISARMED + assert mock_request.call_count == 1 with pytest.raises(HomeAssistantError) as err: await hass.services.async_call( @@ -102,17 +118,16 @@ async def test_arm_home_failure(hass): await hass.async_block_till_done() assert f"{err.value}" == "TotalConnect failed to arm home test." 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.""" - responses = [RESPONSE_DISARMED, RESPONSE_USER_CODE_INVALID, RESPONSE_DISARMED] - with patch( - "homeassistant.components.totalconnect.TotalConnectClient.TotalConnectClient.request", - side_effect=responses, - ): + responses = [RESPONSE_DISARMED, RESPONSE_USER_CODE_INVALID] + with patch(TOTALCONNECT_REQUEST, side_effect=responses) as mock_request: await setup_platform(hass, ALARM_DOMAIN) assert hass.states.get(ENTITY_ID).state == STATE_ALARM_DISARMED + assert mock_request.call_count == 1 with pytest.raises(HomeAssistantError) as err: await hass.services.async_call( @@ -121,34 +136,35 @@ async def test_arm_home_invalid_usercode(hass): await hass.async_block_till_done() assert f"{err.value}" == "TotalConnect failed to arm home test." 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.""" responses = [RESPONSE_DISARMED, RESPONSE_ARM_SUCCESS, RESPONSE_ARMED_AWAY] - with patch( - "homeassistant.components.totalconnect.TotalConnectClient.TotalConnectClient.request", - side_effect=responses, - ): + with patch(TOTALCONNECT_REQUEST, side_effect=responses) as mock_request: await setup_platform(hass, ALARM_DOMAIN) assert hass.states.get(ENTITY_ID).state == STATE_ALARM_DISARMED + assert mock_request.call_count == 1 await hass.services.async_call( 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() + assert mock_request.call_count == 3 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.""" - responses = [RESPONSE_DISARMED, RESPONSE_ARM_FAILURE, RESPONSE_DISARMED] - with patch( - "homeassistant.components.totalconnect.TotalConnectClient.TotalConnectClient.request", - side_effect=responses, - ): + responses = [RESPONSE_DISARMED, RESPONSE_ARM_FAILURE] + with patch(TOTALCONNECT_REQUEST, side_effect=responses) as mock_request: await setup_platform(hass, ALARM_DOMAIN) assert hass.states.get(ENTITY_ID).state == STATE_ALARM_DISARMED + assert mock_request.call_count == 1 with pytest.raises(HomeAssistantError) as err: await hass.services.async_call( @@ -157,34 +173,35 @@ async def test_arm_away_failure(hass): await hass.async_block_till_done() assert f"{err.value}" == "TotalConnect failed to arm away test." 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.""" responses = [RESPONSE_ARMED_AWAY, RESPONSE_DISARM_SUCCESS, RESPONSE_DISARMED] - with patch( - "homeassistant.components.totalconnect.TotalConnectClient.TotalConnectClient.request", - side_effect=responses, - ): + with patch(TOTALCONNECT_REQUEST, side_effect=responses) as mock_request: await setup_platform(hass, ALARM_DOMAIN) assert hass.states.get(ENTITY_ID).state == STATE_ALARM_ARMED_AWAY + assert mock_request.call_count == 1 await hass.services.async_call( 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_DISARMED -async def test_disarm_failure(hass): +async def test_disarm_failure(hass: HomeAssistant) -> None: """Test disarm method failure.""" - responses = [RESPONSE_ARMED_AWAY, RESPONSE_DISARM_FAILURE, RESPONSE_ARMED_AWAY] - with patch( - "homeassistant.components.totalconnect.TotalConnectClient.TotalConnectClient.request", - side_effect=responses, - ): + responses = [RESPONSE_ARMED_AWAY, RESPONSE_DISARM_FAILURE] + with patch(TOTALCONNECT_REQUEST, side_effect=responses) as mock_request: await setup_platform(hass, ALARM_DOMAIN) assert hass.states.get(ENTITY_ID).state == STATE_ALARM_ARMED_AWAY + assert mock_request.call_count == 1 with pytest.raises(HomeAssistantError) as err: await hass.services.async_call( @@ -193,17 +210,16 @@ async def test_disarm_failure(hass): await hass.async_block_till_done() assert f"{err.value}" == "TotalConnect failed to disarm test." 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.""" - responses = [RESPONSE_ARMED_AWAY, RESPONSE_USER_CODE_INVALID, RESPONSE_ARMED_AWAY] - with patch( - "homeassistant.components.totalconnect.TotalConnectClient.TotalConnectClient.request", - side_effect=responses, - ): + responses = [RESPONSE_ARMED_AWAY, RESPONSE_USER_CODE_INVALID] + with patch(TOTALCONNECT_REQUEST, side_effect=responses) as mock_request: await setup_platform(hass, ALARM_DOMAIN) assert hass.states.get(ENTITY_ID).state == STATE_ALARM_ARMED_AWAY + assert mock_request.call_count == 1 with pytest.raises(HomeAssistantError) as err: await hass.services.async_call( @@ -212,35 +228,35 @@ async def test_disarm_invalid_usercode(hass): await hass.async_block_till_done() assert f"{err.value}" == "TotalConnect failed to disarm test." 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.""" responses = [RESPONSE_DISARMED, RESPONSE_ARM_SUCCESS, RESPONSE_ARMED_NIGHT] - with patch( - "homeassistant.components.totalconnect.TotalConnectClient.TotalConnectClient.request", - side_effect=responses, - ): + with patch(TOTALCONNECT_REQUEST, side_effect=responses) as mock_request: await setup_platform(hass, ALARM_DOMAIN) assert hass.states.get(ENTITY_ID).state == STATE_ALARM_DISARMED + assert mock_request.call_count == 1 await hass.services.async_call( 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_ARMED_NIGHT -async def test_arm_night_failure(hass): +async def test_arm_night_failure(hass: HomeAssistant) -> None: """Test arm night method failure.""" - responses = [RESPONSE_DISARMED, RESPONSE_ARM_FAILURE, RESPONSE_DISARMED] - with patch( - "homeassistant.components.totalconnect.TotalConnectClient.TotalConnectClient.request", - side_effect=responses, - ): + responses = [RESPONSE_DISARMED, RESPONSE_ARM_FAILURE] + with patch(TOTALCONNECT_REQUEST, side_effect=responses) as mock_request: await setup_platform(hass, ALARM_DOMAIN) assert hass.states.get(ENTITY_ID).state == STATE_ALARM_DISARMED + assert mock_request.call_count == 1 with pytest.raises(HomeAssistantError) as err: await hass.services.async_call( @@ -249,98 +265,93 @@ async def test_arm_night_failure(hass): await hass.async_block_till_done() assert f"{err.value}" == "TotalConnect failed to arm night test." 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.""" responses = [RESPONSE_DISARMED, RESPONSE_SUCCESS, RESPONSE_ARMING] - with patch( - "homeassistant.components.totalconnect.TotalConnectClient.TotalConnectClient.request", - side_effect=responses, - ): + with patch(TOTALCONNECT_REQUEST, side_effect=responses) as mock_request: await setup_platform(hass, ALARM_DOMAIN) assert hass.states.get(ENTITY_ID).state == STATE_ALARM_DISARMED + assert mock_request.call_count == 1 await hass.services.async_call( 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 -async def test_disarming(hass): +async def test_disarming(hass: HomeAssistant) -> None: """Test disarming.""" responses = [RESPONSE_ARMED_AWAY, RESPONSE_SUCCESS, RESPONSE_DISARMING] - with patch( - "homeassistant.components.totalconnect.TotalConnectClient.TotalConnectClient.request", - side_effect=responses, - ): + with patch(TOTALCONNECT_REQUEST, side_effect=responses) as mock_request: await setup_platform(hass, ALARM_DOMAIN) assert hass.states.get(ENTITY_ID).state == STATE_ALARM_ARMED_AWAY + assert mock_request.call_count == 1 await hass.services.async_call( 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 -async def test_triggered_fire(hass): +async def test_triggered_fire(hass: HomeAssistant) -> None: """Test triggered by fire.""" responses = [RESPONSE_TRIGGERED_FIRE] - with patch( - "homeassistant.components.totalconnect.TotalConnectClient.TotalConnectClient.request", - side_effect=responses, - ): + with patch(TOTALCONNECT_REQUEST, side_effect=responses) as mock_request: await setup_platform(hass, ALARM_DOMAIN) state = hass.states.get(ENTITY_ID) assert state.state == STATE_ALARM_TRIGGERED 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.""" responses = [RESPONSE_TRIGGERED_POLICE] - with patch( - "homeassistant.components.totalconnect.TotalConnectClient.TotalConnectClient.request", - side_effect=responses, - ): + with patch(TOTALCONNECT_REQUEST, side_effect=responses) as mock_request: await setup_platform(hass, ALARM_DOMAIN) state = hass.states.get(ENTITY_ID) assert state.state == STATE_ALARM_TRIGGERED 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.""" responses = [RESPONSE_TRIGGERED_CARBON_MONOXIDE] - with patch( - "homeassistant.components.totalconnect.TotalConnectClient.TotalConnectClient.request", - side_effect=responses, - ): + with patch(TOTALCONNECT_REQUEST, side_effect=responses) as mock_request: await setup_platform(hass, ALARM_DOMAIN) state = hass.states.get(ENTITY_ID) assert state.state == STATE_ALARM_TRIGGERED 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.""" responses = [RESPONSE_ARMED_CUSTOM] - with patch( - "homeassistant.components.totalconnect.TotalConnectClient.TotalConnectClient.request", - side_effect=responses, - ): + with patch(TOTALCONNECT_REQUEST, side_effect=responses) as mock_request: await setup_platform(hass, ALARM_DOMAIN) - state = hass.states.get(ENTITY_ID) - assert state.state == STATE_ALARM_ARMED_CUSTOM_BYPASS + assert hass.states.get(ENTITY_ID).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.""" responses = [RESPONSE_UNKNOWN] - with patch( - "homeassistant.components.totalconnect.TotalConnectClient.TotalConnectClient.request", - side_effect=responses, - ): + with patch(TOTALCONNECT_REQUEST, side_effect=responses) as mock_request: await setup_platform(hass, ALARM_DOMAIN) - state = hass.states.get(ENTITY_ID) - assert state.state == "unknown" + assert hass.states.get(ENTITY_ID).state == STATE_UNAVAILABLE + assert mock_request.call_count == 1 diff --git a/tests/components/totalconnect/test_config_flow.py b/tests/components/totalconnect/test_config_flow.py index 7b80996db14..20497102c6d 100644 --- a/tests/components/totalconnect/test_config_flow.py +++ b/tests/components/totalconnect/test_config_flow.py @@ -11,8 +11,11 @@ from .common import ( CONFIG_DATA_NO_USERCODES, RESPONSE_AUTHENTICATE, RESPONSE_DISARMED, + RESPONSE_GET_ZONE_DETAILS_SUCCESS, + RESPONSE_PARTITION_DETAILS, RESPONSE_SUCCESS, RESPONSE_USER_CODE_INVALID, + TOTALCONNECT_REQUEST, 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 responses = [ RESPONSE_AUTHENTICATE, + RESPONSE_PARTITION_DETAILS, + RESPONSE_GET_ZONE_DETAILS_SUCCESS, RESPONSE_DISARMED, RESPONSE_USER_CODE_INVALID, RESPONSE_SUCCESS, ] - with patch("zeep.Client", autospec=True), 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( + with patch(TOTALCONNECT_REQUEST, side_effect=responses,) as mock_request, patch( "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 assert result["type"] == data_entry_flow.RESULT_TYPE_FORM assert result["step_id"] == "locations" - # client should have sent two requests, authenticate and get status - assert mock_request.call_count == 2 + # client should have sent four requests for init + assert mock_request.call_count == 4 # user enters an invalid usercode 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["step_id"] == "locations" - # client should have sent 3rd request to validate usercode - assert mock_request.call_count == 3 + # client should have sent 5th request to validate usercode + assert mock_request.call_count == 5 # user enters a valid usercode 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 # 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): @@ -94,7 +93,7 @@ async def test_abort_if_already_setup(hass): # Should fail, same USERNAME (flow) with patch( - "homeassistant.components.totalconnect.config_flow.TotalConnectClient.TotalConnectClient" + "homeassistant.components.totalconnect.config_flow.TotalConnectClient" ) as client_mock: client_mock.return_value.is_valid_credentials.return_value = True 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): """Test when we have errors during login.""" with patch( - "homeassistant.components.totalconnect.config_flow.TotalConnectClient.TotalConnectClient" + "homeassistant.components.totalconnect.config_flow.TotalConnectClient" ) as client_mock: client_mock.return_value.is_valid_credentials.return_value = False result = await hass.config_entries.flow.async_init( @@ -139,7 +138,7 @@ async def test_reauth(hass): assert result["step_id"] == "reauth_confirm" with patch( - "homeassistant.components.totalconnect.config_flow.TotalConnectClient.TotalConnectClient" + "homeassistant.components.totalconnect.config_flow.TotalConnectClient" ) as client_mock, patch( "homeassistant.components.totalconnect.async_setup_entry", return_value=True ): diff --git a/tests/components/totalconnect/test_init.py b/tests/components/totalconnect/test_init.py index ba33d996a9b..41cd8bbae90 100644 --- a/tests/components/totalconnect/test_init.py +++ b/tests/components/totalconnect/test_init.py @@ -19,7 +19,7 @@ async def test_reauth_started(hass): mock_entry.add_to_hass(hass) with patch( - "homeassistant.components.totalconnect.TotalConnectClient.TotalConnectClient", + "homeassistant.components.totalconnect.TotalConnectClient", autospec=True, ) as mock_client: mock_client.return_value.is_valid_credentials.return_value = False -- GitLab