diff --git a/homeassistant/components/knx/telegrams.py b/homeassistant/components/knx/telegrams.py index ab08ffd07e97..3d48589d2451 100644 --- a/homeassistant/components/knx/telegrams.py +++ b/homeassistant/components/knx/telegrams.py @@ -339,7 +339,20 @@ class Telegrams: ) return - stored_telegrams = [self.dict_to_model(t) for t in json_data] + # Legacy JSON data from older HA instances might miss fields added later + # (e.g., data_secure was added in 2026.3, value, payload, dpt_main/sub, names might also be missing) + default_migration_values = { + "value": None, + "payload": None, + "dpt_main": None, + "dpt_sub": None, + "source_name": "", + "destination_name": "", + "data_secure": False, + } + stored_telegrams = [ + self.dict_to_model(default_migration_values | t) for t in json_data + ] try: if stored_telegrams: await self.store.store_many(stored_telegrams) diff --git a/tests/components/knx/test_telegrams_migration.py b/tests/components/knx/test_telegrams_migration.py index 69c6330a6557..5421974e9de3 100644 --- a/tests/components/knx/test_telegrams_migration.py +++ b/tests/components/knx/test_telegrams_migration.py @@ -86,3 +86,47 @@ async def test_migrate_telegrams_json_to_sqlite( # Verify legacy file removal assert not os.path.exists(legacy_path) + + +async def test_migrate_telegrams_json_missing_keys( + hass: HomeAssistant, + knx: KNXTestKit, +) -> None: + """Test migrating telegrams from legacy JSON missing fields (e.g., data_secure).""" + store = Store[dict[str, Any]](hass, version=1, key="knx/telegrams_history.json") + legacy_path = store.path + + legacy_data = [ + { + "destination": "3/2/100", + "source": "1.0.6", + "direction": "Incoming", + "payload": [0], + "telegramtype": "GroupValueWrite", + "timestamp": "2026-05-07T23:34:45.127015+02:00", + "value": 0, + } + ] + + await store.async_save(legacy_data) + + await knx.setup_integration() + telegrams_module = hass.data[KNX_MODULE_KEY].telegrams + + await hass.async_block_till_done() + + # Verify migration + assert telegrams_module.store is not None + result = await telegrams_module.store.query(TelegramQuery(order_descending=False)) + assert len(result.telegrams) == 1 + + # Check that data_secure defaults to False/None as appropriate, and others default to None or empty string + assert result.telegrams[0].destination == "3/2/100" + assert result.telegrams[0].data_secure is False + assert result.telegrams[0].source_name == "" + assert result.telegrams[0].destination_name == "" + assert result.telegrams[0].dpt_main is None + assert result.telegrams[0].dpt_sub is None + + # Verify legacy file removal + assert not os.path.exists(legacy_path)