Use constants in Onewire tests (#58017)

* Use ATTR_ENTITY_ID constant

* Add ATTR_UNIQUE_ID constant

* Add new attribute constants

* Fix missing ATTR_DEFAULT_DISABLED

Co-authored-by: epenet <epenet@users.noreply.github.com>
This commit is contained in:
epenet 2021-10-19 21:41:01 +02:00 committed by GitHub
parent ab0247d112
commit 1b0118a81b
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
6 changed files with 595 additions and 547 deletions

View file

@ -8,7 +8,7 @@ from pyownet.protocol import ProtocolError
from homeassistant.components.onewire.const import DEFAULT_SYSBUS_MOUNT_DIR from homeassistant.components.onewire.const import DEFAULT_SYSBUS_MOUNT_DIR
from .const import MOCK_OWPROXY_DEVICES, MOCK_SYSBUS_DEVICES from .const import ATTR_INJECT_READS, MOCK_OWPROXY_DEVICES, MOCK_SYSBUS_DEVICES
def setup_owproxy_mock_devices( def setup_owproxy_mock_devices(
@ -27,13 +27,13 @@ def setup_owproxy_mock_devices(
# Setup device reads # Setup device reads
main_read_side_effect += [device_id[0:2].encode()] main_read_side_effect += [device_id[0:2].encode()]
if "inject_reads" in mock_device: if ATTR_INJECT_READS in mock_device:
main_read_side_effect += mock_device["inject_reads"] main_read_side_effect += mock_device[ATTR_INJECT_READS]
# Setup sub-device reads # Setup sub-device reads
device_sensors = mock_device.get(platform, []) device_sensors = mock_device.get(platform, [])
for expected_sensor in device_sensors: for expected_sensor in device_sensors:
sub_read_side_effect.append(expected_sensor["injected_value"]) sub_read_side_effect.append(expected_sensor[ATTR_INJECT_READS])
# Ensure enough read side effect # Ensure enough read side effect
read_side_effect = ( read_side_effect = (
@ -61,10 +61,10 @@ def setup_sysbus_mock_devices(
# Setup sub-device reads # Setup sub-device reads
device_sensors = mock_device.get(platform, []) device_sensors = mock_device.get(platform, [])
for expected_sensor in device_sensors: for expected_sensor in device_sensors:
if isinstance(expected_sensor["injected_value"], list): if isinstance(expected_sensor[ATTR_INJECT_READS], list):
read_side_effect += expected_sensor["injected_value"] read_side_effect += expected_sensor[ATTR_INJECT_READS]
else: else:
read_side_effect.append(expected_sensor["injected_value"]) read_side_effect.append(expected_sensor[ATTR_INJECT_READS])
# Ensure enough read side effect # Ensure enough read side effect
read_side_effect.extend([FileNotFoundError("Missing injected value")] * 20) read_side_effect.extend([FileNotFoundError("Missing injected value")] * 20)

View file

@ -57,7 +57,7 @@ def get_sysbus_config_entry(hass: HomeAssistant) -> ConfigEntry:
data={ data={
CONF_TYPE: CONF_TYPE_SYSBUS, CONF_TYPE: CONF_TYPE_SYSBUS,
CONF_MOUNT_DIR: DEFAULT_SYSBUS_MOUNT_DIR, CONF_MOUNT_DIR: DEFAULT_SYSBUS_MOUNT_DIR,
"names": { CONF_NAMES: {
"10-111111111111": "My DS18B20", "10-111111111111": "My DS18B20",
}, },
}, },

File diff suppressed because it is too large Load diff

View file

@ -5,10 +5,16 @@ import pytest
from homeassistant.components.binary_sensor import DOMAIN as BINARY_SENSOR_DOMAIN from homeassistant.components.binary_sensor import DOMAIN as BINARY_SENSOR_DOMAIN
from homeassistant.config_entries import ConfigEntry from homeassistant.config_entries import ConfigEntry
from homeassistant.const import ATTR_ENTITY_ID, ATTR_STATE
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant
from . import setup_owproxy_mock_devices from . import setup_owproxy_mock_devices
from .const import ATTR_DEFAULT_DISABLED, MOCK_OWPROXY_DEVICES from .const import (
ATTR_DEFAULT_DISABLED,
ATTR_DEVICE_FILE,
ATTR_UNIQUE_ID,
MOCK_OWPROXY_DEVICES,
)
from tests.common import mock_registry from tests.common import mock_registry
@ -41,7 +47,7 @@ async def test_owserver_binary_sensor(
# Ensure all entities are enabled # Ensure all entities are enabled
for expected_entity in expected_entities: for expected_entity in expected_entities:
if expected_entity.get(ATTR_DEFAULT_DISABLED): if expected_entity.get(ATTR_DEFAULT_DISABLED):
entity_id = expected_entity["entity_id"] entity_id = expected_entity[ATTR_ENTITY_ID]
registry_entry = entity_registry.entities.get(entity_id) registry_entry = entity_registry.entities.get(entity_id)
assert registry_entry.disabled assert registry_entry.disabled
assert registry_entry.disabled_by == "integration" assert registry_entry.disabled_by == "integration"
@ -52,11 +58,12 @@ async def test_owserver_binary_sensor(
await hass.async_block_till_done() await hass.async_block_till_done()
for expected_entity in expected_entities: for expected_entity in expected_entities:
entity_id = expected_entity["entity_id"] entity_id = expected_entity[ATTR_ENTITY_ID]
registry_entry = entity_registry.entities.get(entity_id) registry_entry = entity_registry.entities.get(entity_id)
assert registry_entry is not None assert registry_entry is not None
assert registry_entry.unique_id == expected_entity[ATTR_UNIQUE_ID]
state = hass.states.get(entity_id) state = hass.states.get(entity_id)
assert state.state == expected_entity["result"] assert state.state == expected_entity[ATTR_STATE]
assert state.attributes["device_file"] == expected_entity.get( assert state.attributes[ATTR_DEVICE_FILE] == expected_entity.get(
"device_file", registry_entry.unique_id ATTR_DEVICE_FILE, registry_entry.unique_id
) )

View file

@ -9,16 +9,26 @@ from homeassistant.components.sensor import ATTR_STATE_CLASS, DOMAIN as SENSOR_D
from homeassistant.config_entries import ConfigEntry from homeassistant.config_entries import ConfigEntry
from homeassistant.const import ( from homeassistant.const import (
ATTR_DEVICE_CLASS, ATTR_DEVICE_CLASS,
ATTR_ENTITY_ID,
ATTR_MANUFACTURER, ATTR_MANUFACTURER,
ATTR_MODEL, ATTR_MODEL,
ATTR_NAME, ATTR_NAME,
ATTR_STATE,
ATTR_UNIT_OF_MEASUREMENT, ATTR_UNIT_OF_MEASUREMENT,
) )
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant
from homeassistant.setup import async_setup_component from homeassistant.setup import async_setup_component
from . import setup_owproxy_mock_devices, setup_sysbus_mock_devices from . import setup_owproxy_mock_devices, setup_sysbus_mock_devices
from .const import ATTR_DEFAULT_DISABLED, MOCK_OWPROXY_DEVICES, MOCK_SYSBUS_DEVICES from .const import (
ATTR_DEFAULT_DISABLED,
ATTR_DEVICE_FILE,
ATTR_DEVICE_INFO,
ATTR_INJECT_READS,
ATTR_UNIQUE_ID,
MOCK_OWPROXY_DEVICES,
MOCK_SYSBUS_DEVICES,
)
from tests.common import assert_setup_component, mock_device_registry, mock_registry from tests.common import assert_setup_component, mock_device_registry, mock_registry
@ -86,10 +96,10 @@ async def test_sensors_on_owserver_coupler(
dir_side_effect.append([f"/{device_id}/"]) # dir on root dir_side_effect.append([f"/{device_id}/"]) # dir on root
read_side_effect.append(device_id[0:2].encode()) # read family on root read_side_effect.append(device_id[0:2].encode()) # read family on root
if "inject_reads" in mock_coupler: if ATTR_INJECT_READS in mock_coupler:
read_side_effect += mock_coupler["inject_reads"] read_side_effect += mock_coupler[ATTR_INJECT_READS]
expected_sensors = [] expected_entities = []
for branch, branch_details in mock_coupler["branches"].items(): for branch, branch_details in mock_coupler["branches"].items():
dir_side_effect.append( dir_side_effect.append(
[ # dir on branch [ # dir on branch
@ -100,12 +110,12 @@ async def test_sensors_on_owserver_coupler(
for sub_device_id, sub_device in branch_details.items(): for sub_device_id, sub_device in branch_details.items():
read_side_effect.append(sub_device_id[0:2].encode()) read_side_effect.append(sub_device_id[0:2].encode())
if "inject_reads" in sub_device: if ATTR_INJECT_READS in sub_device:
read_side_effect.extend(sub_device["inject_reads"]) read_side_effect.extend(sub_device[ATTR_INJECT_READS])
expected_sensors += sub_device[SENSOR_DOMAIN] expected_entities += sub_device[SENSOR_DOMAIN]
for expected_sensor in sub_device[SENSOR_DOMAIN]: for expected_entity in sub_device[SENSOR_DOMAIN]:
read_side_effect.append(expected_sensor["injected_value"]) read_side_effect.append(expected_entity[ATTR_INJECT_READS])
# Ensure enough read side effect # Ensure enough read side effect
read_side_effect.extend([ProtocolError("Missing injected value")] * 10) read_side_effect.extend([ProtocolError("Missing injected value")] * 10)
@ -115,18 +125,18 @@ async def test_sensors_on_owserver_coupler(
await hass.config_entries.async_setup(config_entry.entry_id) await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done() 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: for expected_entity in expected_entities:
entity_id = expected_sensor["entity_id"] entity_id = expected_entity[ATTR_ENTITY_ID]
registry_entry = entity_registry.entities.get(entity_id) registry_entry = entity_registry.entities.get(entity_id)
assert registry_entry is not None assert registry_entry is not None
assert registry_entry.unique_id == expected_sensor["unique_id"] assert registry_entry.unique_id == expected_entity[ATTR_UNIQUE_ID]
state = hass.states.get(entity_id) state = hass.states.get(entity_id)
assert state.state == expected_sensor["result"] assert state.state == expected_entity[ATTR_STATE]
for attr in (ATTR_DEVICE_CLASS, ATTR_STATE_CLASS, ATTR_UNIT_OF_MEASUREMENT): for attr in (ATTR_DEVICE_CLASS, ATTR_STATE_CLASS, ATTR_UNIT_OF_MEASUREMENT):
assert state.attributes.get(attr) == expected_sensor[attr] assert state.attributes.get(attr) == expected_entity[attr]
assert state.attributes["device_file"] == expected_sensor["device_file"] assert state.attributes[ATTR_DEVICE_FILE] == expected_entity[ATTR_DEVICE_FILE]
async def test_owserver_setup_valid_device( async def test_owserver_setup_valid_device(
@ -151,7 +161,7 @@ async def test_owserver_setup_valid_device(
# Ensure all entities are enabled # Ensure all entities are enabled
for expected_entity in expected_entities: for expected_entity in expected_entities:
if expected_entity.get(ATTR_DEFAULT_DISABLED): if expected_entity.get(ATTR_DEFAULT_DISABLED):
entity_id = expected_entity["entity_id"] entity_id = expected_entity[ATTR_ENTITY_ID]
registry_entry = entity_registry.entities.get(entity_id) registry_entry = entity_registry.entities.get(entity_id)
assert registry_entry.disabled assert registry_entry.disabled
assert registry_entry.disabled_by == "integration" assert registry_entry.disabled_by == "integration"
@ -162,7 +172,7 @@ async def test_owserver_setup_valid_device(
await hass.async_block_till_done() await hass.async_block_till_done()
if len(expected_entities) > 0: if len(expected_entities) > 0:
device_info = mock_device["device_info"] device_info = mock_device[ATTR_DEVICE_INFO]
assert len(device_registry.devices) == 1 assert len(device_registry.devices) == 1
registry_entry = device_registry.async_get_device({(DOMAIN, device_id)}) registry_entry = device_registry.async_get_device({(DOMAIN, device_id)})
assert registry_entry is not None assert registry_entry is not None
@ -172,16 +182,16 @@ async def test_owserver_setup_valid_device(
assert registry_entry.model == device_info[ATTR_MODEL] assert registry_entry.model == device_info[ATTR_MODEL]
for expected_entity in expected_entities: for expected_entity in expected_entities:
entity_id = expected_entity["entity_id"] entity_id = expected_entity[ATTR_ENTITY_ID]
registry_entry = entity_registry.entities.get(entity_id) registry_entry = entity_registry.entities.get(entity_id)
assert registry_entry is not None assert registry_entry is not None
assert registry_entry.unique_id == expected_entity["unique_id"] assert registry_entry.unique_id == expected_entity[ATTR_UNIQUE_ID]
state = hass.states.get(entity_id) state = hass.states.get(entity_id)
assert state.state == expected_entity["result"] assert state.state == expected_entity[ATTR_STATE]
for attr in (ATTR_DEVICE_CLASS, ATTR_STATE_CLASS, ATTR_UNIT_OF_MEASUREMENT): for attr in (ATTR_DEVICE_CLASS, ATTR_STATE_CLASS, ATTR_UNIT_OF_MEASUREMENT):
assert state.attributes.get(attr) == expected_entity[attr] assert state.attributes.get(attr) == expected_entity[attr]
assert state.attributes["device_file"] == expected_entity.get( assert state.attributes[ATTR_DEVICE_FILE] == expected_entity.get(
"device_file", registry_entry.unique_id ATTR_DEVICE_FILE, registry_entry.unique_id
) )
@ -212,7 +222,7 @@ async def test_onewiredirect_setup_valid_device(
assert len(entity_registry.entities) == len(expected_entities) assert len(entity_registry.entities) == len(expected_entities)
if len(expected_entities) > 0: if len(expected_entities) > 0:
device_info = mock_device["device_info"] device_info = mock_device[ATTR_DEVICE_INFO]
assert len(device_registry.devices) == 1 assert len(device_registry.devices) == 1
registry_entry = device_registry.async_get_device({(DOMAIN, device_id)}) registry_entry = device_registry.async_get_device({(DOMAIN, device_id)})
assert registry_entry is not None assert registry_entry is not None
@ -221,12 +231,12 @@ async def test_onewiredirect_setup_valid_device(
assert registry_entry.name == device_info[ATTR_NAME] assert registry_entry.name == device_info[ATTR_NAME]
assert registry_entry.model == device_info[ATTR_MODEL] assert registry_entry.model == device_info[ATTR_MODEL]
for expected_sensor in expected_entities: for expected_entity in expected_entities:
entity_id = expected_sensor["entity_id"] entity_id = expected_entity[ATTR_ENTITY_ID]
registry_entry = entity_registry.entities.get(entity_id) registry_entry = entity_registry.entities.get(entity_id)
assert registry_entry is not None assert registry_entry is not None
assert registry_entry.unique_id == expected_sensor["unique_id"] assert registry_entry.unique_id == expected_entity[ATTR_UNIQUE_ID]
state = hass.states.get(entity_id) state = hass.states.get(entity_id)
assert state.state == expected_sensor["result"] assert state.state == expected_entity[ATTR_STATE]
for attr in (ATTR_DEVICE_CLASS, ATTR_STATE_CLASS, ATTR_UNIT_OF_MEASUREMENT): for attr in (ATTR_DEVICE_CLASS, ATTR_STATE_CLASS, ATTR_UNIT_OF_MEASUREMENT):
assert state.attributes.get(attr) == expected_sensor[attr] assert state.attributes.get(attr) == expected_entity[attr]

View file

@ -5,11 +5,22 @@ import pytest
from homeassistant.components.switch import DOMAIN as SWITCH_DOMAIN from homeassistant.components.switch import DOMAIN as SWITCH_DOMAIN
from homeassistant.config_entries import ConfigEntry from homeassistant.config_entries import ConfigEntry
from homeassistant.const import ATTR_ENTITY_ID, SERVICE_TOGGLE, STATE_OFF, STATE_ON from homeassistant.const import (
ATTR_ENTITY_ID,
ATTR_STATE,
SERVICE_TOGGLE,
STATE_OFF,
STATE_ON,
)
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant
from . import setup_owproxy_mock_devices from . import setup_owproxy_mock_devices
from .const import ATTR_DEFAULT_DISABLED, MOCK_OWPROXY_DEVICES from .const import (
ATTR_DEFAULT_DISABLED,
ATTR_DEVICE_FILE,
ATTR_UNIQUE_ID,
MOCK_OWPROXY_DEVICES,
)
from tests.common import mock_registry from tests.common import mock_registry
@ -42,7 +53,7 @@ async def test_owserver_switch(
# Ensure all entities are enabled # Ensure all entities are enabled
for expected_entity in expected_entities: for expected_entity in expected_entities:
if expected_entity.get(ATTR_DEFAULT_DISABLED): if expected_entity.get(ATTR_DEFAULT_DISABLED):
entity_id = expected_entity["entity_id"] entity_id = expected_entity[ATTR_ENTITY_ID]
registry_entry = entity_registry.entities.get(entity_id) registry_entry = entity_registry.entities.get(entity_id)
assert registry_entry.disabled assert registry_entry.disabled
assert registry_entry.disabled_by == "integration" assert registry_entry.disabled_by == "integration"
@ -53,18 +64,19 @@ async def test_owserver_switch(
await hass.async_block_till_done() await hass.async_block_till_done()
for expected_entity in expected_entities: for expected_entity in expected_entities:
entity_id = expected_entity["entity_id"] entity_id = expected_entity[ATTR_ENTITY_ID]
registry_entry = entity_registry.entities.get(entity_id) registry_entry = entity_registry.entities.get(entity_id)
assert registry_entry is not None assert registry_entry is not None
assert registry_entry.unique_id == expected_entity[ATTR_UNIQUE_ID]
state = hass.states.get(entity_id) state = hass.states.get(entity_id)
assert state.state == expected_entity["result"] assert state.state == expected_entity[ATTR_STATE]
if state.state == STATE_ON: if state.state == STATE_ON:
owproxy.return_value.read.side_effect = [b" 0"] owproxy.return_value.read.side_effect = [b" 0"]
expected_entity["result"] = STATE_OFF expected_entity[ATTR_STATE] = STATE_OFF
elif state.state == STATE_OFF: elif state.state == STATE_OFF:
owproxy.return_value.read.side_effect = [b" 1"] owproxy.return_value.read.side_effect = [b" 1"]
expected_entity["result"] = STATE_ON expected_entity[ATTR_STATE] = STATE_ON
await hass.services.async_call( await hass.services.async_call(
SWITCH_DOMAIN, SWITCH_DOMAIN,
@ -75,7 +87,7 @@ async def test_owserver_switch(
await hass.async_block_till_done() await hass.async_block_till_done()
state = hass.states.get(entity_id) state = hass.states.get(entity_id)
assert state.state == expected_entity["result"] assert state.state == expected_entity[ATTR_STATE]
assert state.attributes["device_file"] == expected_entity.get( assert state.attributes[ATTR_DEVICE_FILE] == expected_entity.get(
"device_file", registry_entry.unique_id ATTR_DEVICE_FILE, registry_entry.unique_id
) )