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

@@ -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()

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

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