Home-Assistant
/
core
mirror of https://github.com/home-assistant/core.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Cleanup orphan devices in onewire integration (#48581) 

* Cleanup orphan devices (https://github.com/home-assistant/core/issues/47438)

* Refactor unit testing

* Filter device entries for this config entry

* Update logging

* Cleanup check
pull/48782/head
epenet 2021-04-01 15:06:47 +02:00 committed by Paulus Schoutsen
parent f08e7dccdf
commit 5df90b32fc
8 changed files with 419 additions and 409 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

43
homeassistant/components/onewire/__init__.py
View File

@ -1,13 +1,17 @@
"""The 1-Wire component."""
import asyncio
import logging
from homeassistant.config_entries import ConfigEntry
from homeassistant.exceptions import ConfigEntryNotReady
from homeassistant.helpers import device_registry as dr, entity_registry as er
from homeassistant.helpers.typing import HomeAssistantType
from .const import DOMAIN, PLATFORMS
from .onewirehub import CannotConnect, OneWireHub
_LOGGER = logging.getLogger(__name__)
async def async_setup(hass, config):
"""Set up 1-Wire integrations."""
@ -26,10 +30,43 @@ async def async_setup_entry(hass: HomeAssistantType, config_entry: ConfigEntry):
hass.data[DOMAIN][config_entry.unique_id] = onewirehub
for platform in PLATFORMS:
hass.async_create_task(
hass.config_entries.async_forward_entry_setup(config_entry, platform)
async def cleanup_registry() -> None:
# Get registries
device_registry, entity_registry = await asyncio.gather(
hass.helpers.device_registry.async_get_registry(),
hass.helpers.entity_registry.async_get_registry(),
)
# Generate list of all device entries
registry_devices = [
entry.id
for entry in dr.async_entries_for_config_entry(
device_registry, config_entry.entry_id
)
]
# Remove devices that don't belong to any entity
for device_id in registry_devices:
if not er.async_entries_for_device(
entity_registry, device_id, include_disabled_entities=True
):
_LOGGER.debug(
"Removing device `%s` because it does not have any entities",
device_id,
)
device_registry.async_remove_device(device_id)
async def start_platforms() -> None:
"""Start platforms and cleanup devices."""
# wait until all required platforms are ready
await asyncio.gather(
*[
hass.config_entries.async_forward_entry_setup(config_entry, platform)
for platform in PLATFORMS
]
)
await cleanup_registry()
hass.async_create_task(start_platforms())
return True

36
tests/components/onewire/__init__.py
View File

@ -2,6 +2,8 @@
from unittest.mock import patch
from pyownet.protocol import ProtocolError
from homeassistant.components.onewire.const import (
CONF_MOUNT_DIR,
CONF_NAMES,
@ -13,6 +15,8 @@ from homeassistant.components.onewire.const import (
from homeassistant.config_entries import CONN_CLASS_LOCAL_POLL
from homeassistant.const import CONF_HOST, CONF_PORT, CONF_TYPE
from .const import MOCK_OWPROXY_DEVICES
from tests.common import MockConfigEntry
@ -89,3 +93,35 @@ async def setup_onewire_patched_owserver_integration(hass):
await hass.async_block_till_done()
return config_entry
def setup_owproxy_mock_devices(owproxy, domain, device_ids) -> None:
"""Set up mock for owproxy."""
dir_return_value = []
main_read_side_effect = []
sub_read_side_effect = []
for device_id in device_ids:
mock_device = MOCK_OWPROXY_DEVICES[device_id]
# Setup directory listing
dir_return_value += [f"/{device_id}/"]
# Setup device reads
main_read_side_effect += [device_id[0:2].encode()]
if "inject_reads" in mock_device:
main_read_side_effect += mock_device["inject_reads"]
# Setup sub-device reads
device_sensors = mock_device.get(domain, [])
for expected_sensor in device_sensors:
sub_read_side_effect.append(expected_sensor["injected_value"])
# Ensure enough read side effect
read_side_effect = (
main_read_side_effect
+ sub_read_side_effect
+ [ProtocolError("Missing injected value")] * 20
)
owproxy.return_value.dir.return_value = dir_return_value
owproxy.return_value.read.side_effect = read_side_effect

216
tests/components/onewire/test_entity_owserver.py → tests/components/onewire/const.py
View File

@ -1,11 +1,10 @@
"""Tests for 1-Wire devices connected on OWServer."""
from unittest.mock import patch
"""Constants for 1-Wire integration."""
from pi1wire import InvalidCRCException, UnsupportResponseException
from pyownet.protocol import Error as ProtocolError
import pytest
from homeassistant.components.binary_sensor import DOMAIN as BINARY_SENSOR_DOMAIN
from homeassistant.components.onewire.const import DOMAIN, PLATFORMS, PRESSURE_CBAR
from homeassistant.components.onewire.const import DOMAIN, PRESSURE_CBAR
from homeassistant.components.sensor import DOMAIN as SENSOR_DOMAIN
from homeassistant.components.switch import DOMAIN as SWITCH_DOMAIN
from homeassistant.const import (
@ -24,13 +23,8 @@ from homeassistant.const import (
TEMP_CELSIUS,
VOLT,
)
from homeassistant.setup import async_setup_component
from . import setup_onewire_patched_owserver_integration
from tests.common import mock_device_registry, mock_registry
MOCK_DEVICE_SENSORS = {
MOCK_OWPROXY_DEVICES = {
"00.111111111111": {
"inject_reads": [
b"", # read device type
@ -186,7 +180,42 @@ MOCK_DEVICE_SENSORS = {
"model": "DS2409",
"name": "1F.111111111111",
},
SENSOR_DOMAIN: [],
"branches": {
"aux": {},
"main": {
"1D.111111111111": {
"inject_reads": [
b"DS2423", # read device type
],
"device_info": {
"identifiers": {(DOMAIN, "1D.111111111111")},
"manufacturer": "Maxim Integrated",
"model": "DS2423",
"name": "1D.111111111111",
},
SENSOR_DOMAIN: [
{
"entity_id": "sensor.1d_111111111111_counter_a",
"device_file": "/1F.111111111111/main/1D.111111111111/counter.A",
"unique_id": "/1D.111111111111/counter.A",
"injected_value": b" 251123",
"result": "251123",
"unit": "count",
"class": None,
},
{
"entity_id": "sensor.1d_111111111111_counter_b",
"device_file": "/1F.111111111111/main/1D.111111111111/counter.B",
"unique_id": "/1D.111111111111/counter.B",
"injected_value": b" 248125",
"result": "248125",
"unit": "count",
"class": None,
},
],
},
},
},
},
"22.111111111111": {
"inject_reads": [
@ -748,65 +777,106 @@ MOCK_DEVICE_SENSORS = {
},
}
@pytest.mark.parametrize("device_id", MOCK_DEVICE_SENSORS.keys())
@pytest.mark.parametrize("platform", PLATFORMS)
@patch("homeassistant.components.onewire.onewirehub.protocol.proxy")
async def test_owserver_setup_valid_device(owproxy, hass, device_id, platform):
"""Test for 1-Wire device.
As they would be on a clean setup: all binary-sensors and switches disabled.
"""
await async_setup_component(hass, "persistent_notification", {})
entity_registry = mock_registry(hass)
device_registry = mock_device_registry(hass)
mock_device_sensor = MOCK_DEVICE_SENSORS[device_id]
device_family = device_id[0:2]
dir_return_value = [f"/{device_id}/"]
read_side_effect = [device_family.encode()]
if "inject_reads" in mock_device_sensor:
read_side_effect += mock_device_sensor["inject_reads"]
expected_sensors = mock_device_sensor.get(platform, [])
for expected_sensor in expected_sensors:
read_side_effect.append(expected_sensor["injected_value"])
# Ensure enough read side effect
read_side_effect.extend([ProtocolError("Missing injected value")] * 20)
owproxy.return_value.dir.return_value = dir_return_value
owproxy.return_value.read.side_effect = read_side_effect
with patch("homeassistant.components.onewire.PLATFORMS", [platform]):
await setup_onewire_patched_owserver_integration(hass)
await hass.async_block_till_done()
assert len(entity_registry.entities) == len(expected_sensors)
if len(expected_sensors) > 0:
device_info = mock_device_sensor["device_info"]
assert len(device_registry.devices) == 1
registry_entry = device_registry.async_get_device({(DOMAIN, device_id)})
assert registry_entry is not None
assert registry_entry.identifiers == {(DOMAIN, device_id)}
assert registry_entry.manufacturer == device_info["manufacturer"]
assert registry_entry.name == device_info["name"]
assert registry_entry.model == device_info["model"]
for expected_sensor in expected_sensors:
entity_id = expected_sensor["entity_id"]
registry_entry = entity_registry.entities.get(entity_id)
assert registry_entry is not None
assert registry_entry.unique_id == expected_sensor["unique_id"]
assert registry_entry.unit_of_measurement == expected_sensor["unit"]
assert registry_entry.device_class == expected_sensor["class"]
assert registry_entry.disabled == expected_sensor.get("disabled", False)
state = hass.states.get(entity_id)
if registry_entry.disabled:
assert state is None
else:
assert state.state == expected_sensor["result"]
assert state.attributes["device_file"] == expected_sensor.get(
"device_file", registry_entry.unique_id
)
MOCK_SYSBUS_DEVICES = {
"00-111111111111": {"sensors": []},
"10-111111111111": {
"device_info": {
"identifiers": {(DOMAIN, "10-111111111111")},
"manufacturer": "Maxim Integrated",
"model": "10",
"name": "10-111111111111",
},
"sensors": [
{
"entity_id": "sensor.my_ds18b20_temperature",
"unique_id": "/sys/bus/w1/devices/10-111111111111/w1_slave",
"injected_value": 25.123,
"result": "25.1",
"unit": TEMP_CELSIUS,
"class": DEVICE_CLASS_TEMPERATURE,
},
],
},
"12-111111111111": {"sensors": []},
"1D-111111111111": {"sensors": []},
"22-111111111111": {
"device_info": {
"identifiers": {(DOMAIN, "22-111111111111")},
"manufacturer": "Maxim Integrated",
"model": "22",
"name": "22-111111111111",
},
"sensors": [
{
"entity_id": "sensor.22_111111111111_temperature",
"unique_id": "/sys/bus/w1/devices/22-111111111111/w1_slave",
"injected_value": FileNotFoundError,
"result": "unknown",
"unit": TEMP_CELSIUS,
"class": DEVICE_CLASS_TEMPERATURE,
},
],
},
"26-111111111111": {"sensors": []},
"28-111111111111": {
"device_info": {
"identifiers": {(DOMAIN, "28-111111111111")},
"manufacturer": "Maxim Integrated",
"model": "28",
"name": "28-111111111111",
},
"sensors": [
{
"entity_id": "sensor.28_111111111111_temperature",
"unique_id": "/sys/bus/w1/devices/28-111111111111/w1_slave",
"injected_value": InvalidCRCException,
"result": "unknown",
"unit": TEMP_CELSIUS,
"class": DEVICE_CLASS_TEMPERATURE,
},
],
},
"29-111111111111": {"sensors": []},
"3B-111111111111": {
"device_info": {
"identifiers": {(DOMAIN, "3B-111111111111")},
"manufacturer": "Maxim Integrated",
"model": "3B",
"name": "3B-111111111111",
},
"sensors": [
{
"entity_id": "sensor.3b_111111111111_temperature",
"unique_id": "/sys/bus/w1/devices/3B-111111111111/w1_slave",
"injected_value": 29.993,
"result": "30.0",
"unit": TEMP_CELSIUS,
"class": DEVICE_CLASS_TEMPERATURE,
},
],
},
"42-111111111111": {
"device_info": {
"identifiers": {(DOMAIN, "42-111111111111")},
"manufacturer": "Maxim Integrated",
"model": "42",
"name": "42-111111111111",
},
"sensors": [
{
"entity_id": "sensor.42_111111111111_temperature",
"unique_id": "/sys/bus/w1/devices/42-111111111111/w1_slave",
"injected_value": UnsupportResponseException,
"result": "unknown",
"unit": TEMP_CELSIUS,
"class": DEVICE_CLASS_TEMPERATURE,
},
],
},
"EF-111111111111": {
"sensors": [],
},
"EF-111111111112": {
"sensors": [],
},
}

59
tests/components/onewire/test_binary_sensor.py
View File

@ -2,40 +2,25 @@
import copy
from unittest.mock import patch
from pyownet.protocol import Error as ProtocolError
import pytest
from homeassistant.components.binary_sensor import DOMAIN as BINARY_SENSOR_DOMAIN
from homeassistant.components.onewire.binary_sensor import DEVICE_BINARY_SENSORS
from homeassistant.const import STATE_OFF, STATE_ON
from homeassistant.setup import async_setup_component
from . import setup_onewire_patched_owserver_integration
from . import setup_onewire_patched_owserver_integration, setup_owproxy_mock_devices
from .const import MOCK_OWPROXY_DEVICES
from tests.common import mock_registry
MOCK_DEVICE_SENSORS = {
"12.111111111111": {
"inject_reads": [
b"DS2406", # read device type
],
BINARY_SENSOR_DOMAIN: [
{
"entity_id": "binary_sensor.12_111111111111_sensed_a",
"injected_value": b" 1",
"result": STATE_ON,
},
{
"entity_id": "binary_sensor.12_111111111111_sensed_b",
"injected_value": b" 0",
"result": STATE_OFF,
},
],
},
MOCK_BINARY_SENSORS = {
key: value
for (key, value) in MOCK_OWPROXY_DEVICES.items()
if BINARY_SENSOR_DOMAIN in value
}
@pytest.mark.parametrize("device_id", MOCK_DEVICE_SENSORS.keys())
@pytest.mark.parametrize("device_id", MOCK_BINARY_SENSORS.keys())
@patch("homeassistant.components.onewire.onewirehub.protocol.proxy")
async def test_owserver_binary_sensor(owproxy, hass, device_id):
"""Test for 1-Wire binary sensor.
@ -45,26 +30,14 @@ async def test_owserver_binary_sensor(owproxy, hass, device_id):
await async_setup_component(hass, "persistent_notification", {})
entity_registry = mock_registry(hass)
mock_device_sensor = MOCK_DEVICE_SENSORS[device_id]
setup_owproxy_mock_devices(owproxy, BINARY_SENSOR_DOMAIN, [device_id])
device_family = device_id[0:2]
dir_return_value = [f"/{device_id}/"]
read_side_effect = [device_family.encode()]
if "inject_reads" in mock_device_sensor:
read_side_effect += mock_device_sensor["inject_reads"]
expected_sensors = mock_device_sensor[BINARY_SENSOR_DOMAIN]
for expected_sensor in expected_sensors:
read_side_effect.append(expected_sensor["injected_value"])
# Ensure enough read side effect
read_side_effect.extend([ProtocolError("Missing injected value")] * 10)
owproxy.return_value.dir.return_value = dir_return_value
owproxy.return_value.read.side_effect = read_side_effect
mock_device = MOCK_BINARY_SENSORS[device_id]
expected_entities = mock_device[BINARY_SENSOR_DOMAIN]
# Force enable binary sensors
patch_device_binary_sensors = copy.deepcopy(DEVICE_BINARY_SENSORS)
for item in patch_device_binary_sensors[device_family]:
for item in patch_device_binary_sensors[device_id[0:2]]:
item["default_disabled"] = False
with patch(
@ -76,14 +49,14 @@ async def test_owserver_binary_sensor(owproxy, hass, device_id):
await setup_onewire_patched_owserver_integration(hass)
await hass.async_block_till_done()
assert len(entity_registry.entities) == len(expected_sensors)
assert len(entity_registry.entities) == len(expected_entities)
for expected_sensor in expected_sensors:
entity_id = expected_sensor["entity_id"]
for expected_entity in expected_entities:
entity_id = expected_entity["entity_id"]
registry_entry = entity_registry.entities.get(entity_id)
assert registry_entry is not None
state = hass.states.get(entity_id)
assert state.state == expected_sensor["result"]
assert state.attributes["device_file"] == expected_sensor.get(
assert state.state == expected_entity["result"]
assert state.attributes["device_file"] == expected_entity.get(
"device_file", registry_entry.unique_id
)

175
tests/components/onewire/test_entity_sysbus.py
View File

@ -1,175 +0,0 @@
"""Tests for 1-Wire devices connected on SysBus."""
from unittest.mock import patch
from pi1wire import InvalidCRCException, UnsupportResponseException
import pytest
from homeassistant.components.onewire.const import DEFAULT_SYSBUS_MOUNT_DIR, DOMAIN
from homeassistant.components.sensor import DOMAIN as SENSOR_DOMAIN
from homeassistant.const import DEVICE_CLASS_TEMPERATURE, TEMP_CELSIUS
from homeassistant.setup import async_setup_component
from tests.common import mock_device_registry, mock_registry
MOCK_CONFIG = {
SENSOR_DOMAIN: {
"platform": DOMAIN,
"mount_dir": DEFAULT_SYSBUS_MOUNT_DIR,
"names": {
"10-111111111111": "My DS18B20",
},
}
}
MOCK_DEVICE_SENSORS = {
"00-111111111111": {"sensors": []},
"10-111111111111": {
"device_info": {
"identifiers": {(DOMAIN, "10-111111111111")},
"manufacturer": "Maxim Integrated",
"model": "10",
"name": "10-111111111111",
},
"sensors": [
{
"entity_id": "sensor.my_ds18b20_temperature",
"unique_id": "/sys/bus/w1/devices/10-111111111111/w1_slave",
"injected_value": 25.123,
"result": "25.1",
"unit": TEMP_CELSIUS,
"class": DEVICE_CLASS_TEMPERATURE,
},
],
},
"12-111111111111": {"sensors": []},
"1D-111111111111": {"sensors": []},
"22-111111111111": {
"device_info": {
"identifiers": {(DOMAIN, "22-111111111111")},
"manufacturer": "Maxim Integrated",
"model": "22",
"name": "22-111111111111",
},
"sensors": [
{
"entity_id": "sensor.22_111111111111_temperature",
"unique_id": "/sys/bus/w1/devices/22-111111111111/w1_slave",
"injected_value": FileNotFoundError,
"result": "unknown",
"unit": TEMP_CELSIUS,
"class": DEVICE_CLASS_TEMPERATURE,
},
],
},
"26-111111111111": {"sensors": []},
"28-111111111111": {
"device_info": {
"identifiers": {(DOMAIN, "28-111111111111")},
"manufacturer": "Maxim Integrated",
"model": "28",
"name": "28-111111111111",
},
"sensors": [
{
"entity_id": "sensor.28_111111111111_temperature",
"unique_id": "/sys/bus/w1/devices/28-111111111111/w1_slave",
"injected_value": InvalidCRCException,
"result": "unknown",
"unit": TEMP_CELSIUS,
"class": DEVICE_CLASS_TEMPERATURE,
},
],
},
"29-111111111111": {"sensors": []},
"3B-111111111111": {
"device_info": {
"identifiers": {(DOMAIN, "3B-111111111111")},
"manufacturer": "Maxim Integrated",
"model": "3B",
"name": "3B-111111111111",
},
"sensors": [
{
"entity_id": "sensor.3b_111111111111_temperature",
"unique_id": "/sys/bus/w1/devices/3B-111111111111/w1_slave",
"injected_value": 29.993,
"result": "30.0",
"unit": TEMP_CELSIUS,
"class": DEVICE_CLASS_TEMPERATURE,
},
],
},
"42-111111111111": {
"device_info": {
"identifiers": {(DOMAIN, "42-111111111111")},
"manufacturer": "Maxim Integrated",
"model": "42",
"name": "42-111111111111",
},
"sensors": [
{
"entity_id": "sensor.42_111111111111_temperature",
"unique_id": "/sys/bus/w1/devices/42-111111111111/w1_slave",
"injected_value": UnsupportResponseException,
"result": "unknown",
"unit": TEMP_CELSIUS,
"class": DEVICE_CLASS_TEMPERATURE,
},
],
},
"EF-111111111111": {
"sensors": [],
},
"EF-111111111112": {
"sensors": [],
},
}
@pytest.mark.parametrize("device_id", MOCK_DEVICE_SENSORS.keys())
async def test_onewiredirect_setup_valid_device(hass, device_id):
"""Test that sysbus config entry works correctly."""
entity_registry = mock_registry(hass)
device_registry = mock_device_registry(hass)
mock_device_sensor = MOCK_DEVICE_SENSORS[device_id]
glob_result = [f"/{DEFAULT_SYSBUS_MOUNT_DIR}/{device_id}"]
read_side_effect = []
expected_sensors = mock_device_sensor["sensors"]
for expected_sensor in expected_sensors:
read_side_effect.append(expected_sensor["injected_value"])
# Ensure enough read side effect
read_side_effect.extend([FileNotFoundError("Missing injected value")] * 20)
with patch(
"homeassistant.components.onewire.onewirehub.os.path.isdir", return_value=True
), patch("pi1wire._finder.glob.glob", return_value=glob_result,), patch(
"pi1wire.OneWire.get_temperature",
side_effect=read_side_effect,
):
assert await async_setup_component(hass, SENSOR_DOMAIN, MOCK_CONFIG)
await hass.async_block_till_done()
assert len(entity_registry.entities) == len(expected_sensors)
if len(expected_sensors) > 0:
device_info = mock_device_sensor["device_info"]
assert len(device_registry.devices) == 1
registry_entry = device_registry.async_get_device({(DOMAIN, device_id)})
assert registry_entry is not None
assert registry_entry.identifiers == {(DOMAIN, device_id)}
assert registry_entry.manufacturer == device_info["manufacturer"]
assert registry_entry.name == device_info["name"]
assert registry_entry.model == device_info["model"]
for expected_sensor in expected_sensors:
entity_id = expected_sensor["entity_id"]
registry_entry = entity_registry.entities.get(entity_id)
assert registry_entry is not None
assert registry_entry.unique_id == expected_sensor["unique_id"]
assert registry_entry.unit_of_measurement == expected_sensor["unit"]
assert registry_entry.device_class == expected_sensor["class"]
state = hass.states.get(entity_id)
assert state.state == expected_sensor["result"]

50
tests/components/onewire/test_init.py
View File

@ -4,6 +4,7 @@ from unittest.mock import patch
from pyownet.protocol import ConnError, OwnetError
from homeassistant.components.onewire.const import CONF_TYPE_OWSERVER, DOMAIN
from homeassistant.components.sensor import DOMAIN as SENSOR_DOMAIN
from homeassistant.config_entries import (
CONN_CLASS_LOCAL_POLL,
ENTRY_STATE_LOADED,
@ -11,10 +12,17 @@ from homeassistant.config_entries import (
ENTRY_STATE_SETUP_RETRY,
)
from homeassistant.const import CONF_HOST, CONF_PORT, CONF_TYPE
from homeassistant.helpers import device_registry as dr, entity_registry as er
from homeassistant.setup import async_setup_component
from . import setup_onewire_owserver_integration, setup_onewire_sysbus_integration
from . import (
setup_onewire_owserver_integration,
setup_onewire_patched_owserver_integration,
setup_onewire_sysbus_integration,
setup_owproxy_mock_devices,
)
from tests.common import MockConfigEntry
from tests.common import MockConfigEntry, mock_device_registry, mock_registry
async def test_owserver_connect_failure(hass):
@ -87,3 +95,41 @@ async def test_unload_entry(hass):
assert config_entry_owserver.state == ENTRY_STATE_NOT_LOADED
assert config_entry_sysbus.state == ENTRY_STATE_NOT_LOADED
assert not hass.data.get(DOMAIN)
@patch("homeassistant.components.onewire.onewirehub.protocol.proxy")
async def test_registry_cleanup(owproxy, hass):
"""Test for 1-Wire device.
As they would be on a clean setup: all binary-sensors and switches disabled.
"""
await async_setup_component(hass, "persistent_notification", {})
entity_registry = mock_registry(hass)
device_registry = mock_device_registry(hass)
# Initialise with two components
setup_owproxy_mock_devices(
owproxy, SENSOR_DOMAIN, ["10.111111111111", "28.111111111111"]
)
with patch("homeassistant.components.onewire.PLATFORMS", [SENSOR_DOMAIN]):
await setup_onewire_patched_owserver_integration(hass)
await hass.async_block_till_done()
assert len(dr.async_entries_for_config_entry(device_registry, "2")) == 2
assert len(er.async_entries_for_config_entry(entity_registry, "2")) == 2
# Second item has disappeared from bus, and was removed manually from the front-end
setup_owproxy_mock_devices(owproxy, SENSOR_DOMAIN, ["10.111111111111"])
entity_registry.async_remove("sensor.28_111111111111_temperature")
await hass.async_block_till_done()
assert len(er.async_entries_for_config_entry(entity_registry, "2")) == 1
assert len(dr.async_entries_for_config_entry(device_registry, "2")) == 2
# Second item has disappeared from bus, and was removed manually from the front-end
with patch("homeassistant.components.onewire.PLATFORMS", [SENSOR_DOMAIN]):
await hass.config_entries.async_reload("2")
await hass.async_block_till_done()
assert len(er.async_entries_for_config_entry(entity_registry, "2")) == 1
assert len(dr.async_entries_for_config_entry(device_registry, "2")) == 1

159
tests/components/onewire/test_sensor.py
View File

@ -4,54 +4,29 @@ from unittest.mock import patch
from pyownet.protocol import Error as ProtocolError
import pytest
from homeassistant.components.onewire.const import DEFAULT_SYSBUS_MOUNT_DIR, DOMAIN
from homeassistant.components.onewire.const import (
DEFAULT_SYSBUS_MOUNT_DIR,
DOMAIN,
PLATFORMS,
)
from homeassistant.components.sensor import DOMAIN as SENSOR_DOMAIN
from homeassistant.setup import async_setup_component
from . import setup_onewire_patched_owserver_integration
from . import setup_onewire_patched_owserver_integration, setup_owproxy_mock_devices
from .const import MOCK_OWPROXY_DEVICES, MOCK_SYSBUS_DEVICES
from tests.common import assert_setup_component, mock_registry
from tests.common import assert_setup_component, mock_device_registry, mock_registry
MOCK_COUPLERS = {
"1F.111111111111": {
"inject_reads": [
b"DS2409", # read device type
],
"branches": {
"aux": {},
"main": {
"1D.111111111111": {
"inject_reads": [
b"DS2423", # read device type
],
"device_info": {
"identifiers": {(DOMAIN, "1D.111111111111")},
"manufacturer": "Maxim Integrated",
"model": "DS2423",
"name": "1D.111111111111",
},
SENSOR_DOMAIN: [
{
"entity_id": "sensor.1d_111111111111_counter_a",
"device_file": "/1F.111111111111/main/1D.111111111111/counter.A",
"unique_id": "/1D.111111111111/counter.A",
"injected_value": b" 251123",
"result": "251123",
"unit": "count",
"class": None,
},
{
"entity_id": "sensor.1d_111111111111_counter_b",
"device_file": "/1F.111111111111/main/1D.111111111111/counter.B",
"unique_id": "/1D.111111111111/counter.B",
"injected_value": b" 248125",
"result": "248125",
"unit": "count",
"class": None,
},
],
},
},
key: value for (key, value) in MOCK_OWPROXY_DEVICES.items() if "branches" in value
}
MOCK_SYSBUS_CONFIG = {
SENSOR_DOMAIN: {
"platform": DOMAIN,
"mount_dir": DEFAULT_SYSBUS_MOUNT_DIR,
"names": {
"10-111111111111": "My DS18B20",
},
}
}
@ -154,3 +129,103 @@ async def test_sensors_on_owserver_coupler(owproxy, hass, device_id):
else:
assert state.state == expected_sensor["result"]
assert state.attributes["device_file"] == expected_sensor["device_file"]
@pytest.mark.parametrize("device_id", MOCK_OWPROXY_DEVICES.keys())
@pytest.mark.parametrize("platform", PLATFORMS)
@patch("homeassistant.components.onewire.onewirehub.protocol.proxy")
async def test_owserver_setup_valid_device(owproxy, hass, device_id, platform):
"""Test for 1-Wire device.
As they would be on a clean setup: all binary-sensors and switches disabled.
"""
await async_setup_component(hass, "persistent_notification", {})
entity_registry = mock_registry(hass)
device_registry = mock_device_registry(hass)
setup_owproxy_mock_devices(owproxy, platform, [device_id])
mock_device = MOCK_OWPROXY_DEVICES[device_id]
expected_entities = mock_device.get(platform, [])
with patch("homeassistant.components.onewire.PLATFORMS", [platform]):
await setup_onewire_patched_owserver_integration(hass)
await hass.async_block_till_done()
assert len(entity_registry.entities) == len(expected_entities)
if len(expected_entities) > 0:
device_info = mock_device["device_info"]
assert len(device_registry.devices) == 1
registry_entry = device_registry.async_get_device({(DOMAIN, device_id)})
assert registry_entry is not None
assert registry_entry.identifiers == {(DOMAIN, device_id)}
assert registry_entry.manufacturer == device_info["manufacturer"]
assert registry_entry.name == device_info["name"]
assert registry_entry.model == device_info["model"]
for expected_entity in expected_entities:
entity_id = expected_entity["entity_id"]
registry_entry = entity_registry.entities.get(entity_id)
assert registry_entry is not None
assert registry_entry.unique_id == expected_entity["unique_id"]
assert registry_entry.unit_of_measurement == expected_entity["unit"]
assert registry_entry.device_class == expected_entity["class"]
assert registry_entry.disabled == expected_entity.get("disabled", False)
state = hass.states.get(entity_id)
if registry_entry.disabled:
assert state is None
else:
assert state.state == expected_entity["result"]
assert state.attributes["device_file"] == expected_entity.get(
"device_file", registry_entry.unique_id
)
@pytest.mark.parametrize("device_id", MOCK_SYSBUS_DEVICES.keys())
async def test_onewiredirect_setup_valid_device(hass, device_id):
"""Test that sysbus config entry works correctly."""
entity_registry = mock_registry(hass)
device_registry = mock_device_registry(hass)
mock_device_sensor = MOCK_SYSBUS_DEVICES[device_id]
glob_result = [f"/{DEFAULT_SYSBUS_MOUNT_DIR}/{device_id}"]
read_side_effect = []
expected_sensors = mock_device_sensor["sensors"]
for expected_sensor in expected_sensors:
read_side_effect.append(expected_sensor["injected_value"])
# Ensure enough read side effect
read_side_effect.extend([FileNotFoundError("Missing injected value")] * 20)
with patch(
"homeassistant.components.onewire.onewirehub.os.path.isdir", return_value=True
), patch("pi1wire._finder.glob.glob", return_value=glob_result,), patch(
"pi1wire.OneWire.get_temperature",
side_effect=read_side_effect,
):
assert await async_setup_component(hass, SENSOR_DOMAIN, MOCK_SYSBUS_CONFIG)
await hass.async_block_till_done()
assert len(entity_registry.entities) == len(expected_sensors)
if len(expected_sensors) > 0:
device_info = mock_device_sensor["device_info"]
assert len(device_registry.devices) == 1
registry_entry = device_registry.async_get_device({(DOMAIN, device_id)})
assert registry_entry is not None
assert registry_entry.identifiers == {(DOMAIN, device_id)}
assert registry_entry.manufacturer == device_info["manufacturer"]
assert registry_entry.name == device_info["name"]
assert registry_entry.model == device_info["model"]
for expected_sensor in expected_sensors:
entity_id = expected_sensor["entity_id"]
registry_entry = entity_registry.entities.get(entity_id)
assert registry_entry is not None
assert registry_entry.unique_id == expected_sensor["unique_id"]
assert registry_entry.unit_of_measurement == expected_sensor["unit"]
assert registry_entry.device_class == expected_sensor["class"]
state = hass.states.get(entity_id)
assert state.state == expected_sensor["result"]

90
tests/components/onewire/test_switch.py
View File

@ -2,7 +2,6 @@
import copy
from unittest.mock import patch
from pyownet.protocol import Error as ProtocolError
import pytest
from homeassistant.components.onewire.switch import DEVICE_SWITCHES
@ -10,58 +9,19 @@ from homeassistant.components.switch import DOMAIN as SWITCH_DOMAIN
from homeassistant.const import ATTR_ENTITY_ID, SERVICE_TOGGLE, STATE_OFF, STATE_ON
from homeassistant.setup import async_setup_component
from . import setup_onewire_patched_owserver_integration
from . import setup_onewire_patched_owserver_integration, setup_owproxy_mock_devices
from .const import MOCK_OWPROXY_DEVICES
from tests.common import mock_registry
MOCK_DEVICE_SENSORS = {
"12.111111111111": {
"inject_reads": [
b"DS2406", # read device type
],
SWITCH_DOMAIN: [
{
"entity_id": "switch.12_111111111111_pio_a",
"unique_id": "/12.111111111111/PIO.A",
"injected_value": b" 1",
"result": STATE_ON,
"unit": None,
"class": None,
"disabled": True,
},
{
"entity_id": "switch.12_111111111111_pio_b",
"unique_id": "/12.111111111111/PIO.B",
"injected_value": b" 0",
"result": STATE_OFF,
"unit": None,
"class": None,
"disabled": True,
},
{
"entity_id": "switch.12_111111111111_latch_a",
"unique_id": "/12.111111111111/latch.A",
"injected_value": b" 1",
"result": STATE_ON,
"unit": None,
"class": None,
"disabled": True,
},
{
"entity_id": "switch.12_111111111111_latch_b",
"unique_id": "/12.111111111111/latch.B",
"injected_value": b" 0",
"result": STATE_OFF,
"unit": None,
"class": None,
"disabled": True,
},
],
}
MOCK_SWITCHES = {
key: value
for (key, value) in MOCK_OWPROXY_DEVICES.items()
if SWITCH_DOMAIN in value
}
@pytest.mark.parametrize("device_id", ["12.111111111111"])
@pytest.mark.parametrize("device_id", MOCK_SWITCHES.keys())
@patch("homeassistant.components.onewire.onewirehub.protocol.proxy")
async def test_owserver_switch(owproxy, hass, device_id):
"""Test for 1-Wire switch.
@ -71,26 +31,14 @@ async def test_owserver_switch(owproxy, hass, device_id):
await async_setup_component(hass, "persistent_notification", {})
entity_registry = mock_registry(hass)
mock_device_sensor = MOCK_DEVICE_SENSORS[device_id]
setup_owproxy_mock_devices(owproxy, SWITCH_DOMAIN, [device_id])
device_family = device_id[0:2]
dir_return_value = [f"/{device_id}/"]
read_side_effect = [device_family.encode()]
if "inject_reads" in mock_device_sensor:
read_side_effect += mock_device_sensor["inject_reads"]
expected_sensors = mock_device_sensor[SWITCH_DOMAIN]
for expected_sensor in expected_sensors:
read_side_effect.append(expected_sensor["injected_value"])
# Ensure enough read side effect
read_side_effect.extend([ProtocolError("Missing injected value")] * 10)
owproxy.return_value.dir.return_value = dir_return_value
owproxy.return_value.read.side_effect = read_side_effect
mock_device = MOCK_SWITCHES[device_id]
expected_entities = mock_device[SWITCH_DOMAIN]
# Force enable switches
patch_device_switches = copy.deepcopy(DEVICE_SWITCHES)
for item in patch_device_switches[device_family]:
for item in patch_device_switches[device_id[0:2]]:
item["default_disabled"] = False
with patch(
@ -101,21 +49,21 @@ async def test_owserver_switch(owproxy, hass, device_id):
await setup_onewire_patched_owserver_integration(hass)
await hass.async_block_till_done()
assert len(entity_registry.entities) == len(expected_sensors)
assert len(entity_registry.entities) == len(expected_entities)
for expected_sensor in expected_sensors:
entity_id = expected_sensor["entity_id"]
for expected_entity in expected_entities:
entity_id = expected_entity["entity_id"]
registry_entry = entity_registry.entities.get(entity_id)
assert registry_entry is not None
state = hass.states.get(entity_id)
assert state.state == expected_sensor["result"]
assert state.state == expected_entity["result"]
if state.state == STATE_ON:
owproxy.return_value.read.side_effect = [b" 0"]
expected_sensor["result"] = STATE_OFF
expected_entity["result"] = STATE_OFF
elif state.state == STATE_OFF:
owproxy.return_value.read.side_effect = [b" 1"]
expected_sensor["result"] = STATE_ON
expected_entity["result"] = STATE_ON
await hass.services.async_call(
SWITCH_DOMAIN,
@ -126,7 +74,7 @@ async def test_owserver_switch(owproxy, hass, device_id):
await hass.async_block_till_done()
state = hass.states.get(entity_id)
assert state.state == expected_sensor["result"]
assert state.attributes["device_file"] == expected_sensor.get(
assert state.state == expected_entity["result"]
assert state.attributes["device_file"] == expected_entity.get(
"device_file", registry_entry.unique_id
)