mirror of
https://github.com/home-assistant/core.git
synced 2025-04-23 16:57:53 +00:00
Add Config Flow to Modem Caller ID integration (#46677)
* Add phone_modem integration * Use original domain * Add init tests for Modem Caller ID * Clean up tests * Clean up tests * apply suggestions * Fix tests * Make only one instance possible * Allow more than 1 device and remove hangup service * simplify already configured * Update sensor.py * Update config_flow.py * Fix manifest * More cleanup * Fix tests * Ue target * Clean up sensor.py * Minor tweaks * Close modem on restart and unload * Update requirements * fix tests * Bump phone_modem * rework * add typing * use async_setup_platform * typing * tweak * cleanup * fix init * preserve original name * remove callback line * use list of serial devices on host * tweak * rework * Rework for usb dicsovery * Update requirements_test_all.txt * Update config_flow.py * tweaks * tweak * move api out of try statement * suggested tweaks * clean up * typing * tweak * tweak * async name the service
This commit is contained in:
Robert Hillis
GitHub
parent
9bb9f0e070
commit
14aa9c91eb
@ -312,6 +312,7 @@ homeassistant/components/minecraft_server/* @elmurato
|
||||
homeassistant/components/minio/* @tkislan
|
||||
homeassistant/components/mobile_app/* @robbiet480
|
||||
homeassistant/components/modbus/* @adamchengtkc @janiversen @vzahradnik
|
||||
homeassistant/components/modem_callerid/* @tkdrob
|
||||
homeassistant/components/modern_forms/* @wonderslug
|
||||
homeassistant/components/monoprice/* @etsinko @OnFreund
|
||||
homeassistant/components/moon/* @fabaff
|
||||
|
||||
@ -1 +1,37 @@
|
||||
"""The modem_callerid component."""
|
||||
"""The Modem Caller ID integration."""
|
||||
from phone_modem import PhoneModem
|
||||
|
||||
from homeassistant.components.sensor import DOMAIN as SENSOR_DOMAIN
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
from homeassistant.const import CONF_DEVICE
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.exceptions import ConfigEntryNotReady
|
||||
|
||||
from .const import DATA_KEY_API, DOMAIN, EXCEPTIONS
|
||||
|
||||
PLATFORMS = [SENSOR_DOMAIN]
|
||||
|
||||
|
||||
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
|
||||
"""Set up Modem Caller ID from a config entry."""
|
||||
device = entry.data[CONF_DEVICE]
|
||||
api = PhoneModem(device)
|
||||
try:
|
||||
await api.initialize(device)
|
||||
except EXCEPTIONS as ex:
|
||||
raise ConfigEntryNotReady(f"Unable to open port: {device}") from ex
|
||||
|
||||
hass.data.setdefault(DOMAIN, {})[entry.entry_id] = {DATA_KEY_API: api}
|
||||
hass.config_entries.async_setup_platforms(entry, PLATFORMS)
|
||||
|
||||
return True
|
||||
|
||||
|
||||
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
|
||||
"""Unload a config entry."""
|
||||
unload_ok = await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
|
||||
if unload_ok:
|
||||
api = hass.data[DOMAIN].pop(entry.entry_id)[DATA_KEY_API]
|
||||
await api.close()
|
||||
|
||||
return unload_ok
|
||||
|
||||
142
homeassistant/components/modem_callerid/config_flow.py
Normal file
142
homeassistant/components/modem_callerid/config_flow.py
Normal file
@ -0,0 +1,142 @@
|
||||
"""Config flow for Modem Caller ID integration."""
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from typing import Any
|
||||
|
||||
from phone_modem import DEFAULT_PORT, PhoneModem
|
||||
import serial.tools.list_ports
|
||||
import voluptuous as vol
|
||||
|
||||
from homeassistant import config_entries
|
||||
from homeassistant.components import usb
|
||||
from homeassistant.const import CONF_DEVICE, CONF_NAME
|
||||
from homeassistant.data_entry_flow import FlowResult
|
||||
|
||||
from .const import DEFAULT_NAME, DOMAIN, EXCEPTIONS
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
DATA_SCHEMA = vol.Schema({"name": str, "device": str})
|
||||
|
||||
|
||||
def _generate_unique_id(port: Any) -> str:
|
||||
"""Generate unique id from usb attributes."""
|
||||
return f"{port.vid}:{port.pid}_{port.serial_number}_{port.manufacturer}_{port.description}"
|
||||
|
||||
|
||||
class PhoneModemFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
|
||||
"""Handle a config flow for Phone Modem."""
|
||||
|
||||
def __init__(self) -> None:
|
||||
"""Set up flow instance."""
|
||||
self._device: str | None = None
|
||||
|
||||
async def async_step_usb(self, discovery_info: dict[str, str]) -> FlowResult:
|
||||
"""Handle USB Discovery."""
|
||||
device = discovery_info["device"]
|
||||
|
||||
dev_path = await self.hass.async_add_executor_job(usb.get_serial_by_id, device)
|
||||
unique_id = f"{discovery_info['vid']}:{discovery_info['pid']}_{discovery_info['serial_number']}_{discovery_info['manufacturer']}_{discovery_info['description']}"
|
||||
if (
|
||||
await self.validate_device_errors(dev_path=dev_path, unique_id=unique_id)
|
||||
is None
|
||||
):
|
||||
self._device = dev_path
|
||||
return await self.async_step_usb_confirm()
|
||||
return self.async_abort(reason="cannot_connect")
|
||||
|
||||
async def async_step_usb_confirm(
|
||||
self, user_input: dict[str, Any] | None = None
|
||||
) -> FlowResult:
|
||||
"""Handle USB Discovery confirmation."""
|
||||
if user_input is not None:
|
||||
return self.async_create_entry(
|
||||
title=user_input.get(CONF_NAME, DEFAULT_NAME),
|
||||
data={CONF_DEVICE: self._device},
|
||||
)
|
||||
self._set_confirm_only()
|
||||
return self.async_show_form(step_id="usb_confirm")
|
||||
|
||||
async def async_step_user(
|
||||
self, user_input: dict[str, Any] | None = None
|
||||
) -> FlowResult:
|
||||
"""Handle a flow initiated by the user."""
|
||||
if self._async_in_progress():
|
||||
return self.async_abort(reason="already_in_progress")
|
||||
ports = await self.hass.async_add_executor_job(serial.tools.list_ports.comports)
|
||||
existing_devices = [
|
||||
entry.data[CONF_DEVICE] for entry in self._async_current_entries()
|
||||
]
|
||||
unused_ports = [
|
||||
usb.human_readable_device_name(
|
||||
port.device,
|
||||
port.serial_number,
|
||||
port.manufacturer,
|
||||
port.description,
|
||||
port.vid,
|
||||
port.pid,
|
||||
)
|
||||
for port in ports
|
||||
if port.device not in existing_devices
|
||||
]
|
||||
if not unused_ports:
|
||||
return self.async_abort(reason="no_devices_found")
|
||||
|
||||
if user_input is not None:
|
||||
port = ports[unused_ports.index(str(user_input.get(CONF_DEVICE)))]
|
||||
dev_path = await self.hass.async_add_executor_job(
|
||||
usb.get_serial_by_id, port.device
|
||||
)
|
||||
errors: dict | None = await self.validate_device_errors(
|
||||
dev_path=dev_path, unique_id=_generate_unique_id(port)
|
||||
)
|
||||
if errors is None:
|
||||
return self.async_create_entry(
|
||||
title=user_input.get(CONF_NAME, DEFAULT_NAME),
|
||||
data={CONF_DEVICE: dev_path},
|
||||
)
|
||||
user_input = user_input or {}
|
||||
schema = vol.Schema({vol.Required(CONF_DEVICE): vol.In(unused_ports)})
|
||||
return self.async_show_form(
|
||||
step_id="user", data_schema=schema, errors=errors or {}
|
||||
)
|
||||
|
||||
async def async_step_import(self, config: dict[str, Any]) -> FlowResult:
|
||||
"""Import a config entry from configuration.yaml."""
|
||||
if self._async_current_entries():
|
||||
_LOGGER.warning(
|
||||
"Loading Modem_callerid via platform setup is deprecated; Please remove it from your configuration"
|
||||
)
|
||||
if CONF_DEVICE not in config:
|
||||
config[CONF_DEVICE] = DEFAULT_PORT
|
||||
ports = await self.hass.async_add_executor_job(serial.tools.list_ports.comports)
|
||||
for port in ports:
|
||||
if port.device == config[CONF_DEVICE]:
|
||||
if (
|
||||
await self.validate_device_errors(
|
||||
dev_path=port.device,
|
||||
unique_id=_generate_unique_id(port),
|
||||
)
|
||||
is None
|
||||
):
|
||||
return self.async_create_entry(
|
||||
title=config.get(CONF_NAME, DEFAULT_NAME),
|
||||
data={CONF_DEVICE: port.device},
|
||||
)
|
||||
return self.async_abort(reason="cannot_connect")
|
||||
|
||||
async def validate_device_errors(
|
||||
self, dev_path: str, unique_id: str
|
||||
) -> dict[str, str] | None:
|
||||
"""Handle common flow input validation."""
|
||||
self._async_abort_entries_match({CONF_DEVICE: dev_path})
|
||||
await self.async_set_unique_id(unique_id)
|
||||
self._abort_if_unique_id_configured(updates={CONF_DEVICE: dev_path})
|
||||
try:
|
||||
api = PhoneModem()
|
||||
await api.test(dev_path)
|
||||
except EXCEPTIONS:
|
||||
return {"base": "cannot_connect"}
|
||||
else:
|
||||
return None
|
||||
27
homeassistant/components/modem_callerid/const.py
Normal file
27
homeassistant/components/modem_callerid/const.py
Normal file
@ -0,0 +1,27 @@
|
||||
"""Constants for the Modem Caller ID integration."""
|
||||
from typing import Final
|
||||
|
||||
from phone_modem import exceptions
|
||||
from serial import SerialException
|
||||
|
||||
DATA_KEY_API = "api"
|
||||
DATA_KEY_COORDINATOR = "coordinator"
|
||||
DEFAULT_NAME = "Phone Modem"
|
||||
DOMAIN = "modem_callerid"
|
||||
ICON = "mdi:phone-classic"
|
||||
SERVICE_REJECT_CALL = "reject_call"
|
||||
|
||||
EXCEPTIONS: Final = (
|
||||
FileNotFoundError,
|
||||
exceptions.SerialError,
|
||||
exceptions.ResponseError,
|
||||
SerialException,
|
||||
)
|
||||
|
||||
|
||||
class CID:
|
||||
"""CID Attributes."""
|
||||
|
||||
CID_TIME = "cid_time"
|
||||
CID_NUMBER = "cid_number"
|
||||
CID_NAME = "cid_name"
|
||||
@ -1,8 +1,11 @@
|
||||
{
|
||||
"domain": "modem_callerid",
|
||||
"name": "Modem Caller ID",
|
||||
"name": "Phone Modem",
|
||||
"config_flow": true,
|
||||
"documentation": "https://www.home-assistant.io/integrations/modem_callerid",
|
||||
"requirements": ["basicmodem==0.7"],
|
||||
"codeowners": [],
|
||||
"iot_class": "local_polling"
|
||||
"requirements": ["phone_modem==0.1.1"],
|
||||
"codeowners": ["@tkdrob"],
|
||||
"dependencies": ["usb"],
|
||||
"iot_class": "local_polling",
|
||||
"usb": [{"vid":"0572","pid":"1340"}]
|
||||
}
|
||||
|
||||
@ -1,121 +1,126 @@
|
||||
"""A sensor for incoming calls using a USB modem that supports caller ID."""
|
||||
import logging
|
||||
from __future__ import annotations
|
||||
|
||||
from basicmodem.basicmodem import BasicModem as bm
|
||||
from phone_modem import DEFAULT_PORT, PhoneModem
|
||||
import voluptuous as vol
|
||||
|
||||
from homeassistant.components.sensor import PLATFORM_SCHEMA, SensorEntity
|
||||
from homeassistant.config_entries import SOURCE_IMPORT, ConfigEntry
|
||||
from homeassistant.const import (
|
||||
CONF_DEVICE,
|
||||
CONF_NAME,
|
||||
EVENT_HOMEASSISTANT_STOP,
|
||||
STATE_IDLE,
|
||||
)
|
||||
import homeassistant.helpers.config_validation as cv
|
||||
from homeassistant.core import HomeAssistant, callback
|
||||
from homeassistant.helpers import config_validation as cv, entity_platform
|
||||
from homeassistant.helpers.typing import DiscoveryInfoType
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
DEFAULT_NAME = "Modem CallerID"
|
||||
ICON = "mdi:phone-classic"
|
||||
DEFAULT_DEVICE = "/dev/ttyACM0"
|
||||
from .const import CID, DATA_KEY_API, DEFAULT_NAME, DOMAIN, ICON, SERVICE_REJECT_CALL
|
||||
|
||||
STATE_RING = "ring"
|
||||
STATE_CALLERID = "callerid"
|
||||
|
||||
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
|
||||
{
|
||||
vol.Optional(CONF_NAME, default=DEFAULT_NAME): cv.string,
|
||||
vol.Optional(CONF_DEVICE, default=DEFAULT_DEVICE): cv.string,
|
||||
}
|
||||
# Deprecated in Home Assistant 2021.10
|
||||
PLATFORM_SCHEMA = cv.deprecated(
|
||||
vol.All(
|
||||
PLATFORM_SCHEMA.extend(
|
||||
{
|
||||
vol.Optional(CONF_NAME, default=DEFAULT_NAME): cv.string,
|
||||
vol.Optional(CONF_DEVICE, default=DEFAULT_PORT): cv.string,
|
||||
}
|
||||
)
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
def setup_platform(hass, config, add_entities, discovery_info=None):
|
||||
"""Set up modem caller ID sensor platform."""
|
||||
async def async_setup_platform(
|
||||
hass: HomeAssistant,
|
||||
config: ConfigEntry,
|
||||
async_add_entities: entity_platform.AddEntitiesCallback,
|
||||
discovery_info: DiscoveryInfoType | None = None,
|
||||
) -> None:
|
||||
"""Set up the Modem Caller ID component."""
|
||||
hass.async_create_task(
|
||||
hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": SOURCE_IMPORT}, data=config
|
||||
)
|
||||
)
|
||||
|
||||
name = config.get(CONF_NAME)
|
||||
port = config.get(CONF_DEVICE)
|
||||
|
||||
modem = bm(port)
|
||||
if modem.state == modem.STATE_FAILED:
|
||||
_LOGGER.error("Unable to initialize modem")
|
||||
return
|
||||
async def async_setup_entry(
|
||||
hass: HomeAssistant,
|
||||
entry: ConfigEntry,
|
||||
async_add_entities: entity_platform.AddEntitiesCallback,
|
||||
) -> None:
|
||||
"""Set up the Modem Caller ID sensor."""
|
||||
api = hass.data[DOMAIN][entry.entry_id][DATA_KEY_API]
|
||||
async_add_entities(
|
||||
[
|
||||
ModemCalleridSensor(
|
||||
api,
|
||||
entry.title,
|
||||
entry.data[CONF_DEVICE],
|
||||
entry.entry_id,
|
||||
)
|
||||
]
|
||||
)
|
||||
|
||||
add_entities([ModemCalleridSensor(hass, name, port, modem)])
|
||||
async def _async_on_hass_stop(self) -> None:
|
||||
"""HA is shutting down, close modem port."""
|
||||
if hass.data[DOMAIN][entry.entry_id][DATA_KEY_API]:
|
||||
await hass.data[DOMAIN][entry.entry_id][DATA_KEY_API].close()
|
||||
|
||||
entry.async_on_unload(
|
||||
hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP, _async_on_hass_stop)
|
||||
)
|
||||
|
||||
platform = entity_platform.async_get_current_platform()
|
||||
|
||||
platform.async_register_entity_service(SERVICE_REJECT_CALL, {}, "async_reject_call")
|
||||
|
||||
|
||||
class ModemCalleridSensor(SensorEntity):
|
||||
"""Implementation of USB modem caller ID sensor."""
|
||||
|
||||
def __init__(self, hass, name, port, modem):
|
||||
_attr_icon = ICON
|
||||
_attr_should_poll = False
|
||||
|
||||
def __init__(
|
||||
self, api: PhoneModem, name: str, device: str, server_unique_id: str
|
||||
) -> None:
|
||||
"""Initialize the sensor."""
|
||||
self._attributes = {"cid_time": 0, "cid_number": "", "cid_name": ""}
|
||||
self._name = name
|
||||
self.port = port
|
||||
self.modem = modem
|
||||
self._state = STATE_IDLE
|
||||
modem.registercallback(self._incomingcallcallback)
|
||||
hass.bus.listen_once(EVENT_HOMEASSISTANT_STOP, self._stop_modem)
|
||||
self.device = device
|
||||
self.api = api
|
||||
self._attr_name = name
|
||||
self._attr_unique_id = server_unique_id
|
||||
self._attr_native_value = STATE_IDLE
|
||||
self._attr_extra_state_attributes = {
|
||||
CID.CID_TIME: 0,
|
||||
CID.CID_NUMBER: "",
|
||||
CID.CID_NAME: "",
|
||||
}
|
||||
|
||||
def set_state(self, state):
|
||||
"""Set the state."""
|
||||
self._state = state
|
||||
async def async_added_to_hass(self) -> None:
|
||||
"""Call when the modem sensor is added to Home Assistant."""
|
||||
self.api.registercallback(self._async_incoming_call)
|
||||
await super().async_added_to_hass()
|
||||
|
||||
def set_attributes(self, attributes):
|
||||
"""Set the state attributes."""
|
||||
self._attributes = attributes
|
||||
|
||||
@property
|
||||
def should_poll(self):
|
||||
"""No polling needed."""
|
||||
return False
|
||||
|
||||
@property
|
||||
def icon(self):
|
||||
"""Return icon."""
|
||||
return ICON
|
||||
|
||||
@property
|
||||
def native_value(self):
|
||||
"""Return the state of the device."""
|
||||
return self._state
|
||||
|
||||
@property
|
||||
def name(self):
|
||||
"""Return the name of the sensor."""
|
||||
return self._name
|
||||
|
||||
@property
|
||||
def extra_state_attributes(self):
|
||||
"""Return the state attributes."""
|
||||
return self._attributes
|
||||
|
||||
def _stop_modem(self, event):
|
||||
"""HA is shutting down, close modem port."""
|
||||
if self.modem:
|
||||
self.modem.close()
|
||||
self.modem = None
|
||||
|
||||
def _incomingcallcallback(self, newstate):
|
||||
@callback
|
||||
def _async_incoming_call(self, new_state) -> None:
|
||||
"""Handle new states."""
|
||||
if newstate == self.modem.STATE_RING:
|
||||
if self.state == self.modem.STATE_IDLE:
|
||||
att = {
|
||||
"cid_time": self.modem.get_cidtime,
|
||||
"cid_number": "",
|
||||
"cid_name": "",
|
||||
if new_state == PhoneModem.STATE_RING:
|
||||
if self.native_value == PhoneModem.STATE_IDLE:
|
||||
self._attr_extra_state_attributes = {
|
||||
CID.CID_NUMBER: "",
|
||||
CID.CID_NAME: "",
|
||||
}
|
||||
self.set_attributes(att)
|
||||
self._state = STATE_RING
|
||||
self.schedule_update_ha_state()
|
||||
elif newstate == self.modem.STATE_CALLERID:
|
||||
att = {
|
||||
"cid_time": self.modem.get_cidtime,
|
||||
"cid_number": self.modem.get_cidnumber,
|
||||
"cid_name": self.modem.get_cidname,
|
||||
elif new_state == PhoneModem.STATE_CALLERID:
|
||||
self._attr_extra_state_attributes = {
|
||||
CID.CID_NUMBER: self.api.cid_number,
|
||||
CID.CID_NAME: self.api.cid_name,
|
||||
}
|
||||
self.set_attributes(att)
|
||||
self._state = STATE_CALLERID
|
||||
self.schedule_update_ha_state()
|
||||
elif newstate == self.modem.STATE_IDLE:
|
||||
self._state = STATE_IDLE
|
||||
self.schedule_update_ha_state()
|
||||
self._attr_extra_state_attributes[CID.CID_TIME] = self.api.cid_time
|
||||
self._attr_native_value = self.api.state
|
||||
self.async_write_ha_state()
|
||||
|
||||
async def async_reject_call(self) -> None:
|
||||
"""Reject Incoming Call."""
|
||||
await self.api.reject_call(self.device)
|
||||
|
||||
7
homeassistant/components/modem_callerid/services.yaml
Normal file
7
homeassistant/components/modem_callerid/services.yaml
Normal file
@ -0,0 +1,7 @@
|
||||
reject_call:
|
||||
name: Reject Call
|
||||
description: Reject incoming call.
|
||||
target:
|
||||
entity:
|
||||
integration: modem_callerid
|
||||
domain: sensor
|
||||
26
homeassistant/components/modem_callerid/strings.json
Normal file
26
homeassistant/components/modem_callerid/strings.json
Normal file
@ -0,0 +1,26 @@
|
||||
{
|
||||
"config": {
|
||||
"step": {
|
||||
"user": {
|
||||
"title": "Phone Modem",
|
||||
"description": "This is an integration for landline calls using a CX93001 voice modem. This can retrieve caller ID information with an option to reject an incoming call.",
|
||||
"data": {
|
||||
"name": "[%key:common::config_flow::data::name%]",
|
||||
"port": "[%key:common::config_flow::data::port%]"
|
||||
}
|
||||
},
|
||||
"usb_confirm": {
|
||||
"title": "Phone Modem",
|
||||
"description": "This is an integration for landline calls using a CX93001 voice modem. This can retrieve caller ID information with an option to reject an incoming call."
|
||||
}
|
||||
},
|
||||
"error": {
|
||||
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]"
|
||||
},
|
||||
"abort": {
|
||||
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]",
|
||||
"already_in_progress": "[%key:common::config_flow::abort::already_in_progress%]",
|
||||
"no_devices_found": "No remaining devices found"
|
||||
}
|
||||
}
|
||||
}
|
||||
26
homeassistant/components/modem_callerid/translations/en.json
Normal file
26
homeassistant/components/modem_callerid/translations/en.json
Normal file
@ -0,0 +1,26 @@
|
||||
{
|
||||
"config": {
|
||||
"abort": {
|
||||
"already_configured": "Device is already configured",
|
||||
"already_in_progress": "Configuration flow is already in progress",
|
||||
"no_devices_found": "No remaining devices found"
|
||||
},
|
||||
"error": {
|
||||
"cannot_connect": "Failed to connect"
|
||||
},
|
||||
"step": {
|
||||
"user": {
|
||||
"data": {
|
||||
"name": "Name",
|
||||
"port": "Port"
|
||||
},
|
||||
"title": "Phone Modem",
|
||||
"description": "This is an integration for landline calls using a CX93001 voice modem. This can retrieve caller id information with an option to reject an incoming call."
|
||||
},
|
||||
"usb_confirm": {
|
||||
"title": "Phone Modem",
|
||||
"description": "This is an integration for landline calls using a CX93001 voice modem. This can retrieve caller ID information with an option to reject an incoming call."
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -167,6 +167,7 @@ FLOWS = [
|
||||
"mill",
|
||||
"minecraft_server",
|
||||
"mobile_app",
|
||||
"modem_callerid",
|
||||
"modern_forms",
|
||||
"monoprice",
|
||||
"motion_blinds",
|
||||
|
||||
@ -6,6 +6,11 @@ To update, run python3 -m script.hassfest
|
||||
# fmt: off
|
||||
|
||||
USB = [
|
||||
{
|
||||
"domain": "modem_callerid",
|
||||
"vid": "0572",
|
||||
"pid": "1340"
|
||||
},
|
||||
{
|
||||
"domain": "zha",
|
||||
"vid": "10C4",
|
||||
|
||||
@ -356,9 +356,6 @@ baidu-aip==1.6.6
|
||||
# homeassistant.components.homekit
|
||||
base36==0.1.1
|
||||
|
||||
# homeassistant.components.modem_callerid
|
||||
basicmodem==0.7
|
||||
|
||||
# homeassistant.components.linux_battery
|
||||
batinfo==0.4.2
|
||||
|
||||
@ -1169,6 +1166,9 @@ pencompy==0.0.3
|
||||
# homeassistant.components.unifi_direct
|
||||
pexpect==4.6.0
|
||||
|
||||
# homeassistant.components.modem_callerid
|
||||
phone_modem==0.1.1
|
||||
|
||||
# homeassistant.components.onewire
|
||||
pi1wire==0.1.0
|
||||
|
||||
|
||||
@ -662,6 +662,9 @@ pdunehd==1.3.2
|
||||
# homeassistant.components.unifi_direct
|
||||
pexpect==4.6.0
|
||||
|
||||
# homeassistant.components.modem_callerid
|
||||
phone_modem==0.1.1
|
||||
|
||||
# homeassistant.components.onewire
|
||||
pi1wire==0.1.0
|
||||
|
||||
|
||||
25
tests/components/modem_callerid/__init__.py
Normal file
25
tests/components/modem_callerid/__init__.py
Normal file
@ -0,0 +1,25 @@
|
||||
"""Tests for the Modem Caller ID integration."""
|
||||
|
||||
from unittest.mock import patch
|
||||
|
||||
from phone_modem import DEFAULT_PORT
|
||||
|
||||
from homeassistant.const import CONF_DEVICE
|
||||
|
||||
CONF_DATA = {CONF_DEVICE: DEFAULT_PORT}
|
||||
|
||||
IMPORT_DATA = {"sensor": {"platform": "modem_callerid"}}
|
||||
|
||||
|
||||
def _patch_init_modem(mocked_modem):
|
||||
return patch(
|
||||
"homeassistant.components.modem_callerid.PhoneModem",
|
||||
return_value=mocked_modem,
|
||||
)
|
||||
|
||||
|
||||
def _patch_config_flow_modem(mocked_modem):
|
||||
return patch(
|
||||
"homeassistant.components.modem_callerid.config_flow.PhoneModem",
|
||||
return_value=mocked_modem,
|
||||
)
|
||||
204
tests/components/modem_callerid/test_config_flow.py
Normal file
204
tests/components/modem_callerid/test_config_flow.py
Normal file
@ -0,0 +1,204 @@
|
||||
"""Test Modem Caller ID config flow."""
|
||||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
|
||||
import phone_modem
|
||||
import serial.tools.list_ports
|
||||
|
||||
from homeassistant.components import usb
|
||||
from homeassistant.components.modem_callerid.const import DEFAULT_NAME, DOMAIN
|
||||
from homeassistant.config_entries import SOURCE_IMPORT, SOURCE_USB, SOURCE_USER
|
||||
from homeassistant.const import CONF_DEVICE, CONF_SOURCE
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.data_entry_flow import (
|
||||
RESULT_TYPE_ABORT,
|
||||
RESULT_TYPE_CREATE_ENTRY,
|
||||
RESULT_TYPE_FORM,
|
||||
)
|
||||
|
||||
from . import CONF_DATA, IMPORT_DATA, _patch_config_flow_modem
|
||||
|
||||
DISCOVERY_INFO = {
|
||||
"device": phone_modem.DEFAULT_PORT,
|
||||
"pid": "1340",
|
||||
"vid": "0572",
|
||||
"serial_number": "1234",
|
||||
"description": "modem",
|
||||
"manufacturer": "Connexant",
|
||||
}
|
||||
|
||||
|
||||
def _patch_setup():
|
||||
return patch(
|
||||
"homeassistant.components.modem_callerid.async_setup_entry",
|
||||
return_value=True,
|
||||
)
|
||||
|
||||
|
||||
def com_port():
|
||||
"""Mock of a serial port."""
|
||||
port = serial.tools.list_ports_common.ListPortInfo(phone_modem.DEFAULT_PORT)
|
||||
port.serial_number = "1234"
|
||||
port.manufacturer = "Virtual serial port"
|
||||
port.device = phone_modem.DEFAULT_PORT
|
||||
port.description = "Some serial port"
|
||||
|
||||
return port
|
||||
|
||||
|
||||
@patch("serial.tools.list_ports.comports", MagicMock(return_value=[com_port()]))
|
||||
async def test_flow_usb(hass: HomeAssistant):
|
||||
"""Test usb discovery flow."""
|
||||
port = com_port()
|
||||
with _patch_config_flow_modem(AsyncMock()), _patch_setup():
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN,
|
||||
context={CONF_SOURCE: SOURCE_USB},
|
||||
data=DISCOVERY_INFO,
|
||||
)
|
||||
assert result["type"] == RESULT_TYPE_FORM
|
||||
assert result["step_id"] == "usb_confirm"
|
||||
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
user_input={CONF_DEVICE: phone_modem.DEFAULT_PORT},
|
||||
)
|
||||
assert result["type"] == RESULT_TYPE_CREATE_ENTRY
|
||||
assert result["data"] == {CONF_DEVICE: port.device}
|
||||
|
||||
|
||||
@patch("serial.tools.list_ports.comports", MagicMock(return_value=[com_port()]))
|
||||
async def test_flow_usb_cannot_connect(hass: HomeAssistant):
|
||||
"""Test usb flow connection error."""
|
||||
with _patch_config_flow_modem(AsyncMock()) as modemmock:
|
||||
modemmock.side_effect = phone_modem.exceptions.SerialError
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={CONF_SOURCE: SOURCE_USB}, data=DISCOVERY_INFO
|
||||
)
|
||||
assert result["type"] == RESULT_TYPE_ABORT
|
||||
assert result["reason"] == "cannot_connect"
|
||||
|
||||
|
||||
@patch("serial.tools.list_ports.comports", MagicMock(return_value=[com_port()]))
|
||||
async def test_flow_user(hass: HomeAssistant):
|
||||
"""Test user initialized flow."""
|
||||
port = com_port()
|
||||
port_select = usb.human_readable_device_name(
|
||||
port.device,
|
||||
port.serial_number,
|
||||
port.manufacturer,
|
||||
port.description,
|
||||
port.vid,
|
||||
port.pid,
|
||||
)
|
||||
mocked_modem = AsyncMock()
|
||||
with _patch_config_flow_modem(mocked_modem), _patch_setup():
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN,
|
||||
context={CONF_SOURCE: SOURCE_USER},
|
||||
data={CONF_DEVICE: port_select},
|
||||
)
|
||||
assert result["type"] == RESULT_TYPE_CREATE_ENTRY
|
||||
assert result["data"] == {CONF_DEVICE: port.device}
|
||||
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN,
|
||||
context={CONF_SOURCE: SOURCE_USER},
|
||||
data={CONF_DEVICE: port_select},
|
||||
)
|
||||
assert result["type"] == RESULT_TYPE_ABORT
|
||||
assert result["reason"] == "no_devices_found"
|
||||
|
||||
|
||||
@patch("serial.tools.list_ports.comports", MagicMock(return_value=[com_port()]))
|
||||
async def test_flow_user_error(hass: HomeAssistant):
|
||||
"""Test user initialized flow with unreachable device."""
|
||||
port = com_port()
|
||||
port_select = usb.human_readable_device_name(
|
||||
port.device,
|
||||
port.serial_number,
|
||||
port.manufacturer,
|
||||
port.description,
|
||||
port.vid,
|
||||
port.pid,
|
||||
)
|
||||
with _patch_config_flow_modem(AsyncMock()) as modemmock:
|
||||
modemmock.side_effect = phone_modem.exceptions.SerialError
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={CONF_SOURCE: SOURCE_USER}, data={CONF_DEVICE: port_select}
|
||||
)
|
||||
assert result["type"] == RESULT_TYPE_FORM
|
||||
assert result["step_id"] == "user"
|
||||
assert result["errors"] == {"base": "cannot_connect"}
|
||||
|
||||
modemmock.side_effect = None
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
user_input={CONF_DEVICE: port_select},
|
||||
)
|
||||
assert result["type"] == RESULT_TYPE_CREATE_ENTRY
|
||||
assert result["data"] == {CONF_DEVICE: port.device}
|
||||
|
||||
|
||||
@patch("serial.tools.list_ports.comports", MagicMock())
|
||||
async def test_flow_user_no_port_list(hass: HomeAssistant):
|
||||
"""Test user with no list of ports."""
|
||||
with _patch_config_flow_modem(AsyncMock()):
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN,
|
||||
context={CONF_SOURCE: SOURCE_USER},
|
||||
data={CONF_DEVICE: phone_modem.DEFAULT_PORT},
|
||||
)
|
||||
assert result["type"] == RESULT_TYPE_ABORT
|
||||
assert result["reason"] == "no_devices_found"
|
||||
|
||||
|
||||
async def test_abort_user_with_existing_flow(hass: HomeAssistant):
|
||||
"""Test user flow is aborted when another discovery has happened."""
|
||||
with _patch_config_flow_modem(AsyncMock()):
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN,
|
||||
context={CONF_SOURCE: SOURCE_USB},
|
||||
data=DISCOVERY_INFO,
|
||||
)
|
||||
assert result["type"] == RESULT_TYPE_FORM
|
||||
assert result["step_id"] == "usb_confirm"
|
||||
|
||||
result2 = await hass.config_entries.flow.async_init(
|
||||
DOMAIN,
|
||||
context={CONF_SOURCE: SOURCE_USER},
|
||||
data={},
|
||||
)
|
||||
|
||||
assert result2["type"] == RESULT_TYPE_ABORT
|
||||
assert result2["reason"] == "already_in_progress"
|
||||
|
||||
|
||||
@patch("serial.tools.list_ports.comports", MagicMock(return_value=[com_port()]))
|
||||
async def test_flow_import(hass: HomeAssistant):
|
||||
"""Test import step."""
|
||||
with _patch_config_flow_modem(AsyncMock()):
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={CONF_SOURCE: SOURCE_IMPORT}, data=IMPORT_DATA
|
||||
)
|
||||
|
||||
assert result["type"] == RESULT_TYPE_CREATE_ENTRY
|
||||
assert result["title"] == DEFAULT_NAME
|
||||
assert result["data"] == CONF_DATA
|
||||
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={CONF_SOURCE: SOURCE_IMPORT}, data=IMPORT_DATA
|
||||
)
|
||||
|
||||
assert result["type"] == RESULT_TYPE_ABORT
|
||||
assert result["reason"] == "already_configured"
|
||||
|
||||
|
||||
async def test_flow_import_cannot_connect(hass: HomeAssistant):
|
||||
"""Test import connection error."""
|
||||
with _patch_config_flow_modem(AsyncMock()) as modemmock:
|
||||
modemmock.side_effect = phone_modem.exceptions.SerialError
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={CONF_SOURCE: SOURCE_IMPORT}, data=IMPORT_DATA
|
||||
)
|
||||
assert result["type"] == RESULT_TYPE_ABORT
|
||||
assert result["reason"] == "cannot_connect"
|
||||
63
tests/components/modem_callerid/test_init.py
Normal file
63
tests/components/modem_callerid/test_init.py
Normal file
@ -0,0 +1,63 @@
|
||||
"""Test Modem Caller ID integration."""
|
||||
from unittest.mock import AsyncMock, patch
|
||||
|
||||
from phone_modem import exceptions
|
||||
|
||||
from homeassistant.components.modem_callerid.const import DOMAIN
|
||||
from homeassistant.config_entries import ConfigEntryState
|
||||
from homeassistant.core import HomeAssistant
|
||||
|
||||
from . import CONF_DATA, _patch_init_modem
|
||||
|
||||
from tests.common import MockConfigEntry
|
||||
|
||||
|
||||
async def test_setup_config(hass: HomeAssistant):
|
||||
"""Test Modem Caller ID setup."""
|
||||
entry = MockConfigEntry(
|
||||
domain=DOMAIN,
|
||||
data=CONF_DATA,
|
||||
)
|
||||
entry.add_to_hass(hass)
|
||||
mocked_modem = AsyncMock()
|
||||
with _patch_init_modem(mocked_modem):
|
||||
await hass.config_entries.async_setup(entry.entry_id)
|
||||
assert entry.state == ConfigEntryState.LOADED
|
||||
|
||||
|
||||
async def test_async_setup_entry_not_ready(hass: HomeAssistant):
|
||||
"""Test that it throws ConfigEntryNotReady when exception occurs during setup."""
|
||||
entry = MockConfigEntry(
|
||||
domain=DOMAIN,
|
||||
data=CONF_DATA,
|
||||
)
|
||||
entry.add_to_hass(hass)
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.modem_callerid.PhoneModem",
|
||||
side_effect=exceptions.SerialError(),
|
||||
):
|
||||
await hass.config_entries.async_setup(entry.entry_id)
|
||||
assert len(hass.config_entries.async_entries(DOMAIN)) == 1
|
||||
assert entry.state == ConfigEntryState.SETUP_ERROR
|
||||
assert not hass.data.get(DOMAIN)
|
||||
|
||||
|
||||
async def test_unload_config_entry(hass: HomeAssistant):
|
||||
"""Test unload."""
|
||||
entry = MockConfigEntry(
|
||||
domain=DOMAIN,
|
||||
data=CONF_DATA,
|
||||
)
|
||||
entry.add_to_hass(hass)
|
||||
mocked_modem = AsyncMock()
|
||||
with _patch_init_modem(mocked_modem):
|
||||
await hass.config_entries.async_setup(entry.entry_id)
|
||||
assert len(hass.config_entries.async_entries(DOMAIN)) == 1
|
||||
assert entry.state is ConfigEntryState.LOADED
|
||||
|
||||
assert await hass.config_entries.async_unload(entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert entry.state is ConfigEntryState.NOT_LOADED
|
||||
assert not hass.data.get(DOMAIN)
|
||||
Loading…
x
Reference in New Issue
Block a user