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

Add config flow to gogogate2 component (#34709)

Browse Source 
* Add config flow to gogogate2 component.

* Using a more stable gogogate api.

* Getting config flows working better by using different downstream library.

* Fixing options not getting default values.
Adding availability to cover entity.

* Simplifying return types of function.

* Address PR feedback.

* Making user config flow not abort.

* Using DataUpdateCoordinator.

* Addressing PR feedback.

* Using standard method for using hass.data

* Split auth fail test into separate tests.
This commit is contained in:
Robert Van Gorkom 2020-05-16 08:53:11 -07:00 committed by GitHub
parent 6e0359efa6
commit 11b786a4fc
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
18 changed files with 1189 additions and 69 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
.coveragerc
View File

@ -281,7 +281,6 @@ omit =
homeassistant/components/glances/sensor.py
homeassistant/components/gntp/notify.py
homeassistant/components/goalfeed/*
homeassistant/components/gogogate2/cover.py
homeassistant/components/google/*
homeassistant/components/google_cloud/tts.py
homeassistant/components/google_maps/device_tracker.py

1
CODEOWNERS
View File

@ -148,6 +148,7 @@ homeassistant/components/gios/* @bieniu
homeassistant/components/gitter/* @fabaff
homeassistant/components/glances/* @fabaff @engrbm87
homeassistant/components/gntp/* @robbiet480
homeassistant/components/gogogate2/* @vangorra
homeassistant/components/google_assistant/* @home-assistant/cloud
homeassistant/components/google_cloud/* @lufton
homeassistant/components/google_translate/* @awarecan

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

@ -1 +1,36 @@
"""The gogogate2 component."""
from homeassistant.components.cover import DOMAIN as COVER_DOMAIN
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryNotReady
from .common import get_data_update_coordinator
async def async_setup(hass: HomeAssistant, base_config: dict) -> bool:
"""Set up for Gogogate2 controllers."""
return True
async def async_setup_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> bool:
"""Do setup of Gogogate2."""
data_update_coordinator = get_data_update_coordinator(hass, config_entry)
await data_update_coordinator.async_refresh()
if not data_update_coordinator.last_update_success:
raise ConfigEntryNotReady()
hass.async_create_task(
hass.config_entries.async_forward_entry_setup(config_entry, COVER_DOMAIN)
)
return True
async def async_unload_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> bool:
"""Unload Gogogate2 config entry."""
hass.async_create_task(
hass.config_entries.async_forward_entry_unload(config_entry, COVER_DOMAIN)
)
return True

99
homeassistant/components/gogogate2/common.py Normal file
View File

@ -0,0 +1,99 @@
"""Common code for GogoGate2 component."""
from datetime import timedelta
import logging
from typing import Awaitable, Callable, NamedTuple, Optional
import async_timeout
from gogogate2_api import GogoGate2Api
from gogogate2_api.common import Door
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import 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
_LOGGER = logging.getLogger(__name__)
class StateData(NamedTuple):
"""State data for a cover entity."""
config_unique_id: str
unique_id: Optional[str]
door: Optional[Door]
class GogoGateDataUpdateCoordinator(DataUpdateCoordinator):
"""Manages polling for state changes from the device."""
def __init__(
self,
hass: HomeAssistant,
logger: logging.Logger,
api: GogoGate2Api,
*,
name: str,
update_interval: timedelta,
update_method: Optional[Callable[[], Awaitable]] = None,
request_refresh_debouncer: Optional[Debouncer] = None,
):
"""Initialize the data update coordinator."""
DataUpdateCoordinator.__init__(
self,
hass,
logger,
name=name,
update_interval=update_interval,
update_method=update_method,
request_refresh_debouncer=request_refresh_debouncer,
)
self.api = api
def get_data_update_coordinator(
hass: HomeAssistant, config_entry: ConfigEntry
) -> GogoGateDataUpdateCoordinator:
"""Get an update coordinator."""
hass.data.setdefault(DOMAIN, {})
hass.data[DOMAIN].setdefault(config_entry.entry_id, {})
config_entry_data = hass.data[DOMAIN][config_entry.entry_id]
if DATA_UPDATE_COORDINATOR not in config_entry_data:
api = get_api(config_entry.data)
async def async_update_data():
try:
async with async_timeout.timeout(3):
return await hass.async_add_executor_job(api.info)
except Exception as exception:
raise UpdateFailed(f"Error communicating with API: {exception}")
config_entry_data[DATA_UPDATE_COORDINATOR] = GogoGateDataUpdateCoordinator(
hass,
_LOGGER,
api,
# Name of the data. For logging purposes.
name="gogogate2",
update_method=async_update_data,
# Polling interval. Will only be polled if there are subscribers.
update_interval=timedelta(seconds=5),
)
return config_entry_data[DATA_UPDATE_COORDINATOR]
def cover_unique_id(config_entry: ConfigEntry, door: Door) -> str:
"""Generate a cover entity unique id."""
return f"{config_entry.unique_id}_{door.door_id}"
def get_api(config_data: dict) -> GogoGate2Api:
"""Get an api object for config data."""
return GogoGate2Api(
config_data[CONF_IP_ADDRESS],
config_data[CONF_USERNAME],
config_data[CONF_PASSWORD],
)

74
homeassistant/components/gogogate2/config_flow.py Normal file
View File

@ -0,0 +1,74 @@
"""Config flow for Gogogate2."""
import logging
import re
from gogogate2_api.common import ApiError
from gogogate2_api.const import ApiErrorCode
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 .common import get_api
from .const import DOMAIN # pylint: disable=unused-import
_LOGGER = logging.getLogger(__name__)
class Gogogate2FlowHandler(ConfigFlow, domain=DOMAIN):
"""Gogogate2 config flow."""
VERSION = 1
CONNECTION_CLASS = config_entries.CONN_CLASS_LOCAL_POLL
async def async_step_import(self, config_data: dict = None):
"""Handle importing of configuration."""
result = await self.async_step_user(config_data)
self._abort_if_unique_id_configured()
return result
async def async_step_user(self, user_input: dict = None):
"""Handle user initiated flow."""
user_input = user_input or {}
errors = {}
if user_input:
api = get_api(user_input)
try:
data = await self.hass.async_add_executor_job(api.info)
await self.async_set_unique_id(re.sub("\\..*$", "", data.remoteaccess))
return self.async_create_entry(title=data.gogogatename, data=user_input)
except ApiError as api_error:
if api_error.code in (
ApiErrorCode.CREDENTIALS_NOT_SET,
ApiErrorCode.CREDENTIALS_INCORRECT,
):
errors["base"] = "invalid_auth"
else:
errors["base"] = "cannot_connect"
except Exception: # pylint: disable=broad-except
errors["base"] = "cannot_connect"
if errors and self.source == SOURCE_IMPORT:
return self.async_abort(reason="cannot_connect")
return self.async_show_form(
step_id="user",
data_schema=vol.Schema(
{
vol.Required(
CONF_IP_ADDRESS, default=user_input.get(CONF_IP_ADDRESS, "")
): str,
vol.Required(
CONF_USERNAME, default=user_input.get(CONF_USERNAME, "")
): str,
vol.Required(
CONF_PASSWORD, default=user_input.get(CONF_PASSWORD, "")
): str,
}
),
errors=errors,
)

4
homeassistant/components/gogogate2/const.py Normal file
View File

@ -0,0 +1,4 @@
"""Constants for integration."""
DOMAIN = "gogogate2"
DATA_UPDATE_COORDINATOR = "data_update_coordinator"

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

@ -1,112 +1,190 @@
"""Support for Gogogate2 garage Doors."""
from datetime import datetime, timedelta
import logging
from typing import Callable, List, Optional
from pygogogate2 import Gogogate2API as pygogogate2
from gogogate2_api.common import Door, DoorStatus, get_configured_doors, get_door_by_id
import voluptuous as vol
from homeassistant.components.cover import SUPPORT_CLOSE, SUPPORT_OPEN, CoverEntity
from homeassistant.components.cover import (
DEVICE_CLASS_GARAGE,
SUPPORT_CLOSE,
SUPPORT_OPEN,
CoverEntity,
)
from homeassistant.config_entries import SOURCE_IMPORT, ConfigEntry
from homeassistant.const import (
CONF_IP_ADDRESS,
CONF_NAME,
CONF_PASSWORD,
CONF_USERNAME,
STATE_CLOSED,
STATE_CLOSING,
STATE_OPENING,
)
from homeassistant.core import HomeAssistant, callback
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.entity import Entity
from .common import (
GogoGateDataUpdateCoordinator,
cover_unique_id,
get_data_update_coordinator,
)
from .const import DOMAIN
_LOGGER = logging.getLogger(__name__)
DEFAULT_NAME = "gogogate2"
NOTIFICATION_ID = "gogogate2_notification"
NOTIFICATION_TITLE = "Gogogate2 Cover Setup"
COVER_SCHEMA = vol.Schema(
{
vol.Required(CONF_IP_ADDRESS): cv.string,
vol.Required(CONF_PASSWORD): cv.string,
vol.Required(CONF_USERNAME): cv.string,
vol.Optional(CONF_NAME, default=DEFAULT_NAME): cv.string,
}
)
def setup_platform(hass, config, add_entities, discovery_info=None):
"""Set up the Gogogate2 component."""
ip_address = config.get(CONF_IP_ADDRESS)
name = config.get(CONF_NAME)
password = config.get(CONF_PASSWORD)
username = config.get(CONF_USERNAME)
mygogogate2 = pygogogate2(username, password, ip_address)
try:
devices = mygogogate2.get_devices()
if devices is False:
raise ValueError("Username or Password is incorrect or no devices found")
add_entities(MyGogogate2Device(mygogogate2, door, name) for door in devices)
except (TypeError, KeyError, NameError, ValueError) as ex:
_LOGGER.error("%s", ex)
hass.components.persistent_notification.create(
(f"Error: {ex}<br />You will need to restart hass after fixing."),
title=NOTIFICATION_TITLE,
notification_id=NOTIFICATION_ID,
async def async_setup_platform(
hass: HomeAssistant, config: dict, add_entities: Callable, discovery_info=None
) -> None:
"""Convert old style file configs to new style configs."""
hass.async_create_task(
hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_IMPORT}, data=config
)
)
class MyGogogate2Device(CoverEntity):
"""Representation of a Gogogate2 cover."""
async def async_setup_entry(
hass: HomeAssistant,
config_entry: ConfigEntry,
async_add_entities: Callable[[List[Entity], Optional[bool]], None],
) -> None:
"""Set up the config entry."""
data_update_coordinator = get_data_update_coordinator(hass, config_entry)
def __init__(self, mygogogate2, device, name):
"""Initialize with API object, device id."""
self.mygogogate2 = mygogogate2
self.device_id = device["door"]
self._name = name or device["name"]
self._status = device["status"]
self._available = None
async_add_entities(
[
Gogogate2Cover(config_entry, data_update_coordinator, door)
for door in get_configured_doors(data_update_coordinator.data)
]
)
class Gogogate2Cover(CoverEntity):
"""Cover entity for goggate2."""
def __init__(
self,
config_entry: ConfigEntry,
data_update_coordinator: GogoGateDataUpdateCoordinator,
door: Door,
) -> None:
"""Initialize the object."""
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
self._transition_state: Optional[str] = None
self._transition_state_start: Optional[datetime] = None
@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."""
return self._unique_id
@property
def name(self):
"""Return the name of the garage door if any."""
return self._name if self._name else DEFAULT_NAME
"""Return the name of the door."""
return self._door.name
@property
def is_closed(self):
"""Return true if cover is closed, else False."""
return self._status == STATE_CLOSED
if self._door.status == DoorStatus.OPENED:
return False
if self._door.status == DoorStatus.CLOSED:
return True
return None
@property
def is_opening(self):
"""Return if the cover is opening or not."""
return self._transition_state == STATE_OPENING
@property
def is_closing(self):
"""Return if the cover is closing or not."""
return self._transition_state == STATE_CLOSING
@property
def device_class(self):
"""Return the class of this device, from component DEVICE_CLASSES."""
return "garage"
return DEVICE_CLASS_GARAGE
@property
def supported_features(self):
"""Flag supported features."""
return SUPPORT_OPEN | SUPPORT_CLOSE
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)
self._transition_state = STATE_OPENING
self._transition_state_start = datetime.now()
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)
self._transition_state = STATE_CLOSING
self._transition_state_start = datetime.now()
@property
def available(self):
"""Could the device be accessed during the last update call."""
return self._available
def state_attributes(self):
"""Return the state attributes."""
attrs = super().state_attributes
attrs["door_id"] = self._door.door_id
return attrs
def close_cover(self, **kwargs):
"""Issue close command to cover."""
self.mygogogate2.close_device(self.device_id)
@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
def open_cover(self, **kwargs):
"""Issue open command to cover."""
self.mygogogate2.open_device(self.device_id)
door = get_door_by_id(self._door.door_id, self._data_update_coordinator.data)
def update(self):
"""Update status of cover."""
try:
self._status = self.mygogogate2.get_status(self.device_id)
self._available = True
except (TypeError, KeyError, NameError, ValueError) as ex:
_LOGGER.error("%s", ex)
self._status = None
self._available = False
# Check if the transition state should expire.
if self._transition_state:
is_transition_state_expired = (
datetime.now() - self._transition_state_start
) > timedelta(seconds=60)
if is_transition_state_expired or self._door.status != door.status:
self._transition_state = None
self._transition_state_start = None
# 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)
)

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

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

22
homeassistant/components/gogogate2/strings.json Normal file
View File

@ -0,0 +1,22 @@
{
"config": {
"error": {
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"invalid_auth": "[%key:common::config_flow::error::invalid_auth%]"
},
"abort": {
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]"
},
"step": {
"user": {
"title": "Setup GogoGate2",
"description": "Provide requisite information below.",
"data": {
"ip_address": "IP Address",
"username": "[%key:common::config_flow::data::username%]",
"password": "[%key:common::config_flow::data::password%]"
}
}
}
}
}

1
homeassistant/generated/config_flows.py
View File

@ -51,6 +51,7 @@ FLOWS = [
"geonetnz_volcano",
"gios",
"glances",
"gogogate2",
"gpslogger",
"griddy",
"hangouts",

6
requirements_all.txt
View File

@ -646,6 +646,9 @@ glances_api==0.2.0
# homeassistant.components.gntp
gntp==1.0.3
# homeassistant.components.gogogate2
gogogate2-api==1.0.3
# homeassistant.components.google
google-api-python-client==1.6.4
@ -1344,9 +1347,6 @@ pyfttt==0.3
# homeassistant.components.skybeacon
pygatt[GATTTOOL]==4.0.5
# homeassistant.components.gogogate2
pygogogate2==0.1.1
# homeassistant.components.gtfs
pygtfs==0.1.5

3
requirements_test_all.txt
View File

@ -278,6 +278,9 @@ gios==0.1.1
# homeassistant.components.glances
glances_api==0.2.0
# homeassistant.components.gogogate2
gogogate2-api==1.0.3
# homeassistant.components.google
google-api-python-client==1.6.4

1
tests/components/gogogate2/__init__.py Normal file
View File

@ -0,0 +1 @@
"""Tests for the GogoGate2 component."""

162
tests/components/gogogate2/common.py Normal file
View File

@ -0,0 +1,162 @@
"""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")

17
tests/components/gogogate2/conftest.py Normal file
View File

@ -0,0 +1,17 @@
"""Fixtures for tests."""
from mock import patch
import pytest
from homeassistant.core import HomeAssistant
from .common import ComponentFactory
@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)

64
tests/components/gogogate2/test_config_flow.py Normal file
View File

@ -0,0 +1,64 @@
"""Tests for the GogoGate2 component."""
from gogogate2_api import GogoGate2Api
from gogogate2_api.common import ApiError
from gogogate2_api.const import ApiErrorCode
from homeassistant.config_entries import SOURCE_USER
from homeassistant.const import 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
async def test_auth_fail(
hass: HomeAssistant, component_factory: ComponentFactory
) -> None:
"""Test authorization failures."""
api_mock: GogoGate2Api = MagicMock(spec=GogoGate2Api)
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_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"}

531
tests/components/gogogate2/test_cover.py Normal file
View File

@ -0,0 +1,531 @@
"""Tests for the GogoGate2 component."""
from datetime import datetime, timedelta
from unittest.mock import MagicMock, patch
from gogogate2_api import GogoGate2Api
from gogogate2_api.common import (
ActivateResponse,
ApiError,
Door,
DoorMode,
DoorStatus,
InfoResponse,
Network,
Outputs,
Wifi,
)
from homeassistant.components.cover import DOMAIN as COVER_DOMAIN
from homeassistant.components.homeassistant import DOMAIN as HA_DOMAIN
from homeassistant.const import (
CONF_IP_ADDRESS,
CONF_NAME,
CONF_PASSWORD,
CONF_PLATFORM,
CONF_USERNAME,
STATE_CLOSED,
STATE_CLOSING,
STATE_OPEN,
STATE_OPENING,
STATE_UNAVAILABLE,
)
from homeassistant.core import HomeAssistant
from .common import ComponentFactory
async def test_import_fail(
hass: HomeAssistant, component_factory: ComponentFactory
) -> None:
"""Test the failure to import."""
api = MagicMock(spec=GogoGate2Api)
api.info.side_effect = ApiError(22, "Error")
component_factory.api_class_mock.return_value = api
await component_factory.configure_component(
cover_config=[
{
CONF_PLATFORM: "gogogate2",
CONF_NAME: "cover0",
CONF_IP_ADDRESS: "127.0.1.0",
CONF_USERNAME: "user0",
CONF_PASSWORD: "password0",
}
]
)
entity_ids = hass.states.async_entity_ids(COVER_DOMAIN)
assert not entity_ids
async def test_import(hass: HomeAssistant, component_factory: ComponentFactory) -> None:
"""Test importing of file based config."""
api0 = MagicMock(spec=GogoGate2Api)
api0.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=""),
)
api1 = MagicMock(spec=GogoGate2Api)
api1.info.return_value = InfoResponse(
user="user1",
gogogatename="gogogatename0",
model="",
apiversion="",
remoteaccessenabled=False,
remoteaccess="321bca.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=""),
)
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=[
{
CONF_PLATFORM: "gogogate2",
CONF_NAME: "cover0",
CONF_IP_ADDRESS: "127.0.1.0",
CONF_USERNAME: "user0",
CONF_PASSWORD: "password0",
},
{
CONF_PLATFORM: "gogogate2",
CONF_NAME: "cover1",
CONF_IP_ADDRESS: "127.0.1.1",
CONF_USERNAME: "user1",
CONF_PASSWORD: "password1",
},
]
)
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()
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",
}
)
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 hass.states.get("cover.door1").state == STATE_OPEN
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)
await hass.services.async_call(
HA_DOMAIN, "update_entity", service_data={"entity_id": "cover.door1"},
)
await hass.async_block_till_done()
assert hass.states.get("cover.door1").state == STATE_CLOSING
component_data.data_update_coordinator.api.info.return_value = closed_door_response
await component_data.data_update_coordinator.async_refresh()
await hass.async_block_till_done()
assert hass.states.get("cover.door1").state == STATE_CLOSED
# Assert mid state changed when new status is received.
await hass.services.async_call(
COVER_DOMAIN, "open_cover", service_data={"entity_id": "cover.door1"},
)
await hass.async_block_till_done()
component_data.api.open_door.assert_called_with(1)
await hass.services.async_call(
HA_DOMAIN, "update_entity", service_data={"entity_id": "cover.door1"},
)
await hass.async_block_till_done()
assert hass.states.get("cover.door1").state == STATE_OPENING
# 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 hass.services.async_call(
HA_DOMAIN, "update_entity", service_data={"entity_id": "cover.door1"},
)
await hass.async_block_till_done()
assert hass.states.get("cover.door1").state == STATE_OPENING
# Assert the mid state times out.
with patch("homeassistant.components.gogogate2.cover.datetime") as datetime_mock:
datetime_mock.now.return_value = datetime.now() + timedelta(seconds=60.1)
component_data.data_update_coordinator.api.info.return_value = (
closed_door_response
)
await component_data.data_update_coordinator.async_refresh()
await hass.services.async_call(
HA_DOMAIN, "update_entity", service_data={"entity_id": "cover.door1"},
)
await hass.async_block_till_done()
assert hass.states.get("cover.door1").state == STATE_CLOSED
async def test_availability(
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",
}
)
assert hass.states.get("cover.door1").state == STATE_OPEN
component_data.api.info.side_effect = Exception("Error")
await component_data.data_update_coordinator.async_refresh()
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()
await hass.async_block_till_done()
assert hass.states.get("cover.door1").state == STATE_CLOSED

28
tests/components/gogogate2/test_init.py Normal file
View File

@ -0,0 +1,28 @@
"""Tests for the GogoGate2 component."""
import pytest
from homeassistant.components.gogogate2 import async_setup_entry
from homeassistant.components.gogogate2.common import GogoGateDataUpdateCoordinator
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryNotReady
from tests.async_mock import MagicMock, patch
from tests.common import MockConfigEntry
async def test_auth_fail(hass: HomeAssistant) -> None:
"""Test authorization failures."""
coordinator_mock: GogoGateDataUpdateCoordinator = MagicMock(
spec=GogoGateDataUpdateCoordinator
)
coordinator_mock.last_update_success = False
config_entry = MockConfigEntry()
config_entry.add_to_hass(hass)
with patch(
"homeassistant.components.gogogate2.get_data_update_coordinator",
return_value=coordinator_mock,
), pytest.raises(ConfigEntryNotReady):
await async_setup_entry(hass, config_entry)