jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-04-25 01:38:02 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add switch platform to Fritz (#51610)

Browse Source 
* Add switch platform to Fritz

* Fix tests

* Pylint

* Small fix

* Bump fritzprofiles to fix log level and identifier

* Fix different WiFi networks with same name

* Changed exposed attributes

* Moved to extra_state

* Remove redundant lambda

* Add missing wait

* Removed identifiers

* Update homeassistant/components/fritz/switch.py

Co-authored-by: J. Nick Koston <nick@koston.org>

* Update homeassistant/components/fritz/switch.py

Co-authored-by: J. Nick Koston <nick@koston.org>

* Add mapping dict

* Device Profile disabled by default

* Heavy cleanup

* Tweak

* Bug fix

* Update homeassistant/components/fritz/switch.py

Co-authored-by: Aaron David Schneider <aaron.schneider@nbi.ku.dk>

* Fix port forward switching + small log improvement

* Cleanup from old approach

* Handle port mapping hot removal (from device)

* Minor fixes

* Typying

* Removed lambda call

* Last missing strict typing

* Split get entities

* Func rename

* Move FritzBoxBaseSwitch to switch.py

* Removed lambda

* Update homeassistant/components/fritz/common.py

Co-authored-by: J. Nick Koston <nick@koston.org>

* Update homeassistant/components/fritz/common.py

Co-authored-by: J. Nick Koston <nick@koston.org>

* Update homeassistant/components/fritz/switch.py

Co-authored-by: J. Nick Koston <nick@koston.org>

* Update homeassistant/components/fritz/switch.py

Co-authored-by: J. Nick Koston <nick@koston.org>

* Update homeassistant/components/fritz/switch.py

Co-authored-by: J. Nick Koston <nick@koston.org>

* Update homeassistant/components/fritz/switch.py

Co-authored-by: J. Nick Koston <nick@koston.org>

* Fixes after applying comments

* Remvoed redundant try block

* Removed broad-except

* Optimized async/sync switch

* Update homeassistant/components/fritz/switch.py

Co-authored-by: J. Nick Koston <nick@koston.org>

* Update homeassistant/components/fritz/switch.py

Co-authored-by: J. Nick Koston <nick@koston.org>

* Update homeassistant/components/fritz/switch.py

Co-authored-by: J. Nick Koston <nick@koston.org>

* Address remaining comments

* Optimize return list

* More optimization for return lists

* Some missing strict typing

* Redundant typing

* Update homeassistant/components/fritz/switch.py

Co-authored-by: J. Nick Koston <nick@koston.org>

* Wrong if

* Introduce const for profile status

* Update homeassistant/components/fritz/switch.py

Co-authored-by: J. Nick Koston <nick@koston.org>

* Update homeassistant/components/fritz/switch.py

Co-authored-by: J. Nick Koston <nick@koston.org>

* Fix mypy

* Switch back to get_local_ip()

* Address latest comments

Co-authored-by: J. Nick Koston <nick@koston.org>
Co-authored-by: Aaron David Schneider <aaron.schneider@nbi.ku.dk>
This commit is contained in:
Simone Chemelli 2021-06-29 17:57:34 +02:00 committed by GitHub
parent 897f5d9247
commit 7959225fef
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
8 changed files with 735 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
.coveragerc
View File

@ -343,6 +343,7 @@ omit =
homeassistant/components/fritz/device_tracker.py
homeassistant/components/fritz/sensor.py
homeassistant/components/fritz/services.py
homeassistant/components/fritz/switch.py
homeassistant/components/fritzbox_callmonitor/__init__.py
homeassistant/components/fritzbox_callmonitor/const.py
homeassistant/components/fritzbox_callmonitor/base.py

24
homeassistant/components/fritz/common.py
View File

@ -5,7 +5,7 @@ from dataclasses import dataclass, field
from datetime import datetime, timedelta
import logging
from types import MappingProxyType
from typing import Any, TypedDict
from typing import Any, Callable, TypedDict
from fritzconnection import FritzConnection
from fritzconnection.core.exceptions import (
@ -15,6 +15,7 @@ from fritzconnection.core.exceptions import (
)
from fritzconnection.lib.fritzhosts import FritzHosts
from fritzconnection.lib.fritzstatus import FritzStatus
from fritzprofiles import FritzProfileSwitch, get_all_profiles
from homeassistant.components.device_tracker.const import (
CONF_CONSIDER_HOME,
@ -44,7 +45,7 @@ _LOGGER = logging.getLogger(__name__)
class ClassSetupMissing(Exception):
"""Raised when a Class func is called before setup."""
def __init__(self):
def __init__(self) -> None:
"""Init custom exception."""
super().__init__("Function called before Class setup")
@ -85,6 +86,7 @@ class FritzBoxTools:
self._unique_id: str | None = None
self.connection: FritzConnection = None
self.fritz_hosts: FritzHosts = None
self.fritz_profiles: dict[str, FritzProfileSwitch] = {}
self.fritz_status: FritzStatus = None
self.hass = hass
self.host = host
@ -117,6 +119,13 @@ class FritzBoxTools:
self._model = info.get("NewModelName")
self._sw_version = info.get("NewSoftwareVersion")
self.fritz_profiles = {
profile: FritzProfileSwitch(
"http://" + self.host, self.username, self.password, profile
)
for profile in get_all_profiles(self.host, self.username, self.password)
}
async def async_start(self, options: MappingProxyType[str, Any]) -> None:
"""Start FritzHosts connection."""
self.fritz_hosts = FritzHosts(fc=self.connection)
@ -306,6 +315,17 @@ class FritzDevice:
return self._last_activity
class SwitchInfo(TypedDict):
"""FRITZ!Box switch info class."""
description: str
friendly_name: str
icon: str
type: str
callback_update: Callable
callback_switch: Callable
class FritzBoxBaseEntity:
"""Fritz host entity base class."""

10
homeassistant/components/fritz/const.py
View File

@ -2,7 +2,7 @@
DOMAIN = "fritz"
PLATFORMS = ["binary_sensor", "device_tracker", "sensor"]
PLATFORMS = ["binary_sensor", "device_tracker", "sensor", "switch"]
DATA_FRITZ = "fritz_data"
@ -19,6 +19,14 @@ FRITZ_SERVICES = "fritz_services"
SERVICE_REBOOT = "reboot"
SERVICE_RECONNECT = "reconnect"
SWITCH_PROFILE_STATUS_OFF = "never"
SWITCH_PROFILE_STATUS_ON = "unlimited"
SWITCH_TYPE_DEFLECTION = "CallDeflection"
SWITCH_TYPE_DEVICEPROFILE = "DeviceProfile"
SWITCH_TYPE_PORTFORWARD = "PortForward"
SWITCH_TYPE_WIFINETWORK = "WiFiNetwork"
TRACKER_SCAN_INTERVAL = 30
UPTIME_DEVIATION = 5

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

@ -3,8 +3,11 @@
"name": "AVM FRITZ!Box Tools",
"documentation": "https://www.home-assistant.io/integrations/fritz",
"requirements": [
"fritzconnection==1.4.2"
"fritzconnection==1.4.2",
"fritzprofiles==0.6.1",
"xmltodict==0.12.0"
],
"dependencies": ["network"],
"codeowners": [
"@mammuth",
"@AaronDavidSchneider",

639
homeassistant/components/fritz/switch.py Normal file
View File

@ -0,0 +1,639 @@
"""Switches for AVM Fritz!Box functions."""
from __future__ import annotations
from collections import OrderedDict
from functools import partial
import logging
from typing import Any
from fritzconnection.core.exceptions import (
FritzActionError,
FritzActionFailedError,
FritzConnectionException,
FritzSecurityError,
FritzServiceError,
)
import xmltodict
from homeassistant.components.switch import SwitchEntity
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity import Entity
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.util import get_local_ip, slugify
from .common import FritzBoxBaseEntity, FritzBoxTools, SwitchInfo
from .const import (
DOMAIN,
SWITCH_PROFILE_STATUS_OFF,
SWITCH_PROFILE_STATUS_ON,
SWITCH_TYPE_DEFLECTION,
SWITCH_TYPE_DEVICEPROFILE,
SWITCH_TYPE_PORTFORWARD,
SWITCH_TYPE_WIFINETWORK,
)
_LOGGER = logging.getLogger(__name__)
async def async_service_call_action(
fritzbox_tools: FritzBoxTools,
service_name: str,
service_suffix: str | None,
action_name: str,
**kwargs: Any,
) -> None | dict:
"""Return service details."""
return await fritzbox_tools.hass.async_add_executor_job(
partial(
service_call_action,
fritzbox_tools,
service_name,
service_suffix,
action_name,
**kwargs,
)
)
def service_call_action(
fritzbox_tools: FritzBoxTools,
service_name: str,
service_suffix: str | None,
action_name: str,
**kwargs: Any,
) -> dict | None:
"""Return service details."""
if f"{service_name}{service_suffix}" not in fritzbox_tools.connection.services:
return None
try:
return fritzbox_tools.connection.call_action(
f"{service_name}:{service_suffix}",
action_name,
**kwargs,
)
except FritzSecurityError:
_LOGGER.error(
"Authorization Error: Please check the provided credentials and verify that you can log into the web interface",
exc_info=True,
)
return None
except (FritzActionError, FritzActionFailedError, FritzServiceError):
_LOGGER.error(
"Service/Action Error: cannot execute service %s",
service_name,
exc_info=True,
)
return None
except FritzConnectionException:
_LOGGER.error(
"Connection Error: Please check the device is properly configured for remote login",
exc_info=True,
)
return None
def get_deflections(
fritzbox_tools: FritzBoxTools, service_name: str
) -> list[OrderedDict[Any, Any]] | None:
"""Get deflection switch info."""
deflection_list = service_call_action(
fritzbox_tools,
service_name,
"1",
"GetDeflections",
)
if not deflection_list:
return []
return [xmltodict.parse(deflection_list["NewDeflectionList"])["List"]["Item"]]
def deflection_entities_list(
fritzbox_tools: FritzBoxTools, device_friendly_name: str
) -> list[FritzBoxDeflectionSwitch]:
"""Get list of deflection entities."""
_LOGGER.debug("Setting up %s switches", SWITCH_TYPE_DEFLECTION)
service_name = "X_AVM-DE_OnTel"
deflections_response = service_call_action(
fritzbox_tools, service_name, "1", "GetNumberOfDeflections"
)
if not deflections_response:
_LOGGER.debug("The FRITZ!Box has no %s options", SWITCH_TYPE_DEFLECTION)
return []
_LOGGER.debug(
"Specific %s response: GetNumberOfDeflections=%s",
SWITCH_TYPE_DEFLECTION,
deflections_response,
)
if deflections_response["NewNumberOfDeflections"] == 0:
_LOGGER.debug("The FRITZ!Box has no %s options", SWITCH_TYPE_DEFLECTION)
return []
deflection_list = get_deflections(fritzbox_tools, service_name)
if deflection_list is None:
return []
return [
FritzBoxDeflectionSwitch(
fritzbox_tools, device_friendly_name, dict_of_deflection
)
for dict_of_deflection in deflection_list
]
def port_entities_list(
fritzbox_tools: FritzBoxTools, device_friendly_name: str
) -> list[FritzBoxPortSwitch]:
"""Get list of port forwarding entities."""
_LOGGER.debug("Setting up %s switches", SWITCH_TYPE_PORTFORWARD)
entities_list: list = []
service_name = "Layer3Forwarding"
connection_type = service_call_action(
fritzbox_tools, service_name, "1", "GetDefaultConnectionService"
)
if not connection_type:
_LOGGER.debug("The FRITZ!Box has no %s options", SWITCH_TYPE_PORTFORWARD)
return []
# Return NewDefaultConnectionService sample: "1.WANPPPConnection.1"
con_type: str = connection_type["NewDefaultConnectionService"][2:][:-2]
# Query port forwardings and setup a switch for each forward for the current device
resp = service_call_action(
fritzbox_tools, con_type, "1", "GetPortMappingNumberOfEntries"
)
if not resp:
_LOGGER.debug("The FRITZ!Box has no %s options", SWITCH_TYPE_DEFLECTION)
return []
port_forwards_count: int = resp["NewPortMappingNumberOfEntries"]
_LOGGER.debug(
"Specific %s response: GetPortMappingNumberOfEntries=%s",
SWITCH_TYPE_PORTFORWARD,
port_forwards_count,
)
local_ip = get_local_ip()
_LOGGER.debug("IP source for %s is %s", fritzbox_tools.host, local_ip)
for i in range(port_forwards_count):
portmap = service_call_action(
fritzbox_tools,
con_type,
"1",
"GetGenericPortMappingEntry",
NewPortMappingIndex=i,
)
if not portmap:
_LOGGER.debug("The FRITZ!Box has no %s options", SWITCH_TYPE_DEFLECTION)
continue
_LOGGER.debug(
"Specific %s response: GetGenericPortMappingEntry=%s",
SWITCH_TYPE_PORTFORWARD,
portmap,
)
# We can only handle port forwards of the given device
if portmap["NewInternalClient"] == local_ip:
entities_list.append(
FritzBoxPortSwitch(
fritzbox_tools,
device_friendly_name,
portmap,
i,
con_type,
)
)
return entities_list
def profile_entities_list(
fritzbox_tools: FritzBoxTools, device_friendly_name: str
) -> list[FritzBoxProfileSwitch]:
"""Get list of profile entities."""
_LOGGER.debug("Setting up %s switches", SWITCH_TYPE_DEVICEPROFILE)
if len(fritzbox_tools.fritz_profiles) <= 0:
_LOGGER.debug("The FRITZ!Box has no %s options", SWITCH_TYPE_DEVICEPROFILE)
return []
return [
FritzBoxProfileSwitch(fritzbox_tools, device_friendly_name, profile)
for profile in fritzbox_tools.fritz_profiles.keys()
]
def wifi_entities_list(
fritzbox_tools: FritzBoxTools, device_friendly_name: str
) -> list[FritzBoxWifiSwitch]:
"""Get list of wifi entities."""
_LOGGER.debug("Setting up %s switches", SWITCH_TYPE_WIFINETWORK)
std_table = {"ac": "5Ghz", "n": "2.4Ghz"}
networks: dict = {}
for i in range(4):
if not ("WLANConfiguration" + str(i)) in fritzbox_tools.connection.services:
continue
network_info = service_call_action(
fritzbox_tools, "WLANConfiguration", str(i), "GetInfo"
)
if network_info:
ssid = network_info["NewSSID"]
if ssid in networks.values():
networks[i] = f'{ssid} {std_table[network_info["NewStandard"]]}'
else:
networks[i] = ssid
return [
FritzBoxWifiSwitch(fritzbox_tools, device_friendly_name, net, networks[net])
for net in networks
]
def all_entities_list(
fritzbox_tools: FritzBoxTools, device_friendly_name: str
) -> list[Entity]:
"""Get a list of all entities."""
return [
*deflection_entities_list(fritzbox_tools, device_friendly_name),
*port_entities_list(fritzbox_tools, device_friendly_name),
*profile_entities_list(fritzbox_tools, device_friendly_name),
*wifi_entities_list(fritzbox_tools, device_friendly_name),
]
async def async_setup_entry(
hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
) -> None:
"""Set up entry."""
_LOGGER.debug("Setting up switches")
fritzbox_tools: FritzBoxTools = hass.data[DOMAIN][entry.entry_id]
_LOGGER.debug("Fritzbox services: %s", fritzbox_tools.connection.services)
entities_list = await hass.async_add_executor_job(
all_entities_list, fritzbox_tools, entry.title
)
async_add_entities(entities_list)
class FritzBoxBaseSwitch(FritzBoxBaseEntity):
"""Fritz switch base class."""
def __init__(
self,
fritzbox_tools: FritzBoxTools,
device_friendly_name: str,
switch_info: SwitchInfo,
) -> None:
"""Init Fritzbox port switch."""
super().__init__(fritzbox_tools, device_friendly_name)
self._description = switch_info["description"]
self._friendly_name = switch_info["friendly_name"]
self._icon = switch_info["icon"]
self._type = switch_info["type"]
self._update = switch_info["callback_update"]
self._switch = switch_info["callback_switch"]
self._name = f"{self._friendly_name} {self._description}"
self._unique_id = (
f"{self._fritzbox_tools.unique_id}-{slugify(self._description)}"
)
self._attributes: dict[str, str] = {}
self._is_available = True
self._attr_is_on = False
@property
def name(self) -> str:
"""Return name."""
return self._name
@property
def icon(self) -> str:
"""Return name."""
return self._icon
@property
def unique_id(self) -> str:
"""Return unique id."""
return self._unique_id
@property
def available(self) -> bool:
"""Return availability."""
return self._is_available
@property
def extra_state_attributes(self) -> dict[str, str]:
"""Return device attributes."""
return self._attributes
async def async_update(self) -> None:
"""Update data."""
_LOGGER.debug("Updating '%s' (%s) switch state", self.name, self._type)
await self._update()
async def async_turn_on(self, **kwargs: Any) -> None:
"""Turn on switch."""
await self._async_handle_turn_on_off(turn_on=True)
async def async_turn_off(self, **kwargs: Any) -> None:
"""Turn off switch."""
await self._async_handle_turn_on_off(turn_on=False)
async def _async_handle_turn_on_off(self, turn_on: bool) -> bool:
"""Handle switch state change request."""
await self._switch(turn_on)
self._attr_is_on = turn_on
return True
class FritzBoxPortSwitch(FritzBoxBaseSwitch, SwitchEntity):
"""Defines a FRITZ!Box Tools PortForward switch."""
def __init__(
self,
fritzbox_tools: FritzBoxTools,
device_friendly_name: str,
port_mapping: dict[str, Any] | None,
idx: int,
connection_type: str,
) -> None:
"""Init Fritzbox port switch."""
self._fritzbox_tools = fritzbox_tools
self._attributes = {}
self.connection_type = connection_type
self.port_mapping = port_mapping # dict in the format as it comes from fritzconnection. eg: {'NewRemoteHost': '0.0.0.0', 'NewExternalPort': 22, 'NewProtocol': 'TCP', 'NewInternalPort': 22, 'NewInternalClient': '192.168.178.31', 'NewEnabled': True, 'NewPortMappingDescription': 'Beast SSH ', 'NewLeaseDuration': 0}
self._idx = idx # needed for update routine
if port_mapping is None:
return
switch_info = SwitchInfo(
description=f'Port forward {port_mapping["NewPortMappingDescription"]}',
friendly_name=device_friendly_name,
icon="mdi:check-network",
type=SWITCH_TYPE_PORTFORWARD,
callback_update=self._async_fetch_update,
callback_switch=self._async_handle_port_switch_on_off,
)
super().__init__(fritzbox_tools, device_friendly_name, switch_info)
async def _async_fetch_update(self) -> None:
"""Fetch updates."""
self.port_mapping = await async_service_call_action(
self._fritzbox_tools,
self.connection_type,
"1",
"GetGenericPortMappingEntry",
NewPortMappingIndex=self._idx,
)
_LOGGER.debug(
"Specific %s response: %s", SWITCH_TYPE_PORTFORWARD, self.port_mapping
)
if self.port_mapping is None:
self._is_available = False
return
self._attr_is_on = self.port_mapping["NewEnabled"] is True
self._is_available = True
attributes_dict = {
"NewInternalClient": "internalIP",
"NewInternalPort": "internalPort",
"NewExternalPort": "externalPort",
"NewProtocol": "protocol",
"NewPortMappingDescription": "description",
}
for key in attributes_dict:
self._attributes[attributes_dict[key]] = self.port_mapping[key]
async def _async_handle_port_switch_on_off(self, turn_on: bool) -> bool:
if self.port_mapping is None:
return False
self.port_mapping["NewEnabled"] = "1" if turn_on else "0"
resp = await async_service_call_action(
self._fritzbox_tools,
self.connection_type,
"1",
"AddPortMapping",
**self.port_mapping,
)
return bool(resp is not None)
class FritzBoxDeflectionSwitch(FritzBoxBaseSwitch, SwitchEntity):
"""Defines a FRITZ!Box Tools PortForward switch."""
def __init__(
self,
fritzbox_tools: FritzBoxTools,
device_friendly_name: str,
dict_of_deflection: Any,
) -> None:
"""Init Fritxbox Deflection class."""
self._fritzbox_tools: FritzBoxTools = fritzbox_tools
self.dict_of_deflection = dict_of_deflection
self._attributes = {}
self.id = int(self.dict_of_deflection["DeflectionId"])
switch_info = SwitchInfo(
description=f"Call deflection {self.id}",
friendly_name=device_friendly_name,
icon="mdi:phone-forward",
type=SWITCH_TYPE_DEFLECTION,
callback_update=self._async_fetch_update,
callback_switch=self._async_switch_on_off_executor,
)
super().__init__(self._fritzbox_tools, device_friendly_name, switch_info)
async def _async_fetch_update(self) -> None:
"""Fetch updates."""
resp = await async_service_call_action(
self._fritzbox_tools, "X_AVM-DE_OnTel", "1", "GetDeflections"
)
if not resp:
self._is_available = False
return
self.dict_of_deflection = xmltodict.parse(resp["NewDeflectionList"])["List"][
"Item"
]
if isinstance(self.dict_of_deflection, list):
self.dict_of_deflection = self.dict_of_deflection[self.id]
_LOGGER.debug(
"Specific %s response: NewDeflectionList=%s",
SWITCH_TYPE_DEFLECTION,
self.dict_of_deflection,
)
self._attr_is_on = self.dict_of_deflection["Enable"] == "1"
self._is_available = True
self._attributes["Type"] = self.dict_of_deflection["Type"]
self._attributes["Number"] = self.dict_of_deflection["Number"]
self._attributes["DeflectionToNumber"] = self.dict_of_deflection[
"DeflectionToNumber"
]
# Return mode sample: "eImmediately"
self._attributes["Mode"] = self.dict_of_deflection["Mode"][1:]
self._attributes["Outgoing"] = self.dict_of_deflection["Outgoing"]
self._attributes["PhonebookID"] = self.dict_of_deflection["PhonebookID"]
async def _async_switch_on_off_executor(self, turn_on: bool) -> None:
"""Handle deflection switch."""
await async_service_call_action(
self._fritzbox_tools,
"X_AVM-DE_OnTel",
"1",
"SetDeflectionEnable",
NewDeflectionId=self.id,
NewEnable="1" if turn_on else "0",
)
class FritzBoxProfileSwitch(FritzBoxBaseSwitch, SwitchEntity):
"""Defines a FRITZ!Box Tools DeviceProfile switch."""
def __init__(
self, fritzbox_tools: FritzBoxTools, device_friendly_name: str, profile: str
) -> None:
"""Init Fritz profile."""
self._fritzbox_tools: FritzBoxTools = fritzbox_tools
self.profile = profile
switch_info = SwitchInfo(
description=f"Profile {profile}",
friendly_name=device_friendly_name,
icon="mdi:router-wireless-settings",
type=SWITCH_TYPE_DEVICEPROFILE,
callback_update=self._async_fetch_update,
callback_switch=self._async_switch_on_off_executor,
)
super().__init__(self._fritzbox_tools, device_friendly_name, switch_info)
async def _async_fetch_update(self) -> None:
"""Update data."""
try:
status = await self.hass.async_add_executor_job(
self._fritzbox_tools.fritz_profiles[self.profile].get_state
)
_LOGGER.debug(
"Specific %s response: get_State()=%s",
SWITCH_TYPE_DEVICEPROFILE,
status,
)
if status == SWITCH_PROFILE_STATUS_OFF:
self._attr_is_on = False
self._is_available = True
elif status == SWITCH_PROFILE_STATUS_ON:
self._attr_is_on = True
self._is_available = True
else:
self._is_available = False
except Exception: # pylint: disable=broad-except
_LOGGER.error("Could not get %s state", self.name, exc_info=True)
self._is_available = False
async def _async_switch_on_off_executor(self, turn_on: bool) -> None:
"""Handle profile switch."""
state = SWITCH_PROFILE_STATUS_ON if turn_on else SWITCH_PROFILE_STATUS_OFF
await self.hass.async_add_executor_job(
self._fritzbox_tools.fritz_profiles[self.profile].set_state, state
)
@property
def entity_registry_enabled_default(self) -> bool:
"""Return if the entity should be enabled when first added to the entity registry."""
return False
class FritzBoxWifiSwitch(FritzBoxBaseSwitch, SwitchEntity):
"""Defines a FRITZ!Box Tools Wifi switch."""
def __init__(
self,
fritzbox_tools: FritzBoxTools,
device_friendly_name: str,
network_num: int,
network_name: str,
) -> None:
"""Init Fritz Wifi switch."""
self._fritzbox_tools = fritzbox_tools
self._attributes = {}
self._network_num = network_num
switch_info = SwitchInfo(
description=f"Wi-Fi {network_name}",
friendly_name=device_friendly_name,
icon="mdi:wifi",
type=SWITCH_TYPE_WIFINETWORK,
callback_update=self._async_fetch_update,
callback_switch=self._async_switch_on_off_executor,
)
super().__init__(self._fritzbox_tools, device_friendly_name, switch_info)
async def _async_fetch_update(self) -> None:
"""Fetch updates."""
wifi_info = await async_service_call_action(
self._fritzbox_tools,
"WLANConfiguration",
str(self._network_num),
"GetInfo",
)
_LOGGER.debug(
"Specific %s response: GetInfo=%s", SWITCH_TYPE_WIFINETWORK, wifi_info
)
if wifi_info is None:
self._is_available = False
return
self._attr_is_on = wifi_info["NewEnable"] is True
self._is_available = True
std = wifi_info["NewStandard"]
self._attributes["standard"] = std if std else None
self._attributes["BSSID"] = wifi_info["NewBSSID"]
self._attributes["mac_address_control"] = wifi_info[
"NewMACAddressControlEnabled"
]
async def _async_switch_on_off_executor(self, turn_on: bool) -> None:
"""Handle wifi switch."""
await async_service_call_action(
self._fritzbox_tools,
"WLANConfiguration",
str(self._network_num),
"SetEnable",
NewEnable="1" if turn_on else "0",
)

4
requirements_all.txt
View File

@ -640,6 +640,9 @@ freesms==0.2.0
# homeassistant.components.fritzbox_netmonitor
fritzconnection==1.4.2
# homeassistant.components.fritz
fritzprofiles==0.6.1
# homeassistant.components.google_translate
gTTS==2.2.3
@ -2390,6 +2393,7 @@ xboxapi==2.0.1
xknx==0.18.7
# homeassistant.components.bluesound
# homeassistant.components.fritz
# homeassistant.components.rest
# homeassistant.components.startca
# homeassistant.components.ted5000

4
requirements_test_all.txt
View File

@ -349,6 +349,9 @@ freebox-api==0.0.10
# homeassistant.components.fritzbox_netmonitor
fritzconnection==1.4.2
# homeassistant.components.fritz
fritzprofiles==0.6.1
# homeassistant.components.google_translate
gTTS==2.2.3
@ -1308,6 +1311,7 @@ xbox-webapi==2.0.11
xknx==0.18.7
# homeassistant.components.bluesound
# homeassistant.components.fritz
# homeassistant.components.rest
# homeassistant.components.startca
# homeassistant.components.ted5000

57
tests/components/fritz/test_config_flow.py
View File

@ -56,6 +56,8 @@ MOCK_SSDP_DATA = {
ATTR_UPNP_UDN: "uuid:only-a-test",
}
MOCK_REQUEST = b'<?xml version="1.0" encoding="utf-8"?><SessionInfo><SID>xxxxxxxxxxxxxxxx</SID><Challenge>xxxxxxxx</Challenge><BlockTime>0</BlockTime><Rights><Name>Dial</Name><Access>2</Access><Name>App</Name><Access>2</Access><Name>HomeAuto</Name><Access>2</Access><Name>BoxAdmin</Name><Access>2</Access><Name>Phone</Name><Access>2</Access><Name>NAS</Name><Access>2</Access></Rights><Users><User last="1">FakeFritzUser</User></Users></SessionInfo>\n'
@pytest.fixture()
def fc_class_mock():
@ -72,7 +74,16 @@ async def test_user(hass: HomeAssistant, fc_class_mock):
side_effect=fc_class_mock,
), patch("homeassistant.components.fritz.common.FritzStatus"), patch(
"homeassistant.components.fritz.async_setup_entry"
) as mock_setup_entry:
) as mock_setup_entry, patch(
"requests.get"
) as mock_request_get, patch(
"requests.post"
) as mock_request_post:
mock_request_get.return_value.status_code = 200
mock_request_get.return_value.content = MOCK_REQUEST
mock_request_post.return_value.status_code = 200
mock_request_post.return_value.text = MOCK_REQUEST
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
@ -106,7 +117,16 @@ async def test_user_already_configured(hass: HomeAssistant, fc_class_mock):
with patch(
"homeassistant.components.fritz.common.FritzConnection",
side_effect=fc_class_mock,
), patch("homeassistant.components.fritz.common.FritzStatus"):
), patch("homeassistant.components.fritz.common.FritzStatus"), patch(
"requests.get"
) as mock_request_get, patch(
"requests.post"
) as mock_request_post:
mock_request_get.return_value.status_code = 200
mock_request_get.return_value.content = MOCK_REQUEST
mock_request_post.return_value.status_code = 200
mock_request_post.return_value.text = MOCK_REQUEST
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
@ -202,7 +222,16 @@ async def test_reauth_successful(hass: HomeAssistant, fc_class_mock):
side_effect=fc_class_mock,
), patch("homeassistant.components.fritz.common.FritzStatus"), patch(
"homeassistant.components.fritz.async_setup_entry"
) as mock_setup_entry:
) as mock_setup_entry, patch(
"requests.get"
) as mock_request_get, patch(
"requests.post"
) as mock_request_post:
mock_request_get.return_value.status_code = 200
mock_request_get.return_value.content = MOCK_REQUEST
mock_request_post.return_value.status_code = 200
mock_request_post.return_value.text = MOCK_REQUEST
result = await hass.config_entries.flow.async_init(
DOMAIN,
@ -355,7 +384,16 @@ async def test_ssdp(hass: HomeAssistant, fc_class_mock):
side_effect=fc_class_mock,
), patch("homeassistant.components.fritz.common.FritzStatus"), patch(
"homeassistant.components.fritz.async_setup_entry"
) as mock_setup_entry:
) as mock_setup_entry, patch(
"requests.get"
) as mock_request_get, patch(
"requests.post"
) as mock_request_post:
mock_request_get.return_value.status_code = 200
mock_request_get.return_value.content = MOCK_REQUEST
mock_request_post.return_value.status_code = 200
mock_request_post.return_value.text = MOCK_REQUEST
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_SSDP}, data=MOCK_SSDP_DATA
@ -411,7 +449,16 @@ async def test_import(hass: HomeAssistant, fc_class_mock):
side_effect=fc_class_mock,
), patch("homeassistant.components.fritz.common.FritzStatus"), patch(
"homeassistant.components.fritz.async_setup_entry"
) as mock_setup_entry:
) as mock_setup_entry, patch(
"requests.get"
) as mock_request_get, patch(
"requests.post"
) as mock_request_post:
mock_request_get.return_value.status_code = 200
mock_request_get.return_value.content = MOCK_REQUEST
mock_request_post.return_value.status_code = 200
mock_request_post.return_value.text = MOCK_REQUEST
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_IMPORT}, data=MOCK_IMPORT_CONFIG