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

Add config flow and device registry to fritzbox integration (#31240) 

* add config flow

* fix pylint

* update lib

* Update config_flow.py

* remote devices layer in config

* add default host

* avoid double setups of entities

* remove async_setup_platform

* store entities in hass.data

* pass fritz connection together with config_entry

* fritz connections try no4 (or is it even more)

* fix comments

* add unloading

* fixed comments

* Update config_flow.py

* Update const.py

* Update config_flow.py

* Update __init__.py

* Update config_flow.py

* Update __init__.py

* Update __init__.py

* Update config_flow.py

* Update __init__.py

* Update __init__.py

* Update __init__.py

* Update config_flow.py

* add init tests

* test unloading

* add switch tests

* add sensor tests

* add climate tests

* test target temperature

* mock config to package

* comments

* test binary sensor state

* add config flow tests

* comments

* add missing tests

* minor

* remove string title

* deprecate yaml

* don't change yaml

* get devices async

* minor

* add devices again

* comments fixed

* unique_id fixes

* fix patches

* Fix schema

Co-authored-by: Martin Hjelmare <marhje52@gmail.com>
pull/34464/head
escoand 2020-04-20 15:00:07 +02:00 committed by GitHub
parent 2123f6f133
commit c87ecf0ff6
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
22 changed files with 1403 additions and 271 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
.coveragerc
View File

@ -241,7 +241,6 @@ omit =
homeassistant/components/freebox/sensor.py
homeassistant/components/freebox/switch.py
homeassistant/components/fritz/device_tracker.py
homeassistant/components/fritzbox/*
homeassistant/components/fritzbox_callmonitor/sensor.py
homeassistant/components/fritzbox_netmonitor/sensor.py
homeassistant/components/fronius/sensor.py

156
homeassistant/components/fritzbox/__init__.py
View File

@ -1,7 +1,8 @@
"""Support for AVM Fritz!Box smarthome devices."""
import logging
import asyncio
import socket
from pyfritzhome import Fritzhome, LoginError
from pyfritzhome import Fritzhome
import voluptuous as vol
from homeassistant.const import (
@ -11,80 +12,103 @@ from homeassistant.const import (
CONF_USERNAME,
EVENT_HOMEASSISTANT_STOP,
)
from homeassistant.helpers import discovery
import homeassistant.helpers.config_validation as cv
_LOGGER = logging.getLogger(__name__)
from .const import CONF_CONNECTIONS, DEFAULT_HOST, DEFAULT_USERNAME, DOMAIN, PLATFORMS
SUPPORTED_DOMAINS = ["binary_sensor", "climate", "switch", "sensor"]
DOMAIN = "fritzbox"
ATTR_STATE_BATTERY_LOW = "battery_low"
ATTR_STATE_DEVICE_LOCKED = "device_locked"
ATTR_STATE_HOLIDAY_MODE = "holiday_mode"
ATTR_STATE_LOCKED = "locked"
ATTR_STATE_SUMMER_MODE = "summer_mode"
ATTR_STATE_WINDOW_OPEN = "window_open"
def ensure_unique_hosts(value):
"""Validate that all configs have a unique host."""
vol.Schema(vol.Unique("duplicate host entries found"))(
[socket.gethostbyname(entry[CONF_HOST]) for entry in value]
)
return value
CONFIG_SCHEMA = vol.Schema(
{
DOMAIN: vol.Schema(
{
vol.Required(CONF_DEVICES): vol.All(
cv.ensure_list,
[
vol.Schema(
{
vol.Required(CONF_HOST): cv.string,
vol.Required(CONF_PASSWORD): cv.string,
vol.Required(CONF_USERNAME): cv.string,
}
)
],
)
}
)
},
vol.All(
cv.deprecated(DOMAIN),
{
DOMAIN: vol.Schema(
{
vol.Required(CONF_DEVICES): vol.All(
cv.ensure_list,
[
vol.Schema(
{
vol.Required(
CONF_HOST, default=DEFAULT_HOST
): cv.string,
vol.Required(CONF_PASSWORD): cv.string,
vol.Required(
CONF_USERNAME, default=DEFAULT_USERNAME
): cv.string,
}
)
],
ensure_unique_hosts,
)
}
)
},
),
extra=vol.ALLOW_EXTRA,
)
def setup(hass, config):
"""Set up the fritzbox component."""
fritz_list = []
configured_devices = config[DOMAIN].get(CONF_DEVICES)
for device in configured_devices:
host = device.get(CONF_HOST)
username = device.get(CONF_USERNAME)
password = device.get(CONF_PASSWORD)
fritzbox = Fritzhome(host=host, user=username, password=password)
try:
fritzbox.login()
_LOGGER.info("Connected to device %s", device)
except LoginError:
_LOGGER.warning("Login to Fritz!Box %s as %s failed", host, username)
continue
fritz_list.append(fritzbox)
if not fritz_list:
_LOGGER.info("No fritzboxes configured")
return False
hass.data[DOMAIN] = fritz_list
def logout_fritzboxes(event):
"""Close all connections to the fritzboxes."""
for fritz in fritz_list:
fritz.logout()
hass.bus.listen_once(EVENT_HOMEASSISTANT_STOP, logout_fritzboxes)
for domain in SUPPORTED_DOMAINS:
discovery.load_platform(hass, domain, DOMAIN, {}, config)
async def async_setup(hass, config):
"""Set up the AVM Fritz!Box integration."""
if DOMAIN in config:
for entry_config in config[DOMAIN][CONF_DEVICES]:
hass.async_create_task(
hass.config_entries.flow.async_init(
DOMAIN, context={"source": "import"}, data=entry_config
)
)
return True
async def async_setup_entry(hass, entry):
"""Set up the AVM Fritz!Box platforms."""
fritz = Fritzhome(
host=entry.data[CONF_HOST],
user=entry.data[CONF_USERNAME],
password=entry.data[CONF_PASSWORD],
)
await hass.async_add_executor_job(fritz.login)
hass.data.setdefault(DOMAIN, {CONF_CONNECTIONS: {}, CONF_DEVICES: set()})
hass.data[DOMAIN][CONF_CONNECTIONS][entry.entry_id] = fritz
for component in PLATFORMS:
hass.async_create_task(
hass.config_entries.async_forward_entry_setup(entry, component)
)
def logout_fritzbox(event):
"""Close connections to this fritzbox."""
fritz.logout()
hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP, logout_fritzbox)
return True
async def async_unload_entry(hass, entry):
"""Unloading the AVM Fritz!Box platforms."""
fritz = hass.data[DOMAIN][CONF_CONNECTIONS][entry.entry_id]
await hass.async_add_executor_job(fritz.logout)
unload_ok = all(
await asyncio.gather(
*[
hass.config_entries.async_forward_entry_unload(entry, component)
for component in PLATFORMS
]
)
)
if unload_ok:
hass.data[DOMAIN][CONF_CONNECTIONS].pop(entry.entry_id)
return unload_ok

45
homeassistant/components/fritzbox/binary_sensor.py
View File

@ -1,27 +1,24 @@
"""Support for Fritzbox binary sensors."""
import logging
import requests
from homeassistant.components.binary_sensor import BinarySensorDevice
from homeassistant.const import CONF_DEVICES
from . import DOMAIN as FRITZBOX_DOMAIN
_LOGGER = logging.getLogger(__name__)
from .const import CONF_CONNECTIONS, DOMAIN as FRITZBOX_DOMAIN, LOGGER
def setup_platform(hass, config, add_entities, discovery_info=None):
"""Set up the Fritzbox binary sensor platform."""
devices = []
fritz_list = hass.data[FRITZBOX_DOMAIN]
async def async_setup_entry(hass, config_entry, async_add_entities):
"""Set up the Fritzbox binary sensor from config_entry."""
entities = []
devices = hass.data[FRITZBOX_DOMAIN][CONF_DEVICES]
fritz = hass.data[FRITZBOX_DOMAIN][CONF_CONNECTIONS][config_entry.entry_id]
for fritz in fritz_list:
device_list = fritz.get_devices()
for device in device_list:
if device.has_alarm:
devices.append(FritzboxBinarySensor(device, fritz))
for device in await hass.async_add_executor_job(fritz.get_devices):
if device.has_alarm and device.ain not in devices:
entities.append(FritzboxBinarySensor(device, fritz))
devices.add(device.ain)
add_entities(devices, True)
async_add_entities(entities, True)
class FritzboxBinarySensor(BinarySensorDevice):
@ -32,6 +29,22 @@ class FritzboxBinarySensor(BinarySensorDevice):
self._device = device
self._fritz = fritz
@property
def device_info(self):
"""Return device specific attributes."""
return {
"name": self.name,
"identifiers": {(FRITZBOX_DOMAIN, self._device.ain)},
"manufacturer": self._device.manufacturer,
"model": self._device.productname,
"sw_version": self._device.fw_version,
}
@property
def unique_id(self):
"""Return the unique ID of the device."""
return self._device.ain
@property
def name(self):
"""Return the name of the entity."""
@ -54,5 +67,5 @@ class FritzboxBinarySensor(BinarySensorDevice):
try:
self._device.update()
except requests.exceptions.HTTPError as ex:
_LOGGER.warning("Connection error: %s", ex)
LOGGER.warning("Connection error: %s", ex)
self._fritz.login()

47
homeassistant/components/fritzbox/climate.py
View File

@ -1,6 +1,4 @@
"""Support for AVM Fritz!Box smarthome thermostate devices."""
import logging
import requests
from homeassistant.components.climate import ClimateDevice
@ -16,22 +14,23 @@ from homeassistant.components.climate.const import (
from homeassistant.const import (
ATTR_BATTERY_LEVEL,
ATTR_TEMPERATURE,
CONF_DEVICES,
PRECISION_HALVES,
TEMP_CELSIUS,
)
from . import (
from .const import (
ATTR_STATE_BATTERY_LOW,
ATTR_STATE_DEVICE_LOCKED,
ATTR_STATE_HOLIDAY_MODE,
ATTR_STATE_LOCKED,
ATTR_STATE_SUMMER_MODE,
ATTR_STATE_WINDOW_OPEN,
CONF_CONNECTIONS,
DOMAIN as FRITZBOX_DOMAIN,
LOGGER,
)
_LOGGER = logging.getLogger(__name__)
SUPPORT_FLAGS = SUPPORT_TARGET_TEMPERATURE | SUPPORT_PRESET_MODE
OPERATION_LIST = [HVAC_MODE_HEAT, HVAC_MODE_OFF]
@ -48,18 +47,18 @@ ON_REPORT_SET_TEMPERATURE = 30.0
OFF_REPORT_SET_TEMPERATURE = 0.0
def setup_platform(hass, config, add_entities, discovery_info=None):
"""Set up the Fritzbox smarthome thermostat platform."""
devices = []
fritz_list = hass.data[FRITZBOX_DOMAIN]
async def async_setup_entry(hass, config_entry, async_add_entities):
"""Set up the Fritzbox smarthome thermostat from config_entry."""
entities = []
devices = hass.data[FRITZBOX_DOMAIN][CONF_DEVICES]
fritz = hass.data[FRITZBOX_DOMAIN][CONF_CONNECTIONS][config_entry.entry_id]
for fritz in fritz_list:
device_list = fritz.get_devices()
for device in device_list:
if device.has_thermostat:
devices.append(FritzboxThermostat(device, fritz))
for device in await hass.async_add_executor_job(fritz.get_devices):
if device.has_thermostat and device.ain not in devices:
entities.append(FritzboxThermostat(device, fritz))
devices.add(device.ain)
add_entities(devices)
async_add_entities(entities)
class FritzboxThermostat(ClimateDevice):
@ -74,6 +73,22 @@ class FritzboxThermostat(ClimateDevice):
self._comfort_temperature = self._device.comfort_temperature
self._eco_temperature = self._device.eco_temperature
@property
def device_info(self):
"""Return device specific attributes."""
return {
"name": self.name,
"identifiers": {(FRITZBOX_DOMAIN, self._device.ain)},
"manufacturer": self._device.manufacturer,
"model": self._device.productname,
"sw_version": self._device.fw_version,
}
@property
def unique_id(self):
"""Return the unique ID of the device."""
return self._device.ain
@property
def supported_features(self):
"""Return the list of supported features."""
@ -205,5 +220,5 @@ class FritzboxThermostat(ClimateDevice):
self._comfort_temperature = self._device.comfort_temperature
self._eco_temperature = self._device.eco_temperature
except requests.exceptions.HTTPError as ex:
_LOGGER.warning("Fritzbox connection error: %s", ex)
LOGGER.warning("Fritzbox connection error: %s", ex)
self._fritz.login()

151
homeassistant/components/fritzbox/config_flow.py Normal file
View File

@ -0,0 +1,151 @@
"""Config flow for AVM Fritz!Box."""
from urllib.parse import urlparse
from pyfritzhome import Fritzhome, LoginError
import voluptuous as vol
from homeassistant import config_entries
from homeassistant.components.ssdp import ATTR_SSDP_LOCATION, ATTR_UPNP_FRIENDLY_NAME
from homeassistant.const import CONF_HOST, CONF_PASSWORD, CONF_USERNAME
# pylint:disable=unused-import
from .const import DEFAULT_HOST, DEFAULT_USERNAME, DOMAIN
DATA_SCHEMA_USER = vol.Schema(
{
vol.Required(CONF_HOST, default=DEFAULT_HOST): str,
vol.Required(CONF_USERNAME, default=DEFAULT_USERNAME): str,
vol.Required(CONF_PASSWORD): str,
}
)
DATA_SCHEMA_CONFIRM = vol.Schema(
{
vol.Required(CONF_USERNAME, default=DEFAULT_USERNAME): str,
vol.Required(CONF_PASSWORD): str,
}
)
RESULT_AUTH_FAILED = "auth_failed"
RESULT_NOT_FOUND = "not_found"
RESULT_SUCCESS = "success"
class FritzboxConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
"""Handle a AVM Fritz!Box config flow."""
VERSION = 1
CONNECTION_CLASS = config_entries.CONN_CLASS_LOCAL_POLL
# pylint: disable=no-member # https://github.com/PyCQA/pylint/issues/3167
def __init__(self):
"""Initialize flow."""
self._host = None
self._manufacturer = None
self._model = None
self._name = None
self._password = None
self._username = None
def _get_entry(self):
return self.async_create_entry(
title=self._name,
data={
CONF_HOST: self._host,
CONF_PASSWORD: self._password,
CONF_USERNAME: self._username,
},
)
def _try_connect(self):
"""Try to connect and check auth."""
fritzbox = Fritzhome(
host=self._host, user=self._username, password=self._password
)
try:
fritzbox.login()
fritzbox.logout()
return RESULT_SUCCESS
except OSError:
return RESULT_NOT_FOUND
except LoginError:
return RESULT_AUTH_FAILED
async def async_step_import(self, user_input=None):
"""Handle configuration by yaml file."""
return await self.async_step_user(user_input)
async def async_step_user(self, user_input=None):
"""Handle a flow initialized by the user."""
errors = {}
if user_input is not None:
for entry in self.hass.config_entries.async_entries(DOMAIN):
if entry.data[CONF_HOST] == user_input[CONF_HOST]:
if entry.data != user_input:
self.hass.config_entries.async_update_entry(
entry, data=user_input
)
return self.async_abort(reason="already_configured")
self._host = user_input[CONF_HOST]
self._name = user_input[CONF_HOST]
self._password = user_input[CONF_PASSWORD]
self._username = user_input[CONF_USERNAME]
result = await self.hass.async_add_executor_job(self._try_connect)
if result == RESULT_SUCCESS:
return self._get_entry()
if result != RESULT_AUTH_FAILED:
return self.async_abort(reason=result)
errors["base"] = result
return self.async_show_form(
step_id="user", data_schema=DATA_SCHEMA_USER, errors=errors
)
async def async_step_ssdp(self, user_input):
"""Handle a flow initialized by discovery."""
host = urlparse(user_input[ATTR_SSDP_LOCATION]).hostname
self.context[CONF_HOST] = host
for progress in self._async_in_progress():
if progress.get("context", {}).get(CONF_HOST) == host:
return self.async_abort(reason="already_in_progress")
for entry in self.hass.config_entries.async_entries(DOMAIN):
if entry.data[CONF_HOST] == host:
if entry.data != user_input:
self.hass.config_entries.async_update_entry(entry, data=user_input)
return self.async_abort(reason="already_configured")
self._host = host
self._name = user_input[ATTR_UPNP_FRIENDLY_NAME]
self.context["title_placeholders"] = {"name": self._name}
return await self.async_step_confirm()
async def async_step_confirm(self, user_input=None):
"""Handle user-confirmation of discovered node."""
errors = {}
if user_input is not None:
self._password = user_input[CONF_PASSWORD]
self._username = user_input[CONF_USERNAME]
result = await self.hass.async_add_executor_job(self._try_connect)
if result == RESULT_SUCCESS:
return self._get_entry()
if result != RESULT_AUTH_FAILED:
return self.async_abort(reason=result)
errors["base"] = result
return self.async_show_form(
step_id="confirm",
data_schema=DATA_SCHEMA_CONFIRM,
description_placeholders={"name": self._name},
errors=errors,
)

25
homeassistant/components/fritzbox/const.py Normal file
View File

@ -0,0 +1,25 @@
"""Constants for the AVM Fritz!Box integration."""
import logging
ATTR_STATE_BATTERY_LOW = "battery_low"
ATTR_STATE_DEVICE_LOCKED = "device_locked"
ATTR_STATE_HOLIDAY_MODE = "holiday_mode"
ATTR_STATE_LOCKED = "locked"
ATTR_STATE_SUMMER_MODE = "summer_mode"
ATTR_STATE_WINDOW_OPEN = "window_open"
ATTR_TEMPERATURE_UNIT = "temperature_unit"
ATTR_TOTAL_CONSUMPTION = "total_consumption"
ATTR_TOTAL_CONSUMPTION_UNIT = "total_consumption_unit"
CONF_CONNECTIONS = "connections"
DEFAULT_HOST = "fritz.box"
DEFAULT_USERNAME = "admin"
DOMAIN = "fritzbox"
LOGGER = logging.getLogger(__package__)
PLATFORMS = ["binary_sensor", "climate", "switch", "sensor"]

11
homeassistant/components/fritzbox/manifest.json
View File

@ -2,6 +2,13 @@
"domain": "fritzbox",
"name": "AVM FRITZ!Box",
"documentation": "https://www.home-assistant.io/integrations/fritzbox",
"requirements": ["pyfritzhome==0.4.0"],
"codeowners": []
"requirements": ["pyfritzhome==0.4.2"],
"ssdp": [
{
"st": "urn:schemas-upnp-org:device:fritzbox:1"
}
],
"dependencies": [],
"codeowners": [],
"config_flow": true
}

62
homeassistant/components/fritzbox/sensor.py
View File

@ -1,33 +1,35 @@
"""Support for AVM Fritz!Box smarthome temperature sensor only devices."""
import logging
import requests
from homeassistant.const import TEMP_CELSIUS
from homeassistant.const import CONF_DEVICES, TEMP_CELSIUS
from homeassistant.helpers.entity import Entity
from . import ATTR_STATE_DEVICE_LOCKED, ATTR_STATE_LOCKED, DOMAIN as FRITZBOX_DOMAIN
_LOGGER = logging.getLogger(__name__)
from .const import (
ATTR_STATE_DEVICE_LOCKED,
ATTR_STATE_LOCKED,
CONF_CONNECTIONS,
DOMAIN as FRITZBOX_DOMAIN,
LOGGER,
)
def setup_platform(hass, config, add_entities, discovery_info=None):
"""Set up the Fritzbox smarthome sensor platform."""
_LOGGER.debug("Initializing fritzbox temperature sensors")
devices = []
fritz_list = hass.data[FRITZBOX_DOMAIN]
async def async_setup_entry(hass, config_entry, async_add_entities):
"""Set up the Fritzbox smarthome sensor from config_entry."""
entities = []
devices = hass.data[FRITZBOX_DOMAIN][CONF_DEVICES]
fritz = hass.data[FRITZBOX_DOMAIN][CONF_CONNECTIONS][config_entry.entry_id]
for fritz in fritz_list:
device_list = fritz.get_devices()
for device in device_list:
if (
device.has_temperature_sensor
and not device.has_switch
and not device.has_thermostat
):
devices.append(FritzBoxTempSensor(device, fritz))
for device in await hass.async_add_executor_job(fritz.get_devices):
if (
device.has_temperature_sensor
and not device.has_switch
and not device.has_thermostat
and device.ain not in devices
):
entities.append(FritzBoxTempSensor(device, fritz))
devices.add(device.ain)
add_entities(devices)
async_add_entities(entities)
class FritzBoxTempSensor(Entity):
@ -38,6 +40,22 @@ class FritzBoxTempSensor(Entity):
self._device = device
self._fritz = fritz
@property
def device_info(self):
"""Return device specific attributes."""
return {
"name": self.name,
"identifiers": {(FRITZBOX_DOMAIN, self._device.ain)},
"manufacturer": self._device.manufacturer,
"model": self._device.productname,
"sw_version": self._device.fw_version,
}
@property
def unique_id(self):
"""Return the unique ID of the device."""
return self._device.ain
@property
def name(self):
"""Return the name of the device."""
@ -58,7 +76,7 @@ class FritzBoxTempSensor(Entity):
try:
self._device.update()
except requests.exceptions.HTTPError as ex:
_LOGGER.warning("Fritzhome connection error: %s", ex)
LOGGER.warning("Fritzhome connection error: %s", ex)
self._fritz.login()
@property

32
homeassistant/components/fritzbox/strings.json Normal file
View File

@ -0,0 +1,32 @@
{
"config": {
"flow_title": "AVM FRITZ!Box: {name}",
"step": {
"user": {
"title": "AVM FRITZ!Box",
"description": "Enter your AVM FRITZ!Box information.",
"data": {
"host": "Host or IP address",
"username": "Username",
"password": "Password"
}
},
"confirm": {
"title": "AVM FRITZ!Box",
"description": "Do you want to set up {name}?",
"data": {
"username": "Username",
"password": "Password"
}
}
},
"abort": {
"already_in_progress": "AVM FRITZ!Box configuration is already in progress.",
"already_configured": "This AVM FRITZ!Box is already configured.",
"not_found": "No supported AVM FRITZ!Box found on the network."
},
"error": {
"auth_failed": "Username and/or password are incorrect."
}
}
}

64
homeassistant/components/fritzbox/switch.py
View File

@ -1,34 +1,40 @@
"""Support for AVM Fritz!Box smarthome switch devices."""
import logging
import requests
from homeassistant.components.switch import SwitchDevice
from homeassistant.const import ATTR_TEMPERATURE, ENERGY_KILO_WATT_HOUR, TEMP_CELSIUS
from homeassistant.const import (
ATTR_TEMPERATURE,
CONF_DEVICES,
ENERGY_KILO_WATT_HOUR,
TEMP_CELSIUS,
)
from . import ATTR_STATE_DEVICE_LOCKED, ATTR_STATE_LOCKED, DOMAIN as FRITZBOX_DOMAIN
from .const import (
ATTR_STATE_DEVICE_LOCKED,
ATTR_STATE_LOCKED,
ATTR_TEMPERATURE_UNIT,
ATTR_TOTAL_CONSUMPTION,
ATTR_TOTAL_CONSUMPTION_UNIT,
CONF_CONNECTIONS,
DOMAIN as FRITZBOX_DOMAIN,
LOGGER,
)
_LOGGER = logging.getLogger(__name__)
ATTR_TOTAL_CONSUMPTION = "total_consumption"
ATTR_TOTAL_CONSUMPTION_UNIT = "total_consumption_unit"
ATTR_TOTAL_CONSUMPTION_UNIT_VALUE = ENERGY_KILO_WATT_HOUR
ATTR_TEMPERATURE_UNIT = "temperature_unit"
async def async_setup_entry(hass, config_entry, async_add_entities):
"""Set up the Fritzbox smarthome switch from config_entry."""
entities = []
devices = hass.data[FRITZBOX_DOMAIN][CONF_DEVICES]
fritz = hass.data[FRITZBOX_DOMAIN][CONF_CONNECTIONS][config_entry.entry_id]
def setup_platform(hass, config, add_entities, discovery_info=None):
"""Set up the Fritzbox smarthome switch platform."""
devices = []
fritz_list = hass.data[FRITZBOX_DOMAIN]
for device in await hass.async_add_executor_job(fritz.get_devices):
if device.has_switch and device.ain not in devices:
entities.append(FritzboxSwitch(device, fritz))
devices.add(device.ain)
for fritz in fritz_list:
device_list = fritz.get_devices()
for device in device_list:
if device.has_switch:
devices.append(FritzboxSwitch(device, fritz))
add_entities(devices)
async_add_entities(entities)
class FritzboxSwitch(SwitchDevice):
@ -39,6 +45,22 @@ class FritzboxSwitch(SwitchDevice):
self._device = device
self._fritz = fritz
@property
def device_info(self):
"""Return device specific attributes."""
return {
"name": self.name,
"identifiers": {(FRITZBOX_DOMAIN, self._device.ain)},
"manufacturer": self._device.manufacturer,
"model": self._device.productname,
"sw_version": self._device.fw_version,
}
@property
def unique_id(self):
"""Return the unique ID of the device."""
return self._device.ain
@property
def available(self):
"""Return if switch is available."""
@ -67,7 +89,7 @@ class FritzboxSwitch(SwitchDevice):
try:
self._device.update()
except requests.exceptions.HTTPError as ex:
_LOGGER.warning("Fritzhome connection error: %s", ex)
LOGGER.warning("Fritzhome connection error: %s", ex)
self._fritz.login()
@property

1
homeassistant/generated/config_flows.py
View File

@ -35,6 +35,7 @@ FLOWS = [
"flume",
"flunearyou",
"freebox",
"fritzbox",
"garmin_connect",
"gdacs",
"geofency",

5
homeassistant/generated/ssdp.py
View File

@ -17,6 +17,11 @@ SSDP = {
"manufacturer": "DIRECTV"
}
],
"fritzbox": [
{
"st": "urn:schemas-upnp-org:device:fritzbox:1"
}
],
"harmony": [
{
"deviceType": "urn:myharmony-com:device:harmony:1",

2
requirements_all.txt
View File

@ -1298,7 +1298,7 @@ pyflunearyou==1.0.7
pyfnip==0.2
# homeassistant.components.fritzbox
pyfritzhome==0.4.0
pyfritzhome==0.4.2
# homeassistant.components.fronius
pyfronius==0.4.6

2
requirements_test_all.txt
View File

@ -511,7 +511,7 @@ pyflume==0.4.0
pyflunearyou==1.0.7
# homeassistant.components.fritzbox
pyfritzhome==0.4.0
pyfritzhome==0.4.2
# homeassistant.components.ifttt
pyfttt==0.3

100
tests/components/fritzbox/__init__.py
View File

@ -1 +1,99 @@
"""Tests for the FritzBox! integration."""
"""Tests for the AVM Fritz!Box integration."""
from unittest.mock import Mock
from homeassistant.components.fritzbox.const import DOMAIN
from homeassistant.const import CONF_DEVICES, CONF_HOST, CONF_PASSWORD, CONF_USERNAME
MOCK_CONFIG = {
DOMAIN: {
CONF_DEVICES: [
{
CONF_HOST: "fake_host",
CONF_PASSWORD: "fake_pass",
CONF_USERNAME: "fake_user",
}
]
}
}
class FritzDeviceBinarySensorMock(Mock):
"""Mock of a AVM Fritz!Box binary sensor device."""
ain = "fake_ain"
alert_state = "fake_state"
fw_version = "1.2.3"
has_alarm = True
has_switch = False
has_temperature_sensor = False
has_thermostat = False
manufacturer = "fake_manufacturer"
name = "fake_name"
present = True
productname = "fake_productname"
class FritzDeviceClimateMock(Mock):
"""Mock of a AVM Fritz!Box climate device."""
actual_temperature = 18.0
ain = "fake_ain"
alert_state = "fake_state"
battery_level = 23
battery_low = True
comfort_temperature = 22.0
device_lock = "fake_locked_device"
eco_temperature = 16.0
fw_version = "1.2.3"
has_alarm = False
has_switch = False
has_temperature_sensor = False
has_thermostat = True
holiday_active = "fake_holiday"
lock = "fake_locked"
manufacturer = "fake_manufacturer"
name = "fake_name"
present = True
productname = "fake_productname"
summer_active = "fake_summer"
target_temperature = 19.5
window_open = "fake_window"
class FritzDeviceSensorMock(Mock):
"""Mock of a AVM Fritz!Box sensor device."""
ain = "fake_ain"
device_lock = "fake_locked_device"
fw_version = "1.2.3"
has_alarm = False
has_switch = False
has_temperature_sensor = True
has_thermostat = False
lock = "fake_locked"
manufacturer = "fake_manufacturer"
name = "fake_name"
present = True
productname = "fake_productname"
temperature = 1.23
class FritzDeviceSwitchMock(Mock):
"""Mock of a AVM Fritz!Box switch device."""
ain = "fake_ain"
device_lock = "fake_locked_device"
energy = 1234
fw_version = "1.2.3"
has_alarm = False
has_switch = True
has_temperature_sensor = True
has_thermostat = False
switch_state = "fake_state"
lock = "fake_locked"
manufacturer = "fake_manufacturer"
name = "fake_name"
power = 5678
present = True
productname = "fake_productname"
temperature = 135

14
tests/components/fritzbox/conftest.py Normal file
View File

@ -0,0 +1,14 @@
"""Fixtures for the AVM Fritz!Box integration."""
from unittest.mock import Mock, patch
import pytest
@pytest.fixture(name="fritz")
def fritz_fixture() -> Mock:
"""Patch libraries."""
with patch("homeassistant.components.fritzbox.socket") as socket, patch(
"homeassistant.components.fritzbox.Fritzhome"
) as fritz, patch("homeassistant.components.fritzbox.config_flow.Fritzhome"):
socket.gethostbyname.return_value = "FAKE_IP_ADDRESS"
yield fritz

94
tests/components/fritzbox/test_binary_sensor.py Normal file
View File

@ -0,0 +1,94 @@
"""Tests for AVM Fritz!Box binary sensor component."""
from datetime import timedelta
from unittest import mock
from unittest.mock import Mock
from requests.exceptions import HTTPError
from homeassistant.components.binary_sensor import DOMAIN
from homeassistant.components.fritzbox.const import DOMAIN as FB_DOMAIN
from homeassistant.const import (
ATTR_DEVICE_CLASS,
ATTR_FRIENDLY_NAME,
STATE_OFF,
STATE_ON,
)
from homeassistant.helpers.typing import HomeAssistantType
from homeassistant.setup import async_setup_component
import homeassistant.util.dt as dt_util
from . import MOCK_CONFIG, FritzDeviceBinarySensorMock
from tests.common import async_fire_time_changed
ENTITY_ID = f"{DOMAIN}.fake_name"
async def setup_fritzbox(hass: HomeAssistantType, config: dict):
"""Set up mock AVM Fritz!Box."""
assert await async_setup_component(hass, FB_DOMAIN, config)
await hass.async_block_till_done()
async def test_setup(hass: HomeAssistantType, fritz: Mock):
"""Test setup of platform."""
device = FritzDeviceBinarySensorMock()
fritz().get_devices.return_value = [device]
await setup_fritzbox(hass, MOCK_CONFIG)
state = hass.states.get(ENTITY_ID)
assert state
assert state.state == STATE_ON
assert state.attributes[ATTR_FRIENDLY_NAME] == "fake_name"
assert state.attributes[ATTR_DEVICE_CLASS] == "window"
async def test_is_off(hass: HomeAssistantType, fritz: Mock):
"""Test state of platform."""
device = FritzDeviceBinarySensorMock()
device.present = False
fritz().get_devices.return_value = [device]
await setup_fritzbox(hass, MOCK_CONFIG)
state = hass.states.get(ENTITY_ID)
assert state
assert state.state == STATE_OFF
async def test_update(hass: HomeAssistantType, fritz: Mock):
"""Test update with error."""
device = FritzDeviceBinarySensorMock()
fritz().get_devices.return_value = [device]
await setup_fritzbox(hass, MOCK_CONFIG)
assert device.update.call_count == 1
assert fritz().login.call_count == 1
next_update = dt_util.utcnow() + timedelta(seconds=200)
async_fire_time_changed(hass, next_update)
await hass.async_block_till_done()
assert device.update.call_count == 2
assert fritz().login.call_count == 1
async def test_update_error(hass: HomeAssistantType, fritz: Mock):
"""Test update with error."""
device = FritzDeviceBinarySensorMock()
device.update.side_effect = [mock.DEFAULT, HTTPError("Boom")]
fritz().get_devices.return_value = [device]
await setup_fritzbox(hass, MOCK_CONFIG)
assert device.update.call_count == 1
assert fritz().login.call_count == 1
next_update = dt_util.utcnow() + timedelta(seconds=200)
async_fire_time_changed(hass, next_update)
await hass.async_block_till_done()
assert device.update.call_count == 2
assert fritz().login.call_count == 2

403
tests/components/fritzbox/test_climate.py
View File

@ -1,151 +1,306 @@
"""The tests for the demo climate component."""
import unittest
from unittest.mock import Mock, patch
"""Tests for AVM Fritz!Box climate component."""
from datetime import timedelta
from unittest.mock import Mock, call
import requests
from requests.exceptions import HTTPError
from homeassistant.components.fritzbox.climate import FritzboxThermostat
from homeassistant.const import TEMP_CELSIUS
from homeassistant.components.climate.const import (
ATTR_CURRENT_TEMPERATURE,
ATTR_HVAC_MODE,
ATTR_HVAC_MODES,
ATTR_MAX_TEMP,
ATTR_MIN_TEMP,
ATTR_PRESET_MODE,
ATTR_PRESET_MODES,
DOMAIN,
HVAC_MODE_HEAT,
HVAC_MODE_OFF,
PRESET_COMFORT,
PRESET_ECO,
SERVICE_SET_HVAC_MODE,
SERVICE_SET_PRESET_MODE,
SERVICE_SET_TEMPERATURE,
)
from homeassistant.components.fritzbox.const import (
ATTR_STATE_BATTERY_LOW,
ATTR_STATE_DEVICE_LOCKED,
ATTR_STATE_HOLIDAY_MODE,
ATTR_STATE_LOCKED,
ATTR_STATE_SUMMER_MODE,
ATTR_STATE_WINDOW_OPEN,
DOMAIN as FB_DOMAIN,
)
from homeassistant.const import (
ATTR_BATTERY_LEVEL,
ATTR_ENTITY_ID,
ATTR_FRIENDLY_NAME,
ATTR_TEMPERATURE,
)
from homeassistant.helpers.typing import HomeAssistantType
from homeassistant.setup import async_setup_component
import homeassistant.util.dt as dt_util
from . import MOCK_CONFIG, FritzDeviceClimateMock
from tests.common import async_fire_time_changed
ENTITY_ID = f"{DOMAIN}.fake_name"
class TestFritzboxClimate(unittest.TestCase):
"""Test Fritz!Box heating thermostats."""
async def setup_fritzbox(hass: HomeAssistantType, config: dict):
"""Set up mock AVM Fritz!Box."""
assert await async_setup_component(hass, FB_DOMAIN, config) is True
await hass.async_block_till_done()
def setUp(self):
"""Create a mock device to test on."""
self.device = Mock()
self.device.name = "Test Thermostat"
self.device.actual_temperature = 18.0
self.device.target_temperature = 19.5
self.device.comfort_temperature = 22.0
self.device.eco_temperature = 16.0
self.device.present = True
self.device.device_lock = True
self.device.lock = False
self.device.battery_low = True
self.device.set_target_temperature = Mock()
self.device.update = Mock()
mock_fritz = Mock()
mock_fritz.login = Mock()
self.thermostat = FritzboxThermostat(self.device, mock_fritz)
def test_init(self):
"""Test instance creation."""
assert 18.0 == self.thermostat._current_temperature
assert 19.5 == self.thermostat._target_temperature
assert 22.0 == self.thermostat._comfort_temperature
assert 16.0 == self.thermostat._eco_temperature
async def test_setup(hass: HomeAssistantType, fritz: Mock):
"""Test setup of platform."""
device = FritzDeviceClimateMock()
fritz().get_devices.return_value = [device]
def test_supported_features(self):
"""Test supported features property."""
assert self.thermostat.supported_features == 17
await setup_fritzbox(hass, MOCK_CONFIG)
state = hass.states.get(ENTITY_ID)
def test_available(self):
"""Test available property."""
assert self.thermostat.available
self.thermostat._device.present = False
assert not self.thermostat.available
assert state
assert state.attributes[ATTR_BATTERY_LEVEL] == 23
assert state.attributes[ATTR_CURRENT_TEMPERATURE] == 18
assert state.attributes[ATTR_FRIENDLY_NAME] == "fake_name"
assert state.attributes[ATTR_HVAC_MODES] == [HVAC_MODE_HEAT, HVAC_MODE_OFF]
assert state.attributes[ATTR_MAX_TEMP] == 28
assert state.attributes[ATTR_MIN_TEMP] == 8
assert state.attributes[ATTR_PRESET_MODE] is None
assert state.attributes[ATTR_PRESET_MODES] == [PRESET_ECO, PRESET_COMFORT]
assert state.attributes[ATTR_STATE_BATTERY_LOW] is True
assert state.attributes[ATTR_STATE_DEVICE_LOCKED] == "fake_locked_device"
assert state.attributes[ATTR_STATE_HOLIDAY_MODE] == "fake_holiday"
assert state.attributes[ATTR_STATE_LOCKED] == "fake_locked"
assert state.attributes[ATTR_STATE_SUMMER_MODE] == "fake_summer"
assert state.attributes[ATTR_STATE_WINDOW_OPEN] == "fake_window"
assert state.attributes[ATTR_TEMPERATURE] == 19.5
assert state.state == HVAC_MODE_HEAT
def test_name(self):
"""Test name property."""
assert "Test Thermostat" == self.thermostat.name
def test_temperature_unit(self):
"""Test temperature_unit property."""
assert TEMP_CELSIUS == self.thermostat.temperature_unit
async def test_target_temperature_on(hass: HomeAssistantType, fritz: Mock):
"""Test turn device on."""
device = FritzDeviceClimateMock()
fritz().get_devices.return_value = [device]
device.target_temperature = 127.0
def test_precision(self):
"""Test precision property."""
assert 0.5 == self.thermostat.precision
await setup_fritzbox(hass, MOCK_CONFIG)
state = hass.states.get(ENTITY_ID)
assert state
assert state.attributes[ATTR_TEMPERATURE] == 30
def test_current_temperature(self):
"""Test current_temperature property incl. special temperatures."""
assert 18 == self.thermostat.current_temperature
def test_target_temperature(self):
"""Test target_temperature property."""
assert 19.5 == self.thermostat.target_temperature
async def test_target_temperature_off(hass: HomeAssistantType, fritz: Mock):
"""Test turn device on."""
device = FritzDeviceClimateMock()
fritz().get_devices.return_value = [device]
device.target_temperature = 126.5
self.thermostat._target_temperature = 126.5
assert self.thermostat.target_temperature == 0.0
await setup_fritzbox(hass, MOCK_CONFIG)
state = hass.states.get(ENTITY_ID)
assert state
assert state.attributes[ATTR_TEMPERATURE] == 0
self.thermostat._target_temperature = 127.0
assert self.thermostat.target_temperature == 30.0
@patch.object(FritzboxThermostat, "set_hvac_mode")
def test_set_temperature_operation_mode(self, mock_set_op):
"""Test set_temperature by operation_mode."""
self.thermostat.set_temperature(hvac_mode="heat")
mock_set_op.assert_called_once_with("heat")
async def test_update(hass: HomeAssistantType, fritz: Mock):
"""Test update with error."""
device = FritzDeviceClimateMock()
fritz().get_devices.return_value = [device]
def test_set_temperature_temperature(self):
"""Test set_temperature by temperature."""
self.thermostat.set_temperature(temperature=23.0)
self.thermostat._device.set_target_temperature.assert_called_once_with(23.0)
await setup_fritzbox(hass, MOCK_CONFIG)
state = hass.states.get(ENTITY_ID)
@patch.object(FritzboxThermostat, "set_hvac_mode")
def test_set_temperature_none(self, mock_set_op):
"""Test set_temperature with no arguments."""
self.thermostat.set_temperature()
mock_set_op.assert_not_called()
self.thermostat._device.set_target_temperature.assert_not_called()
assert state
assert state.attributes[ATTR_CURRENT_TEMPERATURE] == 18
assert state.attributes[ATTR_MAX_TEMP] == 28
assert state.attributes[ATTR_MIN_TEMP] == 8
assert state.attributes[ATTR_TEMPERATURE] == 19.5
@patch.object(FritzboxThermostat, "set_hvac_mode")
def test_set_temperature_operation_mode_precedence(self, mock_set_op):
"""Test set_temperature for precedence of operation_mode argument."""
self.thermostat.set_temperature(hvac_mode="heat", temperature=23.0)
mock_set_op.assert_called_once_with("heat")
self.thermostat._device.set_target_temperature.assert_not_called()
device.actual_temperature = 19
device.target_temperature = 20
def test_hvac_mode(self):
"""Test operation mode property for different temperatures."""
self.thermostat._target_temperature = 127.0
assert "heat" == self.thermostat.hvac_mode
self.thermostat._target_temperature = 126.5
assert "off" == self.thermostat.hvac_mode
self.thermostat._target_temperature = 22.0
assert "heat" == self.thermostat.hvac_mode
self.thermostat._target_temperature = 16.0
assert "heat" == self.thermostat.hvac_mode
self.thermostat._target_temperature = 12.5
assert "heat" == self.thermostat.hvac_mode
next_update = dt_util.utcnow() + timedelta(seconds=200)
async_fire_time_changed(hass, next_update)
await hass.async_block_till_done()
state = hass.states.get(ENTITY_ID)
def test_operation_list(self):
"""Test operation_list property."""
assert ["heat", "off"] == self.thermostat.hvac_modes
assert device.update.call_count == 1
assert state
assert state.attributes[ATTR_CURRENT_TEMPERATURE] == 19
assert state.attributes[ATTR_TEMPERATURE] == 20
def test_min_max_temperature(self):
"""Test min_temp and max_temp properties."""
assert 8.0 == self.thermostat.min_temp
assert 28.0 == self.thermostat.max_temp
def test_device_state_attributes(self):
"""Test device_state property."""
attr = self.thermostat.device_state_attributes
assert attr["device_locked"] is True
assert attr["locked"] is False
assert attr["battery_low"] is True
async def test_update_error(hass: HomeAssistantType, fritz: Mock):
"""Test update with error."""
device = FritzDeviceClimateMock()
device.update.side_effect = HTTPError("Boom")
fritz().get_devices.return_value = [device]
def test_update(self):
"""Test update function."""
device = Mock()
device.update = Mock()
device.actual_temperature = 10.0
device.target_temperature = 11.0
device.comfort_temperature = 12.0
device.eco_temperature = 13.0
self.thermostat._device = device
await setup_fritzbox(hass, MOCK_CONFIG)
assert device.update.call_count == 0
assert fritz().login.call_count == 1
self.thermostat.update()
next_update = dt_util.utcnow() + timedelta(seconds=200)
async_fire_time_changed(hass, next_update)
await hass.async_block_till_done()
device.update.assert_called_once_with()
assert 10.0 == self.thermostat._current_temperature
assert 11.0 == self.thermostat._target_temperature
assert 12.0 == self.thermostat._comfort_temperature
assert 13.0 == self.thermostat._eco_temperature
assert device.update.call_count == 1
assert fritz().login.call_count == 2
def test_update_http_error(self):
"""Test exception handling of update function."""
self.device.update.side_effect = requests.exceptions.HTTPError
self.thermostat.update()
self.thermostat._fritz.login.assert_called_once_with()
async def test_set_temperature_temperature(hass: HomeAssistantType, fritz: Mock):
"""Test setting temperature by temperature."""
device = FritzDeviceClimateMock()
fritz().get_devices.return_value = [device]
await setup_fritzbox(hass, MOCK_CONFIG)
assert await hass.services.async_call(
DOMAIN,
SERVICE_SET_TEMPERATURE,
{ATTR_ENTITY_ID: ENTITY_ID, ATTR_TEMPERATURE: 123},
True,
)
assert device.set_target_temperature.call_args_list == [call(123)]
async def test_set_temperature_mode_off(hass: HomeAssistantType, fritz: Mock):
"""Test setting temperature by mode."""
device = FritzDeviceClimateMock()
fritz().get_devices.return_value = [device]
await setup_fritzbox(hass, MOCK_CONFIG)
assert await hass.services.async_call(
DOMAIN,
SERVICE_SET_TEMPERATURE,
{
ATTR_ENTITY_ID: ENTITY_ID,
ATTR_HVAC_MODE: HVAC_MODE_OFF,
ATTR_TEMPERATURE: 123,
},
True,
)
assert device.set_target_temperature.call_args_list == [call(0)]
async def test_set_temperature_mode_heat(hass: HomeAssistantType, fritz: Mock):
"""Test setting temperature by mode."""
device = FritzDeviceClimateMock()
fritz().get_devices.return_value = [device]
await setup_fritzbox(hass, MOCK_CONFIG)
assert await hass.services.async_call(
DOMAIN,
SERVICE_SET_TEMPERATURE,
{
ATTR_ENTITY_ID: ENTITY_ID,
ATTR_HVAC_MODE: HVAC_MODE_HEAT,
ATTR_TEMPERATURE: 123,
},
True,
)
assert device.set_target_temperature.call_args_list == [call(22)]
async def test_set_hvac_mode_off(hass: HomeAssistantType, fritz: Mock):
"""Test setting hvac mode."""
device = FritzDeviceClimateMock()
fritz().get_devices.return_value = [device]
await setup_fritzbox(hass, MOCK_CONFIG)
assert await hass.services.async_call(
DOMAIN,
SERVICE_SET_HVAC_MODE,
{ATTR_ENTITY_ID: ENTITY_ID, ATTR_HVAC_MODE: HVAC_MODE_OFF},
True,
)
assert device.set_target_temperature.call_args_list == [call(0)]
async def test_set_hvac_mode_heat(hass: HomeAssistantType, fritz: Mock):
"""Test setting hvac mode."""
device = FritzDeviceClimateMock()
fritz().get_devices.return_value = [device]
await setup_fritzbox(hass, MOCK_CONFIG)
assert await hass.services.async_call(
DOMAIN,
SERVICE_SET_HVAC_MODE,
{ATTR_ENTITY_ID: ENTITY_ID, ATTR_HVAC_MODE: HVAC_MODE_HEAT},
True,
)
assert device.set_target_temperature.call_args_list == [call(22)]
async def test_set_preset_mode_comfort(hass: HomeAssistantType, fritz: Mock):
"""Test setting preset mode."""
device = FritzDeviceClimateMock()
fritz().get_devices.return_value = [device]
await setup_fritzbox(hass, MOCK_CONFIG)
assert await hass.services.async_call(
DOMAIN,
SERVICE_SET_PRESET_MODE,
{ATTR_ENTITY_ID: ENTITY_ID, ATTR_PRESET_MODE: PRESET_COMFORT},
True,
)
assert device.set_target_temperature.call_args_list == [call(22)]
async def test_set_preset_mode_eco(hass: HomeAssistantType, fritz: Mock):
"""Test setting preset mode."""
device = FritzDeviceClimateMock()
fritz().get_devices.return_value = [device]
await setup_fritzbox(hass, MOCK_CONFIG)
assert await hass.services.async_call(
DOMAIN,
SERVICE_SET_PRESET_MODE,
{ATTR_ENTITY_ID: ENTITY_ID, ATTR_PRESET_MODE: PRESET_ECO},
True,
)
assert device.set_target_temperature.call_args_list == [call(16)]
async def test_preset_mode_update(hass: HomeAssistantType, fritz: Mock):
"""Test preset mode."""
device = FritzDeviceClimateMock()
device.comfort_temperature = 98
device.eco_temperature = 99
fritz().get_devices.return_value = [device]
await setup_fritzbox(hass, MOCK_CONFIG)
state = hass.states.get(ENTITY_ID)
assert state
assert state.attributes[ATTR_PRESET_MODE] is None
device.target_temperature = 98
next_update = dt_util.utcnow() + timedelta(seconds=200)
async_fire_time_changed(hass, next_update)
await hass.async_block_till_done()
state = hass.states.get(ENTITY_ID)
assert device.update.call_count == 1
assert state
assert state.attributes[ATTR_PRESET_MODE] == PRESET_COMFORT
device.target_temperature = 99
next_update = dt_util.utcnow() + timedelta(seconds=200)
async_fire_time_changed(hass, next_update)
await hass.async_block_till_done()
state = hass.states.get(ENTITY_ID)
assert device.update.call_count == 2
assert state
assert state.attributes[ATTR_PRESET_MODE] == PRESET_ECO

179
tests/components/fritzbox/test_config_flow.py Normal file
View File

@ -0,0 +1,179 @@
"""Tests for AVM Fritz!Box config flow."""
from unittest import mock
from unittest.mock import Mock, patch
from pyfritzhome import LoginError
import pytest
from homeassistant.components.fritzbox.const import DOMAIN
from homeassistant.components.ssdp import ATTR_SSDP_LOCATION, ATTR_UPNP_FRIENDLY_NAME
from homeassistant.const import CONF_DEVICES, CONF_HOST, CONF_PASSWORD, CONF_USERNAME
from homeassistant.helpers.typing import HomeAssistantType
from . import MOCK_CONFIG
MOCK_USER_DATA = MOCK_CONFIG[DOMAIN][CONF_DEVICES][0]
MOCK_SSDP_DATA = {
ATTR_SSDP_LOCATION: "https://fake_host:12345/test",
ATTR_UPNP_FRIENDLY_NAME: "fake_name",
}
@pytest.fixture(name="fritz")
def fritz_fixture() -> Mock:
"""Patch libraries."""
with patch("homeassistant.components.fritzbox.config_flow.Fritzhome") as fritz:
yield fritz
async def test_user(hass: HomeAssistantType, fritz: Mock):
"""Test starting a flow by user."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": "user"}
)
assert result["type"] == "form"
assert result["step_id"] == "user"
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input=MOCK_USER_DATA
)
assert result["type"] == "create_entry"
assert result["title"] == "fake_host"
assert result["data"][CONF_HOST] == "fake_host"
assert result["data"][CONF_PASSWORD] == "fake_pass"
assert result["data"][CONF_USERNAME] == "fake_user"
async def test_user_auth_failed(hass: HomeAssistantType, fritz: Mock):
"""Test starting a flow by user with authentication failure."""
fritz().login.side_effect = [LoginError("Boom"), mock.DEFAULT]
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": "user"}, data=MOCK_USER_DATA
)
assert result["type"] == "form"
assert result["step_id"] == "user"
assert result["errors"]["base"] == "auth_failed"
async def test_user_not_successful(hass: HomeAssistantType, fritz: Mock):
"""Test starting a flow by user but no connection found."""
fritz().login.side_effect = OSError("Boom")
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": "user"}, data=MOCK_USER_DATA
)
assert result["type"] == "abort"
assert result["reason"] == "not_found"
async def test_user_already_configured(hass: HomeAssistantType, fritz: Mock):
"""Test starting a flow by user when already configured."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": "user"}, data=MOCK_USER_DATA
)
assert result["type"] == "create_entry"
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": "user"}, data=MOCK_USER_DATA
)
assert result["type"] == "abort"
assert result["reason"] == "already_configured"
async def test_import(hass: HomeAssistantType, fritz: Mock):
"""Test starting a flow by import."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": "import"}, data=MOCK_USER_DATA
)
assert result["type"] == "create_entry"
assert result["title"] == "fake_host"
assert result["data"][CONF_HOST] == "fake_host"
assert result["data"][CONF_PASSWORD] == "fake_pass"
assert result["data"][CONF_USERNAME] == "fake_user"
async def test_ssdp(hass: HomeAssistantType, fritz: Mock):
"""Test starting a flow from discovery."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": "ssdp"}, data=MOCK_SSDP_DATA
)
assert result["type"] == "form"
assert result["step_id"] == "confirm"
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={CONF_PASSWORD: "fake_pass", CONF_USERNAME: "fake_user"},
)
assert result["type"] == "create_entry"
assert result["title"] == "fake_name"
assert result["data"][CONF_HOST] == "fake_host"
assert result["data"][CONF_PASSWORD] == "fake_pass"
assert result["data"][CONF_USERNAME] == "fake_user"
async def test_ssdp_auth_failed(hass: HomeAssistantType, fritz: Mock):
"""Test starting a flow from discovery with authentication failure."""
fritz().login.side_effect = LoginError("Boom")
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": "ssdp"}, data=MOCK_SSDP_DATA
)
assert result["type"] == "form"
assert result["step_id"] == "confirm"
assert result["errors"] == {}
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={CONF_PASSWORD: "whatever", CONF_USERNAME: "whatever"},
)
assert result["type"] == "form"
assert result["step_id"] == "confirm"
assert result["errors"]["base"] == "auth_failed"
async def test_ssdp_not_successful(hass: HomeAssistantType, fritz: Mock):
"""Test starting a flow from discovery but no device found."""
fritz().login.side_effect = OSError("Boom")
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": "ssdp"}, data=MOCK_SSDP_DATA
)
assert result["type"] == "form"
assert result["step_id"] == "confirm"
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={CONF_PASSWORD: "whatever", CONF_USERNAME: "whatever"},
)
assert result["type"] == "abort"
assert result["reason"] == "not_found"
async def test_ssdp_already_in_progress(hass: HomeAssistantType, fritz: Mock):
"""Test starting a flow from discovery twice."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": "ssdp"}, data=MOCK_SSDP_DATA
)
assert result["type"] == "form"
assert result["step_id"] == "confirm"
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": "ssdp"}, data=MOCK_SSDP_DATA
)
assert result["type"] == "abort"
assert result["reason"] == "already_in_progress"
async def test_ssdp_already_configured(hass: HomeAssistantType, fritz: Mock):
"""Test starting a flow from discovery when already configured."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": "user"}, data=MOCK_USER_DATA
)
assert result["type"] == "create_entry"
result2 = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": "ssdp"}, data=MOCK_SSDP_DATA
)
assert result2["type"] == "abort"
assert result2["reason"] == "already_configured"

76
tests/components/fritzbox/test_init.py Normal file
View File

@ -0,0 +1,76 @@
"""Tests for the AVM Fritz!Box integration."""
from unittest.mock import Mock, call
from homeassistant.components.fritzbox.const import DOMAIN as FB_DOMAIN
from homeassistant.components.switch import DOMAIN as SWITCH_DOMAIN
from homeassistant.config_entries import ENTRY_STATE_LOADED, ENTRY_STATE_NOT_LOADED
from homeassistant.const import CONF_DEVICES, CONF_HOST, CONF_PASSWORD, CONF_USERNAME
from homeassistant.helpers.typing import HomeAssistantType
from homeassistant.setup import async_setup_component
from . import MOCK_CONFIG, FritzDeviceSwitchMock
from tests.common import MockConfigEntry
async def test_setup(hass: HomeAssistantType, fritz: Mock):
"""Test setup of integration."""
assert await async_setup_component(hass, FB_DOMAIN, MOCK_CONFIG)
await hass.async_block_till_done()
entries = hass.config_entries.async_entries()
assert entries
assert entries[0].data[CONF_HOST] == "fake_host"
assert entries[0].data[CONF_PASSWORD] == "fake_pass"
assert entries[0].data[CONF_USERNAME] == "fake_user"
assert fritz.call_count == 1
assert fritz.call_args_list == [
call(host="fake_host", password="fake_pass", user="fake_user")
]
async def test_setup_duplicate_config(hass: HomeAssistantType, fritz: Mock, caplog):
"""Test duplicate config of integration."""
DUPLICATE = {
FB_DOMAIN: {
CONF_DEVICES: [
MOCK_CONFIG[FB_DOMAIN][CONF_DEVICES][0],
MOCK_CONFIG[FB_DOMAIN][CONF_DEVICES][0],
]
}
}
assert not await async_setup_component(hass, FB_DOMAIN, DUPLICATE)
await hass.async_block_till_done()
assert not hass.states.async_entity_ids()
assert not hass.states.async_all()
assert "duplicate host entries found" in caplog.text
async def test_unload(hass: HomeAssistantType, fritz: Mock):
"""Test unload of integration."""
fritz().get_devices.return_value = [FritzDeviceSwitchMock()]
entity_id = f"{SWITCH_DOMAIN}.fake_name"
entry = MockConfigEntry(
domain=FB_DOMAIN,
data=MOCK_CONFIG[FB_DOMAIN][CONF_DEVICES][0],
unique_id=entity_id,
)
entry.add_to_hass(hass)
config_entries = hass.config_entries.async_entries(FB_DOMAIN)
assert len(config_entries) == 1
assert entry is config_entries[0]
assert await async_setup_component(hass, FB_DOMAIN, {}) is True
await hass.async_block_till_done()
assert entry.state == ENTRY_STATE_LOADED
state = hass.states.get(entity_id)
assert state
await hass.config_entries.async_unload(entry.entry_id)
assert fritz().logout.call_count == 1
assert entry.state == ENTRY_STATE_NOT_LOADED
state = hass.states.get(entity_id)
assert state is None

83
tests/components/fritzbox/test_sensor.py Normal file
View File

@ -0,0 +1,83 @@
"""Tests for AVM Fritz!Box sensor component."""
from datetime import timedelta
from unittest.mock import Mock
from requests.exceptions import HTTPError
from homeassistant.components.fritzbox.const import (
ATTR_STATE_DEVICE_LOCKED,
ATTR_STATE_LOCKED,
DOMAIN as FB_DOMAIN,
)
from homeassistant.components.sensor import DOMAIN
from homeassistant.const import (
ATTR_FRIENDLY_NAME,
ATTR_UNIT_OF_MEASUREMENT,
TEMP_CELSIUS,
)
from homeassistant.helpers.typing import HomeAssistantType
from homeassistant.setup import async_setup_component
import homeassistant.util.dt as dt_util
from . import MOCK_CONFIG, FritzDeviceSensorMock
from tests.common import async_fire_time_changed
ENTITY_ID = f"{DOMAIN}.fake_name"
async def setup_fritzbox(hass: HomeAssistantType, config: dict):
"""Set up mock AVM Fritz!Box."""
assert await async_setup_component(hass, FB_DOMAIN, config)
await hass.async_block_till_done()
async def test_setup(hass: HomeAssistantType, fritz: Mock):
"""Test setup of platform."""
device = FritzDeviceSensorMock()
fritz().get_devices.return_value = [device]
await setup_fritzbox(hass, MOCK_CONFIG)
state = hass.states.get(ENTITY_ID)
assert state
assert state.state == "1.23"
assert state.attributes[ATTR_FRIENDLY_NAME] == "fake_name"
assert state.attributes[ATTR_STATE_DEVICE_LOCKED] == "fake_locked_device"
assert state.attributes[ATTR_STATE_LOCKED] == "fake_locked"
assert state.attributes[ATTR_UNIT_OF_MEASUREMENT] == TEMP_CELSIUS
async def test_update(hass: HomeAssistantType, fritz: Mock):
"""Test update with error."""
device = FritzDeviceSensorMock()
fritz().get_devices.return_value = [device]
await setup_fritzbox(hass, MOCK_CONFIG)
assert device.update.call_count == 0
assert fritz().login.call_count == 1
next_update = dt_util.utcnow() + timedelta(seconds=200)
async_fire_time_changed(hass, next_update)
await hass.async_block_till_done()
assert device.update.call_count == 1
assert fritz().login.call_count == 1
async def test_update_error(hass: HomeAssistantType, fritz: Mock):
"""Test update with error."""
device = FritzDeviceSensorMock()
device.update.side_effect = HTTPError("Boom")
fritz().get_devices.return_value = [device]
await setup_fritzbox(hass, MOCK_CONFIG)
assert device.update.call_count == 0
assert fritz().login.call_count == 1
next_update = dt_util.utcnow() + timedelta(seconds=200)
async_fire_time_changed(hass, next_update)
await hass.async_block_till_done()
assert device.update.call_count == 1
assert fritz().login.call_count == 2

121
tests/components/fritzbox/test_switch.py Normal file
View File

@ -0,0 +1,121 @@
"""Tests for AVM Fritz!Box switch component."""
from datetime import timedelta
from unittest.mock import Mock
from requests.exceptions import HTTPError
from homeassistant.components.fritzbox.const import (
ATTR_STATE_DEVICE_LOCKED,
ATTR_STATE_LOCKED,
ATTR_TEMPERATURE_UNIT,
ATTR_TOTAL_CONSUMPTION,
ATTR_TOTAL_CONSUMPTION_UNIT,
DOMAIN as FB_DOMAIN,
)
from homeassistant.components.switch import ATTR_CURRENT_POWER_W, DOMAIN
from homeassistant.const import (
ATTR_ENTITY_ID,
ATTR_FRIENDLY_NAME,
ATTR_TEMPERATURE,
ENERGY_KILO_WATT_HOUR,
SERVICE_TURN_OFF,
SERVICE_TURN_ON,
STATE_ON,
TEMP_CELSIUS,
)
from homeassistant.helpers.typing import HomeAssistantType
from homeassistant.setup import async_setup_component
import homeassistant.util.dt as dt_util
from . import MOCK_CONFIG, FritzDeviceSwitchMock
from tests.common import async_fire_time_changed
ENTITY_ID = f"{DOMAIN}.fake_name"
async def setup_fritzbox(hass: HomeAssistantType, config: dict):
"""Set up mock AVM Fritz!Box."""
assert await async_setup_component(hass, FB_DOMAIN, config)
await hass.async_block_till_done()
async def test_setup(hass: HomeAssistantType, fritz: Mock):
"""Test setup of platform."""
device = FritzDeviceSwitchMock()
fritz().get_devices.return_value = [device]
await setup_fritzbox(hass, MOCK_CONFIG)
state = hass.states.get(ENTITY_ID)
assert state
assert state.state == STATE_ON
assert state.attributes[ATTR_CURRENT_POWER_W] == 5.678
assert state.attributes[ATTR_FRIENDLY_NAME] == "fake_name"
assert state.attributes[ATTR_STATE_DEVICE_LOCKED] == "fake_locked_device"
assert state.attributes[ATTR_STATE_LOCKED] == "fake_locked"
assert state.attributes[ATTR_TEMPERATURE] == "135"
assert state.attributes[ATTR_TEMPERATURE_UNIT] == TEMP_CELSIUS
assert state.attributes[ATTR_TOTAL_CONSUMPTION] == "1.234"
assert state.attributes[ATTR_TOTAL_CONSUMPTION_UNIT] == ENERGY_KILO_WATT_HOUR
async def test_turn_on(hass: HomeAssistantType, fritz: Mock):
"""Test turn device on."""
device = FritzDeviceSwitchMock()
fritz().get_devices.return_value = [device]
await setup_fritzbox(hass, MOCK_CONFIG)
assert await hass.services.async_call(
DOMAIN, SERVICE_TURN_ON, {ATTR_ENTITY_ID: ENTITY_ID}, True
)
assert device.set_switch_state_on.call_count == 1
async def test_turn_off(hass: HomeAssistantType, fritz: Mock):
"""Test turn device off."""
device = FritzDeviceSwitchMock()
fritz().get_devices.return_value = [device]
await setup_fritzbox(hass, MOCK_CONFIG)
assert await hass.services.async_call(
DOMAIN, SERVICE_TURN_OFF, {ATTR_ENTITY_ID: ENTITY_ID}, True
)
assert device.set_switch_state_off.call_count == 1
async def test_update(hass: HomeAssistantType, fritz: Mock):
"""Test update with error."""
device = FritzDeviceSwitchMock()
fritz().get_devices.return_value = [device]
await setup_fritzbox(hass, MOCK_CONFIG)
assert device.update.call_count == 0
assert fritz().login.call_count == 1
next_update = dt_util.utcnow() + timedelta(seconds=200)
async_fire_time_changed(hass, next_update)
await hass.async_block_till_done()
assert device.update.call_count == 1
assert fritz().login.call_count == 1
async def test_update_error(hass: HomeAssistantType, fritz: Mock):
"""Test update with error."""
device = FritzDeviceSwitchMock()
device.update.side_effect = HTTPError("Boom")
fritz().get_devices.return_value = [device]
await setup_fritzbox(hass, MOCK_CONFIG)
assert device.update.call_count == 0
assert fritz().login.call_count == 1
next_update = dt_util.utcnow() + timedelta(seconds=200)
async_fire_time_changed(hass, next_update)
await hass.async_block_till_done()
assert device.update.call_count == 1
assert fritz().login.call_count == 2