edevau
/
home-assistant-core
mirror of https://github.com/home-assistant/core.git
1
0
Fork 0
Code Issues Projects Releases Activity

Add config flow to uptimerobot (#53938)

This commit is contained in:
Joakim Sørensen 2021-08-04 22:20:03 +02:00 committed by GitHub
parent 4ef859a9a9
commit caf0bdd5b9
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
15 changed files with 464 additions and 81 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
.coveragerc
View File

@ -1113,6 +1113,8 @@ omit =
homeassistant/components/upnp/*
homeassistant/components/upc_connect/*
homeassistant/components/uptimerobot/binary_sensor.py
homeassistant/components/uptimerobot/const.py
homeassistant/components/uptimerobot/entity.py
homeassistant/components/uscis/sensor.py
homeassistant/components/vallox/*
homeassistant/components/vasttrafik/sensor.py

1
.strict-typing
View File

@ -103,6 +103,7 @@ homeassistant.components.tile.*
homeassistant.components.tts.*
homeassistant.components.upcloud.*
homeassistant.components.uptime.*
homeassistant.components.uptimerobot.*
homeassistant.components.vacuum.*
homeassistant.components.water_heater.*
homeassistant.components.weather.*

65
homeassistant/components/uptimerobot/__init__.py
View File

@ -1 +1,64 @@
"""The uptimerobot component."""
"""The Uptime Robot integration."""
from __future__ import annotations
import async_timeout
from pyuptimerobot import UptimeRobot
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_API_KEY
from homeassistant.core import HomeAssistant
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from .const import (
API_ATTR_MONITORS,
API_ATTR_OK,
API_ATTR_STAT,
CONNECTION_ERROR,
COORDINATOR_UPDATE_INTERVAL,
DOMAIN,
LOGGER,
PLATFORMS,
MonitorData,
)
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up Uptime Robot from a config entry."""
hass.data.setdefault(DOMAIN, {})
uptime_robot_api = UptimeRobot()
async def async_update_data() -> list[MonitorData]:
"""Fetch data from API UptimeRobot API."""
async with async_timeout.timeout(10):
monitors = await hass.async_add_executor_job(
uptime_robot_api.getMonitors, entry.data[CONF_API_KEY]
)
if not monitors or monitors.get(API_ATTR_STAT) != API_ATTR_OK:
raise UpdateFailed(CONNECTION_ERROR)
return [
MonitorData.from_dict(monitor)
for monitor in monitors.get(API_ATTR_MONITORS, [])
]
hass.data[DOMAIN][entry.entry_id] = coordinator = DataUpdateCoordinator(
hass,
LOGGER,
name=DOMAIN,
update_method=async_update_data,
update_interval=COORDINATOR_UPDATE_INTERVAL,
)
await coordinator.async_config_entry_first_refresh()
hass.config_entries.async_setup_platforms(entry, PLATFORMS)
return True
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Unload a config entry."""
unload_ok = await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
if unload_ok:
hass.data[DOMAIN].pop(entry.entry_id)
return unload_ok

109
homeassistant/components/uptimerobot/binary_sensor.py
View File

@ -1,9 +1,6 @@
"""A platform that to monitor Uptime Robot monitors."""
from datetime import timedelta
import logging
from __future__ import annotations
import async_timeout
from pyuptimerobot import UptimeRobot
import voluptuous as vol
from homeassistant.components.binary_sensor import (
@ -12,101 +9,61 @@ from homeassistant.components.binary_sensor import (
BinarySensorEntity,
BinarySensorEntityDescription,
)
from homeassistant.const import ATTR_ATTRIBUTION, CONF_API_KEY
from homeassistant.config_entries import SOURCE_IMPORT, ConfigEntry
from homeassistant.const import CONF_API_KEY
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import PlatformNotReady
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.update_coordinator import (
CoordinatorEntity,
DataUpdateCoordinator,
UpdateFailed,
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
from .const import DOMAIN
from .entity import UptimeRobotEntity
PLATFORM_SCHEMA = cv.deprecated(
vol.All(PLATFORM_SCHEMA.extend({vol.Required(CONF_API_KEY): cv.string}))
)
_LOGGER = logging.getLogger(__name__)
ATTR_TARGET = "target"
ATTRIBUTION = "Data provided by Uptime Robot"
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({vol.Required(CONF_API_KEY): cv.string})
async def async_setup_platform(
hass: HomeAssistant, config, async_add_entities, discovery_info=None
):
"""Set up the Uptime Robot binary_sensors."""
uptime_robot_api = UptimeRobot()
api_key = config[CONF_API_KEY]
def api_wrapper():
return uptime_robot_api.getMonitors(api_key)
async def async_update_data():
"""Fetch data from API UptimeRobot API."""
async with async_timeout.timeout(10):
monitors = await hass.async_add_executor_job(api_wrapper)
if not monitors or monitors.get("stat") != "ok":
raise UpdateFailed("Error communicating with Uptime Robot API")
return monitors
coordinator = DataUpdateCoordinator(
hass,
_LOGGER,
name="uptimerobot",
update_method=async_update_data,
update_interval=timedelta(seconds=60),
hass: HomeAssistant,
config: ConfigType,
async_add_entities: AddEntitiesCallback,
discovery_info: DiscoveryInfoType | None = None,
) -> None:
"""Set up the Uptime Robot binary_sensor platform."""
hass.async_create_task(
hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_IMPORT}, data=config
)
)
await coordinator.async_refresh()
if not coordinator.data or coordinator.data.get("stat") != "ok":
_LOGGER.error("Error connecting to Uptime Robot")
raise PlatformNotReady()
async def async_setup_entry(
hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
) -> None:
"""Set up the Uptime Robot binary_sensors."""
coordinator: DataUpdateCoordinator = hass.data[DOMAIN][entry.entry_id]
async_add_entities(
[
UptimeRobotBinarySensor(
coordinator,
BinarySensorEntityDescription(
key=monitor["id"],
name=monitor["friendly_name"],
key=str(monitor.id),
name=monitor.name,
device_class=DEVICE_CLASS_CONNECTIVITY,
),
target=monitor["url"],
target=monitor.url,
)
for monitor in coordinator.data["monitors"]
for monitor in coordinator.data
],
)
class UptimeRobotBinarySensor(BinarySensorEntity, CoordinatorEntity):
class UptimeRobotBinarySensor(UptimeRobotEntity, BinarySensorEntity):
"""Representation of a Uptime Robot binary sensor."""
def __init__(
self,
coordinator: DataUpdateCoordinator,
description: BinarySensorEntityDescription,
target: str,
) -> None:
"""Initialize Uptime Robot the binary sensor."""
super().__init__(coordinator)
self.entity_description = description
self._target = target
self._attr_extra_state_attributes = {
ATTR_ATTRIBUTION: ATTRIBUTION,
ATTR_TARGET: self._target,
}
@property
def is_on(self) -> bool:
"""Return True if the entity is on."""
if monitor := next(
(
monitor
for monitor in self.coordinator.data.get("monitors", [])
if monitor["id"] == self.entity_description.key
),
None,
):
return monitor["status"] == 2
return False
return self.monitor_available

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

@ -0,0 +1,74 @@
"""Config flow for Uptime Robot integration."""
from __future__ import annotations
from pyuptimerobot import UptimeRobot
import voluptuous as vol
from homeassistant import config_entries
from homeassistant.const import CONF_API_KEY
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResult
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.typing import ConfigType
from .const import API_ATTR_OK, API_ATTR_STAT, DOMAIN, LOGGER
STEP_USER_DATA_SCHEMA = vol.Schema({vol.Required(CONF_API_KEY): str})
async def validate_input(hass: HomeAssistant, data: ConfigType) -> None:
"""Validate the user input allows us to connect."""
uptime_robot_api = UptimeRobot()
monitors = await hass.async_add_executor_job(
uptime_robot_api.getMonitors, data[CONF_API_KEY]
)
if not monitors or monitors.get(API_ATTR_STAT) != API_ATTR_OK:
raise CannotConnect("Error communicating with Uptime Robot API")
class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
"""Handle a config flow for Uptime Robot."""
VERSION = 1
async def async_step_user(self, user_input: ConfigType | None = None) -> FlowResult:
"""Handle the initial step."""
errors: dict[str, str] = {}
if user_input is None:
return self.async_show_form(
step_id="user", data_schema=STEP_USER_DATA_SCHEMA, errors=errors
)
try:
await validate_input(self.hass, user_input)
except CannotConnect:
errors["base"] = "cannot_connect"
except Exception: # pylint: disable=broad-except
LOGGER.exception("Unexpected exception")
errors["base"] = "unknown"
else:
return self.async_create_entry(title="", data=user_input)
return self.async_show_form(
step_id="user", data_schema=STEP_USER_DATA_SCHEMA, errors=errors
)
async def async_step_import(self, import_config: ConfigType) -> FlowResult:
"""Import a config entry from configuration.yaml."""
for entry in self._async_current_entries():
if entry.data[CONF_API_KEY] == import_config[CONF_API_KEY]:
LOGGER.warning(
"Already configured. This YAML configuration has already been imported. Please remove it"
)
return self.async_abort(reason="already_configured")
return self.async_create_entry(
title="", data={CONF_API_KEY: import_config[CONF_API_KEY]}
)
class CannotConnect(HomeAssistantError):
"""Error to indicate we cannot connect."""

55
homeassistant/components/uptimerobot/const.py Normal file
View File

@ -0,0 +1,55 @@
"""Constants for the Uptime Robot integration."""
from __future__ import annotations
from dataclasses import dataclass
from datetime import timedelta
from enum import Enum
from logging import Logger, getLogger
from typing import Final
LOGGER: Logger = getLogger(__package__)
COORDINATOR_UPDATE_INTERVAL: timedelta = timedelta(seconds=60)
DOMAIN: Final = "uptimerobot"
PLATFORMS: Final = ["binary_sensor"]
CONNECTION_ERROR: Final = "Error connecting to the Uptime Robot API"
ATTRIBUTION: Final = "Data provided by Uptime Robot"
ATTR_TARGET: Final = "target"
API_ATTR_STAT: Final = "stat"
API_ATTR_OK: Final = "ok"
API_ATTR_MONITORS: Final = "monitors"
class MonitorType(Enum):
"""Monitors type."""
HTTP = 1
keyword = 2
ping = 3
@dataclass
class MonitorData:
"""Dataclass for monitors."""
id: int
status: int
url: str
name: str
type: MonitorType
@staticmethod
def from_dict(monitor: dict) -> MonitorData:
"""Create a new monitor from a dict."""
return MonitorData(
id=monitor["id"],
status=monitor["status"],
url=monitor["url"],
name=monitor["friendly_name"],
type=MonitorType(monitor["type"]),
)

75
homeassistant/components/uptimerobot/entity.py Normal file
View File

@ -0,0 +1,75 @@
"""Base UptimeRobot entity."""
from __future__ import annotations
from homeassistant.const import ATTR_ATTRIBUTION
from homeassistant.helpers.entity import DeviceInfo, EntityDescription
from homeassistant.helpers.update_coordinator import (
CoordinatorEntity,
DataUpdateCoordinator,
)
from .const import ATTR_TARGET, ATTRIBUTION, DOMAIN, MonitorData
class UptimeRobotEntity(CoordinatorEntity):
"""Base UptimeRobot entity."""
def __init__(
self,
coordinator: DataUpdateCoordinator,
description: EntityDescription,
target: str,
) -> None:
"""Initialize Uptime Robot entities."""
super().__init__(coordinator)
self.entity_description = description
self._target = target
self._attr_extra_state_attributes = {
ATTR_ATTRIBUTION: ATTRIBUTION,
ATTR_TARGET: self._target,
}
@property
def unique_id(self) -> str | None:
"""Return the unique_id of the entity."""
return str(self.monitor.id) if self.monitor else None
@property
def device_info(self) -> DeviceInfo:
"""Return device information about this AdGuard Home instance."""
if self.monitor:
return {
"identifiers": {(DOMAIN, str(self.monitor.id))},
"name": "Uptime Robot",
"manufacturer": "Uptime Robot Team",
"entry_type": "service",
"model": self.monitor.type.name,
}
return {}
@property
def monitors(self) -> list[MonitorData]:
"""Return all monitors."""
return self.coordinator.data or []
@property
def monitor(self) -> MonitorData | None:
"""Return the monitor for this entity."""
return next(
(
monitor
for monitor in self.monitors
if str(monitor.id) == self.entity_description.key
),
None,
)
@property
def monitor_available(self) -> bool:
"""Returtn if the monitor is available."""
return self.monitor.status == 2 if self.monitor else False
@property
def available(self) -> bool:
"""Returtn if entity is available."""
return self.monitor is not None

13
homeassistant/components/uptimerobot/manifest.json
View File

@ -2,7 +2,12 @@
"domain": "uptimerobot",
"name": "Uptime Robot",
"documentation": "https://www.home-assistant.io/integrations/uptimerobot",
"requirements": ["pyuptimerobot==0.0.5"],
"codeowners": ["@ludeeus"],
"iot_class": "cloud_polling"
}
"requirements": [
"pyuptimerobot==0.0.5"
],
"codeowners": [
"@ludeeus"
],
"iot_class": "cloud_polling",
"config_flow": true
}

18
homeassistant/components/uptimerobot/strings.json Normal file
View File

@ -0,0 +1,18 @@
{
"config": {
"step": {
"user": {
"data": {
"api_key": "[%key:common::config_flow::data::api_key%]"
}
}
},
"error": {
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"unknown": "[%key:common::config_flow::error::unknown%]"
},
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]"
}
}
}

19
homeassistant/components/uptimerobot/translations/en.json Normal file
View File

@ -0,0 +1,19 @@
{
"config": {
"abort": {
"already_configured": "Device is already configured"
},
"error": {
"cannot_connect": "Failed to connect",
"invalid_auth": "Invalid authentication",
"unknown": "Unexpected error"
},
"step": {
"user": {
"data": {
"api_key": "API Key"
}
}
}
}
}

1
homeassistant/generated/config_flows.py
View File

@ -282,6 +282,7 @@ FLOWS = [
"upb",
"upcloud",
"upnp",
"uptimerobot",
"velbus",
"vera",
"verisure",

11
mypy.ini
View File

@ -1144,6 +1144,17 @@ no_implicit_optional = true
warn_return_any = true
warn_unreachable = true
[mypy-homeassistant.components.uptimerobot.*]
check_untyped_defs = true
disallow_incomplete_defs = true
disallow_subclassing_any = true
disallow_untyped_calls = true
disallow_untyped_decorators = true
disallow_untyped_defs = true
no_implicit_optional = true
warn_return_any = true
warn_unreachable = true
[mypy-homeassistant.components.vacuum.*]
check_untyped_defs = true
disallow_incomplete_defs = true

3
requirements_test_all.txt
View File

@ -1076,6 +1076,9 @@ pytraccar==0.9.0
# homeassistant.components.tradfri
pytradfri[async]==7.0.6
# homeassistant.components.uptimerobot
pyuptimerobot==0.0.5
# homeassistant.components.vera
pyvera==0.3.13

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

@ -0,0 +1 @@
"""Tests for the Uptime Robot integration."""

98
tests/components/uptimerobot/test_config_flow.py Normal file
View File

@ -0,0 +1,98 @@
"""Test the Uptime Robot config flow."""
from unittest.mock import patch
from homeassistant import config_entries, setup
from homeassistant.components.uptimerobot.const import (
API_ATTR_MONITORS,
API_ATTR_OK,
API_ATTR_STAT,
DOMAIN,
)
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import (
RESULT_TYPE_ABORT,
RESULT_TYPE_CREATE_ENTRY,
RESULT_TYPE_FORM,
)
async def test_form(hass: HomeAssistant) -> None:
"""Test we get the form."""
await setup.async_setup_component(hass, "persistent_notification", {})
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == RESULT_TYPE_FORM
assert result["errors"] == {}
with patch(
"pyuptimerobot.UptimeRobot.getMonitors",
return_value={API_ATTR_STAT: API_ATTR_OK, API_ATTR_MONITORS: []},
), patch(
"homeassistant.components.uptimerobot.async_setup_entry",
return_value=True,
) as mock_setup_entry:
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{"api_key": "1234"},
)
await hass.async_block_till_done()
assert result2["type"] == RESULT_TYPE_CREATE_ENTRY
assert result2["title"] == ""
assert result2["data"] == {"api_key": "1234"}
assert len(mock_setup_entry.mock_calls) == 1
async def test_form_cannot_connect(hass: HomeAssistant) -> None:
"""Test we handle cannot connect error."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
with patch("pyuptimerobot.UptimeRobot.getMonitors", return_value=None):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{"api_key": "1234"},
)
assert result2["type"] == RESULT_TYPE_FORM
assert result2["errors"] == {"base": "cannot_connect"}
async def test_flow_import(hass):
"""Test an import flow."""
with patch(
"pyuptimerobot.UptimeRobot.getMonitors",
return_value={API_ATTR_STAT: API_ATTR_OK, API_ATTR_MONITORS: []},
), patch(
"homeassistant.components.uptimerobot.async_setup_entry",
return_value=True,
) as mock_setup_entry:
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_IMPORT},
data={"platform": DOMAIN, "api_key": "1234"},
)
await hass.async_block_till_done()
assert len(mock_setup_entry.mock_calls) == 1
assert result["type"] == RESULT_TYPE_CREATE_ENTRY
assert result["data"] == {"api_key": "1234"}
with patch(
"pyuptimerobot.UptimeRobot.getMonitors",
return_value={API_ATTR_STAT: API_ATTR_OK, API_ATTR_MONITORS: []},
), patch(
"homeassistant.components.uptimerobot.async_setup_entry",
return_value=True,
) as mock_setup_entry:
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_IMPORT},
data={"platform": DOMAIN, "api_key": "1234"},
)
await hass.async_block_till_done()
assert result["type"] == RESULT_TYPE_ABORT
assert result["reason"] == "already_configured"