jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-23 13:17:32 +00:00
Code Issues Packages Projects Releases Wiki Activity

Convert rfxtrx tests to pytest async tests and re-enable (#37206)

Browse Source 
* Rework and re-enable rfxtrx tests

* Add missed change to _signal_event

* Fixup the dummy serial that causes max cpu

* Make sure we cleanup thread here too

* Make sure we always wait for tasks before we check state

* Some more places we need to wait before checking
This commit is contained in:
Joakim Plate 2020-07-03 10:22:02 +02:00 committed by GitHub
parent b859be8cea
commit ab4687d914
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
7 changed files with 1079 additions and 1032 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
tests/components/rfxtrx/__init__.py
View File

@ -1 +1,6 @@
"""Tests for the rfxtrx component.""" """Tests for the rfxtrx component."""
from homeassistant.components import rfxtrx
async def _signal_event(hass, event):
await hass.async_add_executor_job(rfxtrx.RECEIVED_EVT_SUBSCRIBERS[0], event)

75
tests/components/rfxtrx/conftest.py Normal file
View File

@ -0,0 +1,75 @@
"""Common test tools."""
import threading
from unittest import mock
import RFXtrx
import pytest
from homeassistant.components import rfxtrx as rfxtrx_core
from tests.common import mock_component
class FixedDummySerial(RFXtrx._dummySerial): # pylint: disable=protected-access
"""Fixed dummy serial that doesn't cause max CPU usage."""
def __init__(self, *args, **kwargs):
"""Init."""
super().__init__(*args, **kwargs)
self._close_event = threading.Event()
def read(self, data=None):
"""Read."""
res = super().read(data)
if not res and not data:
self._close_event.wait(0.1)
return res
def close(self):
"""Close."""
self._close_event.set()
class FixedDummyTransport(RFXtrx.DummyTransport):
"""Fixed dummy transport that maxes CPU."""
def __init__(self, device="", debug=True):
"""Init."""
super().__init__(device, debug)
self._close_event = threading.Event()
def receive_blocking(self, data=None):
"""Read."""
res = super().receive_blocking(data)
if not res:
self._close_event.wait(0.1)
return res
def close(self):
"""Close."""
self._close_event.set()
@pytest.fixture(autouse=True)
async def rfxtrx_cleanup():
"""Fixture that cleans up threads from integration."""
with mock.patch("RFXtrx._dummySerial", new=FixedDummySerial), mock.patch(
"RFXtrx.DummyTransport", new=FixedDummyTransport
):
yield
rfxtrx_core.RECEIVED_EVT_SUBSCRIBERS.clear()
rfxtrx_core.RFX_DEVICES.clear()
@pytest.fixture(name="rfxtrx")
async def rfxtrx_fixture(hass):
"""Stub out core rfxtrx to test platform."""
mock_component(hass, "rfxtrx")
yield
# These test don't listen for stop to do cleanup.
if rfxtrx_core.DATA_RFXOBJECT in hass.data:
hass.data[rfxtrx_core.DATA_RFXOBJECT].close_connection()

301
tests/components/rfxtrx/test_cover.py
View File

@ -1,37 +1,19 @@
"""The tests for the Rfxtrx cover platform.""" """The tests for the Rfxtrx cover platform."""
import unittest
import RFXtrx as rfxtrxmod import RFXtrx as rfxtrxmod
import pytest
from homeassistant.components import rfxtrx as rfxtrx_core from homeassistant.components import rfxtrx as rfxtrx_core
from homeassistant.setup import setup_component from homeassistant.setup import async_setup_component
from tests.common import get_test_home_assistant, mock_component from . import _signal_event
from tests.common import assert_setup_component
@pytest.mark.skipif("os.environ.get('RFXTRX') != 'RUN'") async def test_valid_config(hass, rfxtrx):
class TestCoverRfxtrx(unittest.TestCase): """Test configuration."""
"""Test the Rfxtrx cover platform.""" with assert_setup_component(1):
assert await async_setup_component(
def setUp(self): hass,
"""Set up things to be run when tests are started."""
self.hass = get_test_home_assistant()
mock_component(self.hass, "rfxtrx")
self.addCleanup(self.tear_down_cleanup)
def tear_down_cleanup(self):
"""Stop everything that was started."""
rfxtrx_core.RECEIVED_EVT_SUBSCRIBERS.clear()
rfxtrx_core.RFX_DEVICES.clear()
if rfxtrx_core.DATA_RFXOBJECT in self.hass.data:
self.hass.data[rfxtrx_core.DATA_RFXOBJECT].close_connection()
self.hass.stop()
def test_valid_config(self):
"""Test configuration."""
assert setup_component(
self.hass,
"cover", "cover",
{ {
"cover": { "cover": {
@ -46,152 +28,157 @@ class TestCoverRfxtrx(unittest.TestCase):
} }
}, },
) )
await hass.async_block_till_done()
def test_default_config(self):
"""Test with 0 cover."""
assert setup_component(
self.hass, "cover", {"cover": {"platform": "rfxtrx", "devices": {}}}
)
assert 0 == len(rfxtrx_core.RFX_DEVICES)
def test_one_cover(self): async def test_default_config(hass, rfxtrx):
"""Test with 1 cover.""" """Test with 0 cover."""
assert setup_component( assert await async_setup_component(
self.hass, hass, "cover", {"cover": {"platform": "rfxtrx", "devices": {}}}
"cover", )
{ await hass.async_block_till_done()
"cover": {
"platform": "rfxtrx",
"devices": {"0b1400cd0213c7f210010f51": {"name": "Test"}},
}
},
)
self.hass.block_till_done()
self.hass.data[rfxtrx_core.DATA_RFXOBJECT] = rfxtrxmod.Core(
"", transport_protocol=rfxtrxmod.DummyTransport
)
assert 1 == len(rfxtrx_core.RFX_DEVICES) assert 0 == len(rfxtrx_core.RFX_DEVICES)
for id in rfxtrx_core.RFX_DEVICES:
entity = rfxtrx_core.RFX_DEVICES[id]
assert entity.signal_repetitions == 1
assert not entity.should_fire_event
assert not entity.should_poll
entity.open_cover()
entity.close_cover()
entity.stop_cover()
def test_several_covers(self):
"""Test with 3 covers."""
assert setup_component(
self.hass,
"cover",
{
"cover": {
"platform": "rfxtrx",
"signal_repetitions": 3,
"devices": {
"0b1100cd0213c7f230010f71": {"name": "Test"},
"0b1100100118cdea02010f70": {"name": "Bath"},
"0b1100101118cdea02010f70": {"name": "Living"},
},
}
},
)
assert 3 == len(rfxtrx_core.RFX_DEVICES) async def test_one_cover(hass, rfxtrx):
device_num = 0 """Test with 1 cover."""
for id in rfxtrx_core.RFX_DEVICES: assert await async_setup_component(
entity = rfxtrx_core.RFX_DEVICES[id] hass,
assert entity.signal_repetitions == 3 "cover",
if entity.name == "Living": {
device_num = device_num + 1 "cover": {
elif entity.name == "Bath": "platform": "rfxtrx",
device_num = device_num + 1 "devices": {"0b1400cd0213c7f210010f51": {"name": "Test"}},
elif entity.name == "Test": }
device_num = device_num + 1 },
)
await hass.async_block_till_done()
assert 3 == device_num hass.data[rfxtrx_core.DATA_RFXOBJECT] = rfxtrxmod.Core(
"", transport_protocol=rfxtrxmod.DummyTransport
)
def test_discover_covers(self): assert 1 == len(rfxtrx_core.RFX_DEVICES)
"""Test with discovery of covers.""" for id in rfxtrx_core.RFX_DEVICES:
assert setup_component( entity = rfxtrx_core.RFX_DEVICES[id]
self.hass, entity.hass = hass
"cover", assert entity.signal_repetitions == 1
{"cover": {"platform": "rfxtrx", "automatic_add": True, "devices": {}}}, assert not entity.should_fire_event
) assert not entity.should_poll
entity.open_cover()
entity.close_cover()
entity.stop_cover()
event = rfxtrx_core.get_rfx_object("0a140002f38cae010f0070")
event.data = bytearray(
[0x0A, 0x14, 0x00, 0x02, 0xF3, 0x8C, 0xAE, 0x01, 0x0F, 0x00, 0x70]
)
for evt_sub in rfxtrx_core.RECEIVED_EVT_SUBSCRIBERS: async def test_several_covers(hass, rfxtrx):
evt_sub(event) """Test with 3 covers."""
assert 1 == len(rfxtrx_core.RFX_DEVICES) assert await async_setup_component(
hass,
"cover",
{
"cover": {
"platform": "rfxtrx",
"signal_repetitions": 3,
"devices": {
"0b1100cd0213c7f230010f71": {"name": "Test"},
"0b1100100118cdea02010f70": {"name": "Bath"},
"0b1100101118cdea02010f70": {"name": "Living"},
},
}
},
)
await hass.async_block_till_done()
event = rfxtrx_core.get_rfx_object("0a1400adf394ab020e0060") assert 3 == len(rfxtrx_core.RFX_DEVICES)
event.data = bytearray( device_num = 0
[0x0A, 0x14, 0x00, 0xAD, 0xF3, 0x94, 0xAB, 0x02, 0x0E, 0x00, 0x60] for id in rfxtrx_core.RFX_DEVICES:
) entity = rfxtrx_core.RFX_DEVICES[id]
assert entity.signal_repetitions == 3
if entity.name == "Living":
device_num = device_num + 1
elif entity.name == "Bath":
device_num = device_num + 1
elif entity.name == "Test":
device_num = device_num + 1
for evt_sub in rfxtrx_core.RECEIVED_EVT_SUBSCRIBERS: assert 3 == device_num
evt_sub(event)
assert 2 == len(rfxtrx_core.RFX_DEVICES)
# Trying to add a sensor
event = rfxtrx_core.get_rfx_object("0a52085e070100b31b0279")
event.data = bytearray(b"\nR\x08^\x07\x01\x00\xb3\x1b\x02y")
for evt_sub in rfxtrx_core.RECEIVED_EVT_SUBSCRIBERS:
evt_sub(event)
assert 2 == len(rfxtrx_core.RFX_DEVICES)
# Trying to add a light async def test_discover_covers(hass, rfxtrx):
event = rfxtrx_core.get_rfx_object("0b1100100118cdea02010f70") """Test with discovery of covers."""
event.data = bytearray( assert await async_setup_component(
[0x0B, 0x11, 0x11, 0x10, 0x01, 0x18, 0xCD, 0xEA, 0x01, 0x02, 0x0F, 0x70] hass,
) "cover",
for evt_sub in rfxtrx_core.RECEIVED_EVT_SUBSCRIBERS: {"cover": {"platform": "rfxtrx", "automatic_add": True, "devices": {}}},
evt_sub(event) )
assert 2 == len(rfxtrx_core.RFX_DEVICES) await hass.async_block_till_done()
def test_discover_cover_noautoadd(self): event = rfxtrx_core.get_rfx_object("0a140002f38cae010f0070")
"""Test with discovery of cover when auto add is False.""" event.data = bytearray(
assert setup_component( [0x0A, 0x14, 0x00, 0x02, 0xF3, 0x8C, 0xAE, 0x01, 0x0F, 0x00, 0x70]
self.hass, )
"cover",
{"cover": {"platform": "rfxtrx", "automatic_add": False, "devices": {}}},
)
event = rfxtrx_core.get_rfx_object("0a1400adf394ab010d0060") await _signal_event(hass, event)
event.data = bytearray( assert 1 == len(rfxtrx_core.RFX_DEVICES)
[0x0A, 0x14, 0x00, 0xAD, 0xF3, 0x94, 0xAB, 0x01, 0x0D, 0x00, 0x60]
)
for evt_sub in rfxtrx_core.RECEIVED_EVT_SUBSCRIBERS: event = rfxtrx_core.get_rfx_object("0a1400adf394ab020e0060")
evt_sub(event) event.data = bytearray(
assert 0 == len(rfxtrx_core.RFX_DEVICES) [0x0A, 0x14, 0x00, 0xAD, 0xF3, 0x94, 0xAB, 0x02, 0x0E, 0x00, 0x60]
)
event = rfxtrx_core.get_rfx_object("0a1400adf394ab020e0060") await _signal_event(hass, event)
event.data = bytearray( assert 2 == len(rfxtrx_core.RFX_DEVICES)
[0x0A, 0x14, 0x00, 0xAD, 0xF3, 0x94, 0xAB, 0x02, 0x0E, 0x00, 0x60]
)
for evt_sub in rfxtrx_core.RECEIVED_EVT_SUBSCRIBERS:
evt_sub(event)
assert 0 == len(rfxtrx_core.RFX_DEVICES)
# Trying to add a sensor # Trying to add a sensor
event = rfxtrx_core.get_rfx_object("0a52085e070100b31b0279") event = rfxtrx_core.get_rfx_object("0a52085e070100b31b0279")
event.data = bytearray(b"\nR\x08^\x07\x01\x00\xb3\x1b\x02y") event.data = bytearray(b"\nR\x08^\x07\x01\x00\xb3\x1b\x02y")
for evt_sub in rfxtrx_core.RECEIVED_EVT_SUBSCRIBERS: await _signal_event(hass, event)
evt_sub(event) assert 2 == len(rfxtrx_core.RFX_DEVICES)
assert 0 == len(rfxtrx_core.RFX_DEVICES)
# Trying to add a light # Trying to add a light
event = rfxtrx_core.get_rfx_object("0b1100100118cdea02010f70") event = rfxtrx_core.get_rfx_object("0b1100100118cdea02010f70")
event.data = bytearray( event.data = bytearray(
[0x0B, 0x11, 0x11, 0x10, 0x01, 0x18, 0xCD, 0xEA, 0x01, 0x02, 0x0F, 0x70] [0x0B, 0x11, 0x11, 0x10, 0x01, 0x18, 0xCD, 0xEA, 0x01, 0x02, 0x0F, 0x70]
) )
for evt_sub in rfxtrx_core.RECEIVED_EVT_SUBSCRIBERS: await _signal_event(hass, event)
evt_sub(event) assert 2 == len(rfxtrx_core.RFX_DEVICES)
assert 0 == len(rfxtrx_core.RFX_DEVICES)
async def test_discover_cover_noautoadd(hass, rfxtrx):
"""Test with discovery of cover when auto add is False."""
assert await async_setup_component(
hass,
"cover",
{"cover": {"platform": "rfxtrx", "automatic_add": False, "devices": {}}},
)
await hass.async_block_till_done()
event = rfxtrx_core.get_rfx_object("0a1400adf394ab010d0060")
event.data = bytearray(
[0x0A, 0x14, 0x00, 0xAD, 0xF3, 0x94, 0xAB, 0x01, 0x0D, 0x00, 0x60]
)
await _signal_event(hass, event)
assert 0 == len(rfxtrx_core.RFX_DEVICES)
event = rfxtrx_core.get_rfx_object("0a1400adf394ab020e0060")
event.data = bytearray(
[0x0A, 0x14, 0x00, 0xAD, 0xF3, 0x94, 0xAB, 0x02, 0x0E, 0x00, 0x60]
)
await _signal_event(hass, event)
assert 0 == len(rfxtrx_core.RFX_DEVICES)
# Trying to add a sensor
event = rfxtrx_core.get_rfx_object("0a52085e070100b31b0279")
event.data = bytearray(b"\nR\x08^\x07\x01\x00\xb3\x1b\x02y")
await _signal_event(hass, event)
assert 0 == len(rfxtrx_core.RFX_DEVICES)
# Trying to add a light
event = rfxtrx_core.get_rfx_object("0b1100100118cdea02010f70")
event.data = bytearray(
[0x0B, 0x11, 0x11, 0x10, 0x01, 0x18, 0xCD, 0xEA, 0x01, 0x02, 0x0F, 0x70]
)
await _signal_event(hass, event)
assert 0 == len(rfxtrx_core.RFX_DEVICES)

345
tests/components/rfxtrx/test_init.py
View File

@ -1,204 +1,201 @@
"""The tests for the Rfxtrx component.""" """The tests for the Rfxtrx component."""
# pylint: disable=protected-access # pylint: disable=protected-access
import time import asyncio
import unittest
import pytest from async_timeout import timeout
from homeassistant.components import rfxtrx from homeassistant.components import rfxtrx
from homeassistant.core import callback from homeassistant.core import callback
from homeassistant.setup import setup_component from homeassistant.setup import async_setup_component
from tests.common import get_test_home_assistant from . import _signal_event
from tests.common import assert_setup_component
@pytest.mark.skipif("os.environ.get('RFXTRX') != 'RUN'") async def test_default_config(hass):
class TestRFXTRX(unittest.TestCase): """Test configuration."""
"""Test the Rfxtrx component.""" assert await async_setup_component(
hass,
"rfxtrx",
{
"rfxtrx": {
"device": "/dev/serial/by-id/usb"
+ "-RFXCOM_RFXtrx433_A1Y0NJGR-if00-port0",
"dummy": True,
}
},
)
def setUp(self): with assert_setup_component(1, "sensor"):
"""Set up things to be run when tests are started.""" await async_setup_component(
self.hass = get_test_home_assistant() hass,
self.addCleanup(self.tear_down_cleanup)
def tear_down_cleanup(self):
"""Stop everything that was started."""
rfxtrx.RECEIVED_EVT_SUBSCRIBERS.clear()
rfxtrx.RFX_DEVICES.clear()
if rfxtrx.DATA_RFXOBJECT in self.hass.data:
self.hass.data[rfxtrx.DATA_RFXOBJECT].close_connection()
self.hass.stop()
def test_default_config(self):
"""Test configuration."""
assert setup_component(
self.hass,
"rfxtrx",
{
"rfxtrx": {
"device": "/dev/serial/by-id/usb"
+ "-RFXCOM_RFXtrx433_A1Y0NJGR-if00-port0",
"dummy": True,
}
},
)
assert setup_component(
self.hass,
"sensor", "sensor",
{"sensor": {"platform": "rfxtrx", "automatic_add": True, "devices": {}}}, {"sensor": {"platform": "rfxtrx", "automatic_add": True, "devices": {}}},
) )
time.sleep(1) # Dummy startup is slow # Dummy startup is slow
async with timeout(10):
while len(hass.data[rfxtrx.DATA_RFXOBJECT].sensors()) < 2:
await asyncio.sleep(0.1)
assert len(self.hass.data[rfxtrx.DATA_RFXOBJECT].sensors()) == 2 assert len(hass.data[rfxtrx.DATA_RFXOBJECT].sensors()) == 2
def test_valid_config(self):
"""Test configuration."""
assert setup_component(
self.hass,
"rfxtrx",
{
"rfxtrx": {
"device": "/dev/serial/by-id/usb"
+ "-RFXCOM_RFXtrx433_A1Y0NJGR-if00-port0",
"dummy": True,
}
},
)
def test_valid_config2(self): async def test_valid_config(hass):
"""Test configuration.""" """Test configuration."""
assert setup_component( assert await async_setup_component(
self.hass, hass,
"rfxtrx", "rfxtrx",
{ {
"rfxtrx": { "rfxtrx": {
"device": "/dev/serial/by-id/usb" "device": "/dev/serial/by-id/usb"
+ "-RFXCOM_RFXtrx433_A1Y0NJGR-if00-port0", + "-RFXCOM_RFXtrx433_A1Y0NJGR-if00-port0",
"dummy": True, "dummy": True,
"debug": True, }
} },
}, )
)
def test_invalid_config(self):
"""Test configuration."""
assert not setup_component(self.hass, "rfxtrx", {"rfxtrx": {}})
assert not setup_component( async def test_valid_config2(hass):
self.hass, """Test configuration."""
"rfxtrx", assert await async_setup_component(
{ hass,
"rfxtrx": { "rfxtrx",
"device": "/dev/serial/by-id/usb" {
+ "-RFXCOM_RFXtrx433_A1Y0NJGR-if00-port0", "rfxtrx": {
"invalid_key": True, "device": "/dev/serial/by-id/usb"
} + "-RFXCOM_RFXtrx433_A1Y0NJGR-if00-port0",
}, "dummy": True,
) "debug": True,
}
},
)
def test_fire_event(self):
"""Test fire event."""
assert setup_component(
self.hass,
"rfxtrx",
{
"rfxtrx": {
"device": "/dev/serial/by-id/usb"
+ "-RFXCOM_RFXtrx433_A1Y0NJGR-if00-port0",
"dummy": True,
}
},
)
assert setup_component(
self.hass,
"switch",
{
"switch": {
"platform": "rfxtrx",
"automatic_add": True,
"devices": {
"0b1100cd0213c7f210010f51": {
"name": "Test",
rfxtrx.ATTR_FIRE_EVENT: True,
}
},
}
},
)
calls = [] async def test_invalid_config(hass):
"""Test configuration."""
assert not await async_setup_component(hass, "rfxtrx", {"rfxtrx": {}})
@callback assert not await async_setup_component(
def record_event(event): hass,
"""Add recorded event to set.""" "rfxtrx",
calls.append(event) {
"rfxtrx": {
"device": "/dev/serial/by-id/usb"
+ "-RFXCOM_RFXtrx433_A1Y0NJGR-if00-port0",
"invalid_key": True,
}
},
)
self.hass.bus.listen(rfxtrx.EVENT_BUTTON_PRESSED, record_event)
self.hass.block_till_done()
entity = rfxtrx.RFX_DEVICES["213c7f2_16"]
entity.update_state(False, 0)
assert "Test" == entity.name
assert "off" == entity.state
assert entity.should_fire_event
event = rfxtrx.get_rfx_object("0b1100cd0213c7f210010f51") async def test_fire_event(hass):
event.data = bytearray( """Test fire event."""
[0x0B, 0x11, 0x00, 0x10, 0x01, 0x18, 0xCD, 0xEA, 0x01, 0x01, 0x0F, 0x70] assert await async_setup_component(
) hass,
rfxtrx.RECEIVED_EVT_SUBSCRIBERS[0](event) "rfxtrx",
self.hass.block_till_done() {
"rfxtrx": {
"device": "/dev/serial/by-id/usb"
+ "-RFXCOM_RFXtrx433_A1Y0NJGR-if00-port0",
"dummy": True,
}
},
)
assert event.values["Command"] == "On" assert await async_setup_component(
assert "on" == entity.state hass,
assert self.hass.states.get("switch.test").state == "on" "switch",
assert 1 == len(calls) {
assert calls[0].data == {"entity_id": "switch.test", "state": "on"} "switch": {
"platform": "rfxtrx",
"automatic_add": True,
"devices": {
"0b1100cd0213c7f210010f51": {
"name": "Test",
rfxtrx.ATTR_FIRE_EVENT: True,
}
},
}
},
)
await hass.async_block_till_done()
def test_fire_event_sensor(self): calls = []
"""Test fire event."""
assert setup_component(
self.hass,
"rfxtrx",
{
"rfxtrx": {
"device": "/dev/serial/by-id/usb"
+ "-RFXCOM_RFXtrx433_A1Y0NJGR-if00-port0",
"dummy": True,
}
},
)
assert setup_component(
self.hass,
"sensor",
{
"sensor": {
"platform": "rfxtrx",
"automatic_add": True,
"devices": {
"0a520802060100ff0e0269": {
"name": "Test",
rfxtrx.ATTR_FIRE_EVENT: True,
}
},
}
},
)
calls = [] @callback
def record_event(event):
"""Add recorded event to set."""
calls.append(event)
@callback hass.bus.async_listen(rfxtrx.EVENT_BUTTON_PRESSED, record_event)
def record_event(event): await hass.async_block_till_done()
"""Add recorded event to set.""" entity = rfxtrx.RFX_DEVICES["213c7f2_16"]
calls.append(event) entity.update_state(False, 0)
assert "Test" == entity.name
assert "off" == entity.state
assert entity.should_fire_event
self.hass.bus.listen("signal_received", record_event) event = rfxtrx.get_rfx_object("0b1100cd0213c7f210010f51")
self.hass.block_till_done() event.data = bytearray(
event = rfxtrx.get_rfx_object("0a520802060101ff0f0269") [0x0B, 0x11, 0x00, 0x10, 0x01, 0x18, 0xCD, 0xEA, 0x01, 0x01, 0x0F, 0x70]
event.data = bytearray(b"\nR\x08\x01\x07\x01\x00\xb8\x1b\x02y") )
rfxtrx.RECEIVED_EVT_SUBSCRIBERS[0](event) await _signal_event(hass, event)
await hass.async_block_till_done()
self.hass.block_till_done() assert event.values["Command"] == "On"
assert 1 == len(calls) assert "on" == entity.state
assert calls[0].data == {"entity_id": "sensor.test_temperature"} assert hass.states.get("switch.test").state == "on"
assert 1 == len(calls)
assert calls[0].data == {"entity_id": "switch.test", "state": "on"}
async def test_fire_event_sensor(hass):
"""Test fire event."""
await async_setup_component(
hass,
"rfxtrx",
{
"rfxtrx": {
"device": "/dev/serial/by-id/usb"
+ "-RFXCOM_RFXtrx433_A1Y0NJGR-if00-port0",
"dummy": True,
}
},
)
await async_setup_component(
hass,
"sensor",
{
"sensor": {
"platform": "rfxtrx",
"automatic_add": True,
"devices": {
"0a520802060100ff0e0269": {
"name": "Test",
rfxtrx.ATTR_FIRE_EVENT: True,
}
},
}
},
)
await hass.async_block_till_done()
calls = []
@callback
def record_event(event):
"""Add recorded event to set."""
calls.append(event)
hass.bus.async_listen("signal_received", record_event)
await hass.async_block_till_done()
event = rfxtrx.get_rfx_object("0a520802060101ff0f0269")
event.data = bytearray(b"\nR\x08\x01\x07\x01\x00\xb8\x1b\x02y")
await _signal_event(hass, event)
await hass.async_block_till_done()
assert 1 == len(calls)
assert calls[0].data == {"entity_id": "sensor.test_temperature"}

417
tests/components/rfxtrx/test_light.py
View File

@ -1,36 +1,17 @@
"""The tests for the Rfxtrx light platform.""" """The tests for the Rfxtrx light platform."""
import unittest
import pytest
from homeassistant.components import rfxtrx as rfxtrx_core from homeassistant.components import rfxtrx as rfxtrx_core
from homeassistant.setup import setup_component from homeassistant.setup import async_setup_component
from tests.common import get_test_home_assistant, mock_component from . import _signal_event
from tests.common import assert_setup_component
@pytest.mark.skipif("os.environ.get('RFXTRX') != 'RUN'") async def test_valid_config(hass, rfxtrx):
class TestLightRfxtrx(unittest.TestCase): """Test configuration."""
"""Test the Rfxtrx light platform.""" with assert_setup_component(1):
assert await async_setup_component(
def setUp(self): hass,
"""Set up things to be run when tests are started."""
self.hass = get_test_home_assistant()
mock_component(self.hass, "rfxtrx")
self.addCleanup(self.tear_down_cleanup)
def tear_down_cleanup(self):
"""Stop everything that was started."""
rfxtrx_core.RECEIVED_EVT_SUBSCRIBERS.clear()
rfxtrx_core.RFX_DEVICES.clear()
if rfxtrx_core.DATA_RFXOBJECT in self.hass.data:
self.hass.data[rfxtrx_core.DATA_RFXOBJECT].close_connection()
self.hass.stop()
def test_valid_config(self):
"""Test configuration."""
assert setup_component(
self.hass,
"light", "light",
{ {
"light": { "light": {
@ -46,8 +27,8 @@ class TestLightRfxtrx(unittest.TestCase):
}, },
) )
assert setup_component( assert await async_setup_component(
self.hass, hass,
"light", "light",
{ {
"light": { "light": {
@ -63,226 +44,238 @@ class TestLightRfxtrx(unittest.TestCase):
}, },
) )
def test_default_config(self):
"""Test with 0 switches.""" async def test_default_config(hass, rfxtrx):
assert setup_component( """Test with 0 switches."""
self.hass, "light", {"light": {"platform": "rfxtrx", "devices": {}}} with assert_setup_component(1):
await async_setup_component(
hass, "light", {"light": {"platform": "rfxtrx", "devices": {}}}
) )
assert 0 == len(rfxtrx_core.RFX_DEVICES) await hass.async_block_till_done()
def test_one_light(self): assert 0 == len(rfxtrx_core.RFX_DEVICES)
"""Test with 1 light."""
assert setup_component(
self.hass,
"light",
{
"light": {
"platform": "rfxtrx",
"devices": {"0b1100cd0213c7f210010f51": {"name": "Test"}},
}
},
)
import RFXtrx as rfxtrxmod
self.hass.data[rfxtrx_core.DATA_RFXOBJECT] = rfxtrxmod.Core( async def test_one_light(hass, rfxtrx):
"", transport_protocol=rfxtrxmod.DummyTransport """Test with 1 light."""
) await async_setup_component(
hass,
"light",
{
"light": {
"platform": "rfxtrx",
"devices": {"0b1100cd0213c7f210010f51": {"name": "Test"}},
}
},
)
await hass.async_block_till_done()
assert 1 == len(rfxtrx_core.RFX_DEVICES) import RFXtrx as rfxtrxmod
entity = rfxtrx_core.RFX_DEVICES["213c7f2_16"]
entity.hass = self.hass
assert "Test" == entity.name
assert "off" == entity.state
assert entity.assumed_state
assert entity.signal_repetitions == 1
assert not entity.should_fire_event
assert not entity.should_poll
assert not entity.is_on hass.data[rfxtrx_core.DATA_RFXOBJECT] = rfxtrxmod.Core(
"", transport_protocol=rfxtrxmod.DummyTransport
)
entity.turn_on() assert 1 == len(rfxtrx_core.RFX_DEVICES)
assert entity.is_on entity = rfxtrx_core.RFX_DEVICES["213c7f2_16"]
assert entity.brightness == 255 entity.hass = hass
assert "Test" == entity.name
assert "off" == entity.state
assert entity.assumed_state
assert entity.signal_repetitions == 1
assert not entity.should_fire_event
assert not entity.should_poll
entity.turn_off() assert not entity.is_on
assert not entity.is_on
assert entity.brightness == 0
entity.turn_on(brightness=100) entity.turn_on()
assert entity.is_on assert entity.is_on
assert entity.brightness == 100 assert entity.brightness == 255
entity.turn_on(brightness=10) entity.turn_off()
assert entity.is_on assert not entity.is_on
assert entity.brightness == 10 assert entity.brightness == 0
entity.turn_on(brightness=255) entity.turn_on(brightness=100)
assert entity.is_on assert entity.is_on
assert entity.brightness == 255 assert entity.brightness == 100
entity.turn_off() entity.turn_on(brightness=10)
assert "Test" == entity.name assert entity.is_on
assert "off" == entity.state assert entity.brightness == 10
entity.turn_on() entity.turn_on(brightness=255)
assert "on" == entity.state assert entity.is_on
assert entity.brightness == 255
entity.turn_off() entity.turn_off()
assert "off" == entity.state assert "Test" == entity.name
assert "off" == entity.state
entity.turn_on(brightness=100) entity.turn_on()
assert "on" == entity.state assert "on" == entity.state
entity.turn_on(brightness=10) entity.turn_off()
assert "on" == entity.state assert "off" == entity.state
entity.turn_on(brightness=255) entity.turn_on(brightness=100)
assert "on" == entity.state assert "on" == entity.state
def test_several_lights(self): entity.turn_on(brightness=10)
"""Test with 3 lights.""" assert "on" == entity.state
assert setup_component(
self.hass,
"light",
{
"light": {
"platform": "rfxtrx",
"signal_repetitions": 3,
"devices": {
"0b1100cd0213c7f230010f71": {"name": "Test"},
"0b1100100118cdea02010f70": {"name": "Bath"},
"0b1100101118cdea02010f70": {"name": "Living"},
},
}
},
)
assert 3 == len(rfxtrx_core.RFX_DEVICES) entity.turn_on(brightness=255)
device_num = 0 assert "on" == entity.state
for id in rfxtrx_core.RFX_DEVICES:
entity = rfxtrx_core.RFX_DEVICES[id]
assert entity.signal_repetitions == 3
if entity.name == "Living":
device_num = device_num + 1
assert "off" == entity.state
assert "<Entity Living: off>" == entity.__str__()
elif entity.name == "Bath":
device_num = device_num + 1
assert "off" == entity.state
assert "<Entity Bath: off>" == entity.__str__()
elif entity.name == "Test":
device_num = device_num + 1
assert "off" == entity.state
assert "<Entity Test: off>" == entity.__str__()
assert 3 == device_num
def test_discover_light(self): async def test_several_lights(hass, rfxtrx):
"""Test with discovery of lights.""" """Test with 3 lights."""
assert setup_component( await async_setup_component(
self.hass, hass,
"light", "light",
{"light": {"platform": "rfxtrx", "automatic_add": True, "devices": {}}}, {
) "light": {
"platform": "rfxtrx",
"signal_repetitions": 3,
"devices": {
"0b1100cd0213c7f230010f71": {"name": "Test"},
"0b1100100118cdea02010f70": {"name": "Bath"},
"0b1100101118cdea02010f70": {"name": "Living"},
},
}
},
)
await hass.async_block_till_done()
event = rfxtrx_core.get_rfx_object("0b11009e00e6116202020070") assert 3 == len(rfxtrx_core.RFX_DEVICES)
event.data = bytearray(b"\x0b\x11\x00\x9e\x00\xe6\x11b\x02\x02\x00p") device_num = 0
for id in rfxtrx_core.RFX_DEVICES:
entity = rfxtrx_core.RFX_DEVICES[id]
assert entity.signal_repetitions == 3
if entity.name == "Living":
device_num = device_num + 1
assert "off" == entity.state
assert "<Entity Living: off>" == entity.__str__()
elif entity.name == "Bath":
device_num = device_num + 1
assert "off" == entity.state
assert "<Entity Bath: off>" == entity.__str__()
elif entity.name == "Test":
device_num = device_num + 1
assert "off" == entity.state
assert "<Entity Test: off>" == entity.__str__()
rfxtrx_core.RECEIVED_EVT_SUBSCRIBERS[0](event) assert 3 == device_num
entity = rfxtrx_core.RFX_DEVICES["0e61162_2"]
assert 1 == len(rfxtrx_core.RFX_DEVICES)
assert "<Entity 0b11009e00e6116202020070: on>" == entity.__str__()
event = rfxtrx_core.get_rfx_object("0b11009e00e6116201010070")
event.data = bytearray(b"\x0b\x11\x00\x9e\x00\xe6\x11b\x01\x01\x00p")
rfxtrx_core.RECEIVED_EVT_SUBSCRIBERS[0](event) async def test_discover_light(hass, rfxtrx):
assert 1 == len(rfxtrx_core.RFX_DEVICES) """Test with discovery of lights."""
await async_setup_component(
hass,
"light",
{"light": {"platform": "rfxtrx", "automatic_add": True, "devices": {}}},
)
await hass.async_block_till_done()
event = rfxtrx_core.get_rfx_object("0b1100120118cdea02020070") event = rfxtrx_core.get_rfx_object("0b11009e00e6116202020070")
event.data = bytearray( event.data = bytearray(b"\x0b\x11\x00\x9e\x00\xe6\x11b\x02\x02\x00p")
[0x0B, 0x11, 0x00, 0x12, 0x01, 0x18, 0xCD, 0xEA, 0x02, 0x02, 0x00, 0x70]
)
rfxtrx_core.RECEIVED_EVT_SUBSCRIBERS[0](event) await _signal_event(hass, event)
entity = rfxtrx_core.RFX_DEVICES["118cdea_2"] entity = rfxtrx_core.RFX_DEVICES["0e61162_2"]
assert 2 == len(rfxtrx_core.RFX_DEVICES) assert 1 == len(rfxtrx_core.RFX_DEVICES)
assert "<Entity 0b1100120118cdea02020070: on>" == entity.__str__() assert "<Entity 0b11009e00e6116202020070: on>" == entity.__str__()
# trying to add a sensor event = rfxtrx_core.get_rfx_object("0b11009e00e6116201010070")
event = rfxtrx_core.get_rfx_object("0a52085e070100b31b0279") event.data = bytearray(b"\x0b\x11\x00\x9e\x00\xe6\x11b\x01\x01\x00p")
event.data = bytearray(b"\nR\x08^\x07\x01\x00\xb3\x1b\x02y")
rfxtrx_core.RECEIVED_EVT_SUBSCRIBERS[0](event)
assert 2 == len(rfxtrx_core.RFX_DEVICES)
# trying to add a switch await _signal_event(hass, event)
event = rfxtrx_core.get_rfx_object("0b1100100118cdea02010f70") assert 1 == len(rfxtrx_core.RFX_DEVICES)
event.data = bytearray(
[0x0B, 0x11, 0x00, 0x10, 0x01, 0x18, 0xCD, 0xEA, 0x01, 0x01, 0x0F, 0x70]
)
rfxtrx_core.RECEIVED_EVT_SUBSCRIBERS[0](event) event = rfxtrx_core.get_rfx_object("0b1100120118cdea02020070")
assert 2 == len(rfxtrx_core.RFX_DEVICES) event.data = bytearray(
[0x0B, 0x11, 0x00, 0x12, 0x01, 0x18, 0xCD, 0xEA, 0x02, 0x02, 0x00, 0x70]
)
# Trying to add a rollershutter await _signal_event(hass, event)
event = rfxtrx_core.get_rfx_object("0a1400adf394ab020e0060") entity = rfxtrx_core.RFX_DEVICES["118cdea_2"]
event.data = bytearray( assert 2 == len(rfxtrx_core.RFX_DEVICES)
[0x0A, 0x14, 0x00, 0xAD, 0xF3, 0x94, 0xAB, 0x02, 0x0E, 0x00, 0x60] assert "<Entity 0b1100120118cdea02020070: on>" == entity.__str__()
)
rfxtrx_core.RECEIVED_EVT_SUBSCRIBERS[0](event)
assert 2 == len(rfxtrx_core.RFX_DEVICES)
def test_discover_light_noautoadd(self): # trying to add a sensor
"""Test with discover of light when auto add is False.""" event = rfxtrx_core.get_rfx_object("0a52085e070100b31b0279")
assert setup_component( event.data = bytearray(b"\nR\x08^\x07\x01\x00\xb3\x1b\x02y")
self.hass, await _signal_event(hass, event)
"light", assert 2 == len(rfxtrx_core.RFX_DEVICES)
{"light": {"platform": "rfxtrx", "automatic_add": False, "devices": {}}},
)
event = rfxtrx_core.get_rfx_object("0b1100120118cdea02020070") # trying to add a switch
event.data = bytearray( event = rfxtrx_core.get_rfx_object("0b1100100118cdea02010f70")
[0x0B, 0x11, 0x00, 0x12, 0x01, 0x18, 0xCD, 0xEA, 0x02, 0x02, 0x00, 0x70] event.data = bytearray(
) [0x0B, 0x11, 0x00, 0x10, 0x01, 0x18, 0xCD, 0xEA, 0x01, 0x01, 0x0F, 0x70]
)
rfxtrx_core.RECEIVED_EVT_SUBSCRIBERS[0](event) await _signal_event(hass, event)
assert 0 == len(rfxtrx_core.RFX_DEVICES) assert 2 == len(rfxtrx_core.RFX_DEVICES)
event = rfxtrx_core.get_rfx_object("0b1100120118cdea02010070") # Trying to add a rollershutter
event.data = bytearray( event = rfxtrx_core.get_rfx_object("0a1400adf394ab020e0060")
[0x0B, 0x11, 0x00, 0x12, 0x01, 0x18, 0xCD, 0xEA, 0x02, 0x01, 0x00, 0x70] event.data = bytearray(
) [0x0A, 0x14, 0x00, 0xAD, 0xF3, 0x94, 0xAB, 0x02, 0x0E, 0x00, 0x60]
)
await _signal_event(hass, event)
assert 2 == len(rfxtrx_core.RFX_DEVICES)
rfxtrx_core.RECEIVED_EVT_SUBSCRIBERS[0](event)
assert 0 == len(rfxtrx_core.RFX_DEVICES)
event = rfxtrx_core.get_rfx_object("0b1100120118cdea02020070") async def test_discover_light_noautoadd(hass, rfxtrx):
event.data = bytearray( """Test with discover of light when auto add is False."""
[0x0B, 0x11, 0x00, 0x12, 0x01, 0x18, 0xCD, 0xEA, 0x02, 0x02, 0x00, 0x70] await async_setup_component(
) hass,
"light",
{"light": {"platform": "rfxtrx", "automatic_add": False, "devices": {}}},
)
await hass.async_block_till_done()
rfxtrx_core.RECEIVED_EVT_SUBSCRIBERS[0](event) event = rfxtrx_core.get_rfx_object("0b1100120118cdea02020070")
assert 0 == len(rfxtrx_core.RFX_DEVICES) event.data = bytearray(
[0x0B, 0x11, 0x00, 0x12, 0x01, 0x18, 0xCD, 0xEA, 0x02, 0x02, 0x00, 0x70]
)
# Trying to add a sensor await _signal_event(hass, event)
event = rfxtrx_core.get_rfx_object("0a52085e070100b31b0279") assert 0 == len(rfxtrx_core.RFX_DEVICES)
event.data = bytearray(b"\nR\x08^\x07\x01\x00\xb3\x1b\x02y")
rfxtrx_core.RECEIVED_EVT_SUBSCRIBERS[0](event)
assert 0 == len(rfxtrx_core.RFX_DEVICES)
# Trying to add a switch event = rfxtrx_core.get_rfx_object("0b1100120118cdea02010070")
event = rfxtrx_core.get_rfx_object("0b1100100118cdea02010f70") event.data = bytearray(
event.data = bytearray( [0x0B, 0x11, 0x00, 0x12, 0x01, 0x18, 0xCD, 0xEA, 0x02, 0x01, 0x00, 0x70]
[0x0B, 0x11, 0x00, 0x10, 0x01, 0x18, 0xCD, 0xEA, 0x01, 0x01, 0x0F, 0x70] )
)
rfxtrx_core.RECEIVED_EVT_SUBSCRIBERS[0](event)
assert 0 == len(rfxtrx_core.RFX_DEVICES)
# Trying to add a rollershutter await _signal_event(hass, event)
event = rfxtrx_core.get_rfx_object("0a1400adf394ab020e0060") assert 0 == len(rfxtrx_core.RFX_DEVICES)
event.data = bytearray(
[0x0A, 0x14, 0x00, 0xAD, 0xF3, 0x94, 0xAB, 0x02, 0x0E, 0x00, 0x60] event = rfxtrx_core.get_rfx_object("0b1100120118cdea02020070")
) event.data = bytearray(
rfxtrx_core.RECEIVED_EVT_SUBSCRIBERS[0](event) [0x0B, 0x11, 0x00, 0x12, 0x01, 0x18, 0xCD, 0xEA, 0x02, 0x02, 0x00, 0x70]
assert 0 == len(rfxtrx_core.RFX_DEVICES) )
await _signal_event(hass, event)
assert 0 == len(rfxtrx_core.RFX_DEVICES)
# Trying to add a sensor
event = rfxtrx_core.get_rfx_object("0a52085e070100b31b0279")
event.data = bytearray(b"\nR\x08^\x07\x01\x00\xb3\x1b\x02y")
await _signal_event(hass, event)
assert 0 == len(rfxtrx_core.RFX_DEVICES)
# Trying to add a switch
event = rfxtrx_core.get_rfx_object("0b1100100118cdea02010f70")
event.data = bytearray(
[0x0B, 0x11, 0x00, 0x10, 0x01, 0x18, 0xCD, 0xEA, 0x01, 0x01, 0x0F, 0x70]
)
await _signal_event(hass, event)
assert 0 == len(rfxtrx_core.RFX_DEVICES)
# Trying to add a rollershutter
event = rfxtrx_core.get_rfx_object("0a1400adf394ab020e0060")
event.data = bytearray(
[0x0A, 0x14, 0x00, 0xAD, 0xF3, 0x94, 0xAB, 0x02, 0x0E, 0x00, 0x60]
)
await _signal_event(hass, event)
assert 0 == len(rfxtrx_core.RFX_DEVICES)

551
tests/components/rfxtrx/test_sensor.py
View File

@ -1,317 +1,308 @@
"""The tests for the Rfxtrx sensor platform.""" """The tests for the Rfxtrx sensor platform."""
import unittest
import pytest
from homeassistant.components import rfxtrx as rfxtrx_core from homeassistant.components import rfxtrx as rfxtrx_core
from homeassistant.const import TEMP_CELSIUS, UNIT_PERCENTAGE from homeassistant.const import TEMP_CELSIUS, UNIT_PERCENTAGE
from homeassistant.setup import setup_component from homeassistant.setup import async_setup_component
from tests.common import get_test_home_assistant, mock_component from . import _signal_event
@pytest.mark.skipif("os.environ.get('RFXTRX') != 'RUN'") async def test_default_config(hass, rfxtrx):
class TestSensorRfxtrx(unittest.TestCase): """Test with 0 sensor."""
"""Test the Rfxtrx sensor platform.""" await async_setup_component(
hass, "sensor", {"sensor": {"platform": "rfxtrx", "devices": {}}}
)
await hass.async_block_till_done()
def setUp(self): assert 0 == len(rfxtrx_core.RFX_DEVICES)
"""Set up things to be run when tests are started."""
self.hass = get_test_home_assistant()
mock_component(self.hass, "rfxtrx")
self.addCleanup(self.tear_down_cleanup)
def tear_down_cleanup(self):
"""Stop everything that was started."""
rfxtrx_core.RECEIVED_EVT_SUBSCRIBERS.clear()
rfxtrx_core.RFX_DEVICES.clear()
if rfxtrx_core.DATA_RFXOBJECT in self.hass.data:
self.hass.data[rfxtrx_core.DATA_RFXOBJECT].close_connection()
self.hass.stop()
def test_default_config(self): async def test_one_sensor(hass, rfxtrx):
"""Test with 0 sensor.""" """Test with 1 sensor."""
assert setup_component( await async_setup_component(
self.hass, "sensor", {"sensor": {"platform": "rfxtrx", "devices": {}}} hass,
) "sensor",
assert 0 == len(rfxtrx_core.RFX_DEVICES) {
"sensor": {
"platform": "rfxtrx",
"devices": {
"0a52080705020095220269": {
"name": "Test",
"data_type": "Temperature",
}
},
}
},
)
await hass.async_block_till_done()
def test_one_sensor(self): assert 1 == len(rfxtrx_core.RFX_DEVICES)
"""Test with 1 sensor.""" entity = rfxtrx_core.RFX_DEVICES["sensor_05_02"]["Temperature"]
assert setup_component( assert "Test Temperature" == entity.name
self.hass, assert TEMP_CELSIUS == entity.unit_of_measurement
"sensor", assert entity.state is None
{
"sensor": {
"platform": "rfxtrx", async def test_one_sensor_no_datatype(hass, rfxtrx):
"devices": { """Test with 1 sensor."""
"0a52080705020095220269": { await async_setup_component(
"name": "Test", hass,
"data_type": "Temperature", "sensor",
} {
"sensor": {
"platform": "rfxtrx",
"devices": {"0a52080705020095220269": {"name": "Test"}},
}
},
)
await hass.async_block_till_done()
assert 1 == len(rfxtrx_core.RFX_DEVICES)
entity = rfxtrx_core.RFX_DEVICES["sensor_05_02"]["Temperature"]
assert "Test Temperature" == entity.name
assert TEMP_CELSIUS == entity.unit_of_measurement
assert entity.state is None
async def test_several_sensors(hass, rfxtrx):
"""Test with 3 sensors."""
await async_setup_component(
hass,
"sensor",
{
"sensor": {
"platform": "rfxtrx",
"devices": {
"0a52080705020095220269": {
"name": "Test",
"data_type": "Temperature",
}, },
} "0a520802060100ff0e0269": {
}, "name": "Bath",
) "data_type": ["Temperature", "Humidity"],
assert 1 == len(rfxtrx_core.RFX_DEVICES)
entity = rfxtrx_core.RFX_DEVICES["sensor_05_02"]["Temperature"]
assert "Test Temperature" == entity.name
assert TEMP_CELSIUS == entity.unit_of_measurement
assert entity.state is None
def test_one_sensor_no_datatype(self):
"""Test with 1 sensor."""
assert setup_component(
self.hass,
"sensor",
{
"sensor": {
"platform": "rfxtrx",
"devices": {"0a52080705020095220269": {"name": "Test"}},
}
},
)
assert 1 == len(rfxtrx_core.RFX_DEVICES)
entity = rfxtrx_core.RFX_DEVICES["sensor_05_02"]["Temperature"]
assert "Test Temperature" == entity.name
assert TEMP_CELSIUS == entity.unit_of_measurement
assert entity.state is None
def test_several_sensors(self):
"""Test with 3 sensors."""
assert setup_component(
self.hass,
"sensor",
{
"sensor": {
"platform": "rfxtrx",
"devices": {
"0a52080705020095220269": {
"name": "Test",
"data_type": "Temperature",
},
"0a520802060100ff0e0269": {
"name": "Bath",
"data_type": ["Temperature", "Humidity"],
},
}, },
} },
}, }
) },
)
await hass.async_block_till_done()
assert 2 == len(rfxtrx_core.RFX_DEVICES) assert 2 == len(rfxtrx_core.RFX_DEVICES)
device_num = 0 device_num = 0
for id in rfxtrx_core.RFX_DEVICES: for id in rfxtrx_core.RFX_DEVICES:
if id == "sensor_06_01": if id == "sensor_06_01":
device_num = device_num + 1 device_num = device_num + 1
assert len(rfxtrx_core.RFX_DEVICES[id]) == 2 assert len(rfxtrx_core.RFX_DEVICES[id]) == 2
_entity_temp = rfxtrx_core.RFX_DEVICES[id]["Temperature"] _entity_temp = rfxtrx_core.RFX_DEVICES[id]["Temperature"]
_entity_hum = rfxtrx_core.RFX_DEVICES[id]["Humidity"] _entity_hum = rfxtrx_core.RFX_DEVICES[id]["Humidity"]
assert UNIT_PERCENTAGE == _entity_hum.unit_of_measurement assert UNIT_PERCENTAGE == _entity_hum.unit_of_measurement
assert "Bath" == _entity_hum.__str__() assert "Bath" == _entity_hum.__str__()
assert _entity_hum.state is None assert _entity_hum.state is None
assert TEMP_CELSIUS == _entity_temp.unit_of_measurement assert TEMP_CELSIUS == _entity_temp.unit_of_measurement
assert "Bath" == _entity_temp.__str__() assert "Bath" == _entity_temp.__str__()
elif id == "sensor_05_02": elif id == "sensor_05_02":
device_num = device_num + 1 device_num = device_num + 1
entity = rfxtrx_core.RFX_DEVICES[id]["Temperature"] entity = rfxtrx_core.RFX_DEVICES[id]["Temperature"]
assert entity.state is None assert entity.state is None
assert TEMP_CELSIUS == entity.unit_of_measurement assert TEMP_CELSIUS == entity.unit_of_measurement
assert "Test" == entity.__str__() assert "Test" == entity.__str__()
assert 2 == device_num assert 2 == device_num
def test_discover_sensor(self):
"""Test with discovery of sensor."""
assert setup_component(
self.hass,
"sensor",
{"sensor": {"platform": "rfxtrx", "automatic_add": True, "devices": {}}},
)
self.hass.block_till_done()
event = rfxtrx_core.get_rfx_object("0a520801070100b81b0279") async def test_discover_sensor(hass, rfxtrx):
event.data = bytearray(b"\nR\x08\x01\x07\x01\x00\xb8\x1b\x02y") """Test with discovery of sensor."""
rfxtrx_core.RECEIVED_EVT_SUBSCRIBERS[0](event) await async_setup_component(
hass,
"sensor",
{"sensor": {"platform": "rfxtrx", "automatic_add": True, "devices": {}}},
)
await hass.async_block_till_done()
entity = rfxtrx_core.RFX_DEVICES["sensor_07_01"]["Temperature"] event = rfxtrx_core.get_rfx_object("0a520801070100b81b0279")
assert 1 == len(rfxtrx_core.RFX_DEVICES) event.data = bytearray(b"\nR\x08\x01\x07\x01\x00\xb8\x1b\x02y")
assert { await _signal_event(hass, event)
"Humidity status": "normal",
"Temperature": 18.4,
"Rssi numeric": 7,
"Humidity": 27,
"Battery numeric": 9,
"Humidity status numeric": 2,
} == entity.device_state_attributes
assert "0a520801070100b81b0279" == entity.__str__()
rfxtrx_core.RECEIVED_EVT_SUBSCRIBERS[0](event) entity = rfxtrx_core.RFX_DEVICES["sensor_07_01"]["Temperature"]
assert 1 == len(rfxtrx_core.RFX_DEVICES) assert 1 == len(rfxtrx_core.RFX_DEVICES)
assert {
"Humidity status": "normal",
"Temperature": 18.4,
"Rssi numeric": 7,
"Humidity": 27,
"Battery numeric": 9,
"Humidity status numeric": 2,
} == entity.device_state_attributes
assert "0a520801070100b81b0279" == entity.__str__()
event = rfxtrx_core.get_rfx_object("0a52080405020095240279") await _signal_event(hass, event)
event.data = bytearray(b"\nR\x08\x04\x05\x02\x00\x95$\x02y") assert 1 == len(rfxtrx_core.RFX_DEVICES)
rfxtrx_core.RECEIVED_EVT_SUBSCRIBERS[0](event)
entity = rfxtrx_core.RFX_DEVICES["sensor_05_02"]["Temperature"]
assert 2 == len(rfxtrx_core.RFX_DEVICES)
assert {
"Humidity status": "normal",
"Temperature": 14.9,
"Rssi numeric": 7,
"Humidity": 36,
"Battery numeric": 9,
"Humidity status numeric": 2,
} == entity.device_state_attributes
assert "0a52080405020095240279" == entity.__str__()
event = rfxtrx_core.get_rfx_object("0a52085e070100b31b0279") event = rfxtrx_core.get_rfx_object("0a52080405020095240279")
event.data = bytearray(b"\nR\x08^\x07\x01\x00\xb3\x1b\x02y") event.data = bytearray(b"\nR\x08\x04\x05\x02\x00\x95$\x02y")
rfxtrx_core.RECEIVED_EVT_SUBSCRIBERS[0](event) await _signal_event(hass, event)
entity = rfxtrx_core.RFX_DEVICES["sensor_07_01"]["Temperature"] entity = rfxtrx_core.RFX_DEVICES["sensor_05_02"]["Temperature"]
assert 2 == len(rfxtrx_core.RFX_DEVICES) assert 2 == len(rfxtrx_core.RFX_DEVICES)
assert { assert {
"Humidity status": "normal", "Humidity status": "normal",
"Temperature": 17.9, "Temperature": 14.9,
"Rssi numeric": 7, "Rssi numeric": 7,
"Humidity": 27, "Humidity": 36,
"Battery numeric": 9, "Battery numeric": 9,
"Humidity status numeric": 2, "Humidity status numeric": 2,
} == entity.device_state_attributes } == entity.device_state_attributes
assert "0a520801070100b81b0279" == entity.__str__() assert "0a52080405020095240279" == entity.__str__()
# trying to add a switch event = rfxtrx_core.get_rfx_object("0a52085e070100b31b0279")
event = rfxtrx_core.get_rfx_object("0b1100cd0213c7f210010f70") event.data = bytearray(b"\nR\x08^\x07\x01\x00\xb3\x1b\x02y")
rfxtrx_core.RECEIVED_EVT_SUBSCRIBERS[0](event) await _signal_event(hass, event)
assert 2 == len(rfxtrx_core.RFX_DEVICES) entity = rfxtrx_core.RFX_DEVICES["sensor_07_01"]["Temperature"]
assert 2 == len(rfxtrx_core.RFX_DEVICES)
assert {
"Humidity status": "normal",
"Temperature": 17.9,
"Rssi numeric": 7,
"Humidity": 27,
"Battery numeric": 9,
"Humidity status numeric": 2,
} == entity.device_state_attributes
assert "0a520801070100b81b0279" == entity.__str__()
def test_discover_sensor_noautoadd(self): # trying to add a switch
"""Test with discover of sensor when auto add is False.""" event = rfxtrx_core.get_rfx_object("0b1100cd0213c7f210010f70")
assert setup_component( await _signal_event(hass, event)
self.hass, assert 2 == len(rfxtrx_core.RFX_DEVICES)
"sensor",
{"sensor": {"platform": "rfxtrx", "automatic_add": False, "devices": {}}},
)
event = rfxtrx_core.get_rfx_object("0a520801070100b81b0279")
event.data = bytearray(b"\nR\x08\x01\x07\x01\x00\xb8\x1b\x02y")
assert 0 == len(rfxtrx_core.RFX_DEVICES) async def test_discover_sensor_noautoadd(hass, rfxtrx):
rfxtrx_core.RECEIVED_EVT_SUBSCRIBERS[0](event) """Test with discover of sensor when auto add is False."""
assert 0 == len(rfxtrx_core.RFX_DEVICES) await async_setup_component(
hass,
"sensor",
{"sensor": {"platform": "rfxtrx", "automatic_add": False, "devices": {}}},
)
await hass.async_block_till_done()
rfxtrx_core.RECEIVED_EVT_SUBSCRIBERS[0](event) event = rfxtrx_core.get_rfx_object("0a520801070100b81b0279")
assert 0 == len(rfxtrx_core.RFX_DEVICES) event.data = bytearray(b"\nR\x08\x01\x07\x01\x00\xb8\x1b\x02y")
event = rfxtrx_core.get_rfx_object("0a52080405020095240279") assert 0 == len(rfxtrx_core.RFX_DEVICES)
event.data = bytearray(b"\nR\x08\x04\x05\x02\x00\x95$\x02y") await _signal_event(hass, event)
rfxtrx_core.RECEIVED_EVT_SUBSCRIBERS[0](event) assert 0 == len(rfxtrx_core.RFX_DEVICES)
assert 0 == len(rfxtrx_core.RFX_DEVICES)
event = rfxtrx_core.get_rfx_object("0a52085e070100b31b0279") await _signal_event(hass, event)
event.data = bytearray(b"\nR\x08^\x07\x01\x00\xb3\x1b\x02y") assert 0 == len(rfxtrx_core.RFX_DEVICES)
rfxtrx_core.RECEIVED_EVT_SUBSCRIBERS[0](event)
assert 0 == len(rfxtrx_core.RFX_DEVICES)
def test_update_of_sensors(self): event = rfxtrx_core.get_rfx_object("0a52080405020095240279")
"""Test with 3 sensors.""" event.data = bytearray(b"\nR\x08\x04\x05\x02\x00\x95$\x02y")
assert setup_component( await _signal_event(hass, event)
self.hass, assert 0 == len(rfxtrx_core.RFX_DEVICES)
"sensor",
{ event = rfxtrx_core.get_rfx_object("0a52085e070100b31b0279")
"sensor": { event.data = bytearray(b"\nR\x08^\x07\x01\x00\xb3\x1b\x02y")
"platform": "rfxtrx", await _signal_event(hass, event)
"devices": { assert 0 == len(rfxtrx_core.RFX_DEVICES)
"0a52080705020095220269": {
"name": "Test",
"data_type": "Temperature", async def test_update_of_sensors(hass, rfxtrx):
}, """Test with 3 sensors."""
"0a520802060100ff0e0269": { await async_setup_component(
"name": "Bath", hass,
"data_type": ["Temperature", "Humidity"], "sensor",
}, {
"sensor": {
"platform": "rfxtrx",
"devices": {
"0a52080705020095220269": {
"name": "Test",
"data_type": "Temperature",
}, },
} "0a520802060100ff0e0269": {
}, "name": "Bath",
) "data_type": ["Temperature", "Humidity"],
},
},
}
},
)
await hass.async_block_till_done()
assert 2 == len(rfxtrx_core.RFX_DEVICES) assert 2 == len(rfxtrx_core.RFX_DEVICES)
device_num = 0 device_num = 0
for id in rfxtrx_core.RFX_DEVICES: for id in rfxtrx_core.RFX_DEVICES:
if id == "sensor_06_01": if id == "sensor_06_01":
device_num = device_num + 1 device_num = device_num + 1
assert len(rfxtrx_core.RFX_DEVICES[id]) == 2 assert len(rfxtrx_core.RFX_DEVICES[id]) == 2
_entity_temp = rfxtrx_core.RFX_DEVICES[id]["Temperature"] _entity_temp = rfxtrx_core.RFX_DEVICES[id]["Temperature"]
_entity_hum = rfxtrx_core.RFX_DEVICES[id]["Humidity"] _entity_hum = rfxtrx_core.RFX_DEVICES[id]["Humidity"]
assert UNIT_PERCENTAGE == _entity_hum.unit_of_measurement assert UNIT_PERCENTAGE == _entity_hum.unit_of_measurement
assert "Bath" == _entity_hum.__str__() assert "Bath" == _entity_hum.__str__()
assert _entity_temp.state is None assert _entity_temp.state is None
assert TEMP_CELSIUS == _entity_temp.unit_of_measurement assert TEMP_CELSIUS == _entity_temp.unit_of_measurement
assert "Bath" == _entity_temp.__str__() assert "Bath" == _entity_temp.__str__()
elif id == "sensor_05_02": elif id == "sensor_05_02":
device_num = device_num + 1 device_num = device_num + 1
entity = rfxtrx_core.RFX_DEVICES[id]["Temperature"] entity = rfxtrx_core.RFX_DEVICES[id]["Temperature"]
assert entity.state is None assert entity.state is None
assert TEMP_CELSIUS == entity.unit_of_measurement assert TEMP_CELSIUS == entity.unit_of_measurement
assert "Test" == entity.__str__() assert "Test" == entity.__str__()
assert 2 == device_num assert 2 == device_num
event = rfxtrx_core.get_rfx_object("0a520802060101ff0f0269") event = rfxtrx_core.get_rfx_object("0a520802060101ff0f0269")
event.data = bytearray(b"\nR\x08\x01\x07\x01\x00\xb8\x1b\x02y") event.data = bytearray(b"\nR\x08\x01\x07\x01\x00\xb8\x1b\x02y")
rfxtrx_core.RECEIVED_EVT_SUBSCRIBERS[0](event) await _signal_event(hass, event)
rfxtrx_core.RECEIVED_EVT_SUBSCRIBERS[0](event) await _signal_event(hass, event)
event = rfxtrx_core.get_rfx_object("0a52080705020085220269") event = rfxtrx_core.get_rfx_object("0a52080705020085220269")
event.data = bytearray(b"\nR\x08\x04\x05\x02\x00\x95$\x02y") event.data = bytearray(b"\nR\x08\x04\x05\x02\x00\x95$\x02y")
rfxtrx_core.RECEIVED_EVT_SUBSCRIBERS[0](event) await _signal_event(hass, event)
assert 2 == len(rfxtrx_core.RFX_DEVICES) assert 2 == len(rfxtrx_core.RFX_DEVICES)
device_num = 0 device_num = 0
for id in rfxtrx_core.RFX_DEVICES: for id in rfxtrx_core.RFX_DEVICES:
if id == "sensor_06_01": if id == "sensor_06_01":
device_num = device_num + 1 device_num = device_num + 1
assert len(rfxtrx_core.RFX_DEVICES[id]) == 2 assert len(rfxtrx_core.RFX_DEVICES[id]) == 2
_entity_temp = rfxtrx_core.RFX_DEVICES[id]["Temperature"] _entity_temp = rfxtrx_core.RFX_DEVICES[id]["Temperature"]
_entity_hum = rfxtrx_core.RFX_DEVICES[id]["Humidity"] _entity_hum = rfxtrx_core.RFX_DEVICES[id]["Humidity"]
assert UNIT_PERCENTAGE == _entity_hum.unit_of_measurement assert UNIT_PERCENTAGE == _entity_hum.unit_of_measurement
assert 15 == _entity_hum.state assert 15 == _entity_hum.state
assert { assert {
"Battery numeric": 9, "Battery numeric": 9,
"Temperature": 51.1, "Temperature": 51.1,
"Humidity": 15, "Humidity": 15,
"Humidity status": "normal", "Humidity status": "normal",
"Humidity status numeric": 2, "Humidity status numeric": 2,
"Rssi numeric": 6, "Rssi numeric": 6,
} == _entity_hum.device_state_attributes } == _entity_hum.device_state_attributes
assert "Bath" == _entity_hum.__str__() assert "Bath" == _entity_hum.__str__()
assert TEMP_CELSIUS == _entity_temp.unit_of_measurement assert TEMP_CELSIUS == _entity_temp.unit_of_measurement
assert 51.1 == _entity_temp.state assert 51.1 == _entity_temp.state
assert { assert {
"Battery numeric": 9, "Battery numeric": 9,
"Temperature": 51.1, "Temperature": 51.1,
"Humidity": 15, "Humidity": 15,
"Humidity status": "normal", "Humidity status": "normal",
"Humidity status numeric": 2, "Humidity status numeric": 2,
"Rssi numeric": 6, "Rssi numeric": 6,
} == _entity_temp.device_state_attributes } == _entity_temp.device_state_attributes
assert "Bath" == _entity_temp.__str__() assert "Bath" == _entity_temp.__str__()
elif id == "sensor_05_02": elif id == "sensor_05_02":
device_num = device_num + 1 device_num = device_num + 1
entity = rfxtrx_core.RFX_DEVICES[id]["Temperature"] entity = rfxtrx_core.RFX_DEVICES[id]["Temperature"]
assert TEMP_CELSIUS == entity.unit_of_measurement assert TEMP_CELSIUS == entity.unit_of_measurement
assert 13.3 == entity.state assert 13.3 == entity.state
assert { assert {
"Humidity status": "normal", "Humidity status": "normal",
"Temperature": 13.3, "Temperature": 13.3,
"Rssi numeric": 6, "Rssi numeric": 6,
"Humidity": 34, "Humidity": 34,
"Battery numeric": 9, "Battery numeric": 9,
"Humidity status numeric": 2, "Humidity status numeric": 2,
} == entity.device_state_attributes } == entity.device_state_attributes
assert "Test" == entity.__str__() assert "Test" == entity.__str__()
assert 2 == device_num assert 2 == device_num
assert 2 == len(rfxtrx_core.RFX_DEVICES) assert 2 == len(rfxtrx_core.RFX_DEVICES)

417
tests/components/rfxtrx/test_switch.py
View File

@ -1,37 +1,19 @@
"""The tests for the RFXtrx switch platform.""" """The tests for the RFXtrx switch platform."""
import unittest
import RFXtrx as rfxtrxmod import RFXtrx as rfxtrxmod
import pytest
from homeassistant.components import rfxtrx as rfxtrx_core from homeassistant.components import rfxtrx as rfxtrx_core
from homeassistant.setup import setup_component from homeassistant.setup import async_setup_component
from tests.common import assert_setup_component, get_test_home_assistant, mock_component from . import _signal_event
from tests.common import assert_setup_component
@pytest.mark.skipif("os.environ.get('RFXTRX') != 'RUN'") async def test_valid_config(hass, rfxtrx):
class TestSwitchRfxtrx(unittest.TestCase): """Test configuration."""
"""Test the RFXtrx switch platform.""" with assert_setup_component(1):
await async_setup_component(
def setUp(self): hass,
"""Set up things to be run when tests are started."""
self.hass = get_test_home_assistant()
mock_component(self.hass, "rfxtrx")
self.addCleanup(self.tear_down_cleanup)
def tear_down_cleanup(self):
"""Stop everything that was started."""
rfxtrx_core.RECEIVED_EVT_SUBSCRIBERS.clear()
rfxtrx_core.RFX_DEVICES.clear()
if rfxtrx_core.DATA_RFXOBJECT in self.hass.data:
self.hass.data[rfxtrx_core.DATA_RFXOBJECT].close_connection()
self.hass.stop()
def test_valid_config(self):
"""Test configuration."""
assert setup_component(
self.hass,
"switch", "switch",
{ {
"switch": { "switch": {
@ -46,11 +28,14 @@ class TestSwitchRfxtrx(unittest.TestCase):
} }
}, },
) )
await hass.async_block_till_done()
def test_valid_config_int_device_id(self):
"""Test configuration.""" async def test_valid_config_int_device_id(hass, rfxtrx):
assert setup_component( """Test configuration."""
self.hass, with assert_setup_component(1):
await async_setup_component(
hass,
"switch", "switch",
{ {
"switch": { "switch": {
@ -65,211 +50,225 @@ class TestSwitchRfxtrx(unittest.TestCase):
} }
}, },
) )
await hass.async_block_till_done()
def test_invalid_config2(self):
"""Test invalid configuration."""
with assert_setup_component(0):
setup_component(
self.hass,
"switch",
{
"switch": {
"platform": "rfxtrx",
"automatic_add": True,
"invalid_key": "afda",
"devices": {
"0b1100cd0213c7f210010f51": {
"name": "Test",
rfxtrx_core.ATTR_FIRE_EVENT: True,
}
},
}
},
)
def test_default_config(self): async def test_invalid_config2(hass, rfxtrx):
"""Test with 0 switches.""" """Test invalid configuration."""
assert setup_component( with assert_setup_component(0):
self.hass, "switch", {"switch": {"platform": "rfxtrx", "devices": {}}} await async_setup_component(
) hass,
assert 0 == len(rfxtrx_core.RFX_DEVICES)
def test_one_switch(self):
"""Test with 1 switch."""
assert setup_component(
self.hass,
"switch", "switch",
{ {
"switch": { "switch": {
"platform": "rfxtrx", "platform": "rfxtrx",
"devices": {"0b1100cd0213c7f210010f51": {"name": "Test"}}, "automatic_add": True,
} "invalid_key": "afda",
},
)
self.hass.data[rfxtrx_core.DATA_RFXOBJECT] = rfxtrxmod.Core(
"", transport_protocol=rfxtrxmod.DummyTransport
)
assert 1 == len(rfxtrx_core.RFX_DEVICES)
entity = rfxtrx_core.RFX_DEVICES["213c7f2_16"]
entity.hass = self.hass
assert "Test" == entity.name
assert "off" == entity.state
assert entity.assumed_state
assert entity.signal_repetitions == 1
assert not entity.should_fire_event
assert not entity.should_poll
assert not entity.is_on
entity.turn_on()
assert entity.is_on
entity.turn_off()
assert not entity.is_on
assert "Test" == entity.name
assert "off" == entity.state
entity.turn_on()
assert "on" == entity.state
entity.turn_off()
assert "off" == entity.state
def test_several_switches(self):
"""Test with 3 switches."""
assert setup_component(
self.hass,
"switch",
{
"switch": {
"platform": "rfxtrx",
"signal_repetitions": 3,
"devices": { "devices": {
"0b1100cd0213c7f230010f71": {"name": "Test"}, "0b1100cd0213c7f210010f51": {
"0b1100100118cdea02010f70": {"name": "Bath"}, "name": "Test",
"0b1100101118cdea02010f70": {"name": "Living"}, rfxtrx_core.ATTR_FIRE_EVENT: True,
}
}, },
} }
}, },
) )
await hass.async_block_till_done()
assert 3 == len(rfxtrx_core.RFX_DEVICES)
device_num = 0
for id in rfxtrx_core.RFX_DEVICES:
entity = rfxtrx_core.RFX_DEVICES[id]
assert entity.signal_repetitions == 3
if entity.name == "Living":
device_num = device_num + 1
assert "off" == entity.state
assert "<Entity Living: off>" == entity.__str__()
elif entity.name == "Bath":
device_num = device_num + 1
assert "off" == entity.state
assert "<Entity Bath: off>" == entity.__str__()
elif entity.name == "Test":
device_num = device_num + 1
assert "off" == entity.state
assert "<Entity Test: off>" == entity.__str__()
assert 3 == device_num async def test_default_config(hass, rfxtrx):
"""Test with 0 switches."""
await async_setup_component(
hass, "switch", {"switch": {"platform": "rfxtrx", "devices": {}}}
)
await hass.async_block_till_done()
def test_discover_switch(self): assert 0 == len(rfxtrx_core.RFX_DEVICES)
"""Test with discovery of switches."""
assert setup_component(
self.hass,
"switch",
{"switch": {"platform": "rfxtrx", "automatic_add": True, "devices": {}}},
)
event = rfxtrx_core.get_rfx_object("0b1100100118cdea02010f70")
event.data = bytearray(
[0x0B, 0x11, 0x00, 0x10, 0x01, 0x18, 0xCD, 0xEA, 0x01, 0x01, 0x0F, 0x70]
)
rfxtrx_core.RECEIVED_EVT_SUBSCRIBERS[0](event) async def test_one_switch(hass, rfxtrx):
entity = rfxtrx_core.RFX_DEVICES["118cdea_2"] """Test with 1 switch."""
assert 1 == len(rfxtrx_core.RFX_DEVICES) await async_setup_component(
assert "<Entity 0b1100100118cdea01010f70: on>" == entity.__str__() hass,
"switch",
{
"switch": {
"platform": "rfxtrx",
"devices": {"0b1100cd0213c7f210010f51": {"name": "Test"}},
}
},
)
await hass.async_block_till_done()
rfxtrx_core.RECEIVED_EVT_SUBSCRIBERS[0](event) hass.data[rfxtrx_core.DATA_RFXOBJECT] = rfxtrxmod.Core(
assert 1 == len(rfxtrx_core.RFX_DEVICES) "", transport_protocol=rfxtrxmod.DummyTransport
)
event = rfxtrx_core.get_rfx_object("0b1100100118cdeb02010f70") assert 1 == len(rfxtrx_core.RFX_DEVICES)
event.data = bytearray( entity = rfxtrx_core.RFX_DEVICES["213c7f2_16"]
[0x0B, 0x11, 0x00, 0x12, 0x01, 0x18, 0xCD, 0xEA, 0x02, 0x00, 0x00, 0x70] entity.hass = hass
) assert "Test" == entity.name
assert "off" == entity.state
assert entity.assumed_state
assert entity.signal_repetitions == 1
assert not entity.should_fire_event
assert not entity.should_poll
rfxtrx_core.RECEIVED_EVT_SUBSCRIBERS[0](event) assert not entity.is_on
entity = rfxtrx_core.RFX_DEVICES["118cdeb_2"] entity.turn_on()
assert 2 == len(rfxtrx_core.RFX_DEVICES) assert entity.is_on
assert "<Entity 0b1100120118cdea02000070: on>" == entity.__str__() entity.turn_off()
assert not entity.is_on
# Trying to add a sensor assert "Test" == entity.name
event = rfxtrx_core.get_rfx_object("0a52085e070100b31b0279") assert "off" == entity.state
event.data = bytearray(b"\nR\x08^\x07\x01\x00\xb3\x1b\x02y") entity.turn_on()
rfxtrx_core.RECEIVED_EVT_SUBSCRIBERS[0](event) assert "on" == entity.state
assert 2 == len(rfxtrx_core.RFX_DEVICES) entity.turn_off()
assert "off" == entity.state
# Trying to add a light
event = rfxtrx_core.get_rfx_object("0b1100100118cdea02010f70")
event.data = bytearray(
[0x0B, 0x11, 0x11, 0x10, 0x01, 0x18, 0xCD, 0xEA, 0x01, 0x02, 0x0F, 0x70]
)
rfxtrx_core.RECEIVED_EVT_SUBSCRIBERS[0](event)
assert 2 == len(rfxtrx_core.RFX_DEVICES)
# Trying to add a rollershutter async def test_several_switches(hass, rfxtrx):
event = rfxtrx_core.get_rfx_object("0a1400adf394ab020e0060") """Test with 3 switches."""
event.data = bytearray( await async_setup_component(
[0x0A, 0x14, 0x00, 0xAD, 0xF3, 0x94, 0xAB, 0x02, 0x0E, 0x00, 0x60] hass,
) "switch",
rfxtrx_core.RECEIVED_EVT_SUBSCRIBERS[0](event) {
assert 2 == len(rfxtrx_core.RFX_DEVICES) "switch": {
"platform": "rfxtrx",
"signal_repetitions": 3,
"devices": {
"0b1100cd0213c7f230010f71": {"name": "Test"},
"0b1100100118cdea02010f70": {"name": "Bath"},
"0b1100101118cdea02010f70": {"name": "Living"},
},
}
},
)
await hass.async_block_till_done()
def test_discover_switch_noautoadd(self): assert 3 == len(rfxtrx_core.RFX_DEVICES)
"""Test with discovery of switch when auto add is False.""" device_num = 0
assert setup_component( for id in rfxtrx_core.RFX_DEVICES:
self.hass, entity = rfxtrx_core.RFX_DEVICES[id]
"switch", assert entity.signal_repetitions == 3
{"switch": {"platform": "rfxtrx", "automatic_add": False, "devices": {}}}, if entity.name == "Living":
) device_num = device_num + 1
assert "off" == entity.state
assert "<Entity Living: off>" == entity.__str__()
elif entity.name == "Bath":
device_num = device_num + 1
assert "off" == entity.state
assert "<Entity Bath: off>" == entity.__str__()
elif entity.name == "Test":
device_num = device_num + 1
assert "off" == entity.state
assert "<Entity Test: off>" == entity.__str__()
event = rfxtrx_core.get_rfx_object("0b1100100118cdea02010f70") assert 3 == device_num
event.data = bytearray(
[0x0B, 0x11, 0x00, 0x10, 0x01, 0x18, 0xCD, 0xEA, 0x01, 0x01, 0x0F, 0x70]
)
rfxtrx_core.RECEIVED_EVT_SUBSCRIBERS[0](event)
assert 0 == len(rfxtrx_core.RFX_DEVICES)
assert 0 == len(rfxtrx_core.RFX_DEVICES)
rfxtrx_core.RECEIVED_EVT_SUBSCRIBERS[0](event) async def test_discover_switch(hass, rfxtrx):
assert 0 == len(rfxtrx_core.RFX_DEVICES) """Test with discovery of switches."""
await async_setup_component(
hass,
"switch",
{"switch": {"platform": "rfxtrx", "automatic_add": True, "devices": {}}},
)
await hass.async_block_till_done()
event = rfxtrx_core.get_rfx_object("0b1100100118cdeb02010f70") event = rfxtrx_core.get_rfx_object("0b1100100118cdea02010f70")
event.data = bytearray( event.data = bytearray(
[0x0B, 0x11, 0x00, 0x12, 0x01, 0x18, 0xCD, 0xEA, 0x02, 0x00, 0x00, 0x70] [0x0B, 0x11, 0x00, 0x10, 0x01, 0x18, 0xCD, 0xEA, 0x01, 0x01, 0x0F, 0x70]
) )
rfxtrx_core.RECEIVED_EVT_SUBSCRIBERS[0](event)
assert 0 == len(rfxtrx_core.RFX_DEVICES)
# Trying to add a sensor await _signal_event(hass, event)
event = rfxtrx_core.get_rfx_object("0a52085e070100b31b0279") entity = rfxtrx_core.RFX_DEVICES["118cdea_2"]
event.data = bytearray(b"\nR\x08^\x07\x01\x00\xb3\x1b\x02y") assert 1 == len(rfxtrx_core.RFX_DEVICES)
rfxtrx_core.RECEIVED_EVT_SUBSCRIBERS[0](event) assert "<Entity 0b1100100118cdea01010f70: on>" == entity.__str__()
assert 0 == len(rfxtrx_core.RFX_DEVICES)
# Trying to add a light await _signal_event(hass, event)
event = rfxtrx_core.get_rfx_object("0b1100100118cdea02010f70") assert 1 == len(rfxtrx_core.RFX_DEVICES)
event.data = bytearray(
[0x0B, 0x11, 0x11, 0x10, 0x01, 0x18, 0xCD, 0xEA, 0x01, 0x02, 0x0F, 0x70]
)
rfxtrx_core.RECEIVED_EVT_SUBSCRIBERS[0](event)
assert 0 == len(rfxtrx_core.RFX_DEVICES)
# Trying to add a rollershutter event = rfxtrx_core.get_rfx_object("0b1100100118cdeb02010f70")
event = rfxtrx_core.get_rfx_object("0a1400adf394ab020e0060") event.data = bytearray(
event.data = bytearray( [0x0B, 0x11, 0x00, 0x12, 0x01, 0x18, 0xCD, 0xEA, 0x02, 0x00, 0x00, 0x70]
[0x0A, 0x14, 0x00, 0xAD, 0xF3, 0x94, 0xAB, 0x02, 0x0E, 0x00, 0x60] )
)
rfxtrx_core.RECEIVED_EVT_SUBSCRIBERS[0](event) await _signal_event(hass, event)
assert 0 == len(rfxtrx_core.RFX_DEVICES) entity = rfxtrx_core.RFX_DEVICES["118cdeb_2"]
assert 2 == len(rfxtrx_core.RFX_DEVICES)
assert "<Entity 0b1100120118cdea02000070: on>" == entity.__str__()
# Trying to add a sensor
event = rfxtrx_core.get_rfx_object("0a52085e070100b31b0279")
event.data = bytearray(b"\nR\x08^\x07\x01\x00\xb3\x1b\x02y")
await _signal_event(hass, event)
assert 2 == len(rfxtrx_core.RFX_DEVICES)
# Trying to add a light
event = rfxtrx_core.get_rfx_object("0b1100100118cdea02010f70")
event.data = bytearray(
[0x0B, 0x11, 0x11, 0x10, 0x01, 0x18, 0xCD, 0xEA, 0x01, 0x02, 0x0F, 0x70]
)
await _signal_event(hass, event)
assert 2 == len(rfxtrx_core.RFX_DEVICES)
# Trying to add a rollershutter
event = rfxtrx_core.get_rfx_object("0a1400adf394ab020e0060")
event.data = bytearray(
[0x0A, 0x14, 0x00, 0xAD, 0xF3, 0x94, 0xAB, 0x02, 0x0E, 0x00, 0x60]
)
await _signal_event(hass, event)
assert 2 == len(rfxtrx_core.RFX_DEVICES)
async def test_discover_switch_noautoadd(hass, rfxtrx):
"""Test with discovery of switch when auto add is False."""
await async_setup_component(
hass,
"switch",
{"switch": {"platform": "rfxtrx", "automatic_add": False, "devices": {}}},
)
await hass.async_block_till_done()
event = rfxtrx_core.get_rfx_object("0b1100100118cdea02010f70")
event.data = bytearray(
[0x0B, 0x11, 0x00, 0x10, 0x01, 0x18, 0xCD, 0xEA, 0x01, 0x01, 0x0F, 0x70]
)
await _signal_event(hass, event)
assert 0 == len(rfxtrx_core.RFX_DEVICES)
assert 0 == len(rfxtrx_core.RFX_DEVICES)
await _signal_event(hass, event)
assert 0 == len(rfxtrx_core.RFX_DEVICES)
event = rfxtrx_core.get_rfx_object("0b1100100118cdeb02010f70")
event.data = bytearray(
[0x0B, 0x11, 0x00, 0x12, 0x01, 0x18, 0xCD, 0xEA, 0x02, 0x00, 0x00, 0x70]
)
await _signal_event(hass, event)
assert 0 == len(rfxtrx_core.RFX_DEVICES)
# Trying to add a sensor
event = rfxtrx_core.get_rfx_object("0a52085e070100b31b0279")
event.data = bytearray(b"\nR\x08^\x07\x01\x00\xb3\x1b\x02y")
await _signal_event(hass, event)
assert 0 == len(rfxtrx_core.RFX_DEVICES)
# Trying to add a light
event = rfxtrx_core.get_rfx_object("0b1100100118cdea02010f70")
event.data = bytearray(
[0x0B, 0x11, 0x11, 0x10, 0x01, 0x18, 0xCD, 0xEA, 0x01, 0x02, 0x0F, 0x70]
)
await _signal_event(hass, event)
assert 0 == len(rfxtrx_core.RFX_DEVICES)
# Trying to add a rollershutter
event = rfxtrx_core.get_rfx_object("0a1400adf394ab020e0060")
event.data = bytearray(
[0x0A, 0x14, 0x00, 0xAD, 0xF3, 0x94, 0xAB, 0x02, 0x0E, 0x00, 0x60]
)
await _signal_event(hass, event)
assert 0 == len(rfxtrx_core.RFX_DEVICES)