Repository: XKNX/xknx
Pull request #1486 (2024-06-15 18:31:33)

Non-awaitable callbacks in TelegramQueue
<!--
  You are awesome! Thanks for contributing to our project!
  Please, DO NOT DELETE ANY TEXT from this template!.
-->
## Description
<!--
Please include a summary of the change and which issue is fixed.
Please also include relevant motivation and context.
List any dependencies that are required for this change.
-->


Fixes # (issue)

## Type of change
<!--
Please tick the applicable options.
NOTE: Ticking multiple options most likely indicates
that your change is to big and it is suggested to split it into several smaller PRs.
-->

- [ ] Bug fix (non-breaking change which fixes an issue)
- [ ] New feature (non-breaking change which adds functionality)
- [x] Breaking change (fix or feature that would cause existing functionality to not work as expected)
- [ ] Code quality improvements to existing code or addition of tests

## Checklist

- [x] The documentation has been adjusted accordingly
- [x] Tests have been added that prove the fix is effective or that the feature works
- [x] The changes are documented in the changelog (docs/changelog.md)

codecov[bot]: ## [Codecov](https://app.codecov.io/gh/XKNX/xknx/pull/1486?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX) Report
All modified and coverable lines are covered by tests :white_check_mark:
> Project coverage is 96.62%. Comparing base [(`05c94ec`)](https://app.codecov.io/gh/XKNX/xknx/commit/05c94ece0077835788f32c1dc7782f167e53e5df?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX) to head [(`a96d643`)](https://app.codecov.io/gh/XKNX/xknx/commit/a96d6439135082098505bf6374b9f3479dfbcc41?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX).

<details><summary>Additional details and impacted files</summary>


[![Impacted file tree graph](https://app.codecov.io/gh/XKNX/xknx/pull/1486/graphs/tree.svg?width=650&height=150&src=pr&token=irWbIygS84&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX)](https://app.codecov.io/gh/XKNX/xknx/pull/1486?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX)

```diff
@@           Coverage Diff           @@
##             main    #1486   +/-   ##
=======================================
  Coverage   96.62%   96.62%           
=======================================
  Files         154      154           
  Lines       10023    10027    +4     
=======================================
+ Hits         9685     9689    +4     
  Misses        338      338           
```


| [Files](https://app.codecov.io/gh/XKNX/xknx/pull/1486?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX) | Coverage Δ | |
|---|---|---|
| [xknx/core/telegram\_queue.py](https://app.codecov.io/gh/XKNX/xknx/pull/1486?src=pr&el=tree&filepath=xknx%2Fcore%2Ftelegram_queue.py&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX#diff-eGtueC9jb3JlL3RlbGVncmFtX3F1ZXVlLnB5) | `98.30% <100.00%> (+0.04%)` | :arrow_up: |
| [xknx/core/value\_reader.py](https://app.codecov.io/gh/XKNX/xknx/pull/1486?src=pr&el=tree&filepath=xknx%2Fcore%2Fvalue_reader.py&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX#diff-eGtueC9jb3JlL3ZhbHVlX3JlYWRlci5weQ==) | `100.00% <100.00%> (ø)` | |
| [xknx/typing/\_\_init\_\_.py](https://app.codecov.io/gh/XKNX/xknx/pull/1486?src=pr&el=tree&filepath=xknx%2Ftyping%2F__init__.py&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX#diff-eGtueC90eXBpbmcvX19pbml0X18ucHk=) | `100.00% <100.00%> (ø)` | |
| [xknx/xknx.py](https://app.codecov.io/gh/XKNX/xknx/pull/1486?src=pr&el=tree&filepath=xknx%2Fxknx.py&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX#diff-eGtueC94a254LnB5) | `88.88% <100.00%> (ø)` | |

</details>

Patch:
diff --git a/docs/changelog.md b/docs/changelog.md
index c4a9f735..796e18af 100644
--- a/docs/changelog.md
+++ b/docs/changelog.md
@@ -11,7 +11,7 @@ nav_order: 2
 ### General
 
 - Drop support for Python 3.9
-- Change callback signatures from awaitable to callable in `XKNX.device_updated_cb`, Device, Devices and RemoteValue.
+- Change callback signatures from awaitable to callable in `XKNX.device_updated_cb`, TelegramQueue, Device, Devices and RemoteValue.
 - Added eager telegram data decoding for GroupValueWrite / GroupValueResponse Telegrams. DPTs for group addresses can be set using `xknx.group_address_dpt.set()`. `Telegram` has a new attribute `decoded_data` which is set when a decoder was found.
 
 ### Devices
diff --git a/docs/xknx.md b/docs/xknx.md
index 0c54c531..40151f74 100644
--- a/docs/xknx.md
+++ b/docs/xknx.md
@@ -135,14 +135,14 @@ await xknx.devices['TestSwitch'].set_off()
 
 # [](#header-2)Callbacks
 
-An awaitable `telegram_received_cb` will be called for each KNX telegram received by the XKNX daemon. Example:
+A callback `telegram_received_cb` will be called for each KNX telegram received by the XKNX daemon. Example:
 
 ```python
 import asyncio
 from xknx import XKNX
 from xknx.telegram import Telegram
 
-async def telegram_received_cb(telegram: Telegram):
+def telegram_received_cb(telegram: Telegram):
     print("Telegram received: {0}".format(telegram))
 
 async def main():
diff --git a/examples/example_telegram_monitor.py b/examples/example_telegram_monitor.py
index df14f14a..babe60de 100644
--- a/examples/example_telegram_monitor.py
+++ b/examples/example_telegram_monitor.py
@@ -8,7 +8,7 @@ from xknx import XKNX
 from xknx.telegram import AddressFilter, Telegram
 
 
-async def telegram_received_cb(telegram: Telegram) -> None:
+def telegram_received_cb(telegram: Telegram) -> None:
     """Do something with the received telegram."""
     print(f"Telegram received: {telegram}")
 
diff --git a/xknx/core/telegram_queue.py b/xknx/core/telegram_queue.py
index 9f4c392e..376e391c 100644
--- a/xknx/core/telegram_queue.py
+++ b/xknx/core/telegram_queue.py
@@ -12,19 +12,18 @@ Telegrams addressed to IndividualAddresses are not processed by this queue.
 from __future__ import annotations
 
 import asyncio
-from collections.abc import Awaitable, Callable
+from collections.abc import Awaitable
 import logging
 from typing import TYPE_CHECKING
 
 from xknx.exceptions import CommunicationError, XKNXException
 from xknx.telegram import AddressFilter, Telegram, TelegramDirection
 from xknx.telegram.address import GroupAddress, InternalGroupAddress
+from xknx.typing import TelegramCallbackType
 
 if TYPE_CHECKING:
     from xknx.xknx import XKNX
 
-    AsyncTelegramCallback = Callable[[Telegram], Awaitable[None]]
-
 logger = logging.getLogger("xknx.log")
 telegram_logger = logging.getLogger("xknx.telegram")
 
@@ -37,7 +36,7 @@ class TelegramQueue:
 
         def __init__(
             self,
-            callback: AsyncTelegramCallback,
+            callback: TelegramCallbackType,
             address_filters: list[AddressFilter] | None = None,
             group_addresses: list[GroupAddress | InternalGroupAddress] | None = None,
             match_for_outgoing_telegrams: bool = False,
@@ -79,7 +78,7 @@ class TelegramQueue:
 
     def register_telegram_received_cb(
         self,
-        telegram_received_cb: AsyncTelegramCallback,
+        telegram_received_cb: TelegramCallbackType,
         address_filters: list[AddressFilter] | None = None,
         group_addresses: list[GroupAddress | InternalGroupAddress] | None = None,
         match_for_outgoing: bool = False,
@@ -218,15 +217,13 @@ class TelegramQueue:
 
     async def _run_telegram_received_cbs(self, telegram: Telegram) -> None:
         """Run registered callbacks. Don't propagate exceptions."""
-        callbacks = [
-            cb.callback(telegram)
-            for cb in self.telegram_received_cbs
-            if cb.is_within_filter(telegram)
-        ]
-        try:
-            await asyncio.gather(*callbacks)
-        except Exception:  # pylint: disable=broad-except
-            logger.exception(
-                "Unexpected error while processing telegram_received_cb for %s",
-                telegram,
-            )
+        for callback in self.telegram_received_cbs:
+            if not callback.is_within_filter(telegram):
+                continue
+            try:
+                callback.callback(telegram)
+            except Exception:  # pylint: disable=broad-except
+                logger.exception(
+                    "Unexpected error while processing telegram_received_cb for %s",
+                    telegram,
+                )
diff --git a/xknx/core/value_reader.py b/xknx/core/value_reader.py
index 2a5c0af6..ed1fffdc 100644
--- a/xknx/core/value_reader.py
+++ b/xknx/core/value_reader.py
@@ -76,7 +76,7 @@ class ValueReader:
         )
         await self.xknx.telegrams.put(telegram)
 
-    async def telegram_received(self, telegram: Telegram) -> None:
+    def telegram_received(self, telegram: Telegram) -> None:
         """Test if telegram has correct group address and trigger event."""
         if telegram.destination_address == self.group_address and isinstance(
             telegram.payload, GroupValueResponse | GroupValueWrite
diff --git a/xknx/typing/__init__.py b/xknx/typing/__init__.py
index 3f716291..b3b190b8 100644
--- a/xknx/typing/__init__.py
+++ b/xknx/typing/__init__.py
@@ -5,8 +5,11 @@ from typing import TYPE_CHECKING, TypeVar
 
 if TYPE_CHECKING:
     from xknx.devices import Device
+    from xknx.telegram import Telegram
 
 CallbackType = Callable[[], None]
 
 DeviceT = TypeVar("DeviceT", bound="Device")
 DeviceCallbackType = Callable[[DeviceT], None]
+
+TelegramCallbackType = Callable[["Telegram"], None]
diff --git a/xknx/xknx.py b/xknx/xknx.py
index ab75c244..138d1d58 100644
--- a/xknx/xknx.py
+++ b/xknx/xknx.py
@@ -29,7 +29,7 @@ from xknx.io import (
 )
 from xknx.management import Management
 from xknx.telegram import GroupAddress, GroupAddressType, IndividualAddress, Telegram
-from xknx.typing import DeviceCallbackType
+from xknx.typing import DeviceCallbackType, TelegramCallbackType
 
 from .__version__ import __version__ as VERSION
 
@@ -42,7 +42,7 @@ class XKNX:
     def __init__(
         self,
         address_format: GroupAddressType = GroupAddressType.LONG,
-        telegram_received_cb: Callable[[Telegram], Awaitable[None]] | None = None,
+        telegram_received_cb: TelegramCallbackType | None = None,
         device_updated_cb: DeviceCallbackType[Device] | None = None,
         connection_state_changed_cb: Callable[[XknxConnectionState], Awaitable[None]]
         | None = None,


Test patch:
diff --git a/test/core_tests/group_address_dpt_test.py b/test/core_tests/group_address_dpt_test.py
index 36945df1..cf28ecf6 100644
--- a/test/core_tests/group_address_dpt_test.py
+++ b/test/core_tests/group_address_dpt_test.py
@@ -1,6 +1,6 @@
 """Test for group address dpt."""
 
-from unittest.mock import patch
+from unittest.mock import Mock, patch
 
 from xknx.dpt import DPTArray, DPTHumidity, DPTScaling, DPTTemperature
 from xknx.telegram import GroupAddress, Telegram, TelegramDirection, apci
@@ -8,15 +8,6 @@ from xknx.telegram import GroupAddress, Telegram, TelegramDirection, apci
 
 async def test_group_address_dpt_in_telegram_queue(xknx_no_interface):
     """Test group address dpt."""
-    telegrams = 0
-    test_telegram: Telegram = None
-
-    async def _telegram_callback(telegram: Telegram):
-        nonlocal telegrams
-        nonlocal test_telegram
-        telegrams += 1
-        test_telegram = telegram
-
     xknx = xknx_no_interface
     telegram = Telegram(
         destination_address=GroupAddress("1/2/3"),
@@ -28,40 +19,38 @@ async def test_group_address_dpt_in_telegram_queue(xknx_no_interface):
         direction=TelegramDirection.INCOMING,
         payload=apci.GroupValueRead(),
     )
-    xknx.telegram_queue.register_telegram_received_cb(_telegram_callback)
+    telegram_callback = Mock()
+    xknx.telegram_queue.register_telegram_received_cb(telegram_callback)
     async with xknx:
         await xknx.telegrams.put(telegram)
         await xknx.telegrams.join()
-        assert telegrams == 1
+        assert telegram_callback.call_count == 1
+        test_telegram = telegram_callback.call_args[0][0]
         assert test_telegram.decoded_data is None
 
         xknx.group_address_dpt.set({GroupAddress("1/2/3"): {"main": 5, "sub": 1}})
         await xknx.telegrams.put(telegram)
         await xknx.telegrams.join()
-        assert telegrams == 2
+        assert telegram_callback.call_count == 2
+        test_telegram = telegram_callback.call_args[0][0]
         assert test_telegram.decoded_data is not None
         assert test_telegram.decoded_data.transcoder is DPTScaling
         assert test_telegram.decoded_data.value == 50
         # do nothing for read-telegrams
         await xknx.telegrams.put(read_telegram)
         await xknx.telegrams.join()
-        assert telegrams == 3
+        assert telegram_callback.call_count == 3
+        test_telegram = telegram_callback.call_args[0][0]
         assert test_telegram.decoded_data is None
 
 
 @patch("logging.Logger.warning")
 async def test_get_invalid_payload(logger_warning_mock, xknx_no_interface):
     """Test processing when DPT doesn't fit payload."""
-
-    test_telegram: Telegram = None
-
-    async def _telegram_callback(telegram: Telegram):
-        nonlocal test_telegram
-        test_telegram = telegram
-
     xknx = xknx_no_interface
     xknx.group_address_dpt.set({GroupAddress("1/2/3"): {"main": 5}})
-    xknx.telegram_queue.register_telegram_received_cb(_telegram_callback)
+    telegram_callback = Mock()
+    xknx.telegram_queue.register_telegram_received_cb(telegram_callback)
 
     telegram = Telegram(
         destination_address=GroupAddress("1/2/3"),
@@ -71,6 +60,8 @@ async def test_get_invalid_payload(logger_warning_mock, xknx_no_interface):
     async with xknx:
         await xknx.telegrams.put(telegram)
         await xknx.telegrams.join()
+        assert telegram_callback.call_count == 1
+        test_telegram = telegram_callback.call_args[0][0]
         assert test_telegram.decoded_data is None
         assert logger_warning_mock.call_count == 1
         assert "DPT decoding error" in logger_warning_mock.call_args[0][0]
diff --git a/test/core_tests/telegram_queue_test.py b/test/core_tests/telegram_queue_test.py
index 20f12c85..f916023e 100644
--- a/test/core_tests/telegram_queue_test.py
+++ b/test/core_tests/telegram_queue_test.py
@@ -1,7 +1,7 @@
 """Unit test for telegram received callback."""
 
 import asyncio
-from unittest.mock import AsyncMock, MagicMock, call, patch
+from unittest.mock import AsyncMock, MagicMock, Mock, call, patch
 
 import pytest
 
@@ -102,8 +102,8 @@ class TestTelegramQueue:
         """Test telegram_received_callback after state of switch was changed."""
 
         xknx = XKNX()
-        async_telegram_received_cb = AsyncMock()
-        xknx.telegram_queue.register_telegram_received_cb(async_telegram_received_cb)
+        telegram_received_cb = Mock()
+        xknx.telegram_queue.register_telegram_received_cb(telegram_received_cb)
 
         telegram = Telegram(
             destination_address=GroupAddress("1/2/3"),
@@ -111,16 +111,16 @@ class TestTelegramQueue:
             payload=GroupValueWrite(DPTBinary(1)),
         )
         await xknx.telegram_queue.process_telegram_incoming(telegram)
-        async_telegram_received_cb.assert_called_once_with(telegram)
+        telegram_received_cb.assert_called_once_with(telegram)
 
     async def test_register_with_outgoing_telegrams(self):
         """Test telegram_received_callback with outgoing telegrams."""
 
         xknx = XKNX()
         xknx.cemi_handler = AsyncMock()
-        async_telegram_received_cb = AsyncMock()
+        telegram_received_cb = Mock()
         xknx.telegram_queue.register_telegram_received_cb(
-            async_telegram_received_cb, None, None, True
+            telegram_received_cb, None, None, True
         )
 
         telegram = Telegram(
@@ -130,15 +130,15 @@ class TestTelegramQueue:
         )
 
         await xknx.telegram_queue.process_telegram_outgoing(telegram)
-        async_telegram_received_cb.assert_called_once_with(telegram)
+        telegram_received_cb.assert_called_once_with(telegram)
 
     async def test_register_with_outgoing_telegrams_does_not_trigger(self):
         """Test telegram_received_callback with outgoing telegrams."""
 
         xknx = XKNX()
         xknx.cemi_handler = AsyncMock()
-        async_telegram_received_cb = AsyncMock()
-        xknx.telegram_queue.register_telegram_received_cb(async_telegram_received_cb)
+        telegram_received_cb = Mock()
+        xknx.telegram_queue.register_telegram_received_cb(telegram_received_cb)
 
         telegram = Telegram(
             destination_address=GroupAddress("1/2/3"),
@@ -147,7 +147,7 @@ class TestTelegramQueue:
         )
 
         await xknx.telegram_queue.process_telegram_outgoing(telegram)
-        async_telegram_received_cb.assert_not_called()
+        telegram_received_cb.assert_not_called()
 
     #
     # TEST UNREGISTER
@@ -156,10 +156,10 @@ class TestTelegramQueue:
         """Test telegram_received_callback after state of switch was changed."""
 
         xknx = XKNX()
-        async_telegram_received_callback = AsyncMock()
+        telegram_received_cb = Mock()
 
         callback = xknx.telegram_queue.register_telegram_received_cb(
-            async_telegram_received_callback
+            telegram_received_cb
         )
         xknx.telegram_queue.unregister_telegram_received_cb(callback)
 
@@ -169,7 +169,7 @@ class TestTelegramQueue:
             payload=GroupValueWrite(DPTBinary(1)),
         )
         await xknx.telegram_queue.process_telegram_incoming(telegram)
-        async_telegram_received_callback.assert_not_called()
+        telegram_received_cb.assert_not_called()
 
     #
     # TEST PROCESS
@@ -196,11 +196,9 @@ class TestTelegramQueue:
         """Test process_telegram_incoming with callback."""
 
         xknx = XKNX()
-        async_telegram_received_callback = AsyncMock()
+        telegram_received_cb = Mock()
 
-        xknx.telegram_queue.register_telegram_received_cb(
-            async_telegram_received_callback
-        )
+        xknx.telegram_queue.register_telegram_received_cb(telegram_received_cb)
 
         telegram = Telegram(
             destination_address=GroupAddress("1/2/3"),
@@ -208,7 +206,7 @@ class TestTelegramQueue:
             payload=GroupValueWrite(DPTBinary(1)),
         )
         await xknx.telegram_queue.process_telegram_incoming(telegram)
-        async_telegram_received_callback.assert_called_once_with(telegram)
+        telegram_received_cb.assert_called_once_with(telegram)
         devices_process.assert_called_once_with(telegram)
 
     async def test_callback_decoded_telegram_data(self):
@@ -216,8 +214,8 @@ class TestTelegramQueue:
 
         xknx = XKNX()
         xknx.group_address_dpt.set({"1/2/3": {"main": 5, "sub": 1}})
-        async_telegram_received_cb = AsyncMock()
-        xknx.telegram_queue.register_telegram_received_cb(async_telegram_received_cb)
+        telegram_received_cb = Mock()
+        xknx.telegram_queue.register_telegram_received_cb(telegram_received_cb)
 
         telegram = Telegram(
             destination_address=GroupAddress("1/2/3"),
@@ -233,8 +231,8 @@ class TestTelegramQueue:
         await xknx.telegrams.join()
         await xknx.telegram_queue.stop()
 
-        assert async_telegram_received_cb.call_count == 1
-        received = async_telegram_received_cb.call_args_list[0][0][0]
+        assert telegram_received_cb.call_count == 1
+        received = telegram_received_cb.call_args_list[0][0][0]
         assert received == telegram
         assert received.decoded_data is not None
         assert received.decoded_data.value == 50
@@ -319,11 +317,9 @@ class TestTelegramQueue:
     async def test_callback_no_filters(self):
         """Test telegram_received_callback after state of switch was changed."""
         xknx = XKNX()
-        async_telegram_received_callback = AsyncMock()
+        telegram_received_cb = Mock()
 
-        xknx.telegram_queue.register_telegram_received_cb(
-            async_telegram_received_callback
-        )
+        xknx.telegram_queue.register_telegram_received_cb(telegram_received_cb)
 
         telegram = Telegram(
             destination_address=GroupAddress("1/2/3"),
@@ -333,7 +329,7 @@ class TestTelegramQueue:
         xknx.telegrams.put_nowait(telegram)
         await xknx.telegram_queue._process_all_telegrams()
 
-        async_telegram_received_callback.assert_called_with(telegram)
+        telegram_received_cb.assert_called_with(telegram)
 
     #
     # TEST POSITIVE FILTERS
@@ -341,10 +337,10 @@ class TestTelegramQueue:
     async def test_callback_positive_address_filters(self):
         """Test telegram_received_callback after state of switch was changed."""
         xknx = XKNX()
-        async_telegram_received_callback = AsyncMock()
+        telegram_received_cb = Mock()
 
         xknx.telegram_queue.register_telegram_received_cb(
-            async_telegram_received_callback,
+            telegram_received_cb,
             address_filters=[AddressFilter("2/4-8/*"), AddressFilter("1/2/-8")],
         )
 
@@ -356,7 +352,7 @@ class TestTelegramQueue:
         xknx.telegrams.put_nowait(telegram)
         await xknx.telegram_queue._process_all_telegrams()
 
-        async_telegram_received_callback.assert_called_with(telegram)
+        telegram_received_cb.assert_called_with(telegram)
 
     #
     # TEST NEGATIVE FILTERS
@@ -364,10 +360,10 @@ class TestTelegramQueue:
     async def test_callback_negative_address_filters(self):
         """Test telegram_received_callback after state of switch was changed."""
         xknx = XKNX()
-        async_telegram_received_callback = AsyncMock()
+        telegram_received_cb = Mock()
 
         xknx.telegram_queue.register_telegram_received_cb(
-            async_telegram_received_callback,
+            telegram_received_cb,
             address_filters=[AddressFilter("2/4-8/*"), AddressFilter("1/2/8-")],
         )
 
@@ -379,21 +375,21 @@ class TestTelegramQueue:
         xknx.telegrams.put_nowait(telegram)
         await xknx.telegram_queue._process_all_telegrams()
 
-        async_telegram_received_callback.assert_not_called()
+        telegram_received_cb.assert_not_called()
 
     async def test_callback_group_addresses(self):
         """Test telegram_received_callback after state of switch was changed."""
         xknx = XKNX()
-        async_telegram_received_cb_one = AsyncMock()
-        async_telegram_received_cb_two = AsyncMock()
+        telegram_received_cb_one = Mock()
+        telegram_received_cb_two = Mock()
 
         callback_one = xknx.telegram_queue.register_telegram_received_cb(
-            async_telegram_received_cb_one,
+            telegram_received_cb_one,
             address_filters=[],
             group_addresses=[GroupAddress("1/2/3")],
         )
         callback_two = xknx.telegram_queue.register_telegram_received_cb(
-            async_telegram_received_cb_two, address_filters=[], group_addresses=[]
+            telegram_received_cb_two, address_filters=[], group_addresses=[]
         )
 
         telegram = Telegram(
@@ -402,16 +398,16 @@ class TestTelegramQueue:
             payload=GroupValueWrite(DPTBinary(1)),
         )
         await xknx.telegram_queue.process_telegram_incoming(telegram)
-        async_telegram_received_cb_one.assert_called_once_with(telegram)
-        async_telegram_received_cb_two.assert_not_called()
+        telegram_received_cb_one.assert_called_once_with(telegram)
+        telegram_received_cb_two.assert_not_called()
 
-        async_telegram_received_cb_one.reset_mock()
+        telegram_received_cb_one.reset_mock()
         # modify the filters - add/remove a GroupAddress
         callback_one.group_addresses.remove(GroupAddress("1/2/3"))
         callback_two.group_addresses.append(GroupAddress("1/2/3"))
         await xknx.telegram_queue.process_telegram_incoming(telegram)
-        async_telegram_received_cb_one.assert_not_called()
-        async_telegram_received_cb_two.assert_called_once_with(telegram)
+        telegram_received_cb_one.assert_not_called()
+        telegram_received_cb_two.assert_called_once_with(telegram)
 
     #
     # TEST EXCEPTION HANDLING
@@ -454,9 +450,9 @@ class TestTelegramQueue:
     async def test_callback_raising(self, logging_exception_mock):
         """Test telegram_received_callback raising an exception."""
         xknx = XKNX()
-        good_callback_1 = AsyncMock()
-        bad_callback = AsyncMock(side_effect=Exception("Boom"))
-        good_callback_2 = AsyncMock()
+        good_callback_1 = Mock()
+        bad_callback = Mock(side_effect=Exception("Boom"))
+        good_callback_2 = Mock()
 
         xknx.telegram_queue.register_telegram_received_cb(good_callback_1)
         xknx.telegram_queue.register_telegram_received_cb(bad_callback)
diff --git a/test/core_tests/value_reader_test.py b/test/core_tests/value_reader_test.py
index 5525fc3f..4c095ac8 100644
--- a/test/core_tests/value_reader_test.py
+++ b/test/core_tests/value_reader_test.py
@@ -27,7 +27,7 @@ class TestValueReader:
 
         value_reader = ValueReader(xknx, test_group_address)
         # receive the response
-        await value_reader.telegram_received(response_telegram)
+        value_reader.telegram_received(response_telegram)
         # and yield the result
         successful_read = await value_reader.read()
 
@@ -122,18 +122,18 @@ class TestValueReader:
 
         value_reader = ValueReader(xknx, test_group_address)
 
-        await value_reader.telegram_received(telegram_wrong_address)
+        value_reader.telegram_received(telegram_wrong_address)
         assert value_reader.received_telegram is None
         assert not value_reader.response_received_event.is_set()
 
-        await value_reader.telegram_received(telegram_wrong_type)
+        value_reader.telegram_received(telegram_wrong_type)
         assert value_reader.received_telegram is None
         assert not value_reader.response_received_event.is_set()
 
-        await value_reader.telegram_received(expected_telegram_1)
+        value_reader.telegram_received(expected_telegram_1)
         assert value_reader.received_telegram == expected_telegram_1
         assert value_reader.response_received_event.is_set()
 
-        await value_reader.telegram_received(expected_telegram_2)
+        value_reader.telegram_received(expected_telegram_2)
         assert value_reader.received_telegram == expected_telegram_2
         assert value_reader.response_received_event.is_set()