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

Refactor Shelly tests - do not access hass.data (#79606)

pull/79628/head
Shay Levy 2022-10-05 12:36:06 +03:00 committed by GitHub
parent 9dd9147343
commit 723b018b63
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
9 changed files with 245 additions and 427 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

25
tests/components/shelly/__init__.py
View File

@ -1 +1,26 @@
"""Tests for the Shelly integration."""
from homeassistant.components.shelly.const import CONF_SLEEP_PERIOD, DOMAIN
from homeassistant.const import CONF_HOST
from homeassistant.core import HomeAssistant
from tests.common import MockConfigEntry
async def init_integration(
hass: HomeAssistant, gen: int, model="SHSW-25"
) -> MockConfigEntry:
"""Set up the Shelly integration in Home Assistant."""
data = {
CONF_HOST: "192.168.1.37",
CONF_SLEEP_PERIOD: 0,
"model": model,
"gen": gen,
}
entry = MockConfigEntry(domain=DOMAIN, data=data, unique_id=DOMAIN)
entry.add_to_hass(hass)
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
return entry

81
tests/components/shelly/conftest.py
View File

@ -3,30 +3,12 @@ from unittest.mock import AsyncMock, Mock, patch
import pytest
from homeassistant.components.shelly import (
BlockDeviceWrapper,
RpcDeviceWrapper,
RpcPollingWrapper,
ShellyDeviceRestWrapper,
)
from homeassistant.components.shelly.const import (
BLOCK,
DATA_CONFIG_ENTRY,
DOMAIN,
EVENT_SHELLY_CLICK,
REST,
REST_SENSORS_UPDATE_INTERVAL,
RPC,
RPC_POLL,
)
from homeassistant.setup import async_setup_component
from tests.common import (
MockConfigEntry,
async_capture_events,
async_mock_service,
mock_device_registry,
)
from tests.common import async_capture_events, async_mock_service, mock_device_registry
MOCK_SETTINGS = {
"name": "Test name",
@ -144,16 +126,14 @@ def events(hass):
@pytest.fixture
async def coap_wrapper(hass):
"""Setups a coap wrapper with mocked device."""
await async_setup_component(hass, "shelly", {})
async def mock_block_device():
"""Mock block (Gen1, CoAP) device."""
with patch("homeassistant.components.shelly.utils.COAP", autospec=True), patch(
"aioshelly.block_device.BlockDevice.create"
) as block_device_mock:
config_entry = MockConfigEntry(
domain=DOMAIN,
data={"sleep_period": 0, "model": "SHSW-25", "host": "1.2.3.4"},
unique_id="12345678",
)
config_entry.add_to_hass(hass)
def update():
block_device_mock.return_value.subscribe_updates.call_args[0][0]({})
device = Mock(
blocks=MOCK_BLOCKS,
@ -165,35 +145,22 @@ async def coap_wrapper(hass):
update_status=AsyncMock(),
trigger_ota_update=AsyncMock(),
trigger_reboot=AsyncMock(),
initialize=AsyncMock(),
initialized=True,
)
block_device_mock.return_value = device
block_device_mock.return_value.mock_update = Mock(side_effect=update)
hass.data[DOMAIN] = {DATA_CONFIG_ENTRY: {}}
hass.data[DOMAIN][DATA_CONFIG_ENTRY][config_entry.entry_id] = {}
hass.data[DOMAIN][DATA_CONFIG_ENTRY][config_entry.entry_id][
REST
] = ShellyDeviceRestWrapper(hass, device, config_entry)
wrapper = hass.data[DOMAIN][DATA_CONFIG_ENTRY][config_entry.entry_id][
BLOCK
] = BlockDeviceWrapper(hass, config_entry, device)
wrapper.async_setup()
return wrapper
yield block_device_mock.return_value
@pytest.fixture
async def rpc_wrapper(hass):
"""Setups a rpc wrapper with mocked device."""
await async_setup_component(hass, "shelly", {})
async def mock_rpc_device():
"""Mock rpc (Gen2, Websocket) device."""
with patch("aioshelly.rpc_device.RpcDevice.create") as rpc_device_mock:
config_entry = MockConfigEntry(
domain=DOMAIN,
data={"sleep_period": 0, "model": "SNSW-001P16EU", "gen": 2, "host": "1.2.3.4"},
unique_id="12345678",
)
config_entry.add_to_hass(hass)
def update():
rpc_device_mock.return_value.subscribe_updates.call_args[0][0]({})
device = Mock(
call_rpc=AsyncMock(),
@ -209,15 +176,7 @@ async def rpc_wrapper(hass):
shutdown=AsyncMock(),
)
hass.data[DOMAIN] = {DATA_CONFIG_ENTRY: {}}
hass.data[DOMAIN][DATA_CONFIG_ENTRY][config_entry.entry_id] = {}
hass.data[DOMAIN][DATA_CONFIG_ENTRY][config_entry.entry_id][
RPC_POLL
] = RpcPollingWrapper(hass, config_entry, device)
rpc_device_mock.return_value = device
rpc_device_mock.return_value.mock_update = Mock(side_effect=update)
wrapper = hass.data[DOMAIN][DATA_CONFIG_ENTRY][config_entry.entry_id][
RPC
] = RpcDeviceWrapper(hass, config_entry, device)
wrapper.async_setup()
return wrapper
yield rpc_device_mock.return_value

55
tests/components/shelly/test_button.py
View File

@ -1,33 +1,17 @@
"""Tests for Shelly button platform."""
from homeassistant.components.button import DOMAIN as BUTTON_DOMAIN, SERVICE_PRESS
from homeassistant.components.shelly.const import DOMAIN
from homeassistant.const import ATTR_ENTITY_ID, STATE_UNKNOWN
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_registry import async_get
from . import init_integration
async def test_block_button(hass: HomeAssistant, coap_wrapper):
async def test_block_button(hass: HomeAssistant, mock_block_device):
"""Test block device reboot button."""
assert coap_wrapper
entity_registry = async_get(hass)
entity_registry.async_get_or_create(
BUTTON_DOMAIN,
DOMAIN,
"test_name_reboot",
suggested_object_id="test_name_reboot",
disabled_by=None,
)
hass.async_create_task(
hass.config_entries.async_forward_entry_setup(coap_wrapper.entry, BUTTON_DOMAIN)
)
await hass.async_block_till_done()
await init_integration(hass, 1)
# reboot button
state = hass.states.get("button.test_name_reboot")
assert state
assert state.state == STATE_UNKNOWN
assert hass.states.get("button.test_name_reboot").state == STATE_UNKNOWN
await hass.services.async_call(
BUTTON_DOMAIN,
@ -35,33 +19,15 @@ async def test_block_button(hass: HomeAssistant, coap_wrapper):
{ATTR_ENTITY_ID: "button.test_name_reboot"},
blocking=True,
)
await hass.async_block_till_done()
assert coap_wrapper.device.trigger_reboot.call_count == 1
assert mock_block_device.trigger_reboot.call_count == 1
async def test_rpc_button(hass: HomeAssistant, rpc_wrapper):
async def test_rpc_button(hass: HomeAssistant, mock_rpc_device):
"""Test rpc device OTA button."""
assert rpc_wrapper
entity_registry = async_get(hass)
entity_registry.async_get_or_create(
BUTTON_DOMAIN,
DOMAIN,
"test_name_reboot",
suggested_object_id="test_name_reboot",
disabled_by=None,
)
hass.async_create_task(
hass.config_entries.async_forward_entry_setup(rpc_wrapper.entry, BUTTON_DOMAIN)
)
await hass.async_block_till_done()
await init_integration(hass, 2)
# reboot button
state = hass.states.get("button.test_name_reboot")
assert state
assert state.state == STATE_UNKNOWN
assert hass.states.get("button.test_name_reboot").state == STATE_UNKNOWN
await hass.services.async_call(
BUTTON_DOMAIN,
@ -69,5 +35,4 @@ async def test_rpc_button(hass: HomeAssistant, rpc_wrapper):
{ATTR_ENTITY_ID: "button.test_name_reboot"},
blocking=True,
)
await hass.async_block_till_done()
assert rpc_wrapper.device.trigger_reboot.call_count == 1
assert mock_rpc_device.trigger_reboot.call_count == 1

110
tests/components/shelly/test_cover.py
View File

@ -13,20 +13,16 @@ from homeassistant.components.cover import (
STATE_OPENING,
)
from homeassistant.const import ATTR_ENTITY_ID
from homeassistant.helpers.entity_component import async_update_entity
from . import init_integration
ROLLER_BLOCK_ID = 1
async def test_block_device_services(hass, coap_wrapper, monkeypatch):
async def test_block_device_services(hass, mock_block_device, monkeypatch):
"""Test block device cover services."""
assert coap_wrapper
monkeypatch.setitem(coap_wrapper.device.settings, "mode", "roller")
hass.async_create_task(
hass.config_entries.async_forward_entry_setup(coap_wrapper.entry, COVER_DOMAIN)
)
await hass.async_block_till_done()
monkeypatch.setitem(mock_block_device.settings, "mode", "roller")
await init_integration(hass, 1)
await hass.services.async_call(
COVER_DOMAIN,
@ -62,46 +58,28 @@ async def test_block_device_services(hass, coap_wrapper, monkeypatch):
assert hass.states.get("cover.test_name").state == STATE_CLOSED
async def test_block_device_update(hass, coap_wrapper, monkeypatch):
async def test_block_device_update(hass, mock_block_device, monkeypatch):
"""Test block device update."""
assert coap_wrapper
monkeypatch.setattr(mock_block_device.blocks[ROLLER_BLOCK_ID], "rollerPos", 0)
await init_integration(hass, 1)
hass.async_create_task(
hass.config_entries.async_forward_entry_setup(coap_wrapper.entry, COVER_DOMAIN)
)
await hass.async_block_till_done()
monkeypatch.setattr(coap_wrapper.device.blocks[ROLLER_BLOCK_ID], "rollerPos", 0)
await async_update_entity(hass, "cover.test_name")
await hass.async_block_till_done()
assert hass.states.get("cover.test_name").state == STATE_CLOSED
monkeypatch.setattr(coap_wrapper.device.blocks[ROLLER_BLOCK_ID], "rollerPos", 100)
await async_update_entity(hass, "cover.test_name")
await hass.async_block_till_done()
monkeypatch.setattr(mock_block_device.blocks[ROLLER_BLOCK_ID], "rollerPos", 100)
mock_block_device.mock_update()
assert hass.states.get("cover.test_name").state == STATE_OPEN
async def test_block_device_no_roller_blocks(hass, coap_wrapper, monkeypatch):
async def test_block_device_no_roller_blocks(hass, mock_block_device, monkeypatch):
"""Test block device without roller blocks."""
assert coap_wrapper
monkeypatch.setattr(coap_wrapper.device.blocks[ROLLER_BLOCK_ID], "type", None)
hass.async_create_task(
hass.config_entries.async_forward_entry_setup(coap_wrapper.entry, COVER_DOMAIN)
)
await hass.async_block_till_done()
monkeypatch.setattr(mock_block_device.blocks[ROLLER_BLOCK_ID], "type", None)
await init_integration(hass, 1)
assert hass.states.get("cover.test_name") is None
async def test_rpc_device_services(hass, rpc_wrapper, monkeypatch):
async def test_rpc_device_services(hass, mock_rpc_device, monkeypatch):
"""Test RPC device cover services."""
assert rpc_wrapper
hass.async_create_task(
hass.config_entries.async_forward_entry_setup(rpc_wrapper.entry, COVER_DOMAIN)
)
await hass.async_block_till_done()
await init_integration(hass, 2)
await hass.services.async_call(
COVER_DOMAIN,
@ -112,81 +90,57 @@ async def test_rpc_device_services(hass, rpc_wrapper, monkeypatch):
state = hass.states.get("cover.test_cover_0")
assert state.attributes[ATTR_CURRENT_POSITION] == 50
monkeypatch.setitem(rpc_wrapper.device.status["cover:0"], "state", "opening")
monkeypatch.setitem(mock_rpc_device.status["cover:0"], "state", "opening")
await hass.services.async_call(
COVER_DOMAIN,
SERVICE_OPEN_COVER,
{ATTR_ENTITY_ID: "cover.test_cover_0"},
blocking=True,
)
rpc_wrapper.async_set_updated_data("")
mock_rpc_device.mock_update()
assert hass.states.get("cover.test_cover_0").state == STATE_OPENING
monkeypatch.setitem(rpc_wrapper.device.status["cover:0"], "state", "closing")
monkeypatch.setitem(mock_rpc_device.status["cover:0"], "state", "closing")
await hass.services.async_call(
COVER_DOMAIN,
SERVICE_CLOSE_COVER,
{ATTR_ENTITY_ID: "cover.test_cover_0"},
blocking=True,
)
rpc_wrapper.async_set_updated_data("")
mock_rpc_device.mock_update()
assert hass.states.get("cover.test_cover_0").state == STATE_CLOSING
monkeypatch.setitem(rpc_wrapper.device.status["cover:0"], "state", "closed")
monkeypatch.setitem(mock_rpc_device.status["cover:0"], "state", "closed")
await hass.services.async_call(
COVER_DOMAIN,
SERVICE_STOP_COVER,
{ATTR_ENTITY_ID: "cover.test_cover_0"},
blocking=True,
)
rpc_wrapper.async_set_updated_data("")
mock_rpc_device.mock_update()
assert hass.states.get("cover.test_cover_0").state == STATE_CLOSED
async def test_rpc_device_no_cover_keys(hass, rpc_wrapper, monkeypatch):
async def test_rpc_device_no_cover_keys(hass, mock_rpc_device, monkeypatch):
"""Test RPC device without cover keys."""
assert rpc_wrapper
monkeypatch.delitem(rpc_wrapper.device.status, "cover:0")
hass.async_create_task(
hass.config_entries.async_forward_entry_setup(rpc_wrapper.entry, COVER_DOMAIN)
)
await hass.async_block_till_done()
monkeypatch.delitem(mock_rpc_device.status, "cover:0")
await init_integration(hass, 2)
assert hass.states.get("cover.test_cover_0") is None
async def test_rpc_device_update(hass, rpc_wrapper, monkeypatch):
async def test_rpc_device_update(hass, mock_rpc_device, monkeypatch):
"""Test RPC device update."""
assert rpc_wrapper
hass.async_create_task(
hass.config_entries.async_forward_entry_setup(rpc_wrapper.entry, COVER_DOMAIN)
)
await hass.async_block_till_done()
monkeypatch.setitem(rpc_wrapper.device.status["cover:0"], "state", "closed")
await async_update_entity(hass, "cover.test_cover_0")
await hass.async_block_till_done()
monkeypatch.setitem(mock_rpc_device.status["cover:0"], "state", "closed")
await init_integration(hass, 2)
assert hass.states.get("cover.test_cover_0").state == STATE_CLOSED
monkeypatch.setitem(rpc_wrapper.device.status["cover:0"], "state", "open")
await async_update_entity(hass, "cover.test_cover_0")
await hass.async_block_till_done()
monkeypatch.setitem(mock_rpc_device.status["cover:0"], "state", "open")
mock_rpc_device.mock_update()
assert hass.states.get("cover.test_cover_0").state == STATE_OPEN
async def test_rpc_device_no_position_control(hass, rpc_wrapper, monkeypatch):
async def test_rpc_device_no_position_control(hass, mock_rpc_device, monkeypatch):
"""Test RPC device with no position control."""
assert rpc_wrapper
monkeypatch.setitem(rpc_wrapper.device.status["cover:0"], "pos_control", False)
hass.async_create_task(
hass.config_entries.async_forward_entry_setup(rpc_wrapper.entry, COVER_DOMAIN)
)
await hass.async_block_till_done()
await async_update_entity(hass, "cover.test_cover_0")
await hass.async_block_till_done()
monkeypatch.setitem(mock_rpc_device.status["cover:0"], "pos_control", False)
await init_integration(hass, 2)
assert hass.states.get("cover.test_cover_0").state == STATE_OPEN

154
tests/components/shelly/test_device_trigger.py
View File

@ -1,6 +1,4 @@
"""The tests for Shelly device triggers."""
from unittest.mock import AsyncMock, Mock
import pytest
from homeassistant.components import automation
@ -8,20 +6,23 @@ from homeassistant.components.device_automation import DeviceAutomationType
from homeassistant.components.device_automation.exceptions import (
InvalidDeviceAutomationConfig,
)
from homeassistant.components.shelly import BlockDeviceWrapper
from homeassistant.components.shelly.const import (
ATTR_CHANNEL,
ATTR_CLICK_TYPE,
BLOCK,
CONF_SUBTYPE,
DATA_CONFIG_ENTRY,
DOMAIN,
EVENT_SHELLY_CLICK,
)
from homeassistant.const import CONF_DEVICE_ID, CONF_DOMAIN, CONF_PLATFORM, CONF_TYPE
from homeassistant.helpers import device_registry
from homeassistant.helpers.device_registry import (
async_entries_for_config_entry,
async_get as async_get_dev_reg,
)
from homeassistant.setup import async_setup_component
from . import init_integration
from tests.common import (
MockConfigEntry,
assert_lists_same,
@ -39,48 +40,52 @@ from tests.common import (
],
)
async def test_get_triggers_block_device(
hass, coap_wrapper, monkeypatch, button_type, is_valid
hass, mock_block_device, monkeypatch, button_type, is_valid
):
"""Test we get the expected triggers from a shelly block device."""
assert coap_wrapper
monkeypatch.setitem(
coap_wrapper.device.settings,
mock_block_device.settings,
"relays",
[
{"btn_type": button_type},
{"btn_type": "toggle"},
],
)
entry = await init_integration(hass, 1)
dev_reg = async_get_dev_reg(hass)
device = async_entries_for_config_entry(dev_reg, entry.entry_id)[0]
expected_triggers = []
if is_valid:
expected_triggers = [
{
CONF_PLATFORM: "device",
CONF_DEVICE_ID: coap_wrapper.device_id,
CONF_DEVICE_ID: device.id,
CONF_DOMAIN: DOMAIN,
CONF_TYPE: type,
CONF_TYPE: type_,
CONF_SUBTYPE: "button1",
"metadata": {},
}
for type in ["single", "long"]
for type_ in ["single", "long"]
]
triggers = await async_get_device_automations(
hass, DeviceAutomationType.TRIGGER, coap_wrapper.device_id
hass, DeviceAutomationType.TRIGGER, device.id
)
triggers = [value for value in triggers if value["domain"] == DOMAIN]
assert_lists_same(triggers, expected_triggers)
async def test_get_triggers_rpc_device(hass, rpc_wrapper):
async def test_get_triggers_rpc_device(hass, mock_rpc_device):
"""Test we get the expected triggers from a shelly RPC device."""
assert rpc_wrapper
entry = await init_integration(hass, 2)
dev_reg = async_get_dev_reg(hass)
device = async_entries_for_config_entry(dev_reg, entry.entry_id)[0]
expected_triggers = [
{
CONF_PLATFORM: "device",
CONF_DEVICE_ID: rpc_wrapper.device_id,
CONF_DEVICE_ID: device.id,
CONF_DOMAIN: DOMAIN,
CONF_TYPE: type,
CONF_SUBTYPE: "button1",
@ -90,43 +95,22 @@ async def test_get_triggers_rpc_device(hass, rpc_wrapper):
]
triggers = await async_get_device_automations(
hass, DeviceAutomationType.TRIGGER, rpc_wrapper.device_id
hass, DeviceAutomationType.TRIGGER, device.id
)
triggers = [value for value in triggers if value["domain"] == DOMAIN]
assert_lists_same(triggers, expected_triggers)
async def test_get_triggers_button(hass):
async def test_get_triggers_button(hass, mock_block_device):
"""Test we get the expected triggers from a shelly button."""
await async_setup_component(hass, "shelly", {})
config_entry = MockConfigEntry(
domain=DOMAIN,
data={"sleep_period": 43200, "model": "SHBTN-1", "host": "1.2.3.4"},
unique_id="12345678",
)
config_entry.add_to_hass(hass)
device = Mock(
blocks=None,
settings=None,
shelly=None,
update=AsyncMock(),
initialized=False,
)
hass.data[DOMAIN] = {DATA_CONFIG_ENTRY: {}}
hass.data[DOMAIN][DATA_CONFIG_ENTRY][config_entry.entry_id] = {}
coap_wrapper = hass.data[DOMAIN][DATA_CONFIG_ENTRY][config_entry.entry_id][
BLOCK
] = BlockDeviceWrapper(hass, config_entry, device)
coap_wrapper.async_setup()
entry = await init_integration(hass, 1, model="SHBTN-1")
dev_reg = async_get_dev_reg(hass)
device = async_entries_for_config_entry(dev_reg, entry.entry_id)[0]
expected_triggers = [
{
CONF_PLATFORM: "device",
CONF_DEVICE_ID: coap_wrapper.device_id,
CONF_DEVICE_ID: device.id,
CONF_DOMAIN: DOMAIN,
CONF_TYPE: type,
CONF_SUBTYPE: "button",
@ -136,51 +120,33 @@ async def test_get_triggers_button(hass):
]
triggers = await async_get_device_automations(
hass, DeviceAutomationType.TRIGGER, coap_wrapper.device_id
hass, DeviceAutomationType.TRIGGER, device.id
)
triggers = [value for value in triggers if value["domain"] == DOMAIN]
assert_lists_same(triggers, expected_triggers)
async def test_get_triggers_non_initialized_devices(hass):
async def test_get_triggers_non_initialized_devices(
hass, mock_block_device, monkeypatch
):
"""Test we get the empty triggers for non-initialized devices."""
await async_setup_component(hass, "shelly", {})
config_entry = MockConfigEntry(
domain=DOMAIN,
data={"sleep_period": 43200, "model": "SHDW-2", "host": "1.2.3.4"},
unique_id="12345678",
)
config_entry.add_to_hass(hass)
device = Mock(
blocks=None,
settings=None,
shelly=None,
update=AsyncMock(),
initialized=False,
)
hass.data[DOMAIN] = {DATA_CONFIG_ENTRY: {}}
hass.data[DOMAIN][DATA_CONFIG_ENTRY][config_entry.entry_id] = {}
coap_wrapper = hass.data[DOMAIN][DATA_CONFIG_ENTRY][config_entry.entry_id][
BLOCK
] = BlockDeviceWrapper(hass, config_entry, device)
coap_wrapper.async_setup()
monkeypatch.setattr(mock_block_device, "initialized", False)
entry = await init_integration(hass, 1)
dev_reg = async_get_dev_reg(hass)
device = async_entries_for_config_entry(dev_reg, entry.entry_id)[0]
expected_triggers = []
triggers = await async_get_device_automations(
hass, DeviceAutomationType.TRIGGER, coap_wrapper.device_id
hass, DeviceAutomationType.TRIGGER, device.id
)
triggers = [value for value in triggers if value["domain"] == DOMAIN]
assert_lists_same(triggers, expected_triggers)
async def test_get_triggers_for_invalid_device_id(hass, device_reg, coap_wrapper):
async def test_get_triggers_for_invalid_device_id(hass, device_reg, mock_block_device):
"""Test error raised for invalid shelly device_id."""
assert coap_wrapper
await init_integration(hass, 1)
config_entry = MockConfigEntry(domain=DOMAIN, data={})
config_entry.add_to_hass(hass)
invalid_device = device_reg.async_get_or_create(
@ -194,9 +160,11 @@ async def test_get_triggers_for_invalid_device_id(hass, device_reg, coap_wrapper
)
async def test_if_fires_on_click_event_block_device(hass, calls, coap_wrapper):
async def test_if_fires_on_click_event_block_device(hass, calls, mock_block_device):
"""Test for click_event trigger firing for block device."""
assert coap_wrapper
entry = await init_integration(hass, 1)
dev_reg = async_get_dev_reg(hass)
device = async_entries_for_config_entry(dev_reg, entry.entry_id)[0]
assert await async_setup_component(
hass,
@ -207,7 +175,7 @@ async def test_if_fires_on_click_event_block_device(hass, calls, coap_wrapper):
"trigger": {
CONF_PLATFORM: "device",
CONF_DOMAIN: DOMAIN,
CONF_DEVICE_ID: coap_wrapper.device_id,
CONF_DEVICE_ID: device.id,
CONF_TYPE: "single",
CONF_SUBTYPE: "button1",
},
@ -221,7 +189,7 @@ async def test_if_fires_on_click_event_block_device(hass, calls, coap_wrapper):
)
message = {
CONF_DEVICE_ID: coap_wrapper.device_id,
CONF_DEVICE_ID: device.id,
ATTR_CLICK_TYPE: "single",
ATTR_CHANNEL: 1,
}
@ -232,9 +200,11 @@ async def test_if_fires_on_click_event_block_device(hass, calls, coap_wrapper):
assert calls[0].data["some"] == "test_trigger_single_click"
async def test_if_fires_on_click_event_rpc_device(hass, calls, rpc_wrapper):
async def test_if_fires_on_click_event_rpc_device(hass, calls, mock_rpc_device):
"""Test for click_event trigger firing for rpc device."""
assert rpc_wrapper
entry = await init_integration(hass, 2)
dev_reg = async_get_dev_reg(hass)
device = async_entries_for_config_entry(dev_reg, entry.entry_id)[0]
assert await async_setup_component(
hass,
@ -245,7 +215,7 @@ async def test_if_fires_on_click_event_rpc_device(hass, calls, rpc_wrapper):
"trigger": {
CONF_PLATFORM: "device",
CONF_DOMAIN: DOMAIN,
CONF_DEVICE_ID: rpc_wrapper.device_id,
CONF_DEVICE_ID: device.id,
CONF_TYPE: "single_push",
CONF_SUBTYPE: "button1",
},
@ -259,7 +229,7 @@ async def test_if_fires_on_click_event_rpc_device(hass, calls, rpc_wrapper):
)
message = {
CONF_DEVICE_ID: rpc_wrapper.device_id,
CONF_DEVICE_ID: device.id,
ATTR_CLICK_TYPE: "single_push",
ATTR_CHANNEL: 1,
}
@ -270,9 +240,9 @@ async def test_if_fires_on_click_event_rpc_device(hass, calls, rpc_wrapper):
assert calls[0].data["some"] == "test_trigger_single_push"
async def test_validate_trigger_block_device_not_ready(hass, calls, coap_wrapper):
async def test_validate_trigger_block_device_not_ready(hass, calls, mock_block_device):
"""Test validate trigger config when block device is not ready."""
assert coap_wrapper
await init_integration(hass, 1)
assert await async_setup_component(
hass,
@ -307,10 +277,8 @@ async def test_validate_trigger_block_device_not_ready(hass, calls, coap_wrapper
assert calls[0].data["some"] == "test_trigger_single_click"
async def test_validate_trigger_rpc_device_not_ready(hass, calls, rpc_wrapper):
async def test_validate_trigger_rpc_device_not_ready(hass, calls, mock_rpc_device):
"""Test validate trigger config when RPC device is not ready."""
assert rpc_wrapper
assert await async_setup_component(
hass,
automation.DOMAIN,
@ -344,9 +312,11 @@ async def test_validate_trigger_rpc_device_not_ready(hass, calls, rpc_wrapper):
assert calls[0].data["some"] == "test_trigger_single_push"
async def test_validate_trigger_invalid_triggers(hass, coap_wrapper):
async def test_validate_trigger_invalid_triggers(hass, mock_block_device):
"""Test for click_event with invalid triggers."""
assert coap_wrapper
entry = await init_integration(hass, 1)
dev_reg = async_get_dev_reg(hass)
device = async_entries_for_config_entry(dev_reg, entry.entry_id)[0]
assert await async_setup_component(
hass,
@ -357,7 +327,7 @@ async def test_validate_trigger_invalid_triggers(hass, coap_wrapper):
"trigger": {
CONF_PLATFORM: "device",
CONF_DOMAIN: DOMAIN,
CONF_DEVICE_ID: coap_wrapper.device_id,
CONF_DEVICE_ID: device.id,
CONF_TYPE: "single",
CONF_SUBTYPE: "button3",
},

21
tests/components/shelly/test_diagnostics.py
View File

@ -6,6 +6,7 @@ from homeassistant.components.shelly.const import DOMAIN
from homeassistant.components.shelly.diagnostics import TO_REDACT
from homeassistant.core import HomeAssistant
from . import init_integration
from .conftest import MOCK_STATUS_COAP
from tests.components.diagnostics import get_diagnostics_for_config_entry
@ -14,10 +15,10 @@ RELAY_BLOCK_ID = 0
async def test_block_config_entry_diagnostics(
hass: HomeAssistant, hass_client: ClientSession, coap_wrapper
hass: HomeAssistant, hass_client: ClientSession, mock_block_device
):
"""Test config entry diagnostics for block device."""
assert coap_wrapper
await init_integration(hass, 1)
entry = hass.config_entries.async_entries(DOMAIN)[0]
entry_dict = entry.as_dict()
@ -30,9 +31,9 @@ async def test_block_config_entry_diagnostics(
assert result == {
"entry": entry_dict,
"device_info": {
"name": coap_wrapper.name,
"model": coap_wrapper.model,
"sw_version": coap_wrapper.sw_version,
"name": "Test name",
"model": "SHSW-25",
"sw_version": "some fw string",
},
"device_settings": {"coiot": {"update_period": 15}},
"device_status": MOCK_STATUS_COAP,
@ -42,10 +43,10 @@ async def test_block_config_entry_diagnostics(
async def test_rpc_config_entry_diagnostics(
hass: HomeAssistant,
hass_client: ClientSession,
rpc_wrapper,
mock_rpc_device,
):
"""Test config entry diagnostics for rpc device."""
assert rpc_wrapper
await init_integration(hass, 2)
entry = hass.config_entries.async_entries(DOMAIN)[0]
entry_dict = entry.as_dict()
@ -58,9 +59,9 @@ async def test_rpc_config_entry_diagnostics(
assert result == {
"entry": entry_dict,
"device_info": {
"name": rpc_wrapper.name,
"model": rpc_wrapper.model,
"sw_version": rpc_wrapper.sw_version,
"name": "Test name",
"model": "SHSW-25",
"sw_version": "some fw string",
},
"device_settings": {},
"device_status": {

24
tests/components/shelly/test_logbook.py
View File

@ -7,14 +7,23 @@ from homeassistant.components.shelly.const import (
EVENT_SHELLY_CLICK,
)
from homeassistant.const import ATTR_DEVICE_ID
from homeassistant.helpers.device_registry import (
async_entries_for_config_entry,
async_get as async_get_dev_reg,
)
from homeassistant.setup import async_setup_component
from . import init_integration
from tests.components.logbook.common import MockRow, mock_humanify
async def test_humanify_shelly_click_event_block_device(hass, coap_wrapper):
async def test_humanify_shelly_click_event_block_device(hass, mock_block_device):
"""Test humanifying Shelly click event for block device."""
assert coap_wrapper
entry = await init_integration(hass, 1)
dev_reg = async_get_dev_reg(hass)
device = async_entries_for_config_entry(dev_reg, entry.entry_id)[0]
hass.config.components.add("recorder")
assert await async_setup_component(hass, "logbook", {})
@ -24,7 +33,7 @@ async def test_humanify_shelly_click_event_block_device(hass, coap_wrapper):
MockRow(
EVENT_SHELLY_CLICK,
{
ATTR_DEVICE_ID: coap_wrapper.device_id,
ATTR_DEVICE_ID: device.id,
ATTR_DEVICE: "shellyix3-12345678",
ATTR_CLICK_TYPE: "single",
ATTR_CHANNEL: 1,
@ -57,9 +66,12 @@ async def test_humanify_shelly_click_event_block_device(hass, coap_wrapper):
)
async def test_humanify_shelly_click_event_rpc_device(hass, rpc_wrapper):
async def test_humanify_shelly_click_event_rpc_device(hass, mock_rpc_device):
"""Test humanifying Shelly click event for rpc device."""
assert rpc_wrapper
entry = await init_integration(hass, 2)
dev_reg = async_get_dev_reg(hass)
device = async_entries_for_config_entry(dev_reg, entry.entry_id)[0]
hass.config.components.add("recorder")
assert await async_setup_component(hass, "logbook", {})
@ -69,7 +81,7 @@ async def test_humanify_shelly_click_event_rpc_device(hass, rpc_wrapper):
MockRow(
EVENT_SHELLY_CLICK,
{
ATTR_DEVICE_ID: rpc_wrapper.device_id,
ATTR_DEVICE_ID: device.id,
ATTR_DEVICE: "shellyplus1pm-12345678",
ATTR_CLICK_TYPE: "single_push",
ATTR_CHANNEL: 1,

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

@ -7,19 +7,15 @@ from homeassistant.const import (
STATE_OFF,
STATE_ON,
)
from homeassistant.helpers.entity_component import async_update_entity
from . import init_integration
RELAY_BLOCK_ID = 0
async def test_block_device_services(hass, coap_wrapper):
async def test_block_device_services(hass, mock_block_device):
"""Test block device turn on/off services."""
assert coap_wrapper
hass.async_create_task(
hass.config_entries.async_forward_entry_setup(coap_wrapper.entry, SWITCH_DOMAIN)
)
await hass.async_block_till_done()
await init_integration(hass, 1)
await hass.services.async_call(
SWITCH_DOMAIN,
@ -38,72 +34,43 @@ async def test_block_device_services(hass, coap_wrapper):
assert hass.states.get("switch.test_name_channel_1").state == STATE_OFF
async def test_block_device_update(hass, coap_wrapper, monkeypatch):
async def test_block_device_update(hass, mock_block_device, monkeypatch):
"""Test block device update."""
assert coap_wrapper
hass.async_create_task(
hass.config_entries.async_forward_entry_setup(coap_wrapper.entry, SWITCH_DOMAIN)
)
await hass.async_block_till_done()
monkeypatch.setattr(coap_wrapper.device.blocks[RELAY_BLOCK_ID], "output", False)
await async_update_entity(hass, "switch.test_name_channel_1")
await hass.async_block_till_done()
monkeypatch.setattr(mock_block_device.blocks[RELAY_BLOCK_ID], "output", False)
await init_integration(hass, 1)
assert hass.states.get("switch.test_name_channel_1").state == STATE_OFF
monkeypatch.setattr(coap_wrapper.device.blocks[RELAY_BLOCK_ID], "output", True)
await async_update_entity(hass, "switch.test_name_channel_1")
await hass.async_block_till_done()
monkeypatch.setattr(mock_block_device.blocks[RELAY_BLOCK_ID], "output", True)
mock_block_device.mock_update()
assert hass.states.get("switch.test_name_channel_1").state == STATE_ON
async def test_block_device_no_relay_blocks(hass, coap_wrapper, monkeypatch):
async def test_block_device_no_relay_blocks(hass, mock_block_device, monkeypatch):
"""Test block device without relay blocks."""
assert coap_wrapper
monkeypatch.setattr(coap_wrapper.device.blocks[RELAY_BLOCK_ID], "type", "roller")
hass.async_create_task(
hass.config_entries.async_forward_entry_setup(coap_wrapper.entry, SWITCH_DOMAIN)
)
await hass.async_block_till_done()
monkeypatch.setattr(mock_block_device.blocks[RELAY_BLOCK_ID], "type", "roller")
await init_integration(hass, 1)
assert hass.states.get("switch.test_name_channel_1") is None
async def test_block_device_mode_roller(hass, coap_wrapper, monkeypatch):
async def test_block_device_mode_roller(hass, mock_block_device, monkeypatch):
"""Test block device in roller mode."""
assert coap_wrapper
monkeypatch.setitem(coap_wrapper.device.settings, "mode", "roller")
hass.async_create_task(
hass.config_entries.async_forward_entry_setup(coap_wrapper.entry, SWITCH_DOMAIN)
)
await hass.async_block_till_done()
monkeypatch.setitem(mock_block_device.settings, "mode", "roller")
await init_integration(hass, 1)
assert hass.states.get("switch.test_name_channel_1") is None
async def test_block_device_app_type_light(hass, coap_wrapper, monkeypatch):
async def test_block_device_app_type_light(hass, mock_block_device, monkeypatch):
"""Test block device in app type set to light mode."""
assert coap_wrapper
monkeypatch.setitem(
coap_wrapper.device.settings["relays"][0], "appliance_type", "light"
mock_block_device.settings["relays"][RELAY_BLOCK_ID], "appliance_type", "light"
)
hass.async_create_task(
hass.config_entries.async_forward_entry_setup(coap_wrapper.entry, SWITCH_DOMAIN)
)
await hass.async_block_till_done()
await init_integration(hass, 1)
assert hass.states.get("switch.test_name_channel_1") is None
async def test_rpc_device_services(hass, rpc_wrapper, monkeypatch):
async def test_rpc_device_services(hass, mock_rpc_device, monkeypatch):
"""Test RPC device turn on/off services."""
assert rpc_wrapper
hass.async_create_task(
hass.config_entries.async_forward_entry_setup(rpc_wrapper.entry, SWITCH_DOMAIN)
)
await hass.async_block_till_done()
await init_integration(hass, 2)
await hass.services.async_call(
SWITCH_DOMAIN,
@ -113,28 +80,21 @@ async def test_rpc_device_services(hass, rpc_wrapper, monkeypatch):
)
assert hass.states.get("switch.test_switch_0").state == STATE_ON
monkeypatch.setitem(rpc_wrapper.device.status["switch:0"], "output", False)
monkeypatch.setitem(mock_rpc_device.status["switch:0"], "output", False)
await hass.services.async_call(
SWITCH_DOMAIN,
SERVICE_TURN_OFF,
{ATTR_ENTITY_ID: "switch.test_switch_0"},
blocking=True,
)
rpc_wrapper.async_set_updated_data("")
mock_rpc_device.mock_update()
assert hass.states.get("switch.test_switch_0").state == STATE_OFF
async def test_rpc_device_switch_type_lights_mode(hass, rpc_wrapper, monkeypatch):
async def test_rpc_device_switch_type_lights_mode(hass, mock_rpc_device, monkeypatch):
"""Test RPC device with switch in consumption type lights mode."""
assert rpc_wrapper
monkeypatch.setitem(
rpc_wrapper.device.config["sys"]["ui_data"],
"consumption_types",
["lights"],
mock_rpc_device.config["sys"]["ui_data"], "consumption_types", ["lights"]
)
hass.async_create_task(
hass.config_entries.async_forward_entry_setup(rpc_wrapper.entry, SWITCH_DOMAIN)
)
await hass.async_block_till_done()
await init_integration(hass, 2)
assert hass.states.get("switch.test_switch_0") is None

62
tests/components/shelly/test_update.py
View File

@ -6,11 +6,11 @@ from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_component import async_update_entity
from homeassistant.helpers.entity_registry import async_get
from . import init_integration
async def test_block_update(hass: HomeAssistant, coap_wrapper, monkeypatch):
async def test_block_update(hass: HomeAssistant, mock_block_device, monkeypatch):
"""Test block device update entity."""
assert coap_wrapper
entity_registry = async_get(hass)
entity_registry.async_get_or_create(
UPDATE_DOMAIN,
@ -19,18 +19,9 @@ async def test_block_update(hass: HomeAssistant, coap_wrapper, monkeypatch):
suggested_object_id="test_name_firmware_update",
disabled_by=None,
)
hass.async_create_task(
hass.config_entries.async_forward_entry_setup(coap_wrapper.entry, UPDATE_DOMAIN)
)
await hass.async_block_till_done()
await init_integration(hass, 1)
# update entity
await async_update_entity(hass, "update.test_name_firmware_update")
await hass.async_block_till_done()
state = hass.states.get("update.test_name_firmware_update")
assert state
assert state.state == STATE_ON
assert hass.states.get("update.test_name_firmware_update").state == STATE_ON
await hass.services.async_call(
UPDATE_DOMAIN,
@ -38,46 +29,30 @@ async def test_block_update(hass: HomeAssistant, coap_wrapper, monkeypatch):
{ATTR_ENTITY_ID: "update.test_name_firmware_update"},
blocking=True,
)
await hass.async_block_till_done()
assert coap_wrapper.device.trigger_ota_update.call_count == 1
assert mock_block_device.trigger_ota_update.call_count == 1
monkeypatch.setitem(coap_wrapper.device.status["update"], "old_version", None)
monkeypatch.setitem(coap_wrapper.device.status["update"], "new_version", None)
monkeypatch.setitem(mock_block_device.status["update"], "old_version", None)
monkeypatch.setitem(mock_block_device.status["update"], "new_version", None)
# update entity
await async_update_entity(hass, "update.test_name_firmware_update")
await hass.async_block_till_done()
state = hass.states.get("update.test_name_firmware_update")
assert state
assert state.state == STATE_UNKNOWN
assert hass.states.get("update.test_name_firmware_update").state == STATE_UNKNOWN
async def test_rpc_update(hass: HomeAssistant, rpc_wrapper, monkeypatch):
async def test_rpc_update(hass: HomeAssistant, mock_rpc_device, monkeypatch):
"""Test rpc device update entity."""
assert rpc_wrapper
entity_registry = async_get(hass)
entity_registry.async_get_or_create(
UPDATE_DOMAIN,
DOMAIN,
"12345678-sys-fwupdate",
"shelly-sys-fwupdate",
suggested_object_id="test_name_firmware_update",
disabled_by=None,
)
await init_integration(hass, 2)
hass.async_create_task(
hass.config_entries.async_forward_entry_setup(rpc_wrapper.entry, UPDATE_DOMAIN)
)
await hass.async_block_till_done()
# update entity
await async_update_entity(hass, "update.test_name_firmware_update")
await hass.async_block_till_done()
state = hass.states.get("update.test_name_firmware_update")
assert state
assert state.state == STATE_ON
assert hass.states.get("update.test_name_firmware_update").state == STATE_ON
await hass.services.async_call(
UPDATE_DOMAIN,
@ -86,15 +61,12 @@ async def test_rpc_update(hass: HomeAssistant, rpc_wrapper, monkeypatch):
blocking=True,
)
await hass.async_block_till_done()
assert rpc_wrapper.device.trigger_ota_update.call_count == 1
assert mock_rpc_device.trigger_ota_update.call_count == 1
monkeypatch.setitem(rpc_wrapper.device.status["sys"], "available_updates", {})
rpc_wrapper.device.shelly = None
monkeypatch.setitem(mock_rpc_device.status["sys"], "available_updates", {})
monkeypatch.setattr(mock_rpc_device, "shelly", None)
# update entity
await async_update_entity(hass, "update.test_name_firmware_update")
await hass.async_block_till_done()
state = hass.states.get("update.test_name_firmware_update")
assert state
assert state.state == STATE_UNKNOWN
assert hass.states.get("update.test_name_firmware_update").state == STATE_UNKNOWN