Move onewire device compatibility checks (#59338)

* Move device compatibility checks to onewirehub

* Add test for dtoverlay warning

* Add tests for unknown device warning

* Move dtoverlay error

* Empty commit to retrigger tests

* Update description

* Patch asyncio.sleep to speed up the tests

Co-authored-by: epenet <epenet@users.noreply.github.com>
This commit is contained in:
epenet 2021-11-09 18:30:05 +01:00 committed by GitHub
parent d5fcf0b622
commit 7e81c6a591
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
8 changed files with 136 additions and 66 deletions

View File

@ -20,6 +20,25 @@ DOMAIN = "onewire"
DEVICE_KEYS_0_7 = range(8)
DEVICE_KEYS_A_B = ("A", "B")
DEVICE_SUPPORT_OWSERVER = {
"05": (),
"10": (),
"12": (),
"1D": (),
"1F": (),
"22": (),
"26": (),
"28": (),
"29": (),
"3A": (),
"3B": (),
"42": (),
"7E": ("EDS0066", "EDS0068"),
"EF": ("HB_MOISTURE_METER", "HobbyBoards_EF"),
}
DEVICE_SUPPORT_SYSBUS = ["10", "22", "28", "3B", "42"]
MANUFACTURER_MAXIM = "Maxim Integrated"
MANUFACTURER_HOBBYBOARDS = "Hobby Boards"
MANUFACTURER_EDS = "Embedded Data Systems"

View File

@ -29,6 +29,8 @@ from .const import (
CONF_MOUNT_DIR,
CONF_TYPE_OWSERVER,
CONF_TYPE_SYSBUS,
DEVICE_SUPPORT_OWSERVER,
DEVICE_SUPPORT_SYSBUS,
DOMAIN,
MANUFACTURER_EDS,
MANUFACTURER_HOBBYBOARDS,
@ -53,6 +55,13 @@ DEVICE_MANUFACTURER = {
_LOGGER = logging.getLogger(__name__)
def _is_known_owserver_device(device_family: str, device_type: str) -> bool:
"""Check if device family/type is known to the library."""
if device_family in ("7E", "EF"): # EDS or HobbyBoard
return device_type in DEVICE_SUPPORT_OWSERVER[device_family]
return device_family in DEVICE_SUPPORT_OWSERVER
class OneWireHub:
"""Hub to communicate with SysBus or OWServer."""
@ -83,10 +92,13 @@ class OneWireHub:
"""Initialize a config entry."""
self.type = config_entry.data[CONF_TYPE]
if self.type == CONF_TYPE_SYSBUS:
await self.check_mount_dir(config_entry.data[CONF_MOUNT_DIR])
mount_dir = config_entry.data[CONF_MOUNT_DIR]
_LOGGER.debug("Initializing using SysBus %s", mount_dir)
await self.check_mount_dir(mount_dir)
elif self.type == CONF_TYPE_OWSERVER:
host = config_entry.data[CONF_HOST]
port = config_entry.data[CONF_PORT]
_LOGGER.debug("Initializing using OWServer %s:%s", host, port)
await self.connect(host, port)
await self.discover_devices()
if TYPE_CHECKING:
@ -120,9 +132,23 @@ class OneWireHub:
"""Discover all sysbus devices."""
devices: list[OWDeviceDescription] = []
assert self.pi1proxy
for interface in self.pi1proxy.find_all_sensors():
all_sensors = self.pi1proxy.find_all_sensors()
if not all_sensors:
_LOGGER.error(
"No onewire sensor found. Check if dtoverlay=w1-gpio "
"is in your /boot/config.txt. "
"Check the mount_dir parameter if it's defined"
)
for interface in all_sensors:
device_family = interface.mac_address[:2]
device_id = f"{device_family}-{interface.mac_address[2:]}"
if device_family not in DEVICE_SUPPORT_SYSBUS:
_LOGGER.warning(
"Ignoring unknown device family (%s) found for device %s",
device_family,
device_id,
)
continue
device_info: DeviceInfo = {
ATTR_IDENTIFIERS: {(DOMAIN, device_id)},
ATTR_MANUFACTURER: DEVICE_MANUFACTURER.get(
@ -149,6 +175,14 @@ class OneWireHub:
device_family = self.owproxy.read(f"{device_path}family").decode()
_LOGGER.debug("read `%sfamily`: %s", device_path, device_family)
device_type = self._get_device_type_owserver(device_path)
if not _is_known_owserver_device(device_family, device_type):
_LOGGER.warning(
"Ignoring unknown device family/type (%s/%s) found for device %s",
device_family,
device_type,
device_id,
)
continue
device_info: DeviceInfo = {
ATTR_IDENTIFIERS: {(DOMAIN, device_id)},
ATTR_MANUFACTURER: DEVICE_MANUFACTURER.get(

View File

@ -43,7 +43,6 @@ from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import StateType
from .const import (
CONF_MOUNT_DIR,
CONF_NAMES,
CONF_TYPE_OWSERVER,
CONF_TYPE_SYSBUS,
@ -212,12 +211,8 @@ DEVICE_SENSORS: dict[str, tuple[OneWireSensorEntityDescription, ...]] = {
state_class=STATE_CLASS_TOTAL_INCREASING,
),
),
"EF": (), # "HobbyBoard": special
"7E": (), # "EDS": special
}
DEVICE_SUPPORT_SYSBUS = ["10", "22", "28", "3B", "42"]
# EF sensors are usually hobbyboards specialized sensors.
# These can only be read by OWFS. Currently this driver only supports them
# via owserver (network protocol)
@ -343,7 +338,9 @@ EDS_SENSORS: dict[str, tuple[OneWireSensorEntityDescription, ...]] = {
}
def get_sensor_types(device_sub_type: str) -> dict[str, Any]:
def get_sensor_types(
device_sub_type: str,
) -> dict[str, tuple[OneWireSensorEntityDescription, ...]]:
"""Return the proper info array for the device type."""
if "HobbyBoard" in device_sub_type:
return HOBBYBOARD_EF
@ -398,11 +395,6 @@ def get_entities(
family = device_type
if family not in get_sensor_types(device_sub_type):
_LOGGER.warning(
"Ignoring unknown family (%s) of sensor found for device: %s",
family,
device_id,
)
continue
for description in get_sensor_types(device_sub_type)[family]:
if description.key.startswith("moisture/"):
@ -434,8 +426,6 @@ def get_entities(
# We have a raw GPIO ow sensor on a Pi
elif conf_type == CONF_TYPE_SYSBUS:
base_dir = config[CONF_MOUNT_DIR]
_LOGGER.debug("Initializing using SysBus %s", base_dir)
for device in onewirehub.devices:
if TYPE_CHECKING:
assert isinstance(device, OWDirectDeviceDescription)
@ -443,14 +433,6 @@ def get_entities(
family = p1sensor.mac_address[:2]
device_id = f"{family}-{p1sensor.mac_address[2:]}"
device_info = device.device_info
if family not in DEVICE_SUPPORT_SYSBUS:
_LOGGER.warning(
"Ignoring unknown family (%s) of sensor found for device: %s",
family,
device_id,
)
continue
description = SIMPLE_TEMPERATURE_SENSOR_DESCRIPTION
device_file = f"/sys/bus/w1/devices/{device_id}/w1_slave"
name = f"{device_names.get(device_id, device_id)} {description.name}"
@ -464,12 +446,6 @@ def get_entities(
owsensor=p1sensor,
)
)
if not entities:
_LOGGER.error(
"No onewire sensor found. Check if dtoverlay=w1-gpio "
"is in your /boot/config.txt. "
"Check the mount_dir parameter if it's defined"
)
return entities

View File

@ -49,6 +49,7 @@ ATTR_DEVICE_FILE = "device_file"
ATTR_DEVICE_INFO = "device_info"
ATTR_INJECT_READS = "inject_reads"
ATTR_UNIQUE_ID = "unique_id"
ATTR_UNKNOWN_DEVICE = "unknown_device"
FIXED_ATTRIBUTES = (
ATTR_DEVICE_CLASS,
@ -62,7 +63,7 @@ MOCK_OWPROXY_DEVICES = {
ATTR_INJECT_READS: [
b"", # read device type
],
SENSOR_DOMAIN: [],
ATTR_UNKNOWN_DEVICE: True,
},
"05.111111111111": {
ATTR_INJECT_READS: [
@ -879,7 +880,7 @@ MOCK_OWPROXY_DEVICES = {
MOCK_SYSBUS_DEVICES = {
"00-111111111111": {
SENSOR_DOMAIN: [],
ATTR_UNKNOWN_DEVICE: True,
},
"10-111111111111": {
ATTR_DEVICE_INFO: {
@ -900,12 +901,6 @@ MOCK_SYSBUS_DEVICES = {
},
],
},
"12-111111111111": {
SENSOR_DOMAIN: [],
},
"1D-111111111111": {
SENSOR_DOMAIN: [],
},
"22-111111111111": {
ATTR_DEVICE_INFO: {
ATTR_IDENTIFIERS: {(DOMAIN, "22-111111111111")},
@ -913,7 +908,7 @@ MOCK_SYSBUS_DEVICES = {
ATTR_MODEL: "22",
ATTR_NAME: "22-111111111111",
},
"sensor": [
SENSOR_DOMAIN: [
{
ATTR_DEVICE_CLASS: DEVICE_CLASS_TEMPERATURE,
ATTR_ENTITY_ID: "sensor.22_111111111111_temperature",
@ -925,9 +920,6 @@ MOCK_SYSBUS_DEVICES = {
},
],
},
"26-111111111111": {
SENSOR_DOMAIN: [],
},
"28-111111111111": {
ATTR_DEVICE_INFO: {
ATTR_IDENTIFIERS: {(DOMAIN, "28-111111111111")},
@ -947,12 +939,6 @@ MOCK_SYSBUS_DEVICES = {
},
],
},
"29-111111111111": {
SENSOR_DOMAIN: [],
},
"3A-111111111111": {
SENSOR_DOMAIN: [],
},
"3B-111111111111": {
ATTR_DEVICE_INFO: {
ATTR_IDENTIFIERS: {(DOMAIN, "3B-111111111111")},
@ -1029,10 +1015,4 @@ MOCK_SYSBUS_DEVICES = {
},
],
},
"EF-111111111111": {
SENSOR_DOMAIN: [],
},
"EF-111111111112": {
SENSOR_DOMAIN: [],
},
}

View File

@ -1,4 +1,5 @@
"""Tests for 1-Wire devices connected on OWServer."""
import logging
from unittest.mock import MagicMock, patch
import pytest
@ -14,7 +15,7 @@ from . import (
check_entities,
setup_owproxy_mock_devices,
)
from .const import ATTR_DEVICE_INFO, MOCK_OWPROXY_DEVICES
from .const import ATTR_DEVICE_INFO, ATTR_UNKNOWN_DEVICE, MOCK_OWPROXY_DEVICES
from tests.common import mock_device_registry, mock_registry
@ -27,7 +28,11 @@ def override_platforms():
async def test_owserver_binary_sensor(
hass: HomeAssistant, config_entry: ConfigEntry, owproxy: MagicMock, device_id: str
hass: HomeAssistant,
config_entry: ConfigEntry,
owproxy: MagicMock,
device_id: str,
caplog: pytest.LogCaptureFixture,
):
"""Test for 1-Wire binary sensor.
@ -41,8 +46,13 @@ async def test_owserver_binary_sensor(
expected_devices = ensure_list(mock_device.get(ATTR_DEVICE_INFO))
setup_owproxy_mock_devices(owproxy, BINARY_SENSOR_DOMAIN, [device_id])
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
with caplog.at_level(logging.WARNING, logger="homeassistant.components.onewire"):
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
if mock_device.get(ATTR_UNKNOWN_DEVICE):
assert "Ignoring unknown device family/type" in caplog.text
else:
assert "Ignoring unknown device family/type" not in caplog.text
check_device_registry(device_registry, expected_devices)
assert len(entity_registry.entities) == len(expected_entities)

View File

@ -1,4 +1,5 @@
"""Tests for 1-Wire config flow."""
import logging
from unittest.mock import MagicMock, patch
import pytest
@ -41,6 +42,19 @@ async def test_unload_entry(hass: HomeAssistant, config_entry: ConfigEntry):
assert not hass.data.get(DOMAIN)
@pytest.mark.usefixtures("sysbus")
async def test_warning_no_devices(
hass: HomeAssistant,
sysbus_config_entry: ConfigEntry,
caplog: pytest.LogCaptureFixture,
):
"""Test warning is generated when no sysbus devices found."""
with caplog.at_level(logging.WARNING, logger="homeassistant.components.onewire"):
await hass.config_entries.async_setup(sysbus_config_entry.entry_id)
await hass.async_block_till_done()
assert "No onewire sensor found. Check if dtoverlay=w1-gpio" in caplog.text
@pytest.mark.usefixtures("sysbus")
async def test_unload_sysbus_entry(
hass: HomeAssistant, sysbus_config_entry: ConfigEntry

View File

@ -1,4 +1,5 @@
"""Tests for 1-Wire sensor platform."""
import logging
from unittest.mock import MagicMock, patch
import pytest
@ -15,7 +16,12 @@ from . import (
setup_owproxy_mock_devices,
setup_sysbus_mock_devices,
)
from .const import ATTR_DEVICE_INFO, MOCK_OWPROXY_DEVICES, MOCK_SYSBUS_DEVICES
from .const import (
ATTR_DEVICE_INFO,
ATTR_UNKNOWN_DEVICE,
MOCK_OWPROXY_DEVICES,
MOCK_SYSBUS_DEVICES,
)
from tests.common import mock_device_registry, mock_registry
@ -28,7 +34,11 @@ def override_platforms():
async def test_owserver_sensor(
hass: HomeAssistant, config_entry: ConfigEntry, owproxy: MagicMock, device_id: str
hass: HomeAssistant,
config_entry: ConfigEntry,
owproxy: MagicMock,
device_id: str,
caplog: pytest.LogCaptureFixture,
):
"""Test for 1-Wire device.
@ -46,8 +56,13 @@ async def test_owserver_sensor(
expected_devices = ensure_list(mock_device.get(ATTR_DEVICE_INFO))
setup_owproxy_mock_devices(owproxy, SENSOR_DOMAIN, [device_id])
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
with caplog.at_level(logging.WARNING, logger="homeassistant.components.onewire"):
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
if mock_device.get(ATTR_UNKNOWN_DEVICE):
assert "Ignoring unknown device family/type" in caplog.text
else:
assert "Ignoring unknown device family/type" not in caplog.text
check_device_registry(device_registry, expected_devices)
assert len(entity_registry.entities) == len(expected_entities)
@ -63,7 +78,10 @@ async def test_owserver_sensor(
@pytest.mark.usefixtures("sysbus")
@pytest.mark.parametrize("device_id", MOCK_SYSBUS_DEVICES.keys(), indirect=True)
async def test_onewiredirect_setup_valid_device(
hass: HomeAssistant, sysbus_config_entry: ConfigEntry, device_id: str
hass: HomeAssistant,
sysbus_config_entry: ConfigEntry,
device_id: str,
caplog: pytest.LogCaptureFixture,
):
"""Test that sysbus config entry works correctly."""
device_registry = mock_device_registry(hass)
@ -80,9 +98,18 @@ async def test_onewiredirect_setup_valid_device(
with patch("pi1wire._finder.glob.glob", return_value=glob_result,), patch(
"pi1wire.OneWire.get_temperature",
side_effect=read_side_effect,
), caplog.at_level(
logging.WARNING, logger="homeassistant.components.onewire"
), patch(
"homeassistant.components.onewire.sensor.asyncio.sleep"
):
await hass.config_entries.async_setup(sysbus_config_entry.entry_id)
await hass.async_block_till_done()
assert "No onewire sensor found. Check if dtoverlay=w1-gpio" not in caplog.text
if mock_device.get(ATTR_UNKNOWN_DEVICE):
assert "Ignoring unknown device family" in caplog.text
else:
assert "Ignoring unknown device family" not in caplog.text
check_device_registry(device_registry, expected_devices)
assert len(entity_registry.entities) == len(expected_entities)

View File

@ -1,4 +1,5 @@
"""Tests for 1-Wire devices connected on OWServer."""
import logging
from unittest.mock import MagicMock, patch
import pytest
@ -21,7 +22,7 @@ from . import (
check_entities,
setup_owproxy_mock_devices,
)
from .const import ATTR_DEVICE_INFO, MOCK_OWPROXY_DEVICES
from .const import ATTR_DEVICE_INFO, ATTR_UNKNOWN_DEVICE, MOCK_OWPROXY_DEVICES
from tests.common import mock_device_registry, mock_registry
@ -34,7 +35,11 @@ def override_platforms():
async def test_owserver_switch(
hass: HomeAssistant, config_entry: ConfigEntry, owproxy: MagicMock, device_id: str
hass: HomeAssistant,
config_entry: ConfigEntry,
owproxy: MagicMock,
device_id: str,
caplog: pytest.LogCaptureFixture,
):
"""Test for 1-Wire switch.
@ -48,8 +53,13 @@ async def test_owserver_switch(
expected_devices = ensure_list(mock_device.get(ATTR_DEVICE_INFO))
setup_owproxy_mock_devices(owproxy, SWITCH_DOMAIN, [device_id])
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
with caplog.at_level(logging.WARNING, logger="homeassistant.components.onewire"):
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
if mock_device.get(ATTR_UNKNOWN_DEVICE):
assert "Ignoring unknown device family/type" in caplog.text
else:
assert "Ignoring unknown device family/type" not in caplog.text
check_device_registry(device_registry, expected_devices)
assert len(entity_registry.entities) == len(expected_entities)