starred/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2026-08-28 10:16:02 -05:00
Code Issues Packages Projects Releases Wiki Activity

Add PostgreSQL backend and storage backend selection for KNX telegrams (#175673)

Browse Source 
Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com>
Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>
This commit is contained in:
Martin Hoefling
2026-07-14 10:09:11 +02:00
committed by GitHub
co-authored by Claude Opus 4.8 Copilot Autofix powered by AI
parent fbc2eb8d27
commit 9eb29b8494
16 changed files with 774 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files
homeassistant/components/knx/__init__.py
+15 -1
View File
@@ -24,11 +24,13 @@ from .const import (
CONF_KNX_KNXKEY_FILENAME,
CONF_KNX_RATE_LIMIT,
CONF_KNX_STATE_UPDATER,
CONF_KNX_TELEGRAM_DB_BACKEND,
CONF_KNX_TELEGRAM_DB_LOAD_HOURS,
CONF_KNX_TELEGRAM_DB_RETENTION_DAYS,
DATA_HASS_CONFIG,
DOMAIN,
KNX_MODULE_KEY,
KNX_TELEGRAM_BACKEND_SQLITE,
KNX_TELEGRAM_DB_PATH_SQLITE,
KNX_TELEGRAM_DB_RETENTION_DEFAULT,
KNX_TELEGRAM_LOAD_HOURS_DEFAULT,
@@ -188,11 +190,23 @@ async def async_migrate_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
new_options.setdefault(CONF_KNX_STATE_UPDATER, CONF_KNX_DEFAULT_STATE_UPDATER)
new_options.setdefault(CONF_KNX_RATE_LIMIT, CONF_KNX_DEFAULT_RATE_LIMIT)
new_options[CONF_KNX_TELEGRAM_DB_BACKEND] = KNX_TELEGRAM_BACKEND_SQLITE
hass.config_entries.async_update_entry(
entry, data=new_data, options=new_options, version=2
entry, data=new_data, options=new_options, version=2, minor_version=2
)
_LOGGER.info("Migration to version 2 successful")
if entry.version == 2 and entry.minor_version < 2:
# version 2.2 introduced in 2026.8
new_options = {**entry.options}
if CONF_KNX_TELEGRAM_DB_BACKEND not in new_options:
new_options[CONF_KNX_TELEGRAM_DB_BACKEND] = KNX_TELEGRAM_BACKEND_SQLITE
hass.config_entries.async_update_entry(
entry, options=new_options, minor_version=2
)
_LOGGER.info("Migration to version 2.2 successful")
return True
homeassistant/components/knx/config_flow.py
+171
View File
@@ -1,8 +1,12 @@
"""Config flow for KNX."""
import asyncio
from collections.abc import AsyncGenerator
from typing import Any, Final, Literal, override
from urllib.parse import quote, unquote, urlparse, urlunparse
from knx_telegram_store import ConnectionErrorKind
from knx_telegram_store.backends.postgres import PostgresStore
import voluptuous as vol
from xknx import XKNX
from xknx.exceptions.exception import (
@@ -49,8 +53,16 @@ from .const import (
CONF_KNX_SECURE_USER_ID,
CONF_KNX_SECURE_USER_PASSWORD,
CONF_KNX_STATE_UPDATER,
CONF_KNX_TELEGRAM_DB_BACKEND,
CONF_KNX_TELEGRAM_DB_DATABASE,
CONF_KNX_TELEGRAM_DB_HOST,
CONF_KNX_TELEGRAM_DB_LOAD_HOURS,
CONF_KNX_TELEGRAM_DB_PASSWORD,
CONF_KNX_TELEGRAM_DB_PORT,
CONF_KNX_TELEGRAM_DB_POSTGRES_DSN,
CONF_KNX_TELEGRAM_DB_RETENTION_DAYS,
CONF_KNX_TELEGRAM_DB_TLS,
CONF_KNX_TELEGRAM_DB_USER,
CONF_KNX_TUNNEL_ENDPOINT_IA,
CONF_KNX_TUNNELING,
CONF_KNX_TUNNELING_TCP,
@@ -58,6 +70,8 @@ from .const import (
DEFAULT_ROUTING_IA,
DOMAIN,
KNX_MODULE_KEY,
KNX_TELEGRAM_BACKEND_POSTGRES,
KNX_TELEGRAM_BACKEND_SQLITE,
KNX_TELEGRAM_DB_RETENTION_DEFAULT,
KNX_TELEGRAM_LOAD_HOURS_DEFAULT,
KNXConfigEntryData,
@@ -82,12 +96,17 @@ DEFAULT_ENTRY_OPTIONS = KNXConfigEntryOptions(
state_updater=CONF_KNX_DEFAULT_STATE_UPDATER,
telegram_db_retention_days=KNX_TELEGRAM_DB_RETENTION_DEFAULT,
telegram_db_load_hours=KNX_TELEGRAM_LOAD_HOURS_DEFAULT,
telegram_db_backend=KNX_TELEGRAM_BACKEND_SQLITE,
)
CONF_KEYRING_FILE: Final = "knxkeys_file"
CONF_KNX_TELEGRAM_STORE_SECTION: Final = "telegram_store_section"
# Timeout for the PostgreSQL connection check, so an unreachable host cannot
# block the options flow until the driver/OS connection timeout expires.
DSN_CHECK_TIMEOUT = 10
CONF_KNX_TUNNELING_TYPE: Final = "tunneling_type"
CONF_KNX_TUNNELING_TYPE_LABELS: Final = {
CONF_KNX_TUNNELING: "UDP (Tunneling v1)",
@@ -113,6 +132,7 @@ class KNXConfigFlow(ConfigFlow, domain=DOMAIN):
"""Handle a KNX config flow."""
VERSION = 2
MINOR_VERSION = 2
def __init__(self) -> None:
"""Initialize KNX config flow."""
@@ -951,6 +971,7 @@ class KNXOptionsFlow(OptionsFlowWithReload):
"""Manage KNX communication settings."""
if user_input is not None:
telegram_store_section = user_input[CONF_KNX_TELEGRAM_STORE_SECTION]
backend = telegram_store_section[CONF_KNX_TELEGRAM_DB_BACKEND]
self.new_entry_options |= KNXConfigEntryOptions(
state_updater=user_input[CONF_KNX_STATE_UPDATER],
rate_limit=user_input[CONF_KNX_RATE_LIMIT],
@@ -960,7 +981,10 @@ class KNXOptionsFlow(OptionsFlowWithReload):
telegram_db_retention_days=telegram_store_section[
CONF_KNX_TELEGRAM_DB_RETENTION_DAYS
],
telegram_db_backend=backend,
)
if backend == KNX_TELEGRAM_BACKEND_POSTGRES:
return await self.async_step_telegram_store_postgres()
return self.finish_flow()
data_schema = {
@@ -1020,6 +1044,22 @@ class KNXOptionsFlow(OptionsFlowWithReload):
),
vol.Coerce(int),
),
vol.Required(
CONF_KNX_TELEGRAM_DB_BACKEND,
default=self.initial_options.get(
CONF_KNX_TELEGRAM_DB_BACKEND,
KNX_TELEGRAM_BACKEND_SQLITE,
),
): selector.SelectSelector(
selector.SelectSelectorConfig(
options=[
KNX_TELEGRAM_BACKEND_SQLITE,
KNX_TELEGRAM_BACKEND_POSTGRES,
],
mode=selector.SelectSelectorMode.DROPDOWN,
translation_key="telegram_backend",
)
),
}
),
),
@@ -1027,5 +1067,136 @@ class KNXOptionsFlow(OptionsFlowWithReload):
return self.async_show_form(
step_id="communication_settings",
data_schema=vol.Schema(data_schema),
last_step=False,
)
async def async_step_telegram_store_postgres(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Collect and validate the PostgreSQL telegram store connection."""
current_dsn = self.initial_options.get(CONF_KNX_TELEGRAM_DB_POSTGRES_DSN, "")
parsed = _parse_dsn(current_dsn)
errors: dict[str, str] = {}
if user_input is not None:
# Reuse the stored password when the field is left blank.
params = {
**user_input,
CONF_KNX_TELEGRAM_DB_PASSWORD: (
user_input.get(CONF_KNX_TELEGRAM_DB_PASSWORD)
or parsed.get(CONF_KNX_TELEGRAM_DB_PASSWORD, "")
),
}
dsn = _build_dsn(params)
errors = await _async_check_postgres_dsn(dsn)
if not errors:
self.new_entry_options |= KNXConfigEntryOptions(
telegram_db_postgres_dsn=dsn
)
return self.finish_flow()
data_schema = vol.Schema(
{
vol.Required(
CONF_KNX_TELEGRAM_DB_HOST,
default=parsed.get(CONF_KNX_TELEGRAM_DB_HOST, "localhost"),
): selector.TextSelector(),
vol.Required(
CONF_KNX_TELEGRAM_DB_PORT,
default=parsed.get(CONF_KNX_TELEGRAM_DB_PORT, 5432),
): vol.All(
selector.NumberSelector(
selector.NumberSelectorConfig(
min=1,
max=65535,
mode=selector.NumberSelectorMode.BOX,
)
),
vol.Coerce(int),
),
vol.Required(
CONF_KNX_TELEGRAM_DB_USER,
default=parsed.get(CONF_KNX_TELEGRAM_DB_USER, ""),
): selector.TextSelector(),
vol.Required(
CONF_KNX_TELEGRAM_DB_PASSWORD, default=""
): selector.TextSelector(
selector.TextSelectorConfig(type=selector.TextSelectorType.PASSWORD)
),
vol.Required(
CONF_KNX_TELEGRAM_DB_DATABASE,
default=parsed.get(CONF_KNX_TELEGRAM_DB_DATABASE, "knx_telegrams"),
): selector.TextSelector(),
vol.Required(
CONF_KNX_TELEGRAM_DB_TLS,
default=parsed.get(CONF_KNX_TELEGRAM_DB_TLS, False),
): selector.BooleanSelector(),
}
)
if user_input is not None:
data_schema = self.add_suggested_values_to_schema(data_schema, user_input)
return self.async_show_form(
step_id="telegram_store_postgres",
data_schema=data_schema,
errors=errors,
last_step=True,
)
async def _async_check_postgres_dsn(dsn: str) -> dict[str, str]:
"""Validate a PostgreSQL DSN, returning form errors on failure."""
connection_errors = {
ConnectionErrorKind.AUTH: "invalid_auth",
ConnectionErrorKind.HOST_UNREACHABLE: "host_unreachable",
ConnectionErrorKind.DATABASE_MISSING: "database_missing",
ConnectionErrorKind.PERMISSION: "permission",
ConnectionErrorKind.TIMEOUT: "timeout",
ConnectionErrorKind.MISSING_DEPENDENCY: "missing_dependency",
}
try:
async with asyncio.timeout(DSN_CHECK_TIMEOUT):
check_result = await PostgresStore.check_config(dsn)
except TimeoutError:
return {"base": "timeout"}
except ValueError:
return {"base": "cannot_connect"}
if not check_result.ok:
return {"base": connection_errors.get(check_result.kind, "cannot_connect")}
return {}
def _build_dsn(params: dict[str, Any]) -> str:
"""Build a PostgreSQL DSN from form params."""
quoted_user = quote(params.get(CONF_KNX_TELEGRAM_DB_USER, ""), safe="")
quoted_password = quote(params.get(CONF_KNX_TELEGRAM_DB_PASSWORD, ""), safe="")
host = params.get(CONF_KNX_TELEGRAM_DB_HOST, "localhost")
if ":" in host and not host.startswith("["):
# IPv6 literals must be bracketed in the URL netloc
host = f"[{host}]"
port = int(params.get(CONF_KNX_TELEGRAM_DB_PORT, 5432))
quoted_database = quote(
params.get(CONF_KNX_TELEGRAM_DB_DATABASE, "knx_telegrams"), safe=""
)
tls = params.get(CONF_KNX_TELEGRAM_DB_TLS, False)
netloc = f"{quoted_user}:{quoted_password}@{host}:{port}"
query = "sslmode=require" if tls else ""
return urlunparse(("postgresql", netloc, f"/{quoted_database}", "", query, ""))
def _parse_dsn(dsn: str) -> dict[str, Any]:
"""Parse a PostgreSQL DSN into form params."""
if not dsn:
return {}
try:
url = urlparse(dsn)
return {
CONF_KNX_TELEGRAM_DB_USER: unquote(url.username or ""),
CONF_KNX_TELEGRAM_DB_PASSWORD: unquote(url.password or ""),
CONF_KNX_TELEGRAM_DB_HOST: url.hostname or "localhost",
CONF_KNX_TELEGRAM_DB_PORT: url.port or 5432,
CONF_KNX_TELEGRAM_DB_DATABASE: unquote(url.path.lstrip("/")),
CONF_KNX_TELEGRAM_DB_TLS: "sslmode=require" in url.query,
}
except ValueError, AttributeError:
return {}
homeassistant/components/knx/const.py
+14
View File
@@ -53,8 +53,20 @@ CONF_KNX_DEFAULT_RATE_LIMIT: Final = 0
DEFAULT_ROUTING_IA: Final = "0.0.240"
CONF_KNX_TELEGRAM_DB_BACKEND: Final = "telegram_db_backend"
CONF_KNX_TELEGRAM_DB_RETENTION_DAYS: Final = "telegram_db_retention_days"
CONF_KNX_TELEGRAM_DB_LOAD_HOURS: Final = "telegram_db_load_hours"
CONF_KNX_TELEGRAM_DB_POSTGRES_DSN: Final = "telegram_db_postgres_dsn"
CONF_KNX_TELEGRAM_DB_HOST: Final = "host"
CONF_KNX_TELEGRAM_DB_PORT: Final = "port"
CONF_KNX_TELEGRAM_DB_USER: Final = "user"
CONF_KNX_TELEGRAM_DB_PASSWORD: Final = "password"
CONF_KNX_TELEGRAM_DB_DATABASE: Final = "database"
CONF_KNX_TELEGRAM_DB_TLS: Final = "tls"
KNX_TELEGRAM_BACKEND_SQLITE: Final = "sqlite"
KNX_TELEGRAM_BACKEND_POSTGRES: Final = "postgres"
KNX_TELEGRAM_DB_RETENTION_DEFAULT: Final = 10 # days
KNX_TELEGRAM_LOAD_HOURS_DEFAULT: Final = 24 # 1 day
@@ -139,6 +151,8 @@ class KNXConfigEntryOptions(TypedDict, total=False):
# Integration only (not forwarded to xknx)
telegram_db_retention_days: int
telegram_db_load_hours: int
telegram_db_backend: str # sqlite | postgres
telegram_db_postgres_dsn: str
class ColorTempModes(Enum):
homeassistant/components/knx/diagnostics.py
+2
View File
@@ -15,6 +15,7 @@ from .const import (
CONF_KNX_ROUTING_BACKBONE_KEY,
CONF_KNX_SECURE_DEVICE_AUTHENTICATION,
CONF_KNX_SECURE_USER_PASSWORD,
CONF_KNX_TELEGRAM_DB_POSTGRES_DSN,
DOMAIN,
KNX_MODULE_KEY,
)
@@ -24,6 +25,7 @@ TO_REDACT = {
CONF_KNX_KNXKEY_PASSWORD,
CONF_KNX_SECURE_USER_PASSWORD,
CONF_KNX_SECURE_DEVICE_AUTHENTICATION,
CONF_KNX_TELEGRAM_DB_POSTGRES_DSN,
}
homeassistant/components/knx/manifest.json
+1 -1
View File
@@ -14,7 +14,7 @@
"xknx==3.16.0",
"xknxproject==3.9.0",
"knx-frontend==2026.6.23.203726",
"knx-telegram-store[sqlite]==0.3.2"
"knx-telegram-store[sqlite,postgres]==0.9.1"
],
"single_config_entry": true
}
homeassistant/components/knx/strings.json
+36
View File
@@ -1162,6 +1162,15 @@
}
},
"options": {
"error": {
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"database_missing": "The specified database does not exist.",
"host_unreachable": "Could not reach the database host.",
"invalid_auth": "[%key:common::config_flow::error::invalid_auth%]",
"missing_dependency": "Required database driver is not installed.",
"permission": "Insufficient privileges to access the database.",
"timeout": "Connection timed out."
},
"step": {
"communication_settings": {
"data": {
@@ -1175,10 +1184,12 @@
"sections": {
"telegram_store_section": {
"data": {
"telegram_db_backend": "Telegram storage backend",
"telegram_db_load_hours": "Group monitor history",
"telegram_db_retention_days": "Retention period"
},
"data_description": {
"telegram_db_backend": "Select where to store KNX telegram history.",
"telegram_db_load_hours": "Number of hours of telegram history to load when the group monitor is opened.",
"telegram_db_retention_days": "Number of days to keep telegram history. Older telegrams are automatically deleted nightly at 3 AM. Set to `0` to delete all telegram history on every nightly run."
},
@@ -1186,6 +1197,25 @@
}
},
"title": "Communication settings"
},
"telegram_store_postgres": {
"data": {
"database": "Database name",
"host": "[%key:common::config_flow::data::host%]",
"password": "[%key:common::config_flow::data::password%]",
"port": "[%key:common::config_flow::data::port%]",
"tls": "Use TLS",
"user": "[%key:common::config_flow::data::username%]"
},
"data_description": {
"database": "Name of the database to store telegrams in.",
"host": "Hostname or IP address of the PostgreSQL server.",
"password": "Password for the PostgreSQL user. Leave blank to keep the current password.",
"port": "Port the PostgreSQL server is listening on.",
"tls": "Encrypt the connection to the PostgreSQL server (`sslmode=require`). Note that the server certificate is not verified.",
"user": "Username to authenticate with the PostgreSQL server."
},
"title": "PostgreSQL connection"
}
}
},
@@ -1260,6 +1290,12 @@
"total": "[%key:component::sensor::entity_component::_::state_attributes::state_class::state::total%]",
"total_increasing": "[%key:component::sensor::entity_component::_::state_attributes::state_class::state::total_increasing%]"
}
},
"telegram_backend": {
"options": {
"postgres": "PostgreSQL (External)",
"sqlite": "Internal storage (Default)"
}
}
},
"services": {
homeassistant/components/knx/telegrams.py
+54 -16
View File
@@ -8,6 +8,7 @@ import os
from typing import Any, TypedDict
from knx_telegram_store import (
BufferedPostgresStore,
BufferedSqliteStore,
KnxTelegramStoreException,
StoredTelegram,
@@ -26,7 +27,10 @@ from homeassistant.helpers.storage import STORAGE_DIR, Store
from homeassistant.util import dt as dt_util
from .const import (
CONF_KNX_TELEGRAM_DB_BACKEND,
CONF_KNX_TELEGRAM_DB_POSTGRES_DSN,
CONF_KNX_TELEGRAM_DB_RETENTION_DAYS,
KNX_TELEGRAM_BACKEND_POSTGRES,
KNX_TELEGRAM_DB_PATH_SQLITE,
SIGNAL_KNX_DATA_SECURE_ISSUE_TELEGRAM,
SIGNAL_KNX_TELEGRAM,
@@ -48,6 +52,15 @@ EVICT_EXPIRED_HOUR = 3
# at risk from a longer interval are those buffered during an ungraceful shutdown.
FLUSH_INTERVAL_SECONDS = 600
# The buffer drops the oldest telegrams when full. Size it to cover a full
# flush interval at ~50 telegrams/s, the maximum rate of a KNX TP line, so
# nothing is dropped while the database is healthy.
MAX_BUFFER_TELEGRAMS = FLUSH_INTERVAL_SECONDS * 50
# Timeout for the migration probe and store initialization, so an unreachable
# database cannot block KNX setup until the driver/OS connection timeout expires.
STORE_INIT_TIMEOUT = 10
class DecodedTelegramPayload(TypedDict):
"""Decoded payload value and metadata."""
@@ -89,19 +102,32 @@ class Telegrams:
self.project = project
self.config = config
self.backend: str = config[CONF_KNX_TELEGRAM_DB_BACKEND]
self.dsn: str = str(config.get(CONF_KNX_TELEGRAM_DB_POSTGRES_DSN, ""))
self.retention_days: int = config[CONF_KNX_TELEGRAM_DB_RETENTION_DAYS]
self.store: BufferedSqliteStore | None = None
self._uninitialized_store: BufferedSqliteStore | None = None
self.store: BufferedSqliteStore | BufferedPostgresStore | None = None
self._uninitialized_store: (
BufferedSqliteStore | BufferedPostgresStore | None
) = None
self._evict_expired_unsub: CALLBACK_TYPE | None = None
full_path = hass.config.path(STORAGE_DIR, KNX_TELEGRAM_DB_PATH_SQLITE)
os.makedirs(os.path.dirname(full_path), exist_ok=True)
self._uninitialized_store = BufferedSqliteStore(
full_path,
retention_days=self.retention_days,
flush_interval=FLUSH_INTERVAL_SECONDS,
)
if self.backend == KNX_TELEGRAM_BACKEND_POSTGRES:
self._uninitialized_store = BufferedPostgresStore(
self.dsn,
retention_days=self.retention_days,
flush_interval=FLUSH_INTERVAL_SECONDS,
max_buffer_size=MAX_BUFFER_TELEGRAMS,
)
else:
full_path = hass.config.path(STORAGE_DIR, KNX_TELEGRAM_DB_PATH_SQLITE)
os.makedirs(os.path.dirname(full_path), exist_ok=True)
self._uninitialized_store = BufferedSqliteStore(
full_path,
retention_days=self.retention_days,
flush_interval=FLUSH_INTERVAL_SECONDS,
max_buffer_size=MAX_BUFFER_TELEGRAMS,
)
self._xknx_telegram_cb_handle = (
xknx.telegram_queue.register_telegram_received_cb(
@@ -121,7 +147,8 @@ class Telegrams:
if self._uninitialized_store is None:
return
try:
needs_migration = await self._uninitialized_store.needs_migration()
async with asyncio.timeout(STORE_INIT_TIMEOUT):
needs_migration = await self._uninitialized_store.needs_migration()
if needs_migration:
_LOGGER.warning(
"KNX telegram history database schema upgrade/migration is required. "
@@ -129,24 +156,35 @@ class Telegrams:
)
await self._uninitialized_store.initialize()
else:
_LOGGER.debug("Initializing KNX telegram storage")
async with asyncio.timeout(10):
_LOGGER.debug(
"Initializing KNX telegram storage backend '%s'",
self.backend,
)
async with asyncio.timeout(STORE_INIT_TIMEOUT):
await self._uninitialized_store.initialize()
_LOGGER.info("Successfully initialized KNX telegram storage")
_LOGGER.info(
"Successfully initialized KNX telegram storage backend '%s'",
self.backend,
)
except TimeoutError:
_LOGGER.error("Timeout initializing KNX telegram storage")
_LOGGER.error(
"Timeout initializing KNX telegram storage backend '%s'",
self.backend,
)
await self._abort_store_init()
return
except KnxTelegramStoreException as err:
_LOGGER.error(
"Database error initializing KNX telegram storage: %s",
"Database error initializing KNX telegram storage backend '%s': %s",
self.backend,
err,
)
await self._abort_store_init()
return
except Exception as err: # noqa: BLE001
_LOGGER.error(
"Error initializing KNX telegram storage: %s",
"Error initializing KNX telegram storage backend '%s': %s",
self.backend,
err,
)
await self._abort_store_init()
homeassistant/components/knx/websocket.py
+11 -2
View File
@@ -8,7 +8,12 @@ import inspect
from typing import TYPE_CHECKING, Any, Final, overload
import knx_frontend as knx_panel
from knx_telegram_store import KnxTelegramStoreException, TelegramQuery
from knx_telegram_store import (
BufferedPostgresStore,
BufferedSqliteStore,
KnxTelegramStoreException,
TelegramQuery,
)
import voluptuous as vol
from xknx.telegram import Telegram
from xknxproject.exceptions import XknxProjectException
@@ -200,7 +205,11 @@ def ws_get_base_data(
"connected": knx.xknx.connection_manager.connected.is_set(),
"current_address": str(knx.xknx.current_address),
"telegram_backend": (
"sqlite" if knx.telegrams.store is not None else "unknown"
"sqlite"
if isinstance(knx.telegrams.store, BufferedSqliteStore)
else "postgres"
if isinstance(knx.telegrams.store, BufferedPostgresStore)
else "unknown"
),
"telegram_retention": knx.telegrams.store.retention_days
if knx.telegrams.store is not None
requirements_all.txt
Generated
+1 -1
View File
@@ -1435,7 +1435,7 @@ knocki==0.4.2
knx-frontend==2026.6.23.203726
# homeassistant.components.knx
knx-telegram-store[sqlite]==0.3.2
knx-telegram-store[sqlite,postgres]==0.9.1
# homeassistant.components.kraken
krakenex==2.2.2
tests/components/knx/conftest.py
+3
View File
@@ -32,10 +32,12 @@ from homeassistant.components.knx.const import (
CONF_KNX_MCAST_PORT,
CONF_KNX_RATE_LIMIT,
CONF_KNX_STATE_UPDATER,
CONF_KNX_TELEGRAM_DB_BACKEND,
CONF_KNX_TELEGRAM_DB_LOAD_HOURS,
CONF_KNX_TELEGRAM_DB_RETENTION_DAYS,
DEFAULT_ROUTING_IA,
DOMAIN,
KNX_TELEGRAM_BACKEND_SQLITE,
KNX_TELEGRAM_DB_RETENTION_DEFAULT,
KNX_TELEGRAM_LOAD_HOURS_DEFAULT,
)
@@ -364,6 +366,7 @@ def mock_config_entry() -> MockConfigEntry:
CONF_KNX_STATE_UPDATER: CONF_KNX_DEFAULT_STATE_UPDATER,
CONF_KNX_TELEGRAM_DB_RETENTION_DAYS: KNX_TELEGRAM_DB_RETENTION_DEFAULT,
CONF_KNX_TELEGRAM_DB_LOAD_HOURS: KNX_TELEGRAM_LOAD_HOURS_DEFAULT,
CONF_KNX_TELEGRAM_DB_BACKEND: KNX_TELEGRAM_BACKEND_SQLITE,
},
)
tests/components/knx/snapshots/test_diagnostic.ambr
+5
View File
@@ -10,6 +10,7 @@
'config_entry_options': dict({
'rate_limit': 0,
'state_updater': True,
'telegram_db_backend': 'sqlite',
'telegram_db_load_hours': 24,
'telegram_db_retention_days': 10,
}),
@@ -48,7 +49,9 @@
'config_entry_options': dict({
'rate_limit': 0,
'state_updater': True,
'telegram_db_backend': 'sqlite',
'telegram_db_load_hours': 24,
'telegram_db_postgres_dsn': '**REDACTED**',
'telegram_db_retention_days': 10,
}),
'config_store': dict({
@@ -79,6 +82,7 @@
'config_entry_options': dict({
'rate_limit': 0,
'state_updater': True,
'telegram_db_backend': 'sqlite',
'telegram_db_load_hours': 24,
'telegram_db_retention_days': 10,
}),
@@ -110,6 +114,7 @@
'config_entry_options': dict({
'rate_limit': 0,
'state_updater': True,
'telegram_db_backend': 'sqlite',
'telegram_db_load_hours': 24,
'telegram_db_retention_days': 10,
}),
tests/components/knx/test_config_flow.py
+319
View File
@@ -1,8 +1,10 @@
"""Test the KNX config flow."""
import asyncio
from contextlib import contextmanager
from unittest.mock import AsyncMock, MagicMock, Mock, patch
from knx_telegram_store.connection import ConnectionCheckResult, ConnectionErrorKind
import pytest
from xknx.exceptions import XKNXException
from xknx.exceptions.exception import CommunicationError, InvalidSecureConfiguration
@@ -21,6 +23,8 @@ from homeassistant.components.knx.config_flow import (
DEFAULT_ENTRY_DATA,
DEFAULT_ENTRY_OPTIONS,
OPTION_MANUAL_TUNNEL,
_build_dsn,
_parse_dsn,
)
from homeassistant.components.knx.const import (
CONF_KNX_AUTOMATIC,
@@ -41,13 +45,17 @@ from homeassistant.components.knx.const import (
CONF_KNX_SECURE_USER_ID,
CONF_KNX_SECURE_USER_PASSWORD,
CONF_KNX_STATE_UPDATER,
CONF_KNX_TELEGRAM_DB_BACKEND,
CONF_KNX_TELEGRAM_DB_LOAD_HOURS,
CONF_KNX_TELEGRAM_DB_POSTGRES_DSN,
CONF_KNX_TELEGRAM_DB_RETENTION_DAYS,
CONF_KNX_TUNNEL_ENDPOINT_IA,
CONF_KNX_TUNNELING,
CONF_KNX_TUNNELING_TCP,
CONF_KNX_TUNNELING_TCP_SECURE,
DOMAIN,
KNX_TELEGRAM_BACKEND_POSTGRES,
KNX_TELEGRAM_BACKEND_SQLITE,
KNX_TELEGRAM_DB_RETENTION_DEFAULT,
KNX_TELEGRAM_LOAD_HOURS_DEFAULT,
)
@@ -1065,6 +1073,7 @@ async def test_form_with_automatic_connection_handling(
CONF_KNX_STATE_UPDATER: True,
CONF_KNX_TELEGRAM_DB_RETENTION_DAYS: KNX_TELEGRAM_DB_RETENTION_DEFAULT,
CONF_KNX_TELEGRAM_DB_LOAD_HOURS: KNX_TELEGRAM_LOAD_HOURS_DEFAULT,
CONF_KNX_TELEGRAM_DB_BACKEND: KNX_TELEGRAM_BACKEND_SQLITE,
}
knx_setup.assert_called_once()
@@ -1690,6 +1699,7 @@ async def test_options_communication_settings(
CONF_KNX_TELEGRAM_STORE_SECTION: {
CONF_KNX_TELEGRAM_DB_LOAD_HOURS: KNX_TELEGRAM_LOAD_HOURS_DEFAULT,
CONF_KNX_TELEGRAM_DB_RETENTION_DAYS: 30,
CONF_KNX_TELEGRAM_DB_BACKEND: KNX_TELEGRAM_BACKEND_SQLITE,
},
},
)
@@ -1699,6 +1709,7 @@ async def test_options_communication_settings(
CONF_KNX_RATE_LIMIT: 40,
CONF_KNX_TELEGRAM_DB_RETENTION_DAYS: 30,
CONF_KNX_TELEGRAM_DB_LOAD_HOURS: KNX_TELEGRAM_LOAD_HOURS_DEFAULT,
CONF_KNX_TELEGRAM_DB_BACKEND: KNX_TELEGRAM_BACKEND_SQLITE,
}
assert mock_config_entry.data == initial_data
assert mock_config_entry.options == {
@@ -1706,5 +1717,313 @@ async def test_options_communication_settings(
CONF_KNX_RATE_LIMIT: 40,
CONF_KNX_TELEGRAM_DB_RETENTION_DAYS: 30,
CONF_KNX_TELEGRAM_DB_LOAD_HOURS: KNX_TELEGRAM_LOAD_HOURS_DEFAULT,
CONF_KNX_TELEGRAM_DB_BACKEND: KNX_TELEGRAM_BACKEND_SQLITE,
}
assert len(knx_setup.mock_calls) == 2
async def _advance_to_postgres_step(
hass: HomeAssistant, flow_id: str, *, retention_days: int = 14
) -> config_entries.ConfigFlowResult:
"""Select the PostgreSQL backend and land on its connection step."""
result = await hass.config_entries.options.async_configure(
flow_id,
user_input={
CONF_KNX_STATE_UPDATER: False,
CONF_KNX_RATE_LIMIT: 40,
CONF_KNX_TELEGRAM_STORE_SECTION: {
CONF_KNX_TELEGRAM_DB_LOAD_HOURS: KNX_TELEGRAM_LOAD_HOURS_DEFAULT,
CONF_KNX_TELEGRAM_DB_RETENTION_DAYS: retention_days,
CONF_KNX_TELEGRAM_DB_BACKEND: KNX_TELEGRAM_BACKEND_POSTGRES,
},
},
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "telegram_store_postgres"
assert not result["errors"]
return result
async def test_options_telegram_store_postgres(
hass: HomeAssistant, knx_setup: AsyncMock, mock_config_entry: MockConfigEntry
) -> None:
"""Test options flow selecting the PostgreSQL telegram store backend."""
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
result = await hass.config_entries.options.async_init(mock_config_entry.entry_id)
result = await _advance_to_postgres_step(hass, result["flow_id"])
with patch(
"knx_telegram_store.backends.postgres.PostgresStore.check_config",
return_value=ConnectionCheckResult.success(),
):
result = await hass.config_entries.options.async_configure(
result["flow_id"],
user_input={
"host": "db.local",
"port": 5432,
"user": "knx",
"password": "s3cret",
"database": "knx_telegrams",
"tls": True,
},
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert (
mock_config_entry.options[CONF_KNX_TELEGRAM_DB_BACKEND]
== KNX_TELEGRAM_BACKEND_POSTGRES
)
assert mock_config_entry.options[CONF_KNX_TELEGRAM_DB_RETENTION_DAYS] == 14
assert (
mock_config_entry.options[CONF_KNX_TELEGRAM_DB_POSTGRES_DSN]
== "postgresql://knx:s3cret@db.local:5432/knx_telegrams?sslmode=require"
)
assert len(knx_setup.mock_calls) == 2
async def test_options_telegram_store_postgres_reuses_password(
hass: HomeAssistant, knx_setup: AsyncMock, mock_config_entry: MockConfigEntry
) -> None:
"""Test the PostgreSQL store reuses the stored password when left blank."""
existing_dsn = "postgresql://olduser:oldpass@old.host:6543/olddb?sslmode=require"
mock_config_entry.add_to_hass(hass)
hass.config_entries.async_update_entry(
mock_config_entry,
options={
**mock_config_entry.options,
CONF_KNX_TELEGRAM_DB_POSTGRES_DSN: existing_dsn,
},
)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
result = await hass.config_entries.options.async_init(mock_config_entry.entry_id)
result = await _advance_to_postgres_step(hass, result["flow_id"], retention_days=7)
# Submit with an empty password - the existing one (parsed from the DSN)
# must be reused.
with patch(
"knx_telegram_store.backends.postgres.PostgresStore.check_config",
return_value=ConnectionCheckResult.success(),
):
result = await hass.config_entries.options.async_configure(
result["flow_id"],
user_input={
"host": "new.host",
"port": 5432,
"user": "newuser",
"password": "",
"database": "newdb",
"tls": False,
},
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert (
mock_config_entry.options[CONF_KNX_TELEGRAM_DB_POSTGRES_DSN]
== "postgresql://newuser:oldpass@new.host:5432/newdb"
)
assert len(knx_setup.mock_calls) == 2
@pytest.mark.parametrize(
("error_kind", "expected_error"),
[
pytest.param(ConnectionErrorKind.AUTH, "invalid_auth", id="invalid_auth"),
pytest.param(
ConnectionErrorKind.HOST_UNREACHABLE,
"host_unreachable",
id="host_unreachable",
),
],
)
async def test_options_telegram_store_postgres_connection_failure(
hass: HomeAssistant,
knx_setup: AsyncMock,
mock_config_entry: MockConfigEntry,
error_kind: ConnectionErrorKind,
expected_error: str,
) -> None:
"""Test the PostgreSQL step maps connection check failures to form errors."""
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
result = await hass.config_entries.options.async_init(mock_config_entry.entry_id)
result = await _advance_to_postgres_step(hass, result["flow_id"])
with patch(
"knx_telegram_store.backends.postgres.PostgresStore.check_config",
return_value=ConnectionCheckResult.failure(error_kind, "check failed"),
):
result = await hass.config_entries.options.async_configure(
result["flow_id"],
user_input={
"host": "db.local",
"port": 5432,
"user": "knx",
"password": "wrong_password",
"database": "knx_telegrams",
"tls": True,
},
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "telegram_store_postgres"
assert result["errors"] == {"base": expected_error}
async def test_options_telegram_store_postgres_timeout(
hass: HomeAssistant, knx_setup: AsyncMock, mock_config_entry: MockConfigEntry
) -> None:
"""Test options flow surfaces a timeout when the connection check hangs."""
async def hanging_check(dsn: str) -> None:
await asyncio.Event().wait()
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
result = await hass.config_entries.options.async_init(mock_config_entry.entry_id)
result = await _advance_to_postgres_step(hass, result["flow_id"])
with (
patch("homeassistant.components.knx.config_flow.DSN_CHECK_TIMEOUT", 0.05),
patch(
"knx_telegram_store.backends.postgres.PostgresStore.check_config",
side_effect=hanging_check,
),
):
result = await hass.config_entries.options.async_configure(
result["flow_id"],
user_input={
"host": "db.local",
"port": 5432,
"user": "knx",
"password": "s3cret",
"database": "knx_telegrams",
"tls": True,
},
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "telegram_store_postgres"
assert result["errors"] == {"base": "timeout"}
async def test_options_telegram_store_postgres_malformed_dsn(
hass: HomeAssistant, knx_setup: AsyncMock, mock_config_entry: MockConfigEntry
) -> None:
"""Test the PostgreSQL step maps a DSN the driver rejects to a form error."""
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
result = await hass.config_entries.options.async_init(mock_config_entry.entry_id)
result = await _advance_to_postgres_step(hass, result["flow_id"])
# An unterminated bracketed IPv6 address makes engine creation
# raise ValueError before any connection attempt.
result = await hass.config_entries.options.async_configure(
result["flow_id"],
user_input={
"host": "[::1",
"port": 5432,
"user": "knx",
"password": "s3cret",
"database": "knx_telegrams",
"tls": False,
},
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "telegram_store_postgres"
assert result["errors"] == {"base": "cannot_connect"}
@pytest.mark.parametrize(
("dsn", "expected"),
[
pytest.param("", {}, id="empty"),
# Invalid port makes urlparse.port raise ValueError -> {}
pytest.param("postgresql://host:notaport/db", {}, id="invalid_port"),
pytest.param(
"postgresql://u:p@h:5432/db?sslmode=require",
{
"user": "u",
"password": "p",
"host": "h",
"port": 5432,
"database": "db",
"tls": True,
},
id="full",
),
pytest.param(
"postgresql://user%40domain:p%40ss%25word@h:5432/db",
{
"user": "user@domain",
"password": "p@ss%word",
"host": "h",
"port": 5432,
"database": "db",
"tls": False,
},
id="percent_encoded_credentials",
),
pytest.param(
"postgresql://u:p@[2001:db8::1]:5432/db",
{
"user": "u",
"password": "p",
"host": "2001:db8::1",
"port": 5432,
"database": "db",
"tls": False,
},
id="ipv6_host",
),
pytest.param(
"postgresql://u:p@h:5432/db%3Fquery%23hash",
{
"user": "u",
"password": "p",
"host": "h",
"port": 5432,
"database": "db?query#hash",
"tls": False,
},
id="percent_encoded_database",
),
],
)
def test_parse_dsn(dsn: str, expected: dict) -> None:
"""Test PostgreSQL DSN parsing, including malformed input."""
assert _parse_dsn(dsn) == expected
@pytest.mark.parametrize(
("user", "password", "host", "database"),
[
pytest.param("simple", "plain", "localhost", "knx", id="plain"),
pytest.param("user@domain", "p@ss", "localhost", "knx", id="at_sign"),
pytest.param("user", "p@ss%word", "localhost", "knx", id="percent_sign"),
pytest.param(
"us:er", "p/a:s@s", "localhost", "knx", id="multiple_special_chars"
),
pytest.param("user", "pass", "2001:db8::1", "knx", id="ipv6_host"),
pytest.param(
"user", "pass", "localhost", "knx?query#hash", id="database_special_chars"
),
],
)
def test_dsn_round_trip(user: str, password: str, host: str, database: str) -> None:
"""Test _build_dsn -> _parse_dsn -> _build_dsn produces identical DSNs.
Catches double percent-encoding: urlparse returns percent-encoded values,
so _parse_dsn must decode them before they are fed back into _build_dsn.
IPv6 hosts must be bracketed in the netloc for the DSN to stay parseable.
Database names with URL delimiters are percent-encoded to prevent truncation.
"""
params = {
"user": user,
"password": password,
"host": host,
"port": 5432,
"database": database,
"tls": False,
}
dsn1 = _build_dsn(params)
parsed = _parse_dsn(dsn1)
dsn2 = _build_dsn(parsed)
assert dsn1 == dsn2
tests/components/knx/test_diagnostic.py
+6
View File
@@ -20,6 +20,7 @@ from homeassistant.components.knx.const import (
CONF_KNX_SECURE_DEVICE_AUTHENTICATION,
CONF_KNX_SECURE_USER_PASSWORD,
CONF_KNX_STATE_UPDATER,
CONF_KNX_TELEGRAM_DB_POSTGRES_DSN,
DEFAULT_ROUTING_IA,
DOMAIN,
)
@@ -100,6 +101,11 @@ async def test_diagnostic_redact(
CONF_KNX_SECURE_DEVICE_AUTHENTICATION: "device_authentication",
CONF_KNX_ROUTING_BACKBONE_KEY: "bbaacc44bbaacc44bbaacc44bbaacc44",
},
options={
CONF_KNX_TELEGRAM_DB_POSTGRES_DSN: (
"postgresql://knx:supersecret@localhost:5432/knx_telegrams"
),
},
)
knx: KNXTestKit = KNXTestKit(hass, mock_config_entry, hass_storage)
await knx.setup_integration()
tests/components/knx/test_init.py
+29
View File
@@ -38,12 +38,14 @@ from homeassistant.components.knx.const import (
CONF_KNX_SECURE_USER_ID,
CONF_KNX_SECURE_USER_PASSWORD,
CONF_KNX_STATE_UPDATER,
CONF_KNX_TELEGRAM_DB_BACKEND,
CONF_KNX_TELEGRAM_DB_LOAD_HOURS,
CONF_KNX_TELEGRAM_DB_RETENTION_DAYS,
CONF_KNX_TUNNELING,
CONF_KNX_TUNNELING_TCP,
CONF_KNX_TUNNELING_TCP_SECURE,
DOMAIN,
KNX_TELEGRAM_BACKEND_SQLITE,
KNX_TELEGRAM_DB_RETENTION_DEFAULT,
KNX_TELEGRAM_LOAD_HOURS_DEFAULT,
KNXConfigEntryData,
@@ -437,3 +439,30 @@ async def test_async_migrate_entry_future_version(hass: HomeAssistant) -> None:
with patch("homeassistant.components.knx.async_setup_entry", return_value=True):
assert not await hass.config_entries.async_setup(config_entry.entry_id)
async def test_async_migrate_entry_v2_to_v2_2(hass: HomeAssistant) -> None:
"""Test KNX config entry migration from v2.x to v2.2."""
config_entry = MockConfigEntry(
title="KNX",
domain=DOMAIN,
version=2,
minor_version=1,
data={
"other_setting": "some_value",
},
options={
"some_option": "value",
},
)
config_entry.add_to_hass(hass)
with patch("homeassistant.components.knx.async_setup_entry", return_value=True):
assert await hass.config_entries.async_setup(config_entry.entry_id)
assert config_entry.version == 2
assert config_entry.minor_version == 2
assert (
config_entry.options[CONF_KNX_TELEGRAM_DB_BACKEND]
== KNX_TELEGRAM_BACKEND_SQLITE
)
tests/components/knx/test_telegrams.py
+68
View File
@@ -2,6 +2,7 @@
from __future__ import annotations
import asyncio
from copy import copy
from datetime import datetime
from unittest.mock import AsyncMock, patch
@@ -11,9 +12,12 @@ from knx_telegram_store import KnxTelegramStoreException, StoredTelegram, Telegr
import pytest
from homeassistant.components.knx.const import (
CONF_KNX_TELEGRAM_DB_BACKEND,
CONF_KNX_TELEGRAM_DB_POSTGRES_DSN,
CONF_KNX_TELEGRAM_DB_RETENTION_DAYS,
DOMAIN,
KNX_MODULE_KEY,
KNX_TELEGRAM_BACKEND_POSTGRES,
REPAIR_ISSUE_TELEGRAM_BACKEND_ERROR,
)
from homeassistant.components.knx.telegrams import TelegramDict
@@ -156,6 +160,34 @@ async def test_store_telegram_history_error_handling(
assert issue is not None
async def test_store_telegram_history_needs_migration_timeout(
hass: HomeAssistant,
knx: KNXTestKit,
) -> None:
"""Test that store initialization is aborted when needs_migration times out."""
async def hanging_probe() -> bool:
await asyncio.Event().wait()
return False
with (
patch("homeassistant.components.knx.telegrams.STORE_INIT_TIMEOUT", 0.05),
patch(
"knx_telegram_store.BufferedSqliteStore.needs_migration",
side_effect=hanging_probe,
),
):
await knx.setup_integration()
telegrams_module = hass.data[KNX_MODULE_KEY].telegrams
assert telegrams_module.store is None
# Check that the repair issue was created
issue_registry = ir.async_get(hass)
issue = issue_registry.async_get_issue(DOMAIN, REPAIR_ISSUE_TELEGRAM_BACKEND_ERROR)
assert issue is not None
async def test_migrate_telegrams_from_json(
hass: HomeAssistant,
knx: KNXTestKit,
@@ -483,3 +515,39 @@ async def test_nightly_eviction_error_handling(
assert "Database error evicting expired KNX telegrams" in caplog.text
# Store remains operational after the failed eviction
assert telegrams_module.store is not None
async def test_postgres_backend_init_error(
hass: HomeAssistant,
knx: KNXTestKit,
) -> None:
"""Test PostgreSQL backend DSN handling and init failure path."""
dsn = "postgresql://user:secret@db.local:5432/knx"
knx.mock_config_entry.add_to_hass(hass)
hass.config_entries.async_update_entry(
knx.mock_config_entry,
options=knx.mock_config_entry.options
| {
CONF_KNX_TELEGRAM_DB_BACKEND: KNX_TELEGRAM_BACKEND_POSTGRES,
CONF_KNX_TELEGRAM_DB_POSTGRES_DSN: dsn,
},
)
# Mock the store to avoid constructing a real SQLAlchemy engine / connecting.
mock_store = AsyncMock()
mock_store.needs_migration.return_value = False
mock_store.initialize.side_effect = KnxTelegramStoreException("no server")
with patch(
"homeassistant.components.knx.telegrams.BufferedPostgresStore",
return_value=mock_store,
):
await knx.setup_integration(add_entry_to_hass=False)
telegrams_module = hass.data[KNX_MODULE_KEY].telegrams
assert telegrams_module.store is None
issue_registry = ir.async_get(hass)
assert (
issue_registry.async_get_issue(DOMAIN, REPAIR_ISSUE_TELEGRAM_BACKEND_ERROR)
is not None
)
tests/components/knx/test_websocket.py
+39
View File
@@ -10,8 +10,11 @@ import pytest
from syrupy.assertion import SnapshotAssertion
from homeassistant.components.knx.const import (
CONF_KNX_TELEGRAM_DB_BACKEND,
CONF_KNX_TELEGRAM_DB_POSTGRES_DSN,
KNX_ADDRESS,
KNX_MODULE_KEY,
KNX_TELEGRAM_BACKEND_POSTGRES,
SUPPORTED_PLATFORMS_UI,
)
from homeassistant.components.knx.project import STORAGE_KEY as KNX_PROJECT_STORAGE_KEY
@@ -37,10 +40,46 @@ async def test_knx_get_base_data_command(
assert res["result"]["connection_info"]["version"] is not None
assert res["result"]["connection_info"]["connected"]
assert res["result"]["connection_info"]["current_address"] == "0.0.0"
assert res["result"]["connection_info"]["telegram_backend"] == "sqlite"
assert res["result"]["project_info"] is None
assert not SUPPORTED_PLATFORMS_UI.difference(res["result"]["supported_platforms"])
async def test_knx_get_base_data_command_postgres(
hass: HomeAssistant, knx: KNXTestKit, hass_ws_client: WebSocketGenerator
) -> None:
"""Test knx/get_base_data reports the PostgreSQL telegram backend."""
knx.mock_config_entry.add_to_hass(hass)
hass.config_entries.async_update_entry(
knx.mock_config_entry,
options=knx.mock_config_entry.options
| {
CONF_KNX_TELEGRAM_DB_BACKEND: KNX_TELEGRAM_BACKEND_POSTGRES,
CONF_KNX_TELEGRAM_DB_POSTGRES_DSN: "postgresql://user:pw@db.local:5432/knx",
},
)
# Patch methods on the real class so the isinstance check in the
# websocket handler still sees a BufferedPostgresStore instance.
with (
patch(
"knx_telegram_store.BufferedPostgresStore.needs_migration",
return_value=False,
),
patch("knx_telegram_store.BufferedPostgresStore.initialize"),
patch(
"knx_telegram_store.BufferedPostgresStore.get_last_unique_telegrams",
return_value=[],
),
):
await knx.setup_integration(add_entry_to_hass=False)
client = await hass_ws_client(hass)
await client.send_json_auto_id({"type": "knx/get_base_data"})
res = await client.receive_json()
assert res["success"], res
assert res["result"]["connection_info"]["telegram_backend"] == "postgres"
@pytest.mark.usefixtures("load_knxproj")
async def test_knx_get_base_data_command_with_project(
hass: HomeAssistant,