From 1d28f485d3125bae9457294f3733f45ffef1bc86 Mon Sep 17 00:00:00 2001 From: mptei Date: Sun, 11 Apr 2021 23:01:30 +0200 Subject: [PATCH] Patch ip interface instead of XKNX in knx (#49064) * knx: Deeper tests. * Set rate_limit to 0; removed waiting for queue --- tests/components/knx/__init__.py | 26 ++++++++++ tests/components/knx/conftest.py | 15 ++++++ tests/components/knx/test_expose.py | 74 ++++++++++------------------- 3 files changed, 66 insertions(+), 49 deletions(-) create mode 100644 tests/components/knx/conftest.py diff --git a/tests/components/knx/__init__.py b/tests/components/knx/__init__.py index eaa84714dc5..1c9bfaf15b8 100644 --- a/tests/components/knx/__init__.py +++ b/tests/components/knx/__init__.py @@ -1 +1,27 @@ """Tests for the KNX integration.""" + +from unittest.mock import DEFAULT, patch + +from homeassistant.components.knx.const import DOMAIN as KNX_DOMAIN +from homeassistant.setup import async_setup_component + + +async def setup_knx_integration(hass, knx_ip_interface, config=None): + """Create the KNX gateway.""" + if config is None: + config = {} + + # To get the XKNX object from the constructor call + def side_effect(*args, **kwargs): + knx_ip_interface.xknx = args[0] + # switch off rate delimiter + knx_ip_interface.xknx.rate_limit = 0 + return DEFAULT + + with patch( + "xknx.xknx.KNXIPInterface", + return_value=knx_ip_interface, + side_effect=side_effect, + ): + await async_setup_component(hass, KNX_DOMAIN, {KNX_DOMAIN: config}) + await hass.async_block_till_done() diff --git a/tests/components/knx/conftest.py b/tests/components/knx/conftest.py new file mode 100644 index 00000000000..b7c27774f78 --- /dev/null +++ b/tests/components/knx/conftest.py @@ -0,0 +1,15 @@ +"""conftest for knx.""" + +from unittest.mock import AsyncMock, Mock + +import pytest + + +@pytest.fixture(autouse=True) +def knx_ip_interface_mock(): + """Create a knx ip interface mock.""" + mock = Mock() + mock.start = AsyncMock() + mock.stop = AsyncMock() + mock.send_telegram = AsyncMock() + return mock diff --git a/tests/components/knx/test_expose.py b/tests/components/knx/test_expose.py index 1590a7bb246..908ef0a56f8 100644 --- a/tests/components/knx/test_expose.py +++ b/tests/components/knx/test_expose.py @@ -1,46 +1,18 @@ """Test knx expose.""" -from unittest.mock import AsyncMock, Mock, patch -import pytest - -from homeassistant.components.knx import ( - CONF_KNX_EXPOSE, - CONFIG_SCHEMA as KNX_CONFIG_SCHEMA, - KNX_ADDRESS, -) -from homeassistant.components.knx.const import DOMAIN as KNX_DOMAIN +from homeassistant.components.knx import CONF_KNX_EXPOSE, KNX_ADDRESS from homeassistant.const import CONF_ATTRIBUTE, CONF_ENTITY_ID, CONF_TYPE -from homeassistant.setup import async_setup_component + +from . import setup_knx_integration -async def setup_knx_integration(hass, knx_mock, config=None): - """Create the KNX gateway.""" - if config is None: - config = {} - with patch("homeassistant.components.knx.XKNX", return_value=knx_mock): - await async_setup_component( - hass, KNX_DOMAIN, KNX_CONFIG_SCHEMA({KNX_DOMAIN: config}) - ) - await hass.async_block_till_done() - - -@pytest.fixture(autouse=True) -def xknx_mock(): - """Create a simple XKNX mock.""" - xknx_mock = Mock() - xknx_mock.telegrams = AsyncMock() - xknx_mock.start = AsyncMock() - xknx_mock.stop = AsyncMock() - return xknx_mock - - -async def test_binary_expose(hass, xknx_mock): +async def test_binary_expose(hass, knx_ip_interface_mock): """Test that a binary expose sends only telegrams on state change.""" entity_id = "fake.entity" await setup_knx_integration( hass, - xknx_mock, + knx_ip_interface_mock, { CONF_KNX_EXPOSE: { CONF_TYPE: "binary", @@ -52,33 +24,37 @@ async def test_binary_expose(hass, xknx_mock): assert not hass.states.async_all() # Change state to on - xknx_mock.reset_mock() + knx_ip_interface_mock.reset_mock() hass.states.async_set(entity_id, "on", {}) await hass.async_block_till_done() - assert xknx_mock.telegrams.put.call_count == 1, "Expected telegram for state change" + assert ( + knx_ip_interface_mock.send_telegram.call_count == 1 + ), "Expected telegram for state change" # Change attribute; keep state - xknx_mock.reset_mock() + knx_ip_interface_mock.reset_mock() hass.states.async_set(entity_id, "on", {"brightness": 180}) await hass.async_block_till_done() assert ( - xknx_mock.telegrams.put.call_count == 0 + knx_ip_interface_mock.send_telegram.call_count == 0 ), "Expected no telegram; state not changed" # Change attribute and state - xknx_mock.reset_mock() + knx_ip_interface_mock.reset_mock() hass.states.async_set(entity_id, "off", {"brightness": 0}) await hass.async_block_till_done() - assert xknx_mock.telegrams.put.call_count == 1, "Expected telegram for state change" + assert ( + knx_ip_interface_mock.send_telegram.call_count == 1 + ), "Expected telegram for state change" -async def test_expose_attribute(hass, xknx_mock): +async def test_expose_attribute(hass, knx_ip_interface_mock): """Test that an expose sends only telegrams on attribute change.""" entity_id = "fake.entity" attribute = "fake_attribute" await setup_knx_integration( hass, - xknx_mock, + knx_ip_interface_mock, { CONF_KNX_EXPOSE: { CONF_TYPE: "percentU8", @@ -91,25 +67,25 @@ async def test_expose_attribute(hass, xknx_mock): assert not hass.states.async_all() # Change state to on; no attribute - xknx_mock.reset_mock() + knx_ip_interface_mock.reset_mock() hass.states.async_set(entity_id, "on", {}) await hass.async_block_till_done() - assert xknx_mock.telegrams.put.call_count == 0 + assert knx_ip_interface_mock.send_telegram.call_count == 0 # Change attribute; keep state - xknx_mock.reset_mock() + knx_ip_interface_mock.reset_mock() hass.states.async_set(entity_id, "on", {attribute: 1}) await hass.async_block_till_done() - assert xknx_mock.telegrams.put.call_count == 1 + assert knx_ip_interface_mock.send_telegram.call_count == 1 # Change state keep attribute - xknx_mock.reset_mock() + knx_ip_interface_mock.reset_mock() hass.states.async_set(entity_id, "off", {attribute: 1}) await hass.async_block_till_done() - assert xknx_mock.telegrams.put.call_count == 0 + assert knx_ip_interface_mock.send_telegram.call_count == 0 # Change state and attribute - xknx_mock.reset_mock() + knx_ip_interface_mock.reset_mock() hass.states.async_set(entity_id, "on", {attribute: 0}) await hass.async_block_till_done() - assert xknx_mock.telegrams.put.call_count == 1 + assert knx_ip_interface_mock.send_telegram.call_count == 1