Speed up KNX tests with an in-memory telegram store (#177328)

Co-authored-by: Joost Lekkerkerker <joostlek@outlook.com>

homeassistant/components/knx/manifest.json

@@ -14,7 +14,7 @@
     "xknx==3.17.0",
     "xknxproject==3.9.0",
     "knx-frontend==2026.7.23.145751",
-    "knx-telegram-store[sqlite,postgres]==0.10.2"
+    "knx-telegram-store[sqlite,postgres]==0.11.1"
   ],
   "single_config_entry": true
 }

requirements_all.txt

@@ -1438,7 +1438,7 @@ knocki==0.4.2
 knx-frontend==2026.7.23.145751
 
 # homeassistant.components.knx
-knx-telegram-store[sqlite,postgres]==0.10.2
+knx-telegram-store[sqlite,postgres]==0.11.1
 
 # homeassistant.components.kraken
 krakenex==2.2.2

tests/components/knx/conftest.py

@@ -3,10 +3,16 @@
 from __future__ import annotations
 
 import asyncio
+from collections.abc import Generator, Iterator
+import contextlib
 from typing import Any
 from unittest.mock import DEFAULT, AsyncMock, Mock, patch
 
-from knx_telegram_store import BufferedSqliteStore
+from knx_telegram_store import (
+    BufferedMemoryStore,
+    BufferedPostgresStore,
+    BufferedSqliteStore,
+)
 import pytest
 from xknx import XKNX
 from xknx.core import XknxConnectionState, XknxConnectionType
@@ -22,6 +28,7 @@ from xknx.telegram.apci import (
     SecureAPDU,
 )
 
+from homeassistant.components.knx import telegrams as knx_telegrams
 from homeassistant.components.knx.const import (
     CONF_KNX_AUTOMATIC,
     CONF_KNX_CONNECTION_TYPE,
@@ -57,6 +64,58 @@ from tests.common import MockConfigEntry, async_load_json_object_fixture
 from tests.typing import WebSocketGenerator
 
 
+class _TestTelegramStore(BufferedMemoryStore):
+    """Drop-in for the SQL-backed stores that keeps telegrams in memory.
+
+    Avoids the per-test cost of creating a SQLAlchemy engine and an aiosqlite
+    background thread. Accepts (and ignores) the ``db_path``/``dsn`` and
+    ``retention_days`` arguments the integration passes to the SQL stores.
+    """
+
+    def __init__(
+        self,
+        *_args: Any,
+        flush_interval: float = 1.0,
+        max_buffer_size: int = 10000,
+        **_kwargs: Any,
+    ) -> None:
+        """Initialize the in-memory test store."""
+        super().__init__(flush_interval=flush_interval, max_buffer_size=max_buffer_size)
+
+
+@contextlib.contextmanager
+def _patch_telegram_store(*, real_store: bool) -> Generator:
+    """Choose the telegram store the integration builds during setup.
+
+    Default (``real_store=False``): swap in the fast in-memory store so tests
+    that don't care about telegram history skip SQLite engine and aiosqlite
+    thread setup. ``real_store=True``: build the real SQLite store (kept
+    in-memory) and the real PostgreSQL store, for tests that exercise the
+    database backends.
+    """
+    if not real_store:
+        with (
+            patch.object(knx_telegrams, "BufferedSqliteStore", _TestTelegramStore),
+            patch.object(knx_telegrams, "BufferedPostgresStore", _TestTelegramStore),
+        ):
+            yield
+        return
+
+    original_init = BufferedSqliteStore.__init__
+
+    def mocked_init(self, db_path: str, *args: Any, **kwargs: Any) -> None:
+        original_init(self, ":memory:", *args, **kwargs)
+
+    # Restore the real store classes (the autouse fixture points them at the
+    # in-memory store) and keep the SQLite database in-memory.
+    with (
+        patch.object(knx_telegrams, "BufferedSqliteStore", BufferedSqliteStore),
+        patch.object(knx_telegrams, "BufferedPostgresStore", BufferedPostgresStore),
+        patch.object(BufferedSqliteStore, "__init__", mocked_init),
+    ):
+        yield
+
+
 class KNXTestKit:
     """Test helper for the KNX integration."""
 
@@ -90,9 +149,14 @@ class KNXTestKit:
         config_store_fixture: str | None = None,
         add_entry_to_hass: bool = True,
         state_updater: bool = True,
+        real_telegram_store: bool = False,
     ) -> None:
-        """Create the KNX integration."""
-        # Force an in-memory telegram store will be done via autouse fixture.
+        """Create the KNX integration.
+
+        By default a fast in-memory telegram store is used. Tests that exercise
+        the SQLite/PostgreSQL telegram backends pass ``real_telegram_store=True``
+        to build the real SQLite store (kept in-memory).
+        """
 
         async def patch_xknx_start():
             """Patch `xknx.start` for unittests."""
@@ -139,10 +203,13 @@ class KNXTestKit:
         ).start()  # keep patched for the whole test run
 
         knx_config = {DOMAIN: yaml_config or {}}
-        with patch(
-            "xknx.xknx.knx_interface_factory",
-            return_value=knx_ip_interface_mock(),
-            side_effect=fish_xknx,
+        with (
+            _patch_telegram_store(real_store=real_telegram_store),
+            patch(
+                "xknx.xknx.knx_interface_factory",
+                return_value=knx_ip_interface_mock(),
+                side_effect=fish_xknx,
+            ),
         ):
             state_updater_patcher = patch(
                 "xknx.xknx.StateUpdater.register_remote_value"
@@ -445,12 +512,14 @@ async def create_ui_entity(
 
 
 @pytest.fixture(autouse=True)
-def mock_knx_telegram_store():
-    """Mock knx-telegram-store to always use an in-memory database."""
-    original_init = BufferedSqliteStore.__init__
+def mock_knx_telegram_store() -> Iterator[None]:
+    """Default every integration setup to the fast in-memory telegram store.
 
-    def mocked_init(self, db_path: str, *args: Any, **kwargs: Any) -> None:
-        original_init(self, ":memory:", *args, **kwargs)
-
-    with patch.object(BufferedSqliteStore, "__init__", mocked_init):
+    The real SQLite store creates a SQLAlchemy engine and an aiosqlite thread
+    on every setup/teardown, which the hundreds of tests that never touch
+    telegram history shouldn't pay for. Tests that do exercise the database
+    backends pass ``real_telegram_store=True`` to ``setup_integration``, which
+    overrides this for the duration of setup.
+    """
+    with _patch_telegram_store(real_store=False):
         yield

tests/components/knx/test_telegrams.py

@@ -90,7 +90,7 @@ async def test_store_telegram_history(
     knx: KNXTestKit,
 ) -> None:
     """Test storing telegram history."""
-    await knx.setup_integration()
+    await knx.setup_integration(real_telegram_store=True)
     telegrams_module = hass.data[KNX_MODULE_KEY].telegrams
 
     await knx.receive_write("1/3/4", True)
@@ -118,7 +118,7 @@ async def test_store_telegram_history_sqlite(
     knx: KNXTestKit,
 ) -> None:
     """Test storing telegram history in SQLite."""
-    await knx.setup_integration()
+    await knx.setup_integration(real_telegram_store=True)
     telegrams_module = hass.data[KNX_MODULE_KEY].telegrams
 
     await knx.receive_write("1/3/4", True)
@@ -151,7 +151,7 @@ async def test_store_telegram_history_error_handling(
         "knx_telegram_store.BufferedSqliteStore.initialize",
         side_effect=side_effect,
     ):
-        await knx.setup_integration()
+        await knx.setup_integration(real_telegram_store=True)
 
     telegrams_module = hass.data[KNX_MODULE_KEY].telegrams
     assert telegrams_module.store is None
@@ -180,7 +180,7 @@ async def test_store_telegram_history_needs_migration_timeout(
             side_effect=hanging_probe,
         ),
     ):
-        await knx.setup_integration()
+        await knx.setup_integration(real_telegram_store=True)
 
     telegrams_module = hass.data[KNX_MODULE_KEY].telegrams
     assert telegrams_module.store is None
@@ -209,7 +209,7 @@ async def test_store_init_timeout_retries_and_succeeds(
             return_value=[],
         ),
     ):
-        await knx.setup_integration()
+        await knx.setup_integration(real_telegram_store=True)
 
         telegrams_module = hass.data[KNX_MODULE_KEY].telegrams
 
@@ -245,7 +245,7 @@ async def test_store_init_timeout_exhausts_retries_and_aborts(
             side_effect=TimeoutError(),
         ),
     ):
-        await knx.setup_integration()
+        await knx.setup_integration(real_telegram_store=True)
 
         telegrams_module = hass.data[KNX_MODULE_KEY].telegrams
 
@@ -277,7 +277,7 @@ async def test_stop_cancels_pending_retry_timer(
         "knx_telegram_store.BufferedSqliteStore.initialize",
         side_effect=TimeoutError(),
     ):
-        await knx.setup_integration()
+        await knx.setup_integration(real_telegram_store=True)
         assert hass.data[KNX_MODULE_KEY].telegrams.store is None
 
         assert await hass.config_entries.async_unload(knx.mock_config_entry.entry_id)
@@ -309,7 +309,7 @@ async def test_stop_cancels_in_flight_retry_task(
         "knx_telegram_store.BufferedSqliteStore.initialize",
         side_effect=initialize_side_effect,
     ):
-        await knx.setup_integration()
+        await knx.setup_integration(real_telegram_store=True)
 
         freezer.tick(STORE_INIT_RETRY_BACKOFF[0] + 1)
         async_fire_time_changed(hass)
@@ -338,7 +338,7 @@ async def test_migrate_telegrams_from_json(
         "data": json_telegrams,
     }
 
-    await knx.setup_integration()
+    await knx.setup_integration(real_telegram_store=True)
 
     telegrams_module = hass.data[KNX_MODULE_KEY].telegrams
     assert telegrams_module.store is not None
@@ -361,7 +361,7 @@ async def test_stop_error_handling(
     side_effect: Exception,
 ) -> None:
     """Test that errors while stopping the store are swallowed."""
-    await knx.setup_integration()
+    await knx.setup_integration(real_telegram_store=True)
 
     telegrams_module = hass.data[KNX_MODULE_KEY].telegrams
     assert telegrams_module.store is not None
@@ -377,7 +377,7 @@ async def test_model_to_dict_resolution(
     knx: KNXTestKit,
 ) -> None:
     """Test model_to_dict name resolution and DPT handling."""
-    await knx.setup_integration()
+    await knx.setup_integration(real_telegram_store=True)
     telegrams_module = hass.data[KNX_MODULE_KEY].telegrams
     assert telegrams_module.project.loaded
 
@@ -458,7 +458,7 @@ async def test_load_history_needs_migration(
         "knx_telegram_store.BufferedSqliteStore.needs_migration",
         return_value=True,
     ):
-        await knx.setup_integration()
+        await knx.setup_integration(real_telegram_store=True)
 
     telegrams_module = hass.data[KNX_MODULE_KEY].telegrams
     assert telegrams_module.store is not None
@@ -481,7 +481,7 @@ async def test_load_history_hydrate_error(
         "knx_telegram_store.BufferedSqliteStore.get_last_unique_telegrams",
         side_effect=side_effect,
     ):
-        await knx.setup_integration()
+        await knx.setup_integration(real_telegram_store=True)
 
     telegrams_module = hass.data[KNX_MODULE_KEY].telegrams
     assert telegrams_module.store is not None
@@ -493,7 +493,7 @@ async def test_migrate_telegrams_no_json(
     knx: KNXTestKit,
 ) -> None:
     """Test migration is a no-op when there is no legacy JSON history."""
-    await knx.setup_integration()
+    await knx.setup_integration(real_telegram_store=True)
 
     telegrams_module = hass.data[KNX_MODULE_KEY].telegrams
     assert telegrams_module.store is not None
@@ -514,7 +514,7 @@ async def test_migrate_telegrams_unexpected_format(
         "data": "not a list or dict",
     }
 
-    await knx.setup_integration()
+    await knx.setup_integration(real_telegram_store=True)
 
     telegrams_module = hass.data[KNX_MODULE_KEY].telegrams
     assert telegrams_module.store is not None
@@ -556,7 +556,7 @@ async def test_migrate_telegrams_store_error(
         "knx_telegram_store.BufferedSqliteStore.store_many",
         side_effect=KnxTelegramStoreException("write failed"),
     ):
-        await knx.setup_integration()
+        await knx.setup_integration(real_telegram_store=True)
 
     # Setup still succeeds even though migration failed
     telegrams_module = hass.data[KNX_MODULE_KEY].telegrams
@@ -571,7 +571,7 @@ async def test_nightly_eviction_calls_evict_expired(
     """Test expired telegrams are evicted on the nightly 3 AM run."""
     await hass.config.async_set_time_zone("UTC")
     freezer.move_to("2024-01-01 12:00:00+00:00")
-    await knx.setup_integration()
+    await knx.setup_integration(real_telegram_store=True)
     telegrams_module = hass.data[KNX_MODULE_KEY].telegrams
     assert telegrams_module.store is not None
 
@@ -606,7 +606,7 @@ async def test_nightly_eviction_zero_retention_deletes_all(
         options=knx.mock_config_entry.options
         | {CONF_KNX_TELEGRAM_DB_RETENTION_DAYS: 0},
     )
-    await knx.setup_integration(add_entry_to_hass=False)
+    await knx.setup_integration(add_entry_to_hass=False, real_telegram_store=True)
     telegrams_module = hass.data[KNX_MODULE_KEY].telegrams
     assert telegrams_module.store is not None
 
@@ -632,7 +632,7 @@ async def test_nightly_eviction_error_handling(
     """Test a store error during nightly eviction is logged and does not raise."""
     await hass.config.async_set_time_zone("UTC")
     freezer.move_to("2024-01-01 12:00:00+00:00")
-    await knx.setup_integration()
+    await knx.setup_integration(real_telegram_store=True)
     telegrams_module = hass.data[KNX_MODULE_KEY].telegrams
     assert telegrams_module.store is not None
 
@@ -674,7 +674,7 @@ async def test_postgres_backend_init_error(
         "homeassistant.components.knx.telegrams.BufferedPostgresStore",
         return_value=mock_store,
     ):
-        await knx.setup_integration(add_entry_to_hass=False)
+        await knx.setup_integration(add_entry_to_hass=False, real_telegram_store=True)
 
     telegrams_module = hass.data[KNX_MODULE_KEY].telegrams
     assert telegrams_module.store is None

tests/components/knx/test_telegrams_migration.py

@@ -32,7 +32,7 @@ async def test_migrate_telegrams_json_to_sqlite(
     # returns the list. Save the inner list, not the fixture wrapper dict.
     await store.async_save(legacy_data["data"])
 
-    await knx.setup_integration()
+    await knx.setup_integration(real_telegram_store=True)
     telegrams_module = hass.data[KNX_MODULE_KEY].telegrams
 
     await hass.async_block_till_done()
@@ -110,7 +110,7 @@ async def test_migrate_telegrams_json_missing_keys(
 
     await store.async_save(legacy_data)
 
-    await knx.setup_integration()
+    await knx.setup_integration(real_telegram_store=True)
     telegrams_module = hass.data[KNX_MODULE_KEY].telegrams
 
     await hass.async_block_till_done()

tests/components/knx/test_websocket.py

@@ -31,7 +31,7 @@ async def test_knx_get_base_data_command(
     hass: HomeAssistant, knx: KNXTestKit, hass_ws_client: WebSocketGenerator
 ) -> None:
     """Test knx/get_base_data command."""
-    await knx.setup_integration()
+    await knx.setup_integration(real_telegram_store=True)
     client = await hass_ws_client(hass)
     await client.send_json_auto_id({"type": "knx/get_base_data"})
 
@@ -71,7 +71,7 @@ async def test_knx_get_base_data_command_postgres(
             return_value=[],
         ),
     ):
-        await knx.setup_integration(add_entry_to_hass=False)
+        await knx.setup_integration(add_entry_to_hass=False, real_telegram_store=True)
         client = await hass_ws_client(hass)
         await client.send_json_auto_id({"type": "knx/get_base_data"})
         res = await client.receive_json()
@@ -236,7 +236,7 @@ async def test_knx_group_monitor_info_command(
     hass: HomeAssistant, knx: KNXTestKit, hass_ws_client: WebSocketGenerator
 ) -> None:
     """Test knx/group_monitor_info command."""
-    await knx.setup_integration()
+    await knx.setup_integration(real_telegram_store=True)
     client = await hass_ws_client(hass)
 
     await client.send_json_auto_id({"type": "knx/group_monitor_info"})
@@ -251,7 +251,7 @@ async def test_knx_query_telegrams_command(
     hass: HomeAssistant, knx: KNXTestKit, hass_ws_client: WebSocketGenerator
 ) -> None:
     """Test knx/query_telegrams command."""
-    await knx.setup_integration()
+    await knx.setup_integration(real_telegram_store=True)
     client = await hass_ws_client(hass)
 
     # get some telegrams to populate the store
@@ -312,7 +312,7 @@ async def test_telegram_store_not_initialized(
         "knx_telegram_store.BufferedSqliteStore.initialize",
         side_effect=KnxTelegramStoreException("init failed"),
     ):
-        await knx.setup_integration()
+        await knx.setup_integration(real_telegram_store=True)
     client = await hass_ws_client(hass)
 
     assert hass.data[KNX_MODULE_KEY].telegrams.store is None
@@ -334,7 +334,7 @@ async def test_telegram_store_query_database_error(
     command: str,
 ) -> None:
     """Test telegram commands when the store query raises a database error."""
-    await knx.setup_integration()
+    await knx.setup_integration(real_telegram_store=True)
     client = await hass_ws_client(hass)
 
     store = hass.data[KNX_MODULE_KEY].telegrams.store
@@ -387,7 +387,8 @@ async def test_knx_subscribe_telegrams_command_recent_telegrams(
                 CONF_NAME: "test",
                 KNX_ADDRESS: "1/2/4",
             }
-        }
+        },
+        real_telegram_store=True,
     )
 
     # send incoming telegram