Use StrEnum [fritzbox_callmonitor] (#70289)

This commit is contained in:
Marc Mueller 2022-04-26 09:40:08 +02:00 committed by GitHub
parent 196c60bdca
commit eeecdf213d
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
4 changed files with 42 additions and 39 deletions

View File

@ -1,4 +1,5 @@
"""Config flow for fritzbox_callmonitor."""
from __future__ import annotations
from fritzconnection import FritzConnection
from fritzconnection.core.exceptions import FritzConnectionException, FritzSecurityError
@ -6,6 +7,7 @@ from requests.exceptions import ConnectionError as RequestsConnectionError
import voluptuous as vol
from homeassistant import config_entries
from homeassistant.backports.enum import StrEnum
from homeassistant.const import (
CONF_HOST,
CONF_NAME,
@ -40,11 +42,15 @@ DATA_SCHEMA_USER = vol.Schema(
}
)
RESULT_INVALID_AUTH = "invalid_auth"
RESULT_INSUFFICIENT_PERMISSIONS = "insufficient_permissions"
RESULT_MALFORMED_PREFIXES = "malformed_prefixes"
RESULT_NO_DEVIES_FOUND = "no_devices_found"
RESULT_SUCCESS = "success"
class ConnectResult(StrEnum):
"""FritzBoxPhonebook connection result."""
INVALID_AUTH = "invalid_auth"
INSUFFICIENT_PERMISSIONS = "insufficient_permissions"
MALFORMED_PREFIXES = "malformed_prefixes"
NO_DEVIES_FOUND = "no_devices_found"
SUCCESS = "success"
class FritzBoxCallMonitorConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
@ -81,7 +87,7 @@ class FritzBoxCallMonitorConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
},
)
def _try_connect(self):
def _try_connect(self) -> ConnectResult:
"""Try to connect and check auth."""
self._fritzbox_phonebook = FritzBoxPhonebook(
host=self._host,
@ -103,13 +109,13 @@ class FritzBoxCallMonitorConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
)
self._serial_number = device_info[FRITZ_ATTR_SERIAL_NUMBER]
return RESULT_SUCCESS
return ConnectResult.SUCCESS
except RequestsConnectionError:
return RESULT_NO_DEVIES_FOUND
return ConnectResult.NO_DEVIES_FOUND
except FritzSecurityError:
return RESULT_INSUFFICIENT_PERMISSIONS
return ConnectResult.INSUFFICIENT_PERMISSIONS
except FritzConnectionException:
return RESULT_INVALID_AUTH
return ConnectResult.INVALID_AUTH
async def _get_name_of_phonebook(self, phonebook_id):
"""Return name of phonebook for given phonebook_id."""
@ -152,14 +158,14 @@ class FritzBoxCallMonitorConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
result = await self.hass.async_add_executor_job(self._try_connect)
if result == RESULT_INVALID_AUTH:
if result == ConnectResult.INVALID_AUTH:
return self.async_show_form(
step_id="user",
data_schema=DATA_SCHEMA_USER,
errors={"base": RESULT_INVALID_AUTH},
errors={"base": ConnectResult.INVALID_AUTH},
)
if result != RESULT_SUCCESS:
if result != ConnectResult.SUCCESS:
return self.async_abort(reason=result)
if self.context["source"] == config_entries.SOURCE_IMPORT:
@ -252,7 +258,7 @@ class FritzBoxCallMonitorOptionsFlowHandler(config_entries.OptionsFlow):
return self.async_show_form(
step_id="init",
data_schema=option_schema_prefixes,
errors={"base": RESULT_MALFORMED_PREFIXES},
errors={"base": ConnectResult.MALFORMED_PREFIXES},
)
return self.async_create_entry(

View File

@ -1,11 +1,6 @@
"""Constants for the AVM Fritz!Box call monitor integration."""
from homeassistant.const import Platform
STATE_RINGING = "ringing"
STATE_DIALING = "dialing"
STATE_TALKING = "talking"
STATE_IDLE = "idle"
FRITZ_STATE_RING = "RING"
FRITZ_STATE_CALL = "CALL"
FRITZ_STATE_CONNECT = "CONNECT"

View File

@ -10,6 +10,7 @@ from time import sleep
from fritzconnection.core.fritzmonitor import FritzMonitor
import voluptuous as vol
from homeassistant.backports.enum import StrEnum
from homeassistant.components.sensor import PLATFORM_SCHEMA, SensorEntity
from homeassistant.config_entries import SOURCE_IMPORT, ConfigEntry
from homeassistant.const import (
@ -44,10 +45,6 @@ from .const import (
ICON_PHONE,
MANUFACTURER,
SERIAL_NUMBER,
STATE_DIALING,
STATE_IDLE,
STATE_RINGING,
STATE_TALKING,
UNKNOWN_NAME,
)
@ -55,6 +52,16 @@ _LOGGER = logging.getLogger(__name__)
SCAN_INTERVAL = timedelta(hours=3)
class CallState(StrEnum):
"""Fritz sensor call states."""
RINGING = "ringing"
DIALING = "dialing"
TALKING = "talking"
IDLE = "idle"
# Deprecated in Home Assistant 2022.3
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
{
@ -124,7 +131,7 @@ class FritzBoxCallSensor(SensorEntity):
def __init__(self, name, unique_id, fritzbox_phonebook, prefixes, host, port):
"""Initialize the sensor."""
self._state = STATE_IDLE
self._state: CallState = CallState.IDLE
self._attributes = {}
self._name = name.title()
self._unique_id = unique_id
@ -172,7 +179,7 @@ class FritzBoxCallSensor(SensorEntity):
self._monitor.connection.stop()
_LOGGER.debug("Stopped monitor for: %s", self.entity_id)
def set_state(self, state):
def set_state(self, state: CallState) -> None:
"""Set the state."""
self._state = state
@ -282,7 +289,7 @@ class FritzBoxCallMonitor:
df_out = "%Y-%m-%dT%H:%M:%S"
isotime = datetime.strptime(line[0], df_in).strftime(df_out)
if line[1] == FRITZ_STATE_RING:
self._sensor.set_state(STATE_RINGING)
self._sensor.set_state(CallState.RINGING)
att = {
"type": "incoming",
"from": line[3],
@ -293,7 +300,7 @@ class FritzBoxCallMonitor:
}
self._sensor.set_attributes(att)
elif line[1] == FRITZ_STATE_CALL:
self._sensor.set_state(STATE_DIALING)
self._sensor.set_state(CallState.DIALING)
att = {
"type": "outgoing",
"from": line[4],
@ -304,7 +311,7 @@ class FritzBoxCallMonitor:
}
self._sensor.set_attributes(att)
elif line[1] == FRITZ_STATE_CONNECT:
self._sensor.set_state(STATE_TALKING)
self._sensor.set_state(CallState.TALKING)
att = {
"with": line[4],
"device": line[3],
@ -313,7 +320,7 @@ class FritzBoxCallMonitor:
}
self._sensor.set_attributes(att)
elif line[1] == FRITZ_STATE_DISCONNECT:
self._sensor.set_state(STATE_IDLE)
self._sensor.set_state(CallState.IDLE)
att = {"duration": line[3], "closed": isotime}
self._sensor.set_attributes(att)
self._sensor.schedule_update_ha_state()

View File

@ -4,12 +4,7 @@ from unittest.mock import PropertyMock
from fritzconnection.core.exceptions import FritzConnectionException, FritzSecurityError
from requests.exceptions import ConnectionError as RequestsConnectionError
from homeassistant.components.fritzbox_callmonitor.config_flow import (
RESULT_INSUFFICIENT_PERMISSIONS,
RESULT_INVALID_AUTH,
RESULT_MALFORMED_PREFIXES,
RESULT_NO_DEVIES_FOUND,
)
from homeassistant.components.fritzbox_callmonitor.config_flow import ConnectResult
from homeassistant.components.fritzbox_callmonitor.const import (
CONF_PHONEBOOK,
CONF_PREFIXES,
@ -230,7 +225,7 @@ async def test_setup_cannot_connect(hass: HomeAssistant) -> None:
)
assert result["type"] == RESULT_TYPE_ABORT
assert result["reason"] == RESULT_NO_DEVIES_FOUND
assert result["reason"] == ConnectResult.NO_DEVIES_FOUND
async def test_setup_insufficient_permissions(hass: HomeAssistant) -> None:
@ -249,7 +244,7 @@ async def test_setup_insufficient_permissions(hass: HomeAssistant) -> None:
)
assert result["type"] == RESULT_TYPE_ABORT
assert result["reason"] == RESULT_INSUFFICIENT_PERMISSIONS
assert result["reason"] == ConnectResult.INSUFFICIENT_PERMISSIONS
async def test_setup_invalid_auth(hass: HomeAssistant) -> None:
@ -268,7 +263,7 @@ async def test_setup_invalid_auth(hass: HomeAssistant) -> None:
)
assert result["type"] == RESULT_TYPE_FORM
assert result["errors"] == {"base": RESULT_INVALID_AUTH}
assert result["errors"] == {"base": ConnectResult.INVALID_AUTH}
async def test_options_flow_correct_prefixes(hass: HomeAssistant) -> None:
@ -326,7 +321,7 @@ async def test_options_flow_incorrect_prefixes(hass: HomeAssistant) -> None:
)
assert result["type"] == RESULT_TYPE_FORM
assert result["errors"] == {"base": RESULT_MALFORMED_PREFIXES}
assert result["errors"] == {"base": ConnectResult.MALFORMED_PREFIXES}
async def test_options_flow_no_prefixes(hass: HomeAssistant) -> None: