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

Add iSmartGate support (#39437)

Browse Source 
* Add iSmartGate support.

* Addressing PR feedback.

* More PR feedback cleanups.
This commit is contained in:
Robert Van Gorkom 2020-09-05 07:26:01 -07:00 committed by GitHub
parent 52c09396e0
commit b860caa631
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
14 changed files with 458 additions and 598 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
homeassistant/components/gogogate2/__init__.py
View File

@ -1,10 +1,12 @@
"""The gogogate2 component."""
from homeassistant.components.cover import DOMAIN as COVER_DOMAIN
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_DEVICE
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryNotReady
from .common import get_data_update_coordinator
from .const import DEVICE_TYPE_GOGOGATE2
async def async_setup(hass: HomeAssistant, base_config: dict) -> bool:
@ -14,6 +16,18 @@ async def async_setup(hass: HomeAssistant, base_config: dict) -> bool:
async def async_setup_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> bool:
"""Do setup of Gogogate2."""
# Update the config entry.
config_updates = {}
if CONF_DEVICE not in config_entry.data:
config_updates["data"] = {
**config_entry.data,
**{CONF_DEVICE: DEVICE_TYPE_GOGOGATE2},
}
if config_updates:
hass.config_entries.async_update_entry(config_entry, **config_updates)
data_update_coordinator = get_data_update_coordinator(hass, config_entry)
await data_update_coordinator.async_refresh()

34
homeassistant/components/gogogate2/common.py
View File

@ -4,16 +4,21 @@ import logging
from typing import Awaitable, Callable, NamedTuple, Optional
import async_timeout
from gogogate2_api import GogoGate2Api
from gogogate2_api.common import Door
from gogogate2_api import AbstractGateApi, GogoGate2Api, ISmartGateApi
from gogogate2_api.common import AbstractDoor
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_IP_ADDRESS, CONF_PASSWORD, CONF_USERNAME
from homeassistant.const import (
CONF_DEVICE,
CONF_IP_ADDRESS,
CONF_PASSWORD,
CONF_USERNAME,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers.debounce import Debouncer
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from .const import DATA_UPDATE_COORDINATOR, DOMAIN
from .const import DATA_UPDATE_COORDINATOR, DEVICE_TYPE_ISMARTGATE, DOMAIN
_LOGGER = logging.getLogger(__name__)
@ -23,17 +28,17 @@ class StateData(NamedTuple):
config_unique_id: str
unique_id: Optional[str]
door: Optional[Door]
door: Optional[AbstractDoor]
class GogoGateDataUpdateCoordinator(DataUpdateCoordinator):
class DeviceDataUpdateCoordinator(DataUpdateCoordinator):
"""Manages polling for state changes from the device."""
def __init__(
self,
hass: HomeAssistant,
logger: logging.Logger,
api: GogoGate2Api,
api: AbstractGateApi,
*,
name: str,
update_interval: timedelta,
@ -55,7 +60,7 @@ class GogoGateDataUpdateCoordinator(DataUpdateCoordinator):
def get_data_update_coordinator(
hass: HomeAssistant, config_entry: ConfigEntry
) -> GogoGateDataUpdateCoordinator:
) -> DeviceDataUpdateCoordinator:
"""Get an update coordinator."""
hass.data.setdefault(DOMAIN, {})
hass.data[DOMAIN].setdefault(config_entry.entry_id, {})
@ -73,7 +78,7 @@ def get_data_update_coordinator(
f"Error communicating with API: {exception}"
) from exception
config_entry_data[DATA_UPDATE_COORDINATOR] = GogoGateDataUpdateCoordinator(
config_entry_data[DATA_UPDATE_COORDINATOR] = DeviceDataUpdateCoordinator(
hass,
_LOGGER,
api,
@ -87,14 +92,19 @@ def get_data_update_coordinator(
return config_entry_data[DATA_UPDATE_COORDINATOR]
def cover_unique_id(config_entry: ConfigEntry, door: Door) -> str:
def cover_unique_id(config_entry: ConfigEntry, door: AbstractDoor) -> str:
"""Generate a cover entity unique id."""
return f"{config_entry.unique_id}_{door.door_id}"
def get_api(config_data: dict) -> GogoGate2Api:
def get_api(config_data: dict) -> AbstractGateApi:
"""Get an api object for config data."""
return GogoGate2Api(
gate_class = GogoGate2Api
if config_data[CONF_DEVICE] == DEVICE_TYPE_ISMARTGATE:
gate_class = ISmartGateApi
return gate_class(
config_data[CONF_IP_ADDRESS],
config_data[CONF_USERNAME],
config_data[CONF_PASSWORD],

49
homeassistant/components/gogogate2/config_flow.py
View File

@ -1,16 +1,23 @@
"""Config flow for Gogogate2."""
import dataclasses
import logging
import re
from gogogate2_api.common import ApiError
from gogogate2_api.const import ApiErrorCode
from gogogate2_api.common import AbstractInfoResponse, ApiError
from gogogate2_api.const import GogoGate2ApiErrorCode, ISmartGateApiErrorCode
import voluptuous as vol
from homeassistant import config_entries
from homeassistant.config_entries import SOURCE_IMPORT, ConfigFlow
from homeassistant.const import CONF_IP_ADDRESS, CONF_PASSWORD, CONF_USERNAME
from homeassistant.const import (
CONF_DEVICE,
CONF_IP_ADDRESS,
CONF_PASSWORD,
CONF_USERNAME,
)
from .common import get_api
from .const import DEVICE_TYPE_GOGOGATE2, DEVICE_TYPE_ISMARTGATE
from .const import DOMAIN # pylint: disable=unused-import
_LOGGER = logging.getLogger(__name__)
@ -36,15 +43,35 @@ class Gogogate2FlowHandler(ConfigFlow, domain=DOMAIN):
if user_input:
api = get_api(user_input)
try:
data = await self.hass.async_add_executor_job(api.info)
data: AbstractInfoResponse = await self.hass.async_add_executor_job(
api.info
)
data_dict = dataclasses.asdict(data)
title = data_dict.get(
"gogogatename", data_dict.get("ismartgatename", "Cover")
)
await self.async_set_unique_id(re.sub("\\..*$", "", data.remoteaccess))
return self.async_create_entry(title=data.gogogatename, data=user_input)
return self.async_create_entry(title=title, data=user_input)
except ApiError as api_error:
if api_error.code in (
ApiErrorCode.CREDENTIALS_NOT_SET,
ApiErrorCode.CREDENTIALS_INCORRECT,
):
device_type = user_input[CONF_DEVICE]
is_invalid_auth = (
device_type == DEVICE_TYPE_GOGOGATE2
and api_error.code
in (
GogoGate2ApiErrorCode.CREDENTIALS_NOT_SET,
GogoGate2ApiErrorCode.CREDENTIALS_INCORRECT,
)
) or (
device_type == DEVICE_TYPE_ISMARTGATE
and api_error.code
in (
ISmartGateApiErrorCode.CREDENTIALS_NOT_SET,
ISmartGateApiErrorCode.CREDENTIALS_INCORRECT,
)
)
if is_invalid_auth:
errors["base"] = "invalid_auth"
else:
errors["base"] = "cannot_connect"
@ -59,6 +86,10 @@ class Gogogate2FlowHandler(ConfigFlow, domain=DOMAIN):
step_id="user",
data_schema=vol.Schema(
{
vol.Required(
CONF_DEVICE,
default=user_input.get(CONF_DEVICE, DEVICE_TYPE_GOGOGATE2),
): vol.In((DEVICE_TYPE_GOGOGATE2, DEVICE_TYPE_ISMARTGATE)),
vol.Required(
CONF_IP_ADDRESS, default=user_input.get(CONF_IP_ADDRESS, "")
): str,

2
homeassistant/components/gogogate2/const.py
View File

@ -2,3 +2,5 @@
DOMAIN = "gogogate2"
DATA_UPDATE_COORDINATOR = "data_update_coordinator"
DEVICE_TYPE_GOGOGATE2 = "gogogate2"
DEVICE_TYPE_ISMARTGATE = "ismartgate"

86
homeassistant/components/gogogate2/cover.py
View File

@ -2,7 +2,12 @@
import logging
from typing import Callable, List, Optional
from gogogate2_api.common import Door, DoorStatus, get_configured_doors, get_door_by_id
from gogogate2_api.common import (
AbstractDoor,
DoorStatus,
get_configured_doors,
get_door_by_id,
)
import voluptuous as vol
from homeassistant.components.cover import (
@ -12,17 +17,23 @@ from homeassistant.components.cover import (
CoverEntity,
)
from homeassistant.config_entries import SOURCE_IMPORT, ConfigEntry
from homeassistant.const import CONF_IP_ADDRESS, CONF_PASSWORD, CONF_USERNAME
from homeassistant.core import HomeAssistant, callback
from homeassistant.const import (
CONF_DEVICE,
CONF_IP_ADDRESS,
CONF_PASSWORD,
CONF_USERNAME,
)
from homeassistant.core import HomeAssistant
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.entity import Entity
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .common import (
GogoGateDataUpdateCoordinator,
DeviceDataUpdateCoordinator,
cover_unique_id,
get_data_update_coordinator,
)
from .const import DOMAIN
from .const import DEVICE_TYPE_GOGOGATE2, DEVICE_TYPE_ISMARTGATE, DOMAIN
_LOGGER = logging.getLogger(__name__)
@ -30,6 +41,9 @@ _LOGGER = logging.getLogger(__name__)
COVER_SCHEMA = vol.Schema(
{
vol.Required(CONF_IP_ADDRESS): cv.string,
vol.Required(CONF_DEVICE, default=DEVICE_TYPE_GOGOGATE2): vol.In(
(DEVICE_TYPE_GOGOGATE2, DEVICE_TYPE_ISMARTGATE)
),
vol.Required(CONF_PASSWORD): cv.string,
vol.Required(CONF_USERNAME): cv.string,
}
@ -57,39 +71,29 @@ async def async_setup_entry(
async_add_entities(
[
Gogogate2Cover(config_entry, data_update_coordinator, door)
DeviceCover(config_entry, data_update_coordinator, door)
for door in get_configured_doors(data_update_coordinator.data)
]
)
class Gogogate2Cover(CoverEntity):
class DeviceCover(CoordinatorEntity, CoverEntity):
"""Cover entity for goggate2."""
def __init__(
self,
config_entry: ConfigEntry,
data_update_coordinator: GogoGateDataUpdateCoordinator,
door: Door,
data_update_coordinator: DeviceDataUpdateCoordinator,
door: AbstractDoor,
) -> None:
"""Initialize the object."""
super().__init__(data_update_coordinator)
self._config_entry = config_entry
self._data_update_coordinator = data_update_coordinator
self._door = door
self._api = data_update_coordinator.api
self._unique_id = cover_unique_id(config_entry, door)
self._is_available = True
@property
def available(self) -> bool:
"""Return True if entity is available."""
return self._is_available
@property
def should_poll(self) -> bool:
"""Return False as the data manager handles dispatching data."""
return False
@property
def unique_id(self) -> Optional[str]:
"""Return a unique ID."""
@ -98,14 +102,16 @@ class Gogogate2Cover(CoverEntity):
@property
def name(self):
"""Return the name of the door."""
return self._door.name
return self._get_door().name
@property
def is_closed(self):
"""Return true if cover is closed, else False."""
if self._door.status == DoorStatus.OPENED:
door = self._get_door()
if door.status == DoorStatus.OPENED:
return False
if self._door.status == DoorStatus.CLOSED:
if door.status == DoorStatus.CLOSED:
return True
return None
@ -122,36 +128,24 @@ class Gogogate2Cover(CoverEntity):
async def async_open_cover(self, **kwargs):
"""Open the door."""
await self.hass.async_add_executor_job(self._api.open_door, self._door.door_id)
await self.hass.async_add_executor_job(
self._api.open_door, self._get_door().door_id
)
async def async_close_cover(self, **kwargs):
"""Close the door."""
await self.hass.async_add_executor_job(self._api.close_door, self._door.door_id)
await self.hass.async_add_executor_job(
self._api.close_door, self._get_door().door_id
)
@property
def state_attributes(self):
"""Return the state attributes."""
attrs = super().state_attributes
attrs["door_id"] = self._door.door_id
attrs["door_id"] = self._get_door().door_id
return attrs
@callback
def async_on_data_updated(self) -> None:
"""Receive data from data dispatcher."""
if not self._data_update_coordinator.last_update_success:
self._is_available = False
self.async_write_ha_state()
return
door = get_door_by_id(self._door.door_id, self._data_update_coordinator.data)
# Set the state.
self._door = door
self._is_available = True
self.async_write_ha_state()
async def async_added_to_hass(self) -> None:
"""Register update dispatcher."""
self.async_on_remove(
self._data_update_coordinator.async_add_listener(self.async_on_data_updated)
)
def _get_door(self) -> AbstractDoor:
door = get_door_by_id(self._door.door_id, self.coordinator.data)
self._door = door or self._door
return self._door

4
homeassistant/components/gogogate2/manifest.json
View File

@ -1,8 +1,8 @@
{
"domain": "gogogate2",
"name": "Gogogate2",
"name": "Gogogate2 and iSmartGate",
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/gogogate2",
"requirements": ["gogogate2-api==1.0.4"],
"requirements": ["gogogate2-api==2.0.0"],
"codeowners": ["@vangorra"]
}

2
homeassistant/components/gogogate2/strings.json
View File

@ -9,7 +9,7 @@
},
"step": {
"user": {
"title": "Setup GogoGate2",
"title": "Setup GogoGate2 or iSmartGate",
"description": "Provide requisite information below.",
"data": {
"ip_address": "IP Address",

2
requirements_all.txt
View File

@ -668,7 +668,7 @@ glances_api==0.2.0
gntp==1.0.3
# homeassistant.components.gogogate2
gogogate2-api==1.0.4
gogogate2-api==2.0.0
# homeassistant.components.google
google-api-python-client==1.6.4

2
requirements_test_all.txt
View File

@ -334,7 +334,7 @@ gios==0.1.4
glances_api==0.2.0
# homeassistant.components.gogogate2
gogogate2-api==1.0.4
gogogate2-api==2.0.0
# homeassistant.components.google
google-api-python-client==1.6.4

163
tests/components/gogogate2/common.py
View File

@ -1,163 +0,0 @@
"""Common test code."""
from typing import List, NamedTuple, Optional
from unittest.mock import MagicMock, Mock
from gogogate2_api import GogoGate2Api, InfoResponse
from gogogate2_api.common import Door, DoorMode, DoorStatus, Network, Outputs, Wifi
from homeassistant.components import persistent_notification
from homeassistant.components.cover import DOMAIN as COVER_DOMAIN
from homeassistant.components.gogogate2 import async_unload_entry
from homeassistant.components.gogogate2.common import (
GogoGateDataUpdateCoordinator,
get_data_update_coordinator,
)
import homeassistant.components.gogogate2.const as const
from homeassistant.components.homeassistant import DOMAIN as HA_DOMAIN
from homeassistant.config import async_process_ha_core_config
from homeassistant.config_entries import SOURCE_USER
from homeassistant.const import CONF_UNIT_SYSTEM, CONF_UNIT_SYSTEM_METRIC
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import RESULT_TYPE_CREATE_ENTRY, RESULT_TYPE_FORM
from homeassistant.setup import async_setup_component
INFO_RESPONSE = InfoResponse(
user="user1",
gogogatename="gogogatename1",
model="",
apiversion="",
remoteaccessenabled=False,
remoteaccess="abcdefg.my-gogogate.com",
firmwareversion="",
apicode="API_CODE",
door1=Door(
door_id=1,
permission=True,
name="Door1",
mode=DoorMode.GARAGE,
status=DoorStatus.OPENED,
sensor=True,
sensorid=None,
camera=False,
events=2,
temperature=None,
),
door2=Door(
door_id=2,
permission=True,
name=None,
mode=DoorMode.GARAGE,
status=DoorStatus.OPENED,
sensor=True,
sensorid=None,
camera=False,
events=2,
temperature=None,
),
door3=Door(
door_id=3,
permission=True,
name="Door3",
mode=DoorMode.GARAGE,
status=DoorStatus.OPENED,
sensor=True,
sensorid=None,
camera=False,
events=2,
temperature=None,
),
outputs=Outputs(output1=True, output2=False, output3=True),
network=Network(ip=""),
wifi=Wifi(SSID="", linkquality="", signal=""),
)
class ComponentData(NamedTuple):
"""Test data for a mocked component."""
api: GogoGate2Api
data_update_coordinator: GogoGateDataUpdateCoordinator
class ComponentFactory:
"""Manages the setup and unloading of the withing component and profiles."""
def __init__(self, hass: HomeAssistant, gogogate_api_mock: Mock) -> None:
"""Initialize the object."""
self._hass = hass
self._gogogate_api_mock = gogogate_api_mock
@property
def api_class_mock(self):
"""Get the api class mock."""
return self._gogogate_api_mock
async def configure_component(
self, cover_config: Optional[List[dict]] = None
) -> None:
"""Configure the component."""
hass_config = {
"homeassistant": {CONF_UNIT_SYSTEM: CONF_UNIT_SYSTEM_METRIC},
"cover": cover_config or [],
}
await async_process_ha_core_config(self._hass, hass_config.get("homeassistant"))
assert await async_setup_component(self._hass, HA_DOMAIN, {})
assert await async_setup_component(
self._hass, persistent_notification.DOMAIN, {}
)
assert await async_setup_component(self._hass, COVER_DOMAIN, hass_config)
assert await async_setup_component(self._hass, const.DOMAIN, hass_config)
await self._hass.async_block_till_done()
async def run_config_flow(
self, config_data: dict, api_mock: Optional[GogoGate2Api] = None
) -> ComponentData:
"""Run a config flow."""
if api_mock is None:
api_mock: GogoGate2Api = MagicMock(spec=GogoGate2Api)
api_mock.info.return_value = INFO_RESPONSE
self._gogogate_api_mock.reset_mocks()
self._gogogate_api_mock.return_value = api_mock
result = await self._hass.config_entries.flow.async_init(
const.DOMAIN, context={"source": SOURCE_USER}
)
assert result
assert result["type"] == RESULT_TYPE_FORM
assert result["step_id"] == "user"
result = await self._hass.config_entries.flow.async_configure(
result["flow_id"],
user_input=config_data,
)
assert result
assert result["type"] == RESULT_TYPE_CREATE_ENTRY
assert result["data"] == config_data
await self._hass.async_block_till_done()
config_entry = next(
iter(
entry
for entry in self._hass.config_entries.async_entries(const.DOMAIN)
if entry.unique_id == "abcdefg"
)
)
return ComponentData(
api=api_mock,
data_update_coordinator=get_data_update_coordinator(
self._hass, config_entry
),
)
async def unload(self) -> None:
"""Unload all config entries."""
config_entries = self._hass.config_entries.async_entries(const.DOMAIN)
for config_entry in config_entries:
await async_unload_entry(self._hass, config_entry)
await self._hass.async_block_till_done()
assert not self._hass.states.async_entity_ids("gogogate")

18
tests/components/gogogate2/conftest.py
View File

@ -1,18 +0,0 @@
"""Fixtures for tests."""
import pytest
from homeassistant.core import HomeAssistant
from .common import ComponentFactory
from tests.async_mock import patch
@pytest.fixture()
def component_factory(hass: HomeAssistant):
"""Return a factory for initializing the gogogate2 api."""
with patch(
"homeassistant.components.gogogate2.common.GogoGate2Api"
) as gogogate2_api_mock:
yield ComponentFactory(hass, gogogate2_api_mock)

99
tests/components/gogogate2/test_config_flow.py
View File

@ -1,65 +1,66 @@
"""Tests for the GogoGate2 component."""
from gogogate2_api import GogoGate2Api
from gogogate2_api.common import ApiError
from gogogate2_api.const import ApiErrorCode
from gogogate2_api.const import GogoGate2ApiErrorCode
from homeassistant.components.gogogate2.const import DEVICE_TYPE_GOGOGATE2
from homeassistant.config_entries import SOURCE_USER
from homeassistant.const import CONF_IP_ADDRESS, CONF_PASSWORD, CONF_USERNAME
from homeassistant.const import (
CONF_DEVICE,
CONF_IP_ADDRESS,
CONF_PASSWORD,
CONF_USERNAME,
)
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import RESULT_TYPE_FORM
from .common import ComponentFactory
from tests.async_mock import MagicMock, patch
@patch("homeassistant.components.gogogate2.async_setup", return_value=True)
@patch("homeassistant.components.gogogate2.async_setup_entry", return_value=True)
@patch("homeassistant.components.gogogate2.common.GogoGate2Api")
async def test_auth_fail(
hass: HomeAssistant, component_factory: ComponentFactory
gogogate2api_mock, async_setup_entry_mock, async_setup_mock, hass: HomeAssistant
) -> None:
"""Test authorization failures."""
api_mock: GogoGate2Api = MagicMock(spec=GogoGate2Api)
api: GogoGate2Api = MagicMock(spec=GogoGate2Api)
gogogate2api_mock.return_value = api
with patch(
"homeassistant.components.gogogate2.async_setup", return_value=True
), patch(
"homeassistant.components.gogogate2.async_setup_entry",
return_value=True,
):
await component_factory.configure_component()
component_factory.api_class_mock.return_value = api_mock
api.reset_mock()
api.info.side_effect = ApiError(GogoGate2ApiErrorCode.CREDENTIALS_INCORRECT, "blah")
result = await hass.config_entries.flow.async_init(
"gogogate2", context={"source": SOURCE_USER}
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={
CONF_DEVICE: DEVICE_TYPE_GOGOGATE2,
CONF_IP_ADDRESS: "127.0.0.2",
CONF_USERNAME: "user0",
CONF_PASSWORD: "password0",
},
)
assert result
assert result["type"] == RESULT_TYPE_FORM
assert result["errors"] == {
"base": "invalid_auth",
}
api_mock.reset_mock()
api_mock.info.side_effect = ApiError(ApiErrorCode.CREDENTIALS_INCORRECT, "blah")
result = await hass.config_entries.flow.async_init(
"gogogate2", context={"source": SOURCE_USER}
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={
CONF_IP_ADDRESS: "127.0.0.2",
CONF_USERNAME: "user0",
CONF_PASSWORD: "password0",
},
)
assert result
assert result["type"] == RESULT_TYPE_FORM
assert result["errors"] == {
"base": "invalid_auth",
}
api_mock.reset_mock()
api_mock.info.side_effect = Exception("Generic connection error.")
result = await hass.config_entries.flow.async_init(
"gogogate2", context={"source": SOURCE_USER}
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={
CONF_IP_ADDRESS: "127.0.0.2",
CONF_USERNAME: "user0",
CONF_PASSWORD: "password0",
},
)
assert result
assert result["type"] == RESULT_TYPE_FORM
assert result["errors"] == {"base": "cannot_connect"}
api.reset_mock()
api.info.side_effect = Exception("Generic connection error.")
result = await hass.config_entries.flow.async_init(
"gogogate2", context={"source": SOURCE_USER}
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={
CONF_DEVICE: DEVICE_TYPE_GOGOGATE2,
CONF_IP_ADDRESS: "127.0.0.2",
CONF_USERNAME: "user0",
CONF_PASSWORD: "password0",
},
)
assert result
assert result["type"] == RESULT_TYPE_FORM
assert result["errors"] == {"base": "cannot_connect"}

506
tests/components/gogogate2/test_cover.py
View File

@ -1,65 +1,90 @@
"""Tests for the GogoGate2 component."""
from gogogate2_api import GogoGate2Api
from datetime import timedelta
from gogogate2_api import GogoGate2Api, ISmartGateApi
from gogogate2_api.common import (
ActivateResponse,
ApiError,
Door,
DoorMode,
DoorStatus,
InfoResponse,
GogoGate2ActivateResponse,
GogoGate2Door,
GogoGate2InfoResponse,
ISmartGateDoor,
ISmartGateInfoResponse,
Network,
Outputs,
Wifi,
)
from homeassistant.components.cover import DOMAIN as COVER_DOMAIN
from homeassistant.components.gogogate2.const import (
DEVICE_TYPE_GOGOGATE2,
DEVICE_TYPE_ISMARTGATE,
DOMAIN,
)
from homeassistant.components.homeassistant import DOMAIN as HA_DOMAIN
from homeassistant.config import async_process_ha_core_config
from homeassistant.config_entries import SOURCE_USER
from homeassistant.const import (
CONF_DEVICE,
CONF_IP_ADDRESS,
CONF_NAME,
CONF_PASSWORD,
CONF_PLATFORM,
CONF_UNIT_SYSTEM,
CONF_UNIT_SYSTEM_METRIC,
CONF_USERNAME,
STATE_CLOSED,
STATE_OPEN,
STATE_UNAVAILABLE,
STATE_UNKNOWN,
)
from homeassistant.core import HomeAssistant
from homeassistant.setup import async_setup_component
from homeassistant.util.dt import utcnow
from .common import ComponentFactory
from tests.async_mock import MagicMock
from tests.async_mock import MagicMock, patch
from tests.common import MockConfigEntry, async_fire_time_changed
async def test_import_fail(
hass: HomeAssistant, component_factory: ComponentFactory
) -> None:
@patch("homeassistant.components.gogogate2.common.GogoGate2Api")
async def test_import_fail(gogogate2api_mock, hass: HomeAssistant) -> None:
"""Test the failure to import."""
api = MagicMock(spec=GogoGate2Api)
api.info.side_effect = ApiError(22, "Error")
gogogate2api_mock.return_value = api
component_factory.api_class_mock.return_value = api
await component_factory.configure_component(
cover_config=[
hass_config = {
HA_DOMAIN: {CONF_UNIT_SYSTEM: CONF_UNIT_SYSTEM_METRIC},
COVER_DOMAIN: [
{
CONF_PLATFORM: "gogogate2",
CONF_NAME: "cover0",
CONF_DEVICE: DEVICE_TYPE_GOGOGATE2,
CONF_IP_ADDRESS: "127.0.1.0",
CONF_USERNAME: "user0",
CONF_PASSWORD: "password0",
}
]
)
],
}
await async_process_ha_core_config(hass, hass_config[HA_DOMAIN])
assert await async_setup_component(hass, HA_DOMAIN, {})
assert await async_setup_component(hass, COVER_DOMAIN, hass_config)
await hass.async_block_till_done()
entity_ids = hass.states.async_entity_ids(COVER_DOMAIN)
assert not entity_ids
async def test_import(hass: HomeAssistant, component_factory: ComponentFactory) -> None:
@patch("homeassistant.components.gogogate2.common.GogoGate2Api")
@patch("homeassistant.components.gogogate2.common.ISmartGateApi")
async def test_import(
ismartgateapi_mock, gogogate2api_mock, hass: HomeAssistant
) -> None:
"""Test importing of file based config."""
api0 = MagicMock(spec=GogoGate2Api)
api0.info.return_value = InfoResponse(
api0.info.return_value = GogoGate2InfoResponse(
user="user1",
gogogatename="gogogatename0",
model="",
@ -68,7 +93,7 @@ async def test_import(hass: HomeAssistant, component_factory: ComponentFactory)
remoteaccess="abc123.blah.blah",
firmwareversion="",
apicode="",
door1=Door(
door1=GogoGate2Door(
door_id=1,
permission=True,
name="Door1",
@ -80,7 +105,7 @@ async def test_import(hass: HomeAssistant, component_factory: ComponentFactory)
events=2,
temperature=None,
),
door2=Door(
door2=GogoGate2Door(
door_id=2,
permission=True,
name=None,
@ -92,7 +117,7 @@ async def test_import(hass: HomeAssistant, component_factory: ComponentFactory)
events=0,
temperature=None,
),
door3=Door(
door3=GogoGate2Door(
door_id=3,
permission=True,
name=None,
@ -108,18 +133,21 @@ async def test_import(hass: HomeAssistant, component_factory: ComponentFactory)
network=Network(ip=""),
wifi=Wifi(SSID="", linkquality="", signal=""),
)
gogogate2api_mock.return_value = api0
api1 = MagicMock(spec=GogoGate2Api)
api1.info.return_value = InfoResponse(
api1 = MagicMock(spec=ISmartGateApi)
api1.info.return_value = ISmartGateInfoResponse(
user="user1",
gogogatename="gogogatename0",
ismartgatename="ismartgatename0",
model="",
apiversion="",
remoteaccessenabled=False,
remoteaccess="321bca.blah.blah",
remoteaccess="abc321.blah.blah",
firmwareversion="",
apicode="",
door1=Door(
pin=123,
lang="en",
newfirmware=False,
door1=ISmartGateDoor(
door_id=1,
permission=True,
name="Door1",
@ -130,50 +158,52 @@ async def test_import(hass: HomeAssistant, component_factory: ComponentFactory)
camera=False,
events=2,
temperature=None,
enabled=True,
apicode="apicode0",
customimage=False,
),
door2=Door(
door_id=2,
door2=ISmartGateDoor(
door_id=1,
permission=True,
name=None,
mode=DoorMode.GARAGE,
status=DoorStatus.UNDEFINED,
status=DoorStatus.CLOSED,
sensor=True,
sensorid=None,
camera=False,
events=0,
events=2,
temperature=None,
enabled=True,
apicode="apicode0",
customimage=False,
),
door3=Door(
door_id=3,
door3=ISmartGateDoor(
door_id=1,
permission=True,
name=None,
mode=DoorMode.GARAGE,
status=DoorStatus.UNDEFINED,
status=DoorStatus.CLOSED,
sensor=True,
sensorid=None,
camera=False,
events=0,
events=2,
temperature=None,
enabled=True,
apicode="apicode0",
customimage=False,
),
outputs=Outputs(output1=True, output2=False, output3=True),
network=Network(ip=""),
wifi=Wifi(SSID="", linkquality="", signal=""),
)
ismartgateapi_mock.return_value = api1
def new_api(ip_address: str, username: str, password: str) -> GogoGate2Api:
if ip_address == "127.0.1.0":
return api0
if ip_address == "127.0.1.1":
return api1
raise Exception(f"Untested ip address {ip_address}")
component_factory.api_class_mock.side_effect = new_api
await component_factory.configure_component(
cover_config=[
hass_config = {
HA_DOMAIN: {CONF_UNIT_SYSTEM: CONF_UNIT_SYSTEM_METRIC},
COVER_DOMAIN: [
{
CONF_PLATFORM: "gogogate2",
CONF_NAME: "cover0",
CONF_DEVICE: DEVICE_TYPE_GOGOGATE2,
CONF_IP_ADDRESS: "127.0.1.0",
CONF_USERNAME: "user0",
CONF_PASSWORD: "password0",
@ -181,274 +211,148 @@ async def test_import(hass: HomeAssistant, component_factory: ComponentFactory)
{
CONF_PLATFORM: "gogogate2",
CONF_NAME: "cover1",
CONF_DEVICE: DEVICE_TYPE_ISMARTGATE,
CONF_IP_ADDRESS: "127.0.1.1",
CONF_USERNAME: "user1",
CONF_PASSWORD: "password1",
},
]
)
],
}
await async_process_ha_core_config(hass, hass_config[HA_DOMAIN])
assert await async_setup_component(hass, HA_DOMAIN, {})
assert await async_setup_component(hass, COVER_DOMAIN, hass_config)
await hass.async_block_till_done()
entity_ids = hass.states.async_entity_ids(COVER_DOMAIN)
assert entity_ids is not None
assert len(entity_ids) == 2
assert "cover.door1" in entity_ids
assert "cover.door1_2" in entity_ids
await component_factory.unload()
@patch("homeassistant.components.gogogate2.common.GogoGate2Api")
async def test_open_close_update(gogogat2api_mock, hass: HomeAssistant) -> None:
"""Test open and close and data update."""
async def test_cover_update(
hass: HomeAssistant, component_factory: ComponentFactory
) -> None:
"""Test cover."""
await component_factory.configure_component()
component_data = await component_factory.run_config_flow(
config_data={
CONF_IP_ADDRESS: "127.0.0.2",
CONF_USERNAME: "user0",
CONF_PASSWORD: "password0",
}
def info_response(door_status: DoorStatus) -> GogoGate2InfoResponse:
return GogoGate2InfoResponse(
user="user1",
gogogatename="gogogatename0",
model="",
apiversion="",
remoteaccessenabled=False,
remoteaccess="abc123.blah.blah",
firmwareversion="",
apicode="",
door1=GogoGate2Door(
door_id=1,
permission=True,
name="Door1",
mode=DoorMode.GARAGE,
status=door_status,
sensor=True,
sensorid=None,
camera=False,
events=2,
temperature=None,
),
door2=GogoGate2Door(
door_id=2,
permission=True,
name=None,
mode=DoorMode.GARAGE,
status=DoorStatus.UNDEFINED,
sensor=True,
sensorid=None,
camera=False,
events=0,
temperature=None,
),
door3=GogoGate2Door(
door_id=3,
permission=True,
name=None,
mode=DoorMode.GARAGE,
status=DoorStatus.UNDEFINED,
sensor=True,
sensorid=None,
camera=False,
events=0,
temperature=None,
),
outputs=Outputs(output1=True, output2=False, output3=True),
network=Network(ip=""),
wifi=Wifi(SSID="", linkquality="", signal=""),
)
api = MagicMock(GogoGate2Api)
api.activate.return_value = GogoGate2ActivateResponse(result=True)
api.info.return_value = info_response(DoorStatus.OPENED)
gogogat2api_mock.return_value = api
config_entry = MockConfigEntry(
domain=DOMAIN,
source=SOURCE_USER,
data={
CONF_IP_ADDRESS: "127.0.0.1",
CONF_USERNAME: "admin",
CONF_PASSWORD: "password",
},
)
config_entry.add_to_hass(hass)
assert hass.states.async_entity_ids(COVER_DOMAIN)
state = hass.states.get("cover.door1")
assert state
assert state.state == STATE_OPEN
assert state.attributes["friendly_name"] == "Door1"
assert state.attributes["supported_features"] == 3
assert state.attributes["device_class"] == "garage"
component_data.data_update_coordinator.api.info.return_value = InfoResponse(
user="user1",
gogogatename="gogogatename0",
model="",
apiversion="",
remoteaccessenabled=False,
remoteaccess="abc123.blah.blah",
firmwareversion="",
apicode="",
door1=Door(
door_id=1,
permission=True,
name="Door1",
mode=DoorMode.GARAGE,
status=DoorStatus.OPENED,
sensor=True,
sensorid=None,
camera=False,
events=2,
temperature=None,
),
door2=Door(
door_id=2,
permission=True,
name=None,
mode=DoorMode.GARAGE,
status=DoorStatus.UNDEFINED,
sensor=True,
sensorid=None,
camera=False,
events=0,
temperature=None,
),
door3=Door(
door_id=3,
permission=True,
name=None,
mode=DoorMode.GARAGE,
status=DoorStatus.UNDEFINED,
sensor=True,
sensorid=None,
camera=False,
events=0,
temperature=None,
),
outputs=Outputs(output1=True, output2=False, output3=True),
network=Network(ip=""),
wifi=Wifi(SSID="", linkquality="", signal=""),
)
await component_data.data_update_coordinator.async_refresh()
await hass.async_block_till_done()
state = hass.states.get("cover.door1")
assert state
assert state.state == STATE_OPEN
component_data.data_update_coordinator.api.info.return_value = InfoResponse(
user="user1",
gogogatename="gogogatename0",
model="",
apiversion="",
remoteaccessenabled=False,
remoteaccess="abc123.blah.blah",
firmwareversion="",
apicode="",
door1=Door(
door_id=1,
permission=True,
name="Door1",
mode=DoorMode.GARAGE,
status=DoorStatus.CLOSED,
sensor=True,
sensorid=None,
camera=False,
events=2,
temperature=None,
),
door2=Door(
door_id=2,
permission=True,
name=None,
mode=DoorMode.GARAGE,
status=DoorStatus.UNDEFINED,
sensor=True,
sensorid=None,
camera=False,
events=0,
temperature=None,
),
door3=Door(
door_id=3,
permission=True,
name=None,
mode=DoorMode.GARAGE,
status=DoorStatus.UNDEFINED,
sensor=True,
sensorid=None,
camera=False,
events=0,
temperature=None,
),
outputs=Outputs(output1=True, output2=False, output3=True),
network=Network(ip=""),
wifi=Wifi(SSID="", linkquality="", signal=""),
)
await component_data.data_update_coordinator.async_refresh()
await hass.async_block_till_done()
state = hass.states.get("cover.door1")
assert state
assert state.state == STATE_CLOSED
async def test_open_close(
hass: HomeAssistant, component_factory: ComponentFactory
) -> None:
"""Test open and close."""
closed_door_response = InfoResponse(
user="user1",
gogogatename="gogogatename0",
model="",
apiversion="",
remoteaccessenabled=False,
remoteaccess="abc123.blah.blah",
firmwareversion="",
apicode="",
door1=Door(
door_id=1,
permission=True,
name="Door1",
mode=DoorMode.GARAGE,
status=DoorStatus.CLOSED,
sensor=True,
sensorid=None,
camera=False,
events=2,
temperature=None,
),
door2=Door(
door_id=2,
permission=True,
name=None,
mode=DoorMode.GARAGE,
status=DoorStatus.UNDEFINED,
sensor=True,
sensorid=None,
camera=False,
events=0,
temperature=None,
),
door3=Door(
door_id=3,
permission=True,
name=None,
mode=DoorMode.GARAGE,
status=DoorStatus.UNDEFINED,
sensor=True,
sensorid=None,
camera=False,
events=0,
temperature=None,
),
outputs=Outputs(output1=True, output2=False, output3=True),
network=Network(ip=""),
wifi=Wifi(SSID="", linkquality="", signal=""),
)
await component_factory.configure_component()
assert hass.states.get("cover.door1") is None
component_data = await component_factory.run_config_flow(
config_data={
CONF_IP_ADDRESS: "127.0.0.2",
CONF_USERNAME: "user0",
CONF_PASSWORD: "password0",
}
)
component_data.api.activate.return_value = ActivateResponse(result=True)
assert await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
assert hass.states.get("cover.door1").state == STATE_OPEN
api.info.return_value = info_response(DoorStatus.CLOSED)
await hass.services.async_call(
COVER_DOMAIN,
"close_cover",
service_data={"entity_id": "cover.door1"},
)
await hass.async_block_till_done()
component_data.api.close_door.assert_called_with(1)
component_data.data_update_coordinator.api.info.return_value = closed_door_response
await component_data.data_update_coordinator.async_refresh()
async_fire_time_changed(hass, utcnow() + timedelta(hours=2))
await hass.async_block_till_done()
assert hass.states.get("cover.door1").state == STATE_CLOSED
api.close_door.assert_called_with(1)
# Assert mid state changed when new status is received.
api.info.return_value = info_response(DoorStatus.OPENED)
await hass.services.async_call(
COVER_DOMAIN,
"open_cover",
service_data={"entity_id": "cover.door1"},
)
async_fire_time_changed(hass, utcnow() + timedelta(hours=2))
await hass.async_block_till_done()
component_data.api.open_door.assert_called_with(1)
assert hass.states.get("cover.door1").state == STATE_OPEN
api.open_door.assert_called_with(1)
# Assert the mid state does not change when the same status is returned.
component_data.data_update_coordinator.api.info.return_value = closed_door_response
await component_data.data_update_coordinator.async_refresh()
component_data.data_update_coordinator.api.info.return_value = closed_door_response
await component_data.data_update_coordinator.async_refresh()
await component_data.data_update_coordinator.async_refresh()
await hass.services.async_call(
HA_DOMAIN,
"update_entity",
service_data={"entity_id": "cover.door1"},
)
api.info.return_value = info_response(DoorStatus.UNDEFINED)
async_fire_time_changed(hass, utcnow() + timedelta(hours=2))
await hass.async_block_till_done()
assert hass.states.get("cover.door1").state == STATE_CLOSED
assert hass.states.get("cover.door1").state == STATE_UNKNOWN
assert await hass.config_entries.async_unload(config_entry.entry_id)
assert not hass.states.async_entity_ids(DOMAIN)
async def test_availability(
hass: HomeAssistant, component_factory: ComponentFactory
) -> None:
"""Test open and close."""
closed_door_response = InfoResponse(
@patch("homeassistant.components.gogogate2.common.ISmartGateApi")
async def test_availability(ismartgateapi_mock, hass: HomeAssistant) -> None:
"""Test availability."""
closed_door_response = ISmartGateInfoResponse(
user="user1",
gogogatename="gogogatename0",
ismartgatename="ismartgatename0",
model="",
apiversion="",
remoteaccessenabled=False,
remoteaccess="abc123.blah.blah",
firmwareversion="",
apicode="",
door1=Door(
pin=123,
lang="en",
newfirmware=False,
door1=ISmartGateDoor(
door_id=1,
permission=True,
name="Door1",
@ -459,8 +363,11 @@ async def test_availability(
camera=False,
events=2,
temperature=None,
enabled=True,
apicode="apicode0",
customimage=False,
),
door2=Door(
door2=ISmartGateDoor(
door_id=2,
permission=True,
name=None,
@ -471,8 +378,11 @@ async def test_availability(
camera=False,
events=0,
temperature=None,
enabled=True,
apicode="apicode0",
customimage=False,
),
door3=Door(
door3=ISmartGateDoor(
door_id=3,
permission=True,
name=None,
@ -483,31 +393,43 @@ async def test_availability(
camera=False,
events=0,
temperature=None,
enabled=True,
apicode="apicode0",
customimage=False,
),
outputs=Outputs(output1=True, output2=False, output3=True),
network=Network(ip=""),
wifi=Wifi(SSID="", linkquality="", signal=""),
)
await component_factory.configure_component()
assert hass.states.get("cover.door1") is None
api = MagicMock(ISmartGateApi)
api.info.return_value = closed_door_response
ismartgateapi_mock.return_value = api
component_data = await component_factory.run_config_flow(
config_data={
CONF_IP_ADDRESS: "127.0.0.2",
CONF_USERNAME: "user0",
CONF_PASSWORD: "password0",
}
config_entry = MockConfigEntry(
domain=DOMAIN,
source=SOURCE_USER,
data={
CONF_DEVICE: DEVICE_TYPE_ISMARTGATE,
CONF_IP_ADDRESS: "127.0.0.1",
CONF_USERNAME: "admin",
CONF_PASSWORD: "password",
},
)
assert hass.states.get("cover.door1").state == STATE_OPEN
config_entry.add_to_hass(hass)
component_data.api.info.side_effect = Exception("Error")
await component_data.data_update_coordinator.async_refresh()
assert hass.states.get("cover.door1") is None
assert await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
assert hass.states.get("cover.door1")
api.info.side_effect = Exception("Error")
async_fire_time_changed(hass, utcnow() + timedelta(hours=2))
await hass.async_block_till_done()
assert hass.states.get("cover.door1").state == STATE_UNAVAILABLE
component_data.api.info.side_effect = None
component_data.api.info.return_value = closed_door_response
await component_data.data_update_coordinator.async_refresh()
api.info.side_effect = None
api.info.return_value = closed_door_response
async_fire_time_changed(hass, utcnow() + timedelta(hours=2))
await hass.async_block_till_done()
assert hass.states.get("cover.door1").state == STATE_CLOSED

75
tests/components/gogogate2/test_init.py
View File

@ -1,8 +1,17 @@
"""Tests for the GogoGate2 component."""
from gogogate2_api import GogoGate2Api
import pytest
from homeassistant.components.gogogate2 import async_setup_entry
from homeassistant.components.gogogate2.common import GogoGateDataUpdateCoordinator
from homeassistant.components.gogogate2 import DEVICE_TYPE_GOGOGATE2, async_setup_entry
from homeassistant.components.gogogate2.common import DeviceDataUpdateCoordinator
from homeassistant.components.gogogate2.const import DEVICE_TYPE_ISMARTGATE, DOMAIN
from homeassistant.config_entries import SOURCE_USER
from homeassistant.const import (
CONF_DEVICE,
CONF_IP_ADDRESS,
CONF_PASSWORD,
CONF_USERNAME,
)
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryNotReady
@ -10,11 +19,69 @@ from tests.async_mock import MagicMock, patch
from tests.common import MockConfigEntry
@patch("homeassistant.components.gogogate2.common.GogoGate2Api")
async def test_config_update(gogogate2api_mock, hass: HomeAssistant) -> None:
"""Test config setup where the config is updated."""
api = MagicMock(GogoGate2Api)
api.info.side_effect = Exception("Error")
gogogate2api_mock.return_value = api
config_entry = MockConfigEntry(
domain=DOMAIN,
source=SOURCE_USER,
data={
CONF_IP_ADDRESS: "127.0.0.1",
CONF_USERNAME: "admin",
CONF_PASSWORD: "password",
},
)
config_entry.add_to_hass(hass)
assert not await hass.config_entries.async_setup(entry_id=config_entry.entry_id)
await hass.async_block_till_done()
assert config_entry.data == {
CONF_DEVICE: DEVICE_TYPE_GOGOGATE2,
CONF_IP_ADDRESS: "127.0.0.1",
CONF_USERNAME: "admin",
CONF_PASSWORD: "password",
}
@patch("homeassistant.components.gogogate2.common.ISmartGateApi")
async def test_config_no_update(ismartgateapi_mock, hass: HomeAssistant) -> None:
"""Test config setup where the data is not updated."""
api = MagicMock(GogoGate2Api)
api.info.side_effect = Exception("Error")
ismartgateapi_mock.return_value = api
config_entry = MockConfigEntry(
domain=DOMAIN,
source=SOURCE_USER,
data={
CONF_DEVICE: DEVICE_TYPE_ISMARTGATE,
CONF_IP_ADDRESS: "127.0.0.1",
CONF_USERNAME: "admin",
CONF_PASSWORD: "password",
},
)
config_entry.add_to_hass(hass)
assert not await hass.config_entries.async_setup(entry_id=config_entry.entry_id)
await hass.async_block_till_done()
assert config_entry.data == {
CONF_DEVICE: DEVICE_TYPE_ISMARTGATE,
CONF_IP_ADDRESS: "127.0.0.1",
CONF_USERNAME: "admin",
CONF_PASSWORD: "password",
}
async def test_auth_fail(hass: HomeAssistant) -> None:
"""Test authorization failures."""
coordinator_mock: GogoGateDataUpdateCoordinator = MagicMock(
spec=GogoGateDataUpdateCoordinator
coordinator_mock: DeviceDataUpdateCoordinator = MagicMock(
spec=DeviceDataUpdateCoordinator
)
coordinator_mock.last_update_success = False