Refactor Freebox : add config flow + temperature sensor + signal dispatch (#30334)
* Add config flow to Freebox * Add manufacturer in device_tracker info * Add device_info to sensor + switch * Add device_info: connections * Add config_flow test + update .coveragerc * Typing * Add device_type icon * Remove one error log * Fix pylint * Add myself as CODEOWNER * Handle sync in one place * Separate the Freebox[Router/Device/Sensor] from __init__.py * Add link step to config flow * Make temperature sensors auto-discovered * Use device activity instead of reachablility for device_tracker * Store token file in .storage Depending on host if list of Freebox integration on the future without breaking change * Remove IP sensors + add Freebox router as a device with attrs : IPs, conection type, uptime, version & serial * Add sensor should_poll=False * Test typing * Handle devices with no name * None is the default for data * Fix comment * Use config_entry.unique_id * Add async_unload_entry with asyncio * Add and use bunch of data size and rate related constants (#31781) * Review * Remove useless "already_configured" error string * Review : merge 2 device & 2 sensor classes * Entities from platforms * Fix unload + add device after setup + clean loggers * async_add_entities True * Review * Use pathlib + refactor get_api * device_tracker set + tests with CoroutineMock() * Removing active & reachable from tracker attrs * Review * Fix pipeline * typing * typing * typing * Raise ConfigEntryNotReady when HttpRequestError at setup * Review * Multiple Freebox s * Review: store sensors in router * Freebox: a sensor storypull/32708/head
Quentame
GitHub
parent
180bcad477
commit
19be4a5d6d
|
|
@ -242,7 +242,11 @@ omit =
|
|||
homeassistant/components/foscam/const.py
|
||||
homeassistant/components/foursquare/*
|
||||
homeassistant/components/free_mobile/notify.py
|
||||
homeassistant/components/freebox/*
|
||||
homeassistant/components/freebox/__init__.py
|
||||
homeassistant/components/freebox/device_tracker.py
|
||||
homeassistant/components/freebox/router.py
|
||||
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
|
||||
|
|
|
|||
|
|
@ -122,7 +122,7 @@ homeassistant/components/fortigate/* @kifeo
|
|||
homeassistant/components/fortios/* @kimfrellsen
|
||||
homeassistant/components/foscam/* @skgsergio
|
||||
homeassistant/components/foursquare/* @robbiet480
|
||||
homeassistant/components/freebox/* @snoof85
|
||||
homeassistant/components/freebox/* @snoof85 @Quentame
|
||||
homeassistant/components/fronius/* @nielstron
|
||||
homeassistant/components/frontend/* @home-assistant/frontend
|
||||
homeassistant/components/garmin_connect/* @cyberjunky
|
||||
|
|
|
|||
|
|
@ -0,0 +1,26 @@
|
|||
{
|
||||
"config": {
|
||||
"abort": {
|
||||
"already_configured": "Host already configured"
|
||||
},
|
||||
"error": {
|
||||
"connection_failed": "Failed to connect, please try again",
|
||||
"register_failed": "Failed to register, please try again",
|
||||
"unknown": "Unknown error: please retry later"
|
||||
},
|
||||
"step": {
|
||||
"link": {
|
||||
"description": "Click \"Submit\", then touch the right arrow on the router to register Freebox with Home Assistant.\n\n",
|
||||
"title": "Link Freebox router"
|
||||
},
|
||||
"user": {
|
||||
"data": {
|
||||
"host": "Host",
|
||||
"port": "Port"
|
||||
},
|
||||
"title": "Freebox"
|
||||
}
|
||||
},
|
||||
"title": "Freebox"
|
||||
}
|
||||
}
|
||||
|
|
@ -1,29 +1,26 @@
|
|||
"""Support for Freebox devices (Freebox v6 and Freebox mini 4K)."""
|
||||
import asyncio
|
||||
import logging
|
||||
import socket
|
||||
|
||||
from aiofreepybox import Freepybox
|
||||
from aiofreepybox.exceptions import HttpRequestError
|
||||
import voluptuous as vol
|
||||
|
||||
from homeassistant.components.discovery import SERVICE_FREEBOX
|
||||
from homeassistant.config_entries import SOURCE_DISCOVERY, SOURCE_IMPORT, ConfigEntry
|
||||
from homeassistant.const import CONF_HOST, CONF_PORT, EVENT_HOMEASSISTANT_STOP
|
||||
from homeassistant.helpers import config_validation as cv, discovery
|
||||
from homeassistant.helpers.discovery import async_load_platform
|
||||
from homeassistant.helpers.typing import HomeAssistantType
|
||||
|
||||
from .const import DOMAIN, PLATFORMS
|
||||
from .router import FreeboxRouter
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
DOMAIN = "freebox"
|
||||
DATA_FREEBOX = DOMAIN
|
||||
|
||||
FREEBOX_CONFIG_FILE = "freebox.conf"
|
||||
FREEBOX_SCHEMA = vol.Schema(
|
||||
{vol.Required(CONF_HOST): cv.string, vol.Required(CONF_PORT): cv.port}
|
||||
)
|
||||
|
||||
CONFIG_SCHEMA = vol.Schema(
|
||||
{
|
||||
DOMAIN: vol.Schema(
|
||||
{vol.Required(CONF_HOST): cv.string, vol.Required(CONF_PORT): cv.port}
|
||||
)
|
||||
},
|
||||
{DOMAIN: vol.Schema(vol.All(cv.ensure_list, [FREEBOX_SCHEMA]))},
|
||||
extra=vol.ALLOW_EXTRA,
|
||||
)
|
||||
|
||||
|
|
@ -37,54 +34,70 @@ async def async_setup(hass, config):
|
|||
host = discovery_info.get("properties", {}).get("api_domain")
|
||||
port = discovery_info.get("properties", {}).get("https_port")
|
||||
_LOGGER.info("Discovered Freebox server: %s:%s", host, port)
|
||||
await async_setup_freebox(hass, config, host, port)
|
||||
hass.async_create_task(
|
||||
hass.config_entries.flow.async_init(
|
||||
DOMAIN,
|
||||
context={"source": SOURCE_DISCOVERY},
|
||||
data={CONF_HOST: host, CONF_PORT: port},
|
||||
)
|
||||
)
|
||||
|
||||
discovery.async_listen(hass, SERVICE_FREEBOX, discovery_dispatch)
|
||||
|
||||
if conf is not None:
|
||||
host = conf.get(CONF_HOST)
|
||||
port = conf.get(CONF_PORT)
|
||||
await async_setup_freebox(hass, config, host, port)
|
||||
if conf is None:
|
||||
return True
|
||||
|
||||
for freebox_conf in conf:
|
||||
hass.async_create_task(
|
||||
hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": SOURCE_IMPORT}, data=freebox_conf,
|
||||
)
|
||||
)
|
||||
|
||||
return True
|
||||
|
||||
|
||||
async def async_setup_freebox(hass, config, host, port):
|
||||
"""Start up the Freebox component platforms."""
|
||||
async def async_setup_entry(hass: HomeAssistantType, entry: ConfigEntry):
|
||||
"""Set up Freebox component."""
|
||||
router = FreeboxRouter(hass, entry)
|
||||
await router.setup()
|
||||
|
||||
app_desc = {
|
||||
"app_id": "hass",
|
||||
"app_name": "Home Assistant",
|
||||
"app_version": "0.65",
|
||||
"device_name": socket.gethostname(),
|
||||
}
|
||||
hass.data.setdefault(DOMAIN, {})
|
||||
hass.data[DOMAIN][entry.unique_id] = router
|
||||
|
||||
token_file = hass.config.path(FREEBOX_CONFIG_FILE)
|
||||
api_version = "v6"
|
||||
|
||||
fbx = Freepybox(app_desc=app_desc, token_file=token_file, api_version=api_version)
|
||||
|
||||
try:
|
||||
await fbx.open(host, port)
|
||||
except HttpRequestError:
|
||||
_LOGGER.exception("Failed to connect to Freebox")
|
||||
else:
|
||||
hass.data[DATA_FREEBOX] = fbx
|
||||
|
||||
async def async_freebox_reboot(call):
|
||||
"""Handle reboot service call."""
|
||||
await fbx.system.reboot()
|
||||
|
||||
hass.services.async_register(DOMAIN, "reboot", async_freebox_reboot)
|
||||
|
||||
hass.async_create_task(async_load_platform(hass, "sensor", DOMAIN, {}, config))
|
||||
for platform in PLATFORMS:
|
||||
hass.async_create_task(
|
||||
async_load_platform(hass, "device_tracker", DOMAIN, {}, config)
|
||||
hass.config_entries.async_forward_entry_setup(entry, platform)
|
||||
)
|
||||
hass.async_create_task(async_load_platform(hass, "switch", DOMAIN, {}, config))
|
||||
|
||||
async def close_fbx(event):
|
||||
"""Close Freebox connection on HA Stop."""
|
||||
await fbx.close()
|
||||
# Services
|
||||
async def async_reboot(call):
|
||||
"""Handle reboot service call."""
|
||||
await router.reboot()
|
||||
|
||||
hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP, close_fbx)
|
||||
hass.services.async_register(DOMAIN, "reboot", async_reboot)
|
||||
|
||||
async def async_close_connection(event):
|
||||
"""Close Freebox connection on HA Stop."""
|
||||
await router.close()
|
||||
|
||||
hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP, async_close_connection)
|
||||
|
||||
return True
|
||||
|
||||
|
||||
async def async_unload_entry(hass: HomeAssistantType, entry: ConfigEntry):
|
||||
"""Unload a config entry."""
|
||||
unload_ok = all(
|
||||
await asyncio.gather(
|
||||
*[
|
||||
hass.config_entries.async_forward_entry_unload(entry, platform)
|
||||
for platform in PLATFORMS
|
||||
]
|
||||
)
|
||||
)
|
||||
if unload_ok:
|
||||
router = hass.data[DOMAIN].pop(entry.unique_id)
|
||||
await router.close()
|
||||
|
||||
return unload_ok
|
||||
|
|
|
|||
|
|
@ -0,0 +1,110 @@
|
|||
"""Config flow to configure the Freebox integration."""
|
||||
import logging
|
||||
|
||||
from aiofreepybox.exceptions import AuthorizationError, HttpRequestError
|
||||
import voluptuous as vol
|
||||
|
||||
from homeassistant import config_entries
|
||||
from homeassistant.const import CONF_HOST, CONF_PORT
|
||||
|
||||
from .const import DOMAIN # pylint: disable=unused-import
|
||||
from .router import get_api
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class FreeboxFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
|
||||
"""Handle a config flow."""
|
||||
|
||||
VERSION = 1
|
||||
CONNECTION_CLASS = config_entries.CONN_CLASS_LOCAL_POLL
|
||||
|
||||
def __init__(self):
|
||||
"""Initialize Freebox config flow."""
|
||||
self._host = None
|
||||
self._port = None
|
||||
|
||||
def _show_setup_form(self, user_input=None, errors=None):
|
||||
"""Show the setup form to the user."""
|
||||
|
||||
if user_input is None:
|
||||
user_input = {}
|
||||
|
||||
return self.async_show_form(
|
||||
step_id="user",
|
||||
data_schema=vol.Schema(
|
||||
{
|
||||
vol.Required(CONF_HOST, default=user_input.get(CONF_HOST, "")): str,
|
||||
vol.Required(CONF_PORT, default=user_input.get(CONF_PORT, "")): int,
|
||||
}
|
||||
),
|
||||
errors=errors or {},
|
||||
)
|
||||
|
||||
async def async_step_user(self, user_input=None):
|
||||
"""Handle a flow initiated by the user."""
|
||||
errors = {}
|
||||
|
||||
if user_input is None:
|
||||
return self._show_setup_form(user_input, errors)
|
||||
|
||||
self._host = user_input[CONF_HOST]
|
||||
self._port = user_input[CONF_PORT]
|
||||
|
||||
# Check if already configured
|
||||
await self.async_set_unique_id(self._host)
|
||||
self._abort_if_unique_id_configured()
|
||||
|
||||
return await self.async_step_link()
|
||||
|
||||
async def async_step_link(self, user_input=None):
|
||||
"""Attempt to link with the Freebox router.
|
||||
|
||||
Given a configured host, will ask the user to press the button
|
||||
to connect to the router.
|
||||
"""
|
||||
if user_input is None:
|
||||
return self.async_show_form(step_id="link")
|
||||
|
||||
errors = {}
|
||||
|
||||
fbx = await get_api(self.hass, self._host)
|
||||
try:
|
||||
# Open connection and check authentication
|
||||
await fbx.open(self._host, self._port)
|
||||
|
||||
# Check permissions
|
||||
await fbx.system.get_config()
|
||||
await fbx.lan.get_hosts_list()
|
||||
await self.hass.async_block_till_done()
|
||||
|
||||
# Close connection
|
||||
await fbx.close()
|
||||
|
||||
return self.async_create_entry(
|
||||
title=self._host, data={CONF_HOST: self._host, CONF_PORT: self._port},
|
||||
)
|
||||
|
||||
except AuthorizationError as error:
|
||||
_LOGGER.error(error)
|
||||
errors["base"] = "register_failed"
|
||||
|
||||
except HttpRequestError:
|
||||
_LOGGER.error("Error connecting to the Freebox router at %s", self._host)
|
||||
errors["base"] = "connection_failed"
|
||||
|
||||
except Exception: # pylint: disable=broad-except
|
||||
_LOGGER.exception(
|
||||
"Unknown error connecting with Freebox router at %s", self._host
|
||||
)
|
||||
errors["base"] = "unknown"
|
||||
|
||||
return self.async_show_form(step_id="link", errors=errors)
|
||||
|
||||
async def async_step_import(self, user_input=None):
|
||||
"""Import a config entry."""
|
||||
return await self.async_step_user(user_input)
|
||||
|
||||
async def async_step_discovery(self, user_input=None):
|
||||
"""Initialize step from discovery."""
|
||||
return await self.async_step_user(user_input)
|
||||
|
|
@ -0,0 +1,75 @@
|
|||
"""Freebox component constants."""
|
||||
import socket
|
||||
|
||||
from homeassistant.const import (
|
||||
DATA_RATE_KILOBYTES_PER_SECOND,
|
||||
DEVICE_CLASS_TEMPERATURE,
|
||||
TEMP_CELSIUS,
|
||||
)
|
||||
|
||||
DOMAIN = "freebox"
|
||||
|
||||
APP_DESC = {
|
||||
"app_id": "hass",
|
||||
"app_name": "Home Assistant",
|
||||
"app_version": "0.106",
|
||||
"device_name": socket.gethostname(),
|
||||
}
|
||||
API_VERSION = "v6"
|
||||
|
||||
PLATFORMS = ["device_tracker", "sensor", "switch"]
|
||||
|
||||
DEFAULT_DEVICE_NAME = "Unknown device"
|
||||
|
||||
# to store the cookie
|
||||
STORAGE_KEY = DOMAIN
|
||||
STORAGE_VERSION = 1
|
||||
|
||||
# Sensor
|
||||
SENSOR_NAME = "name"
|
||||
SENSOR_UNIT = "unit"
|
||||
SENSOR_ICON = "icon"
|
||||
SENSOR_DEVICE_CLASS = "device_class"
|
||||
|
||||
CONNECTION_SENSORS = {
|
||||
"rate_down": {
|
||||
SENSOR_NAME: "Freebox download speed",
|
||||
SENSOR_UNIT: DATA_RATE_KILOBYTES_PER_SECOND,
|
||||
SENSOR_ICON: "mdi:download-network",
|
||||
SENSOR_DEVICE_CLASS: None,
|
||||
},
|
||||
"rate_up": {
|
||||
SENSOR_NAME: "Freebox upload speed",
|
||||
SENSOR_UNIT: DATA_RATE_KILOBYTES_PER_SECOND,
|
||||
SENSOR_ICON: "mdi:upload-network",
|
||||
SENSOR_DEVICE_CLASS: None,
|
||||
},
|
||||
}
|
||||
|
||||
TEMPERATURE_SENSOR_TEMPLATE = {
|
||||
SENSOR_NAME: None,
|
||||
SENSOR_UNIT: TEMP_CELSIUS,
|
||||
SENSOR_ICON: "mdi:thermometer",
|
||||
SENSOR_DEVICE_CLASS: DEVICE_CLASS_TEMPERATURE,
|
||||
}
|
||||
|
||||
# Icons
|
||||
DEVICE_ICONS = {
|
||||
"freebox_delta": "mdi:television-guide",
|
||||
"freebox_hd": "mdi:television-guide",
|
||||
"freebox_mini": "mdi:television-guide",
|
||||
"freebox_player": "mdi:television-guide",
|
||||
"ip_camera": "mdi:cctv",
|
||||
"ip_phone": "mdi:phone-voip",
|
||||
"laptop": "mdi:laptop",
|
||||
"multimedia_device": "mdi:play-network",
|
||||
"nas": "mdi:nas",
|
||||
"networking_device": "mdi:network",
|
||||
"printer": "mdi:printer",
|
||||
"router": "mdi:router-wireless",
|
||||
"smartphone": "mdi:cellphone",
|
||||
"tablet": "mdi:tablet",
|
||||
"television": "mdi:television",
|
||||
"vg_console": "mdi:gamepad-variant",
|
||||
"workstation": "mdi:desktop-tower-monitor",
|
||||
}
|
||||
|
|
@ -1,65 +1,148 @@
|
|||
"""Support for Freebox devices (Freebox v6 and Freebox mini 4K)."""
|
||||
from collections import namedtuple
|
||||
from datetime import datetime
|
||||
import logging
|
||||
from typing import Dict
|
||||
|
||||
from homeassistant.components.device_tracker import DeviceScanner
|
||||
from homeassistant.components.device_tracker import SOURCE_TYPE_ROUTER
|
||||
from homeassistant.components.device_tracker.config_entry import ScannerEntity
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
from homeassistant.core import callback
|
||||
from homeassistant.helpers.device_registry import CONNECTION_NETWORK_MAC
|
||||
from homeassistant.helpers.dispatcher import async_dispatcher_connect
|
||||
from homeassistant.helpers.typing import HomeAssistantType
|
||||
|
||||
from . import DATA_FREEBOX
|
||||
from .const import DEFAULT_DEVICE_NAME, DEVICE_ICONS, DOMAIN
|
||||
from .router import FreeboxRouter
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
||||
async def async_get_scanner(hass, config):
|
||||
"""Validate the configuration and return a Freebox scanner."""
|
||||
scanner = FreeboxDeviceScanner(hass.data[DATA_FREEBOX])
|
||||
await scanner.async_connect()
|
||||
return scanner if scanner.success_init else None
|
||||
async def async_setup_entry(
|
||||
hass: HomeAssistantType, entry: ConfigEntry, async_add_entities
|
||||
) -> None:
|
||||
"""Set up device tracker for Freebox component."""
|
||||
router = hass.data[DOMAIN][entry.unique_id]
|
||||
tracked = set()
|
||||
|
||||
@callback
|
||||
def update_router():
|
||||
"""Update the values of the router."""
|
||||
add_entities(router, async_add_entities, tracked)
|
||||
|
||||
Device = namedtuple("Device", ["id", "name", "ip"])
|
||||
|
||||
|
||||
def _build_device(device_dict):
|
||||
return Device(
|
||||
device_dict["l2ident"]["id"],
|
||||
device_dict["primary_name"],
|
||||
device_dict["l3connectivities"][0]["addr"],
|
||||
router.listeners.append(
|
||||
async_dispatcher_connect(hass, router.signal_device_new, update_router)
|
||||
)
|
||||
|
||||
update_router()
|
||||
|
||||
class FreeboxDeviceScanner(DeviceScanner):
|
||||
"""Queries the Freebox device."""
|
||||
|
||||
def __init__(self, fbx):
|
||||
"""Initialize the scanner."""
|
||||
self.last_results = {}
|
||||
self.success_init = False
|
||||
self.connection = fbx
|
||||
@callback
|
||||
def add_entities(router, async_add_entities, tracked):
|
||||
"""Add new tracker entities from the router."""
|
||||
new_tracked = []
|
||||
|
||||
async def async_connect(self):
|
||||
"""Initialize connection to the router."""
|
||||
# Test the router is accessible.
|
||||
data = await self.connection.lan.get_hosts_list()
|
||||
self.success_init = data is not None
|
||||
for mac, device in router.devices.items():
|
||||
if mac in tracked:
|
||||
continue
|
||||
|
||||
async def async_scan_devices(self):
|
||||
"""Scan for new devices and return a list with found device IDs."""
|
||||
await self.async_update_info()
|
||||
return [device.id for device in self.last_results]
|
||||
new_tracked.append(FreeboxDevice(router, device))
|
||||
tracked.add(mac)
|
||||
|
||||
async def get_device_name(self, device):
|
||||
"""Return the name of the given device or None if we don't know."""
|
||||
name = next(
|
||||
(result.name for result in self.last_results if result.id == device), None
|
||||
if new_tracked:
|
||||
async_add_entities(new_tracked, True)
|
||||
|
||||
|
||||
class FreeboxDevice(ScannerEntity):
|
||||
"""Representation of a Freebox device."""
|
||||
|
||||
def __init__(self, router: FreeboxRouter, device: Dict[str, any]) -> None:
|
||||
"""Initialize a Freebox device."""
|
||||
self._router = router
|
||||
self._name = device["primary_name"].strip() or DEFAULT_DEVICE_NAME
|
||||
self._mac = device["l2ident"]["id"]
|
||||
self._manufacturer = device["vendor_name"]
|
||||
self._icon = icon_for_freebox_device(device)
|
||||
self._active = False
|
||||
self._attrs = {}
|
||||
|
||||
self._unsub_dispatcher = None
|
||||
|
||||
def update(self) -> None:
|
||||
"""Update the Freebox device."""
|
||||
device = self._router.devices[self._mac]
|
||||
self._active = device["active"]
|
||||
if device.get("attrs") is None:
|
||||
# device
|
||||
self._attrs = {
|
||||
"last_time_reachable": datetime.fromtimestamp(
|
||||
device["last_time_reachable"]
|
||||
),
|
||||
"last_time_activity": datetime.fromtimestamp(device["last_activity"]),
|
||||
}
|
||||
else:
|
||||
# router
|
||||
self._attrs = device["attrs"]
|
||||
|
||||
@property
|
||||
def unique_id(self) -> str:
|
||||
"""Return a unique ID."""
|
||||
return self._mac
|
||||
|
||||
@property
|
||||
def name(self) -> str:
|
||||
"""Return the name."""
|
||||
return self._name
|
||||
|
||||
@property
|
||||
def is_connected(self):
|
||||
"""Return true if the device is connected to the network."""
|
||||
return self._active
|
||||
|
||||
@property
|
||||
def source_type(self) -> str:
|
||||
"""Return the source type."""
|
||||
return SOURCE_TYPE_ROUTER
|
||||
|
||||
@property
|
||||
def icon(self) -> str:
|
||||
"""Return the icon."""
|
||||
return self._icon
|
||||
|
||||
@property
|
||||
def device_state_attributes(self) -> Dict[str, any]:
|
||||
"""Return the attributes."""
|
||||
return self._attrs
|
||||
|
||||
@property
|
||||
def device_info(self) -> Dict[str, any]:
|
||||
"""Return the device information."""
|
||||
return {
|
||||
"connections": {(CONNECTION_NETWORK_MAC, self._mac)},
|
||||
"identifiers": {(DOMAIN, self.unique_id)},
|
||||
"name": self.name,
|
||||
"manufacturer": self._manufacturer,
|
||||
}
|
||||
|
||||
@property
|
||||
def should_poll(self) -> bool:
|
||||
"""No polling needed."""
|
||||
return False
|
||||
|
||||
async def async_on_demand_update(self):
|
||||
"""Update state."""
|
||||
self.async_schedule_update_ha_state(True)
|
||||
|
||||
async def async_added_to_hass(self):
|
||||
"""Register state update callback."""
|
||||
self._unsub_dispatcher = async_dispatcher_connect(
|
||||
self.hass, self._router.signal_device_update, self.async_on_demand_update
|
||||
)
|
||||
return name
|
||||
|
||||
async def async_update_info(self):
|
||||
"""Ensure the information from the Freebox router is up to date."""
|
||||
_LOGGER.debug("Checking Devices")
|
||||
async def async_will_remove_from_hass(self):
|
||||
"""Clean up after entity before removal."""
|
||||
self._unsub_dispatcher()
|
||||
|
||||
hosts = await self.connection.lan.get_hosts_list()
|
||||
|
||||
last_results = [_build_device(device) for device in hosts if device["active"]]
|
||||
|
||||
self.last_results = last_results
|
||||
def icon_for_freebox_device(device) -> str:
|
||||
"""Return a host icon from his type."""
|
||||
return DEVICE_ICONS.get(device["host_type"], "mdi:help-network")
|
||||
|
|
|
|||
|
|
@ -1,9 +1,10 @@
|
|||
{
|
||||
"domain": "freebox",
|
||||
"name": "Freebox",
|
||||
"config_flow": true,
|
||||
"documentation": "https://www.home-assistant.io/integrations/freebox",
|
||||
"requirements": ["aiofreepybox==0.0.8"],
|
||||
"dependencies": [],
|
||||
"after_dependencies": ["discovery"],
|
||||
"codeowners": ["@snoof85"]
|
||||
"codeowners": ["@snoof85", "@Quentame"]
|
||||
}
|
||||
|
|
|
|||
|
|
@ -0,0 +1,193 @@
|
|||
"""Represent the Freebox router and its devices and sensors."""
|
||||
from datetime import datetime, timedelta
|
||||
import logging
|
||||
from pathlib import Path
|
||||
from typing import Dict, Optional
|
||||
|
||||
from aiofreepybox import Freepybox
|
||||
from aiofreepybox.api.wifi import Wifi
|
||||
from aiofreepybox.exceptions import HttpRequestError
|
||||
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
from homeassistant.const import CONF_HOST, CONF_PORT
|
||||
from homeassistant.exceptions import ConfigEntryNotReady
|
||||
from homeassistant.helpers.device_registry import CONNECTION_NETWORK_MAC
|
||||
from homeassistant.helpers.dispatcher import async_dispatcher_send
|
||||
from homeassistant.helpers.event import async_track_time_interval
|
||||
from homeassistant.helpers.typing import HomeAssistantType
|
||||
from homeassistant.util import slugify
|
||||
|
||||
from .const import (
|
||||
API_VERSION,
|
||||
APP_DESC,
|
||||
CONNECTION_SENSORS,
|
||||
DOMAIN,
|
||||
STORAGE_KEY,
|
||||
STORAGE_VERSION,
|
||||
)
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
SCAN_INTERVAL = timedelta(seconds=30)
|
||||
|
||||
|
||||
class FreeboxRouter:
|
||||
"""Representation of a Freebox router."""
|
||||
|
||||
def __init__(self, hass: HomeAssistantType, entry: ConfigEntry) -> None:
|
||||
"""Initialize a Freebox router."""
|
||||
self.hass = hass
|
||||
self._entry = entry
|
||||
self._host = entry.data[CONF_HOST]
|
||||
self._port = entry.data[CONF_PORT]
|
||||
|
||||
self._api: Freepybox = None
|
||||
self._name = None
|
||||
self.mac = None
|
||||
self._sw_v = None
|
||||
self._attrs = {}
|
||||
|
||||
self.devices: Dict[str, any] = {}
|
||||
self.sensors_temperature: Dict[str, int] = {}
|
||||
self.sensors_connection: Dict[str, float] = {}
|
||||
|
||||
self.listeners = []
|
||||
|
||||
async def setup(self) -> None:
|
||||
"""Set up a Freebox router."""
|
||||
self._api = await get_api(self.hass, self._host)
|
||||
|
||||
try:
|
||||
await self._api.open(self._host, self._port)
|
||||
except HttpRequestError:
|
||||
_LOGGER.exception("Failed to connect to Freebox")
|
||||
return ConfigEntryNotReady
|
||||
|
||||
# System
|
||||
fbx_config = await self._api.system.get_config()
|
||||
self.mac = fbx_config["mac"]
|
||||
self._name = fbx_config["model_info"]["pretty_name"]
|
||||
self._sw_v = fbx_config["firmware_version"]
|
||||
|
||||
# Devices & sensors
|
||||
await self.update_all()
|
||||
async_track_time_interval(self.hass, self.update_all, SCAN_INTERVAL)
|
||||
|
||||
async def update_all(self, now: Optional[datetime] = None) -> None:
|
||||
"""Update all Freebox platforms."""
|
||||
await self.update_sensors()
|
||||
await self.update_devices()
|
||||
|
||||
async def update_devices(self) -> None:
|
||||
"""Update Freebox devices."""
|
||||
new_device = False
|
||||
fbx_devices: Dict[str, any] = await self._api.lan.get_hosts_list()
|
||||
|
||||
# Adds the Freebox itself
|
||||
fbx_devices.append(
|
||||
{
|
||||
"primary_name": self._name,
|
||||
"l2ident": {"id": self.mac},
|
||||
"vendor_name": "Freebox SAS",
|
||||
"host_type": "router",
|
||||
"active": True,
|
||||
"attrs": self._attrs,
|
||||
}
|
||||
)
|
||||
|
||||
for fbx_device in fbx_devices:
|
||||
device_mac = fbx_device["l2ident"]["id"]
|
||||
|
||||
if self.devices.get(device_mac) is None:
|
||||
new_device = True
|
||||
|
||||
self.devices[device_mac] = fbx_device
|
||||
|
||||
async_dispatcher_send(self.hass, self.signal_device_update)
|
||||
|
||||
if new_device:
|
||||
async_dispatcher_send(self.hass, self.signal_device_new)
|
||||
|
||||
async def update_sensors(self) -> None:
|
||||
"""Update Freebox sensors."""
|
||||
# System sensors
|
||||
syst_datas: Dict[str, any] = await self._api.system.get_config()
|
||||
|
||||
# According to the doc `syst_datas["sensors"]` is temperature sensors in celsius degree.
|
||||
# Name and id of sensors may vary under Freebox devices.
|
||||
for sensor in syst_datas["sensors"]:
|
||||
self.sensors_temperature[sensor["name"]] = sensor["value"]
|
||||
|
||||
# Connection sensors
|
||||
connection_datas: Dict[str, any] = await self._api.connection.get_status()
|
||||
for sensor_key in CONNECTION_SENSORS:
|
||||
self.sensors_connection[sensor_key] = connection_datas[sensor_key]
|
||||
|
||||
self._attrs = {
|
||||
"IPv4": connection_datas.get("ipv4"),
|
||||
"IPv6": connection_datas.get("ipv6"),
|
||||
"connection_type": connection_datas["media"],
|
||||
"uptime": datetime.fromtimestamp(
|
||||
round(datetime.now().timestamp()) - syst_datas["uptime_val"]
|
||||
),
|
||||
"firmware_version": self._sw_v,
|
||||
"serial": syst_datas["serial"],
|
||||
}
|
||||
|
||||
async_dispatcher_send(self.hass, self.signal_sensor_update)
|
||||
|
||||
async def reboot(self) -> None:
|
||||
"""Reboot the Freebox."""
|
||||
await self._api.system.reboot()
|
||||
|
||||
async def close(self) -> None:
|
||||
"""Close the connection."""
|
||||
if self._api is not None:
|
||||
await self._api.close()
|
||||
self._api = None
|
||||
|
||||
@property
|
||||
def device_info(self) -> Dict[str, any]:
|
||||
"""Return the device information."""
|
||||
return {
|
||||
"connections": {(CONNECTION_NETWORK_MAC, self.mac)},
|
||||
"identifiers": {(DOMAIN, self.mac)},
|
||||
"name": self._name,
|
||||
"manufacturer": "Freebox SAS",
|
||||
"sw_version": self._sw_v,
|
||||
}
|
||||
|
||||
@property
|
||||
def signal_device_new(self) -> str:
|
||||
"""Event specific per Freebox entry to signal new device."""
|
||||
return f"{DOMAIN}-{self._host}-device-new"
|
||||
|
||||
@property
|
||||
def signal_device_update(self) -> str:
|
||||
"""Event specific per Freebox entry to signal updates in devices."""
|
||||
return f"{DOMAIN}-{self._host}-device-update"
|
||||
|
||||
@property
|
||||
def signal_sensor_update(self) -> str:
|
||||
"""Event specific per Freebox entry to signal updates in sensors."""
|
||||
return f"{DOMAIN}-{self._host}-sensor-update"
|
||||
|
||||
@property
|
||||
def sensors(self) -> Wifi:
|
||||
"""Return the wifi."""
|
||||
return {**self.sensors_temperature, **self.sensors_connection}
|
||||
|
||||
@property
|
||||
def wifi(self) -> Wifi:
|
||||
"""Return the wifi."""
|
||||
return self._api.wifi
|
||||
|
||||
|
||||
async def get_api(hass: HomeAssistantType, host: str) -> Freepybox:
|
||||
"""Get the Freebox API."""
|
||||
freebox_path = Path(hass.helpers.storage.Store(STORAGE_VERSION, STORAGE_KEY).path)
|
||||
freebox_path.mkdir(exist_ok=True)
|
||||
|
||||
token_file = Path(f"{freebox_path}/{slugify(host)}.conf")
|
||||
|
||||
return Freepybox(APP_DESC, token_file, API_VERSION)
|
||||
|
|
@ -1,81 +1,127 @@
|
|||
"""Support for Freebox devices (Freebox v6 and Freebox mini 4K)."""
|
||||
import logging
|
||||
from typing import Dict
|
||||
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
from homeassistant.const import DATA_RATE_KILOBYTES_PER_SECOND
|
||||
from homeassistant.helpers.dispatcher import async_dispatcher_connect
|
||||
from homeassistant.helpers.entity import Entity
|
||||
from homeassistant.helpers.typing import HomeAssistantType
|
||||
|
||||
from . import DATA_FREEBOX
|
||||
from .const import (
|
||||
CONNECTION_SENSORS,
|
||||
DOMAIN,
|
||||
SENSOR_DEVICE_CLASS,
|
||||
SENSOR_ICON,
|
||||
SENSOR_NAME,
|
||||
SENSOR_UNIT,
|
||||
TEMPERATURE_SENSOR_TEMPLATE,
|
||||
)
|
||||
from .router import FreeboxRouter
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
||||
async def async_setup_platform(hass, config, async_add_entities, discovery_info=None):
|
||||
async def async_setup_entry(
|
||||
hass: HomeAssistantType, entry: ConfigEntry, async_add_entities
|
||||
) -> None:
|
||||
"""Set up the sensors."""
|
||||
fbx = hass.data[DATA_FREEBOX]
|
||||
async_add_entities([FbxRXSensor(fbx), FbxTXSensor(fbx)], True)
|
||||
router = hass.data[DOMAIN][entry.unique_id]
|
||||
entities = []
|
||||
|
||||
for sensor_name in router.sensors_temperature:
|
||||
entities.append(
|
||||
FreeboxSensor(
|
||||
router,
|
||||
sensor_name,
|
||||
{**TEMPERATURE_SENSOR_TEMPLATE, SENSOR_NAME: f"Freebox {sensor_name}"},
|
||||
)
|
||||
)
|
||||
|
||||
for sensor_key in CONNECTION_SENSORS:
|
||||
entities.append(
|
||||
FreeboxSensor(router, sensor_key, CONNECTION_SENSORS[sensor_key])
|
||||
)
|
||||
|
||||
async_add_entities(entities, True)
|
||||
|
||||
|
||||
class FbxSensor(Entity):
|
||||
"""Representation of a freebox sensor."""
|
||||
class FreeboxSensor(Entity):
|
||||
"""Representation of a Freebox sensor."""
|
||||
|
||||
_name = "generic"
|
||||
_unit = None
|
||||
_icon = None
|
||||
|
||||
def __init__(self, fbx):
|
||||
"""Initialize the sensor."""
|
||||
self._fbx = fbx
|
||||
def __init__(
|
||||
self, router: FreeboxRouter, sensor_type: str, sensor: Dict[str, any]
|
||||
) -> None:
|
||||
"""Initialize a Freebox sensor."""
|
||||
self._state = None
|
||||
self._datas = None
|
||||
self._router = router
|
||||
self._sensor_type = sensor_type
|
||||
self._name = sensor[SENSOR_NAME]
|
||||
self._unit = sensor[SENSOR_UNIT]
|
||||
self._icon = sensor[SENSOR_ICON]
|
||||
self._device_class = sensor[SENSOR_DEVICE_CLASS]
|
||||
self._unique_id = f"{self._router.mac} {self._name}"
|
||||
|
||||
self._unsub_dispatcher = None
|
||||
|
||||
def update(self) -> None:
|
||||
"""Update the Freebox sensor."""
|
||||
state = self._router.sensors[self._sensor_type]
|
||||
if self._unit == DATA_RATE_KILOBYTES_PER_SECOND:
|
||||
self._state = round(state / 1000, 2)
|
||||
else:
|
||||
self._state = state
|
||||
|
||||
@property
|
||||
def name(self):
|
||||
"""Return the name of the sensor."""
|
||||
def unique_id(self) -> str:
|
||||
"""Return a unique ID."""
|
||||
return self._unique_id
|
||||
|
||||
@property
|
||||
def name(self) -> str:
|
||||
"""Return the name."""
|
||||
return self._name
|
||||
|
||||
@property
|
||||
def unit_of_measurement(self):
|
||||
"""Return the unit of the sensor."""
|
||||
def state(self) -> str:
|
||||
"""Return the state."""
|
||||
return self._state
|
||||
|
||||
@property
|
||||
def unit_of_measurement(self) -> str:
|
||||
"""Return the unit."""
|
||||
return self._unit
|
||||
|
||||
@property
|
||||
def icon(self):
|
||||
"""Return the icon of the sensor."""
|
||||
def icon(self) -> str:
|
||||
"""Return the icon."""
|
||||
return self._icon
|
||||
|
||||
@property
|
||||
def state(self):
|
||||
"""Return the state of the sensor."""
|
||||
return self._state
|
||||
def device_class(self) -> str:
|
||||
"""Return the device_class."""
|
||||
return self._device_class
|
||||
|
||||
async def async_update(self):
|
||||
"""Fetch status from freebox."""
|
||||
self._datas = await self._fbx.connection.get_status()
|
||||
@property
|
||||
def device_info(self) -> Dict[str, any]:
|
||||
"""Return the device information."""
|
||||
return self._router.device_info
|
||||
|
||||
@property
|
||||
def should_poll(self) -> bool:
|
||||
"""No polling needed."""
|
||||
return False
|
||||
|
||||
class FbxRXSensor(FbxSensor):
|
||||
"""Update the Freebox RxSensor."""
|
||||
async def async_on_demand_update(self):
|
||||
"""Update state."""
|
||||
self.async_schedule_update_ha_state(True)
|
||||
|
||||
_name = "Freebox download speed"
|
||||
_unit = DATA_RATE_KILOBYTES_PER_SECOND
|
||||
_icon = "mdi:download-network"
|
||||
async def async_added_to_hass(self):
|
||||
"""Register state update callback."""
|
||||
self._unsub_dispatcher = async_dispatcher_connect(
|
||||
self.hass, self._router.signal_sensor_update, self.async_on_demand_update
|
||||
)
|
||||
|
||||
async def async_update(self):
|
||||
"""Get the value from fetched datas."""
|
||||
await super().async_update()
|
||||
if self._datas is not None:
|
||||
self._state = round(self._datas["rate_down"] / 1000, 2)
|
||||
|
||||
|
||||
class FbxTXSensor(FbxSensor):
|
||||
"""Update the Freebox TxSensor."""
|
||||
|
||||
_name = "Freebox upload speed"
|
||||
_unit = DATA_RATE_KILOBYTES_PER_SECOND
|
||||
_icon = "mdi:upload-network"
|
||||
|
||||
async def async_update(self):
|
||||
"""Get the value from fetched datas."""
|
||||
await super().async_update()
|
||||
if self._datas is not None:
|
||||
self._state = round(self._datas["rate_up"] / 1000, 2)
|
||||
async def async_will_remove_from_hass(self):
|
||||
"""Clean up after entity before removal."""
|
||||
self._unsub_dispatcher()
|
||||
|
|
|
|||
|
|
@ -0,0 +1,26 @@
|
|||
{
|
||||
"config": {
|
||||
"title": "Freebox",
|
||||
"step": {
|
||||
"user": {
|
||||
"title": "Freebox",
|
||||
"data": {
|
||||
"host": "Host",
|
||||
"port": "Port"
|
||||
}
|
||||
},
|
||||
"link": {
|
||||
"title": "Link Freebox router",
|
||||
"description": "Click \"Submit\", then touch the right arrow on the router to register Freebox with Home Assistant.\n\n"
|
||||
}
|
||||
},
|
||||
"error":{
|
||||
"register_failed": "Failed to register, please try again",
|
||||
"connection_failed": "Failed to connect, please try again",
|
||||
"unknown": "Unknown error: please retry later"
|
||||
},
|
||||
"abort":{
|
||||
"already_configured": "Host already configured"
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -1,50 +1,65 @@
|
|||
"""Support for Freebox Delta, Revolution and Mini 4K."""
|
||||
import logging
|
||||
from typing import Dict
|
||||
|
||||
from aiofreepybox.exceptions import InsufficientPermissionsError
|
||||
|
||||
from homeassistant.components.switch import SwitchDevice
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
from homeassistant.helpers.typing import HomeAssistantType
|
||||
|
||||
from . import DATA_FREEBOX
|
||||
from .const import DOMAIN
|
||||
from .router import FreeboxRouter
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
||||
async def async_setup_platform(hass, config, async_add_entities, discovery_info=None):
|
||||
async def async_setup_entry(
|
||||
hass: HomeAssistantType, entry: ConfigEntry, async_add_entities
|
||||
) -> None:
|
||||
"""Set up the switch."""
|
||||
fbx = hass.data[DATA_FREEBOX]
|
||||
async_add_entities([FbxWifiSwitch(fbx)], True)
|
||||
router = hass.data[DOMAIN][entry.unique_id]
|
||||
async_add_entities([FreeboxWifiSwitch(router)], True)
|
||||
|
||||
|
||||
class FbxWifiSwitch(SwitchDevice):
|
||||
class FreeboxWifiSwitch(SwitchDevice):
|
||||
"""Representation of a freebox wifi switch."""
|
||||
|
||||
def __init__(self, fbx):
|
||||
def __init__(self, router: FreeboxRouter) -> None:
|
||||
"""Initialize the Wifi switch."""
|
||||
self._name = "Freebox WiFi"
|
||||
self._state = None
|
||||
self._fbx = fbx
|
||||
self._router = router
|
||||
self._unique_id = f"{self._router.mac} {self._name}"
|
||||
|
||||
@property
|
||||
def name(self):
|
||||
def unique_id(self) -> str:
|
||||
"""Return a unique ID."""
|
||||
return self._unique_id
|
||||
|
||||
@property
|
||||
def name(self) -> str:
|
||||
"""Return the name of the switch."""
|
||||
return self._name
|
||||
|
||||
@property
|
||||
def is_on(self):
|
||||
def is_on(self) -> bool:
|
||||
"""Return true if device is on."""
|
||||
return self._state
|
||||
|
||||
async def _async_set_state(self, enabled):
|
||||
"""Turn the switch on or off."""
|
||||
from aiofreepybox.exceptions import InsufficientPermissionsError
|
||||
@property
|
||||
def device_info(self) -> Dict[str, any]:
|
||||
"""Return the device information."""
|
||||
return self._router.device_info
|
||||
|
||||
async def _async_set_state(self, enabled: bool):
|
||||
"""Turn the switch on or off."""
|
||||
wifi_config = {"enabled": enabled}
|
||||
try:
|
||||
await self._fbx.wifi.set_global_config(wifi_config)
|
||||
await self._router.wifi.set_global_config(wifi_config)
|
||||
except InsufficientPermissionsError:
|
||||
_LOGGER.warning(
|
||||
"Home Assistant does not have permissions to"
|
||||
" modify the Freebox settings. Please refer"
|
||||
" to documentation."
|
||||
"Home Assistant does not have permissions to modify the Freebox settings. Please refer to documentation."
|
||||
)
|
||||
|
||||
async def async_turn_on(self, **kwargs):
|
||||
|
|
@ -57,6 +72,6 @@ class FbxWifiSwitch(SwitchDevice):
|
|||
|
||||
async def async_update(self):
|
||||
"""Get the state and update it."""
|
||||
datas = await self._fbx.wifi.get_global_config()
|
||||
datas = await self._router.wifi.get_global_config()
|
||||
active = datas["enabled"]
|
||||
self._state = bool(active)
|
||||
|
|
|
|||
|
|
@ -29,6 +29,7 @@ FLOWS = [
|
|||
"elgato",
|
||||
"emulated_roku",
|
||||
"esphome",
|
||||
"freebox",
|
||||
"garmin_connect",
|
||||
"gdacs",
|
||||
"geofency",
|
||||
|
|
|
|||
|
|
@ -61,6 +61,9 @@ aiobotocore==0.11.1
|
|||
# homeassistant.components.esphome
|
||||
aioesphomeapi==2.6.1
|
||||
|
||||
# homeassistant.components.freebox
|
||||
aiofreepybox==0.0.8
|
||||
|
||||
# homeassistant.components.homekit_controller
|
||||
aiohomekit[IP]==0.2.29
|
||||
|
||||
|
|
|
|||
|
|
@ -0,0 +1 @@
|
|||
"""Tests for the Freebox component."""
|
||||
|
|
@ -0,0 +1,11 @@
|
|||
"""Test helpers for Freebox."""
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def mock_path():
|
||||
"""Mock path lib."""
|
||||
with patch("homeassistant.components.freebox.router.Path"):
|
||||
yield
|
||||
|
|
@ -0,0 +1,144 @@
|
|||
"""Tests for the Freebox config flow."""
|
||||
from aiofreepybox.exceptions import (
|
||||
AuthorizationError,
|
||||
HttpRequestError,
|
||||
InvalidTokenError,
|
||||
)
|
||||
from asynctest import CoroutineMock, patch
|
||||
import pytest
|
||||
|
||||
from homeassistant import data_entry_flow
|
||||
from homeassistant.components.freebox.const import DOMAIN
|
||||
from homeassistant.config_entries import SOURCE_DISCOVERY, SOURCE_IMPORT, SOURCE_USER
|
||||
from homeassistant.const import CONF_HOST, CONF_PORT
|
||||
|
||||
from tests.common import MockConfigEntry
|
||||
|
||||
HOST = "myrouter.freeboxos.fr"
|
||||
PORT = 1234
|
||||
|
||||
|
||||
@pytest.fixture(name="connect")
|
||||
def mock_controller_connect():
|
||||
"""Mock a successful connection."""
|
||||
with patch("homeassistant.components.freebox.router.Freepybox") as service_mock:
|
||||
service_mock.return_value.open = CoroutineMock()
|
||||
service_mock.return_value.system.get_config = CoroutineMock()
|
||||
service_mock.return_value.lan.get_hosts_list = CoroutineMock()
|
||||
service_mock.return_value.connection.get_status = CoroutineMock()
|
||||
service_mock.return_value.close = CoroutineMock()
|
||||
yield service_mock
|
||||
|
||||
|
||||
async def test_user(hass):
|
||||
"""Test user config."""
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": SOURCE_USER}
|
||||
)
|
||||
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
|
||||
assert result["step_id"] == "user"
|
||||
|
||||
# test with all provided
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN,
|
||||
context={"source": SOURCE_USER},
|
||||
data={CONF_HOST: HOST, CONF_PORT: PORT},
|
||||
)
|
||||
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
|
||||
assert result["step_id"] == "link"
|
||||
|
||||
|
||||
async def test_import(hass):
|
||||
"""Test import step."""
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN,
|
||||
context={"source": SOURCE_IMPORT},
|
||||
data={CONF_HOST: HOST, CONF_PORT: PORT},
|
||||
)
|
||||
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
|
||||
assert result["step_id"] == "link"
|
||||
|
||||
|
||||
async def test_discovery(hass):
|
||||
"""Test discovery step."""
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN,
|
||||
context={"source": SOURCE_DISCOVERY},
|
||||
data={CONF_HOST: HOST, CONF_PORT: PORT},
|
||||
)
|
||||
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
|
||||
assert result["step_id"] == "link"
|
||||
|
||||
|
||||
async def test_link(hass, connect):
|
||||
"""Test linking."""
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN,
|
||||
context={"source": SOURCE_USER},
|
||||
data={CONF_HOST: HOST, CONF_PORT: PORT},
|
||||
)
|
||||
|
||||
result = await hass.config_entries.flow.async_configure(result["flow_id"], {})
|
||||
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
|
||||
assert result["result"].unique_id == HOST
|
||||
assert result["title"] == HOST
|
||||
assert result["data"][CONF_HOST] == HOST
|
||||
assert result["data"][CONF_PORT] == PORT
|
||||
|
||||
|
||||
async def test_abort_if_already_setup(hass):
|
||||
"""Test we abort if component is already setup."""
|
||||
MockConfigEntry(
|
||||
domain=DOMAIN, data={CONF_HOST: HOST, CONF_PORT: PORT}, unique_id=HOST
|
||||
).add_to_hass(hass)
|
||||
|
||||
# Should fail, same HOST (import)
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN,
|
||||
context={"source": SOURCE_IMPORT},
|
||||
data={CONF_HOST: HOST, CONF_PORT: PORT},
|
||||
)
|
||||
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
|
||||
assert result["reason"] == "already_configured"
|
||||
|
||||
# Should fail, same HOST (flow)
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN,
|
||||
context={"source": SOURCE_USER},
|
||||
data={CONF_HOST: HOST, CONF_PORT: PORT},
|
||||
)
|
||||
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
|
||||
assert result["reason"] == "already_configured"
|
||||
|
||||
|
||||
async def test_on_link_failed(hass):
|
||||
"""Test when we have errors during linking the router."""
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN,
|
||||
context={"source": SOURCE_USER},
|
||||
data={CONF_HOST: HOST, CONF_PORT: PORT},
|
||||
)
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.freebox.router.Freepybox.open",
|
||||
side_effect=AuthorizationError(),
|
||||
):
|
||||
result = await hass.config_entries.flow.async_configure(result["flow_id"], {})
|
||||
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
|
||||
assert result["errors"] == {"base": "register_failed"}
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.freebox.router.Freepybox.open",
|
||||
side_effect=HttpRequestError(),
|
||||
):
|
||||
result = await hass.config_entries.flow.async_configure(result["flow_id"], {})
|
||||
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
|
||||
assert result["errors"] == {"base": "connection_failed"}
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.freebox.router.Freepybox.open",
|
||||
side_effect=InvalidTokenError(),
|
||||
):
|
||||
result = await hass.config_entries.flow.async_configure(result["flow_id"], {})
|
||||
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
|
||||
assert result["errors"] == {"base": "unknown"}
|
||||
Loading…
Reference in New Issue