Add tests for KNX diagnostic and expose (#64938)

* Add test for KNX diagnostic

* Add test for KNX expose

* Apply review suggestions
This commit is contained in:
Marvin Wichmann 2022-01-27 23:03:20 +01:00 committed by Paulus Schoutsen
parent cd6c182c07
commit 3dde12f887
5 changed files with 102 additions and 9 deletions

View File

@ -560,12 +560,7 @@ omit =
homeassistant/components/knx/__init__.py homeassistant/components/knx/__init__.py
homeassistant/components/knx/climate.py homeassistant/components/knx/climate.py
homeassistant/components/knx/cover.py homeassistant/components/knx/cover.py
homeassistant/components/knx/diagnostics.py
homeassistant/components/knx/expose.py
homeassistant/components/knx/knx_entity.py
homeassistant/components/knx/light.py
homeassistant/components/knx/notify.py homeassistant/components/knx/notify.py
homeassistant/components/knx/schema.py
homeassistant/components/kodi/__init__.py homeassistant/components/kodi/__init__.py
homeassistant/components/kodi/browse_media.py homeassistant/components/kodi/browse_media.py
homeassistant/components/kodi/const.py homeassistant/components/kodi/const.py

View File

@ -148,8 +148,6 @@ class KNXExposeSensor:
async def _async_set_knx_value(self, value: StateType) -> None: async def _async_set_knx_value(self, value: StateType) -> None:
"""Set new value on xknx ExposeSensor.""" """Set new value on xknx ExposeSensor."""
if value is None:
return
await self.device.set(value) await self.device.set(value)

View File

@ -13,7 +13,7 @@ from xknx.telegram import Telegram, TelegramDirection
from xknx.telegram.address import GroupAddress, IndividualAddress from xknx.telegram.address import GroupAddress, IndividualAddress
from xknx.telegram.apci import APCI, GroupValueRead, GroupValueResponse, GroupValueWrite from xknx.telegram.apci import APCI, GroupValueRead, GroupValueResponse, GroupValueWrite
from homeassistant.components.knx import ConnectionSchema from homeassistant.components.knx import ConnectionSchema, KNXModule
from homeassistant.components.knx.const import ( from homeassistant.components.knx.const import (
CONF_KNX_AUTOMATIC, CONF_KNX_AUTOMATIC,
CONF_KNX_CONNECTION_TYPE, CONF_KNX_CONNECTION_TYPE,
@ -40,6 +40,11 @@ class KNXTestKit:
# telegrams to an InternalGroupAddress won't be queued here # telegrams to an InternalGroupAddress won't be queued here
self._outgoing_telegrams: asyncio.Queue = asyncio.Queue() self._outgoing_telegrams: asyncio.Queue = asyncio.Queue()
@property
def knx_module(self) -> KNXModule:
"""Get the KNX module."""
return self.hass.data[KNX_DOMAIN]
def assert_state(self, entity_id: str, state: str, **attributes) -> None: def assert_state(self, entity_id: str, state: str, **attributes) -> None:
"""Assert the state of an entity.""" """Assert the state of an entity."""
test_state = self.hass.states.get(entity_id) test_state = self.hass.states.get(entity_id)

View File

@ -0,0 +1,67 @@
"""Tests for the diagnostics data provided by the KNX integration."""
from unittest.mock import patch
from aiohttp import ClientSession
from homeassistant.core import HomeAssistant
from tests.common import MockConfigEntry
from tests.components.diagnostics import get_diagnostics_for_config_entry
from tests.components.knx.conftest import KNXTestKit
async def test_diagnostics(
hass: HomeAssistant,
hass_client: ClientSession,
mock_config_entry: MockConfigEntry,
knx: KNXTestKit,
):
"""Test diagnostics."""
await knx.setup_integration({})
with patch("homeassistant.config.async_hass_config_yaml", return_value={}):
# Overwrite the version for this test since we don't want to change this with every library bump
knx.xknx.version = "1.0.0"
assert await get_diagnostics_for_config_entry(
hass, hass_client, mock_config_entry
) == {
"config_entry_data": {
"connection_type": "automatic",
"individual_address": "15.15.250",
"multicast_group": "224.0.23.12",
"multicast_port": 3671,
},
"configuration_error": None,
"configuration_yaml": None,
"xknx": {"current_address": "0.0.0", "version": "1.0.0"},
}
async def test_diagnostic_config_error(
hass: HomeAssistant,
hass_client: ClientSession,
mock_config_entry: MockConfigEntry,
knx: KNXTestKit,
):
"""Test diagnostics."""
await knx.setup_integration({})
with patch(
"homeassistant.config.async_hass_config_yaml",
return_value={"knx": {"wrong_key": {}}},
):
# Overwrite the version for this test since we don't want to change this with every library bump
knx.xknx.version = "1.0.0"
assert await get_diagnostics_for_config_entry(
hass, hass_client, mock_config_entry
) == {
"config_entry_data": {
"connection_type": "automatic",
"individual_address": "15.15.250",
"multicast_group": "224.0.23.12",
"multicast_port": 3671,
},
"configuration_error": "extra keys not allowed @ data['knx']['wrong_key']",
"configuration_yaml": {"wrong_key": {}},
"xknx": {"current_address": "0.0.0", "version": "1.0.0"},
}

View File

@ -1,5 +1,8 @@
"""Test KNX expose.""" """Test KNX expose."""
from homeassistant.components.knx import CONF_KNX_EXPOSE, KNX_ADDRESS import time
from unittest.mock import patch
from homeassistant.components.knx import CONF_KNX_EXPOSE, DOMAIN, KNX_ADDRESS
from homeassistant.components.knx.schema import ExposeSchema from homeassistant.components.knx.schema import ExposeSchema
from homeassistant.const import CONF_ATTRIBUTE, CONF_ENTITY_ID, CONF_TYPE from homeassistant.const import CONF_ATTRIBUTE, CONF_ENTITY_ID, CONF_TYPE
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant
@ -123,3 +126,28 @@ async def test_expose_attribute_with_default(hass: HomeAssistant, knx: KNXTestKi
# Change state to "off"; no attribute # Change state to "off"; no attribute
hass.states.async_set(entity_id, "off", {}) hass.states.async_set(entity_id, "off", {})
await knx.assert_write("1/1/8", (0,)) await knx.assert_write("1/1/8", (0,))
@patch("time.localtime")
async def test_expose_with_date(localtime, hass: HomeAssistant, knx: KNXTestKit):
"""Test an expose with a date."""
localtime.return_value = time.struct_time([2022, 1, 7, 9, 13, 14, 6, 0, 0])
await knx.setup_integration(
{
CONF_KNX_EXPOSE: {
CONF_TYPE: "datetime",
KNX_ADDRESS: "1/1/8",
}
},
)
assert not hass.states.async_all()
await knx.assert_write("1/1/8", (0x7A, 0x1, 0x7, 0xE9, 0xD, 0xE, 0x20, 0x80))
await knx.receive_read("1/1/8")
await knx.assert_response("1/1/8", (0x7A, 0x1, 0x7, 0xE9, 0xD, 0xE, 0x20, 0x80))
entries = hass.config_entries.async_entries(DOMAIN)
assert len(entries) == 1
assert await hass.config_entries.async_unload(entries[0].entry_id)