Repository: XKNX/xknx
Pull request #768 (2021-10-03 09:47:16)

Small documentation change in cemi_frame
<!--
  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.
-->
This is a small change in the comments in cemi_frame

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.
-->

- [x] Bug fix (non-breaking change which fixes an issue)
- [ ] New feature (non-breaking change which adds functionality)
- [ ] Breaking change (fix or feature that would cause existing functionality to not work as expected)
- [ ] This change requires a documentation update

## Checklist:

- [x] I have performed a self-review of my own code
- [x] The documentation has been adjusted accordingly
- [x] The changes generate no new warnings
- [x] Tests have been added that prove the fix is effective or that the feature works
- [ ] The changes are documented in the changelog
- [ ] The Homeassistant plugin has been adjusted in case of new config options



Patch:
diff --git a/changelog.md b/changelog.md
index 6011bc92..412b0484 100644
--- a/changelog.md
+++ b/changelog.md
@@ -4,6 +4,8 @@
 
 ### Internals
 
+- Catch and log exceptions raised in callbacks to not stall the TelegramQueue
+- Handle callbacks in separate asyncio Tasks
 - GatewayScanFilter: Ignore non-gateway KNX/IP devices
 
 ## 0.18.9 HS-color 2021-07-26
diff --git a/xknx/core/telegram_queue.py b/xknx/core/telegram_queue.py
index bd4a4e2c..1e019149 100644
--- a/xknx/core/telegram_queue.py
+++ b/xknx/core/telegram_queue.py
@@ -126,6 +126,12 @@ class TelegramQueue:
                     await self.process_telegram_incoming(telegram)
                 except XKNXException as ex:
                     logger.error("Error while processing incoming telegram %s", ex)
+                except Exception:  # pylint: disable=broad-except
+                    # prevent the parser Task from stalling when unexpected errors occur
+                    logger.exception(
+                        "Unexpected error while processing incoming telegram %s",
+                        telegram,
+                    )
                 finally:
                     self.xknx.telegrams.task_done()
             elif telegram.direction == TelegramDirection.OUTGOING:
@@ -148,6 +154,11 @@ class TelegramQueue:
                     logger.warning(ex)
             except XKNXException as ex:
                 logger.error("Error while processing outgoing telegram %s", ex)
+            except Exception:  # pylint: disable=broad-except
+                # prevent the sender Task from stalling when unexpected errors occur (eg. ValueError from creating KNXIPFrames)
+                logger.exception(
+                    "Unexpected error while processing outgoing telegram %s", telegram
+                )
             finally:
                 self.outgoing_queue.task_done()
                 self.xknx.telegrams.task_done()
@@ -183,14 +194,25 @@ class TelegramQueue:
             await self.xknx.knxip_interface.send_telegram(telegram)
 
         await self.xknx.devices.process(telegram)
-        for telegram_received_cb in self.telegram_received_cbs:
-            if telegram_received_cb.is_within_filter(telegram):
-                await telegram_received_cb.callback(telegram)
+        await self._run_telegram_received_cbs(telegram)
 
     async def process_telegram_incoming(self, telegram: Telegram) -> None:
         """Process incoming telegram."""
         telegram_logger.debug(telegram)
-        for telegram_received_cb in self.telegram_received_cbs:
-            if telegram_received_cb.is_within_filter(telegram):
-                await telegram_received_cb.callback(telegram)
+        await self._run_telegram_received_cbs(telegram)
         await self.xknx.devices.process(telegram)
+
+    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,
+            )
diff --git a/xknx/devices/device.py b/xknx/devices/device.py
index d9e42155..b862e481 100644
--- a/xknx/devices/device.py
+++ b/xknx/devices/device.py
@@ -6,6 +6,7 @@ It provides basis functionality for reading the state from the KNX bus.
 from __future__ import annotations
 
 from abc import ABC, abstractmethod
+import asyncio
 import logging
 from typing import TYPE_CHECKING, Any, Awaitable, Callable, Iterator
 
@@ -73,8 +74,13 @@ class Device(ABC):
 
     async def after_update(self) -> None:
         """Execute callbacks after internal state has been changed."""
-        for device_updated_cb in self.device_updated_cbs:
-            await device_updated_cb(self)
+        try:
+            await asyncio.gather(*[cb(self) for cb in self.device_updated_cbs])
+        except Exception:  # pylint: disable=broad-except
+            logger.exception(
+                "Unexpected error while processing device_updated_cb for %s",
+                self,
+            )
 
     async def sync(self, wait_for_result: bool = False) -> None:
         """Read states of device from KNX bus."""
diff --git a/xknx/knxip/cemi_frame.py b/xknx/knxip/cemi_frame.py
index f32f2610..ac66875d 100644
--- a/xknx/knxip/cemi_frame.py
+++ b/xknx/knxip/cemi_frame.py
@@ -135,7 +135,8 @@ class CEMIFrame:
     def from_knx_data_link_layer(self, cemi: bytes) -> int:
         """Parse L_DATA_IND, CEMIMessageCode.L_DATA_REQ, CEMIMessageCode.L_DATA_CON."""
         if len(cemi) < 11:
-            # eg. ETS Line-Scan issues L_DATA_IND with length 10
+            # eg. ETS Line-Scan issues L_DATA_IND with length 10 or checking for individual addresses
+            # at start of ETS programming procedure
             raise UnsupportedCEMIMessage(
                 "CEMI too small. Length: {}; CEMI: {}".format(len(cemi), cemi.hex())
             )


Test patch:
diff --git a/test/core_tests/telegram_queue_test.py b/test/core_tests/telegram_queue_test.py
index 74481726..10016bb4 100644
--- a/test/core_tests/telegram_queue_test.py
+++ b/test/core_tests/telegram_queue_test.py
@@ -1,6 +1,6 @@
 """Unit test for telegram received callback."""
 import asyncio
-from unittest.mock import AsyncMock, MagicMock, patch
+from unittest.mock import AsyncMock, MagicMock, call, patch
 
 import pytest
 from xknx import XKNX
@@ -238,7 +238,7 @@ class TestTelegramQueue:
 
         async def process_exception():
             raise CouldNotParseTelegram(
-                "Something went wrong when receiving the telegram." ""
+                "Something went wrong when receiving the telegram."
             )
 
         process_tg_in_mock.return_value = asyncio.ensure_future(process_exception())
@@ -254,9 +254,7 @@ class TestTelegramQueue:
 
         logging_error_mock.assert_called_once_with(
             "Error while processing telegram %s",
-            CouldNotParseTelegram(
-                "Something went wrong when receiving the telegram." ""
-            ),
+            CouldNotParseTelegram("Something went wrong when receiving the telegram."),
         )
 
     @patch("xknx.core.TelegramQueue.process_telegram_outgoing", new_callable=AsyncMock)
@@ -385,3 +383,68 @@ class TestTelegramQueue:
         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)
+
+    #
+    # TEST EXCEPTION HANDLING
+    #
+    @patch("logging.Logger.exception")
+    @patch("xknx.xknx.Devices.process", side_effect=Exception)
+    async def test_process_raising(self, process_mock, logging_exception_mock):
+        """Test unexpected exception handling in telegram queues."""
+        xknx = XKNX()
+        telegram_in = Telegram(
+            destination_address=GroupAddress("1/2/3"),
+            direction=TelegramDirection.INCOMING,
+            payload=GroupValueWrite(DPTBinary(1)),
+        )
+        # InternalGroupAddress to avoid CommunicationError for missing knxip_interface
+        telegram_out = Telegram(
+            destination_address=InternalGroupAddress("i-outgoing"),
+            direction=TelegramDirection.OUTGOING,
+            payload=GroupValueWrite(DPTBinary(0)),
+        )
+        xknx.telegrams.put_nowait(telegram_in)
+        xknx.telegrams.put_nowait(telegram_out)
+
+        await xknx.telegram_queue.start()
+        await xknx.telegram_queue.stop()
+
+        log_calls = [
+            call(
+                "Unexpected error while processing incoming telegram %s",
+                telegram_in,
+            ),
+            call(
+                "Unexpected error while processing outgoing telegram %s",
+                telegram_out,
+            ),
+        ]
+        logging_exception_mock.assert_has_calls(log_calls)
+
+    @patch("logging.Logger.exception")
+    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()
+
+        xknx.telegram_queue.register_telegram_received_cb(good_callback_1)
+        xknx.telegram_queue.register_telegram_received_cb(bad_callback)
+        xknx.telegram_queue.register_telegram_received_cb(good_callback_2)
+
+        telegram = Telegram(
+            destination_address=GroupAddress("1/2/3"),
+            direction=TelegramDirection.INCOMING,
+            payload=GroupValueWrite(DPTBinary(1)),
+        )
+        await xknx.telegram_queue.process_telegram_incoming(telegram)
+
+        good_callback_1.assert_called_once_with(telegram)
+        bad_callback.assert_called_once_with(telegram)
+        good_callback_2.assert_called_once_with(telegram)
+
+        logging_exception_mock.assert_called_once_with(
+            "Unexpected error while processing telegram_received_cb for %s",
+            telegram,
+        )
diff --git a/test/devices_tests/device_test.py b/test/devices_tests/device_test.py
index eb20f4e7..d853fb8a 100644
--- a/test/devices_tests/device_test.py
+++ b/test/devices_tests/device_test.py
@@ -60,6 +60,29 @@ class TestDevice:
         after_update_callback1.assert_not_called()
         after_update_callback2.assert_not_called()
 
+    @patch("logging.Logger.exception")
+    async def test_bad_callback(self, logging_exception_mock):
+        """Test handling callback raising an exception"""
+        xknx = XKNX()
+        device = Device(xknx, "TestDevice")
+        good_callback_1 = AsyncMock()
+        bad_callback = AsyncMock(side_effect=Exception("Boom"))
+        good_callback_2 = AsyncMock()
+
+        device.register_device_updated_cb(good_callback_1)
+        device.register_device_updated_cb(bad_callback)
+        device.register_device_updated_cb(good_callback_2)
+
+        await device.after_update()
+        good_callback_1.assert_called_with(device)
+        bad_callback.assert_called_with(device)
+        good_callback_2.assert_called_with(device)
+
+        logging_exception_mock.assert_called_once_with(
+            "Unexpected error while processing device_updated_cb for %s",
+            device,
+        )
+
     async def test_process(self):
         """Test if telegram is handled by the correct process_* method."""
         xknx = XKNX()