mirror of
https://github.com/elisspace/core.git
synced 2026-10-03 07:49:57 +00:00
Test invalid payload type for websocket subscription
This commit is contained in:
committed by
Marvin Wichmann
parent
78524d8f83
commit
09d1c90e67
@@ -11,7 +11,13 @@ from xknx.dpt import DPTArray, DPTBinary
|
||||
from xknx.io import DEFAULT_MCAST_GRP, DEFAULT_MCAST_PORT
|
||||
from xknx.telegram import Telegram, TelegramDirection
|
||||
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,
|
||||
IndividualAddressRead,
|
||||
)
|
||||
|
||||
from homeassistant.components.knx.const import (
|
||||
CONF_KNX_AUTOMATIC,
|
||||
@@ -188,6 +194,19 @@ class KNXTestKit:
|
||||
await self.xknx.telegrams.join()
|
||||
await self.hass.async_block_till_done()
|
||||
|
||||
async def receive_individual_address_read(self):
|
||||
"""Inject incoming IndividualAddressRead telegram."""
|
||||
self.xknx.telegrams.put_nowait(
|
||||
Telegram(
|
||||
destination_address=IndividualAddress(self.INDIVIDUAL_ADDRESS),
|
||||
direction=TelegramDirection.INCOMING,
|
||||
payload=IndividualAddressRead(),
|
||||
source_address=IndividualAddress("1.3.5"),
|
||||
)
|
||||
)
|
||||
await self.xknx.telegrams.join()
|
||||
await self.hass.async_block_till_done()
|
||||
|
||||
async def receive_read(
|
||||
self,
|
||||
group_address: str,
|
||||
|
||||
@@ -46,6 +46,7 @@ async def test_knx_subscribe_telegrams_command(
|
||||
await knx.receive_read("1/2/3")
|
||||
await knx.receive_write("1/3/4", True)
|
||||
await knx.receive_write("1/3/4", False)
|
||||
await knx.receive_individual_address_read()
|
||||
await knx.receive_write("1/3/8", (0x34, 0x45))
|
||||
# send outgoing events
|
||||
await hass.services.async_call(
|
||||
@@ -92,6 +93,6 @@ async def test_knx_subscribe_telegrams_command(
|
||||
assert res["event"]["type"] == "GroupValueWrite"
|
||||
assert (
|
||||
res["event"]["source_address"] == "0.0.0"
|
||||
) # needs to be the currently connected IA connected to
|
||||
) # needs to be the IA currently connected to
|
||||
assert res["event"]["direction"] == "label.outgoing"
|
||||
assert res["event"]["timestamp"] is not None
|
||||
|
||||
Reference in New Issue
Block a user