Repository: XKNX/xknx
Pull request #523 (2020-12-07 22:46:01)

Refactor timeout handling
<!--
  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.
-->
- Refactor timeout handling of GatewayScanner, RequestResponse and ValueReader to
  - handle Task cancellation without leaving ghost callbacks
  - be shorter and better readable with fewer methods
- add type-checking for GatewayScanner and ValueReader 
  - (RequestResponse seems to be a bigger task - thats for another weekend)

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)
- [ ] 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
- [ ] 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
- [x] The changes are documented in the changelog
- [ ] The Homeassistant plugin has been adjusted in case of new config options


coveralls: 
[![Coverage Status](https://coveralls.io/builds/35548036/badge)](https://coveralls.io/builds/35548036)

Coverage decreased (-0.06%) to 94.518% when pulling **69435c9825299a06c586b3137adfb881f3578cc7 on farmio:timeout-refactor** into **6c3f2d43488e786d6e9ae04a0226b5b3a16445be on XKNX:master**.


Patch:
diff --git a/.coveragerc b/.coveragerc
index 8b652710..519778a6 100644
--- a/.coveragerc
+++ b/.coveragerc
@@ -7,3 +7,10 @@ omit =
 
 [report]
 precision = 2
+# Regexes for lines to exclude from consideration
+exclude_lines =
+    # Don't complain about type checking imports:
+    if TYPE_CHECKING:
+
+    # Don't complain if tests don't hit defensive assertion code:
+    raise NotImplementedError
diff --git a/changelog.md b/changelog.md
index 9474a656..dd67d200 100644
--- a/changelog.md
+++ b/changelog.md
@@ -8,6 +8,7 @@
 
 ### Internals
 
+- refactored timeout handling in GatewayScanner, RequestResponse and ValueReader.
 - renamed "PhysicalAddress" to "IndividualAddress"
 - Telegram: `group_address` renamed to `destination_address`, to prepare support for other APCI services and add `source_address`
 - Telegram: remove `Telegram.telegramtype` and replace with payload object derived from `xknx.telegram.apci.APCI`.
diff --git a/setup.cfg b/setup.cfg
index d53268da..dfed14ef 100644
--- a/setup.cfg
+++ b/setup.cfg
@@ -42,7 +42,7 @@ warn_redundant_casts = true
 warn_unused_configs = true
 
 # add the modules below once we add typing for them so that we fail the build in the future if someone changes something without updating the typings
-[mypy-xknx.telegram.*,xknx.exceptions.*,xknx.devices.travelcalculator]
+[mypy-xknx.core.value_reader,xknx.devices.travelcalculator,xknx.exceptions.*,xknx.io.gateway_scanner,xknx.telegram.*,]
 strict = true
 ignore_errors = false
 warn_unreachable = true
diff --git a/xknx/core/value_reader.py b/xknx/core/value_reader.py
index 928d05d9..b8a61729 100644
--- a/xknx/core/value_reader.py
+++ b/xknx/core/value_reader.py
@@ -9,52 +9,65 @@ The module will
 """
 import asyncio
 import logging
+from typing import TYPE_CHECKING, Optional
 
-from xknx.telegram import Telegram
+from xknx.telegram import GroupAddress, Telegram
 from xknx.telegram.apci import GroupValueRead, GroupValueResponse, GroupValueWrite
 
+if TYPE_CHECKING:
+    from xknx.xknx import XKNX
+
 logger = logging.getLogger("xknx.log")
 
 
 class ValueReader:
     """Class for reading the value of a specific KNX group address from KNX bus."""
 
-    # pylint: disable=too-many-instance-attributes
-
-    def __init__(self, xknx, group_address, timeout_in_seconds=2):
+    def __init__(
+        self, xknx: "XKNX", group_address: GroupAddress, timeout_in_seconds: float = 2.0
+    ):
         """Initialize ValueReader class."""
         self.xknx = xknx
-        self.group_address = group_address
+        self.group_address: GroupAddress = group_address
         self.response_received_or_timeout = asyncio.Event()
-        self.success = False
-        self.timeout_in_seconds = timeout_in_seconds
-        self.timeout_handle = None
-        self.received_telegram = None
+        self.success: bool = False
+        self.timeout_in_seconds: float = timeout_in_seconds
+        self.received_telegram: Optional[Telegram] = None
 
-    async def read(self):
+    async def read(self) -> Optional[Telegram]:
         """Send group read and wait for response."""
         cb_obj = self.xknx.telegram_queue.register_telegram_received_cb(
             self.telegram_received
         )
-
         await self.send_group_read()
-        await self.start_timeout()
-        await self.response_received_or_timeout.wait()
-        await self.stop_timeout()
 
-        self.xknx.telegram_queue.unregister_telegram_received_cb(cb_obj)
+        try:
+            await asyncio.wait_for(
+                self.response_received_or_timeout.wait(),
+                timeout=self.timeout_in_seconds,
+            )
+        except asyncio.TimeoutError:
+            logger.warning(
+                "Error: KNX bus did not respond in time (%s secs) to GroupValueRead request for: %s",
+                self.timeout_in_seconds,
+                self.group_address,
+            )
+        finally:
+            # cleanup to not leave callbacks (for asyncio.CancelledError)
+            self.xknx.telegram_queue.unregister_telegram_received_cb(cb_obj)
+
         if not self.success:
             return None
         return self.received_telegram
 
-    async def send_group_read(self):
+    async def send_group_read(self) -> None:
         """Send group read."""
         telegram = Telegram(
             destination_address=self.group_address, payload=GroupValueRead()
         )
         await self.xknx.telegrams.put(telegram)
 
-    async def telegram_received(self, telegram):
+    async 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)
@@ -62,21 +75,3 @@ class ValueReader:
             self.success = True
             self.received_telegram = telegram
             self.response_received_or_timeout.set()
-
-    def timeout(self):
-        """Handle timeout for not having received expected group response."""
-        logger.warning(
-            "Error: KNX bus did not respond in time (%s secs) to GroupValueRead request for: %s",
-            self.timeout_in_seconds,
-            self.group_address,
-        )
-        self.response_received_or_timeout.set()
-
-    async def start_timeout(self):
-        """Start timeout. Register callback for no answer received within timeout."""
-        loop = asyncio.get_running_loop()
-        self.timeout_handle = loop.call_later(self.timeout_in_seconds, self.timeout)
-
-    async def stop_timeout(self):
-        """Stop timeout."""
-        self.timeout_handle.cancel()
diff --git a/xknx/io/gateway_scanner.py b/xknx/io/gateway_scanner.py
index 5b15bf5b..77470c54 100644
--- a/xknx/io/gateway_scanner.py
+++ b/xknx/io/gateway_scanner.py
@@ -7,8 +7,9 @@ GatewayScanner is an abstraction for searching for KNX/IP devices on the local n
 """
 
 import asyncio
+from functools import partial
 import logging
-from typing import List, Optional
+from typing import TYPE_CHECKING, List, Optional
 
 import netifaces
 from xknx.knxip import (
@@ -23,6 +24,9 @@ from xknx.knxip import (
 
 from .udp_client import UDPClient
 
+if TYPE_CHECKING:
+    from xknx.xknx import XKNX
+
 logger = logging.getLogger("xknx.log")
 
 
@@ -40,7 +44,7 @@ class GatewayDescriptor:
         local_ip: str,
         supports_tunnelling: bool = False,
         supports_routing: bool = False,
-    ) -> None:
+    ):
         """Initialize GatewayDescriptor class."""
         # pylint: disable=too-many-arguments
         self.name = name
@@ -51,7 +55,7 @@ class GatewayDescriptor:
         self.supports_routing = supports_routing
         self.supports_tunnelling = supports_tunnelling
 
-    def __str__(self):
+    def __str__(self) -> str:
         """Return object as readable string."""
         return '<GatewayDescriptor name="{}" addr="{}:{}" local="{}@{}" routing="{}" tunnelling="{}" />'.format(
             self.name,
@@ -70,8 +74,11 @@ class GatewayScanFilter:
     # pylint: disable=too-few-public-methods
 
     def __init__(
-        self, name: str = None, tunnelling: bool = None, routing: bool = None
-    ) -> None:
+        self,
+        name: Optional[str] = None,
+        tunnelling: Optional[bool] = None,
+        routing: Optional[bool] = None,
+    ):
         """Initialize GatewayScanFilter class."""
         self.name = name
         self.tunnelling = tunnelling
@@ -98,20 +105,19 @@ class GatewayScanner:
     # pylint: disable=too-many-instance-attributes
     def __init__(
         self,
-        xknx,
-        timeout_in_seconds: int = 4,
+        xknx: "XKNX",
+        timeout_in_seconds: float = 4.0,
         stop_on_found: Optional[int] = 1,
         scan_filter: GatewayScanFilter = GatewayScanFilter(),
-    ) -> None:
+    ):
         """Initialize GatewayScanner class."""
         self.xknx = xknx
         self.timeout_in_seconds = timeout_in_seconds
         self.stop_on_found = stop_on_found
         self.scan_filter = scan_filter
-        self.found_gateways = []  # type: List[GatewayDescriptor]
-        self._udp_clients = []
+        self.found_gateways: List[GatewayDescriptor] = []
+        self._udp_clients: List[UDPClient] = []
         self._response_received_or_timeout = asyncio.Event()
-        self._timeout_handle = None
         self._count_upper_bound = 0
         """Clean value of self.stop_on_found, computed when ``scan`` is called."""
 
@@ -122,18 +128,23 @@ class GatewayScanner:
         else:
             self._count_upper_bound = max(0, self.stop_on_found)
         await self._send_search_requests()
-        await self._start_timeout()
-        await self._response_received_or_timeout.wait()
-        await self._stop()
-        await self._stop_timeout()
+        try:
+            await asyncio.wait_for(
+                self._response_received_or_timeout.wait(),
+                timeout=self.timeout_in_seconds,
+            )
+        except asyncio.TimeoutError:
+            pass
+        finally:
+            await self._stop()
         return self.found_gateways
 
-    async def _stop(self):
+    async def _stop(self) -> None:
         """Stop tearing down udpclient."""
         for udp_client in self._udp_clients:
             await udp_client.stop()
 
-    async def _send_search_requests(self):
+    async def _send_search_requests(self) -> None:
         """Find all interfaces with active IPv4 connection to search for gateways."""
         # pylint: disable=no-member
         for interface in netifaces.interfaces():
@@ -145,19 +156,20 @@ class GatewayScanner:
                 logger.info("Could not connect to an KNX/IP device on %s", interface)
                 continue
 
-    async def _search_interface(self, interface, ip_addr):
+    async def _search_interface(self, interface: str, ip_addr: str) -> None:
         """Send a search request on a specific interface."""
         logger.debug("Searching on %s / %s", interface, ip_addr)
 
         udp_client = UDPClient(
             self.xknx,
-            (ip_addr, 0, interface),
+            (ip_addr, 0),
             (self.xknx.multicast_group, self.xknx.multicast_port),
             multicast=True,
         )
 
         udp_client.register_callback(
-            self._response_rec_callback, [KNXIPServiceType.SEARCH_RESPONSE]
+            partial(self._response_rec_callback, interface=interface),
+            [KNXIPServiceType.SEARCH_RESPONSE],
         )
         await udp_client.connect()
 
@@ -170,7 +182,7 @@ class GatewayScanner:
         udp_client.send(KNXIPFrame.init_from_body(search_request))
 
     def _response_rec_callback(
-        self, knx_ip_frame: KNXIPFrame, udp_client: UDPClient
+        self, knx_ip_frame: KNXIPFrame, udp_client: UDPClient, interface: str = ""
     ) -> None:
         """Verify and handle knxipframe. Callback from internal udpclient."""
         if not isinstance(knx_ip_frame.body, SearchResponse):
@@ -181,7 +193,7 @@ class GatewayScanner:
             name=knx_ip_frame.body.device_name,
             ip_addr=knx_ip_frame.body.control_endpoint.ip_addr,
             port=knx_ip_frame.body.control_endpoint.port,
-            local_interface=udp_client.local_addr[2],
+            local_interface=interface,
             local_ip=udp_client.local_addr[0],
         )
         try:
@@ -193,21 +205,8 @@ class GatewayScanner:
 
         self._add_found_gateway(gateway)
 
-    def _add_found_gateway(self, gateway):
+    def _add_found_gateway(self, gateway: GatewayDescriptor) -> None:
         if self.scan_filter.match(gateway):
             self.found_gateways.append(gateway)
             if 0 < self._count_upper_bound <= len(self.found_gateways):
                 self._response_received_or_timeout.set()
-
-    def _timeout(self):
-        """Handle timeout for not having received enough SearchResponse."""
-        self._response_received_or_timeout.set()
-
-    async def _start_timeout(self):
-        """Start time out."""
-        loop = asyncio.get_running_loop()
-        self._timeout_handle = loop.call_later(self.timeout_in_seconds, self._timeout)
-
-    async def _stop_timeout(self):
-        """Stop/cancel timeout."""
-        self._timeout_handle.cancel()
diff --git a/xknx/io/request_response.py b/xknx/io/request_response.py
index 805753fd..eed530d7 100644
--- a/xknx/io/request_response.py
+++ b/xknx/io/request_response.py
@@ -6,7 +6,7 @@ Will report if the corresponding answer was not received.
 import asyncio
 import logging
 
-from xknx.knxip import ErrorCode, KNXIPFrame
+from xknx.knxip import ErrorCode, KNXIPBody, KNXIPFrame
 
 logger = logging.getLogger("xknx.log")
 
@@ -14,38 +14,52 @@ logger = logging.getLogger("xknx.log")
 class RequestResponse:
     """Class for sending a specific type of KNX/IP Packet to a KNX/IP device and wait for the corresponding answer."""
 
-    # pylint: disable=too-many-instance-attributes
-
-    def __init__(self, xknx, udp_client, awaited_response_class, timeout_in_seconds=1):
+    def __init__(
+        self,
+        xknx,
+        udp_client,
+        awaited_response_class: KNXIPBody,
+        timeout_in_seconds: float = 1.0,
+    ):
         """Initialize RequstResponse class."""
         self.xknx = xknx
         self.udpclient = udp_client
-        self.awaited_response_class = awaited_response_class
+        self.awaited_response_class: KNXIPBody = awaited_response_class
         self.response_received_or_timeout = asyncio.Event()
         self.success = False
         self.timeout_in_seconds = timeout_in_seconds
-        self.timeout_handle = None
 
     def create_knxipframe(self) -> KNXIPFrame:
         """Create KNX/IP Frame object to be sent to device."""
         raise NotImplementedError("create_knxipframe has to be implemented")
 
-    async def start(self):
+    async def start(self) -> None:
         """Start. Send request and wait for an answer."""
         callb = self.udpclient.register_callback(
             self.response_rec_callback, [self.awaited_response_class.service_type]
         )
         await self.send_request()
-        await self.start_timeout()
-        await self.response_received_or_timeout.wait()
-        await self.stop_timeout()
-        self.udpclient.unregister_callback(callb)
 
-    async def send_request(self):
+        try:
+            await asyncio.wait_for(
+                self.response_received_or_timeout.wait(),
+                timeout=self.timeout_in_seconds,
+            )
+        except asyncio.TimeoutError:
+            logger.debug(
+                "Error: KNX bus did not respond in time (%s secs) to request of type '%s'",
+                self.timeout_in_seconds,
+                self.__class__.__name__,
+            )
+        finally:
+            # cleanup to not leave callbacks (for asyncio.CancelledError)
+            self.udpclient.unregister_callback(callb)
+
+    async def send_request(self) -> None:
         """Build knxipframe (within derived class) and send via UDP."""
         self.udpclient.send(self.create_knxipframe())
 
-    def response_rec_callback(self, knxipframe, _):
+    def response_rec_callback(self, knxipframe: KNXIPFrame, _) -> None:
         """Verify and handle knxipframe. Callback from internal udpclient."""
         if not isinstance(knxipframe.body, self.awaited_response_class):
             logger.warning("Could not understand knxipframe")
@@ -57,10 +71,10 @@ class RequestResponse:
         else:
             self.on_error_hook(knxipframe)
 
-    def on_success_hook(self, knxipframe):
+    def on_success_hook(self, knxipframe: KNXIPFrame) -> None:
         """Do something after having received a valid answer. May be overwritten in derived class."""
 
-    def on_error_hook(self, knxipframe):
+    def on_error_hook(self, knxipframe: KNXIPFrame) -> None:
         """Do something after having received error within given time. May be overwritten in derived class."""
         logger.debug(
             "Error: KNX bus responded to request of type '%s' with error in '%s': %s",
@@ -68,21 +82,3 @@ class RequestResponse:
             self.awaited_response_class.__name__,
             knxipframe.body.status_code,
         )
-
-    def timeout(self):
-        """Handle timeout for not having received expected knxipframe."""
-        logger.debug(
-            "Error: KNX bus did not respond in time (%s secs) to request of type '%s'",
-            self.timeout_in_seconds,
-            self.__class__.__name__,
-        )
-        self.response_received_or_timeout.set()
-
-    async def start_timeout(self):
-        """Start timeout."""
-        loop = asyncio.get_running_loop()
-        self.timeout_handle = loop.call_later(self.timeout_in_seconds, self.timeout)
-
-    async def stop_timeout(self):
-        """Stop timeout."""
-        self.timeout_handle.cancel()
diff --git a/xknx/io/udp_client.py b/xknx/io/udp_client.py
index 1bed0091..d9ef021e 100644
--- a/xknx/io/udp_client.py
+++ b/xknx/io/udp_client.py
@@ -8,10 +8,14 @@ import asyncio
 import logging
 import socket
 from sys import platform
+from typing import TYPE_CHECKING, Optional, Tuple
 
 from xknx.exceptions import CouldNotParseKNXIP, XKNXException
 from xknx.knxip import KNXIPFrame
 
+if TYPE_CHECKING:
+    from xknx.xknx import XKNX
+
 raw_socket_logger = logging.getLogger("xknx.raw_socket")
 logger = logging.getLogger("xknx.log")
 knx_logger = logging.getLogger("xknx.knx")
@@ -64,7 +68,13 @@ class UDPClient:
             if hasattr(self, "xknx") and exc is not None:
                 logger.info("Closing transport.")
 
-    def __init__(self, xknx, local_addr, remote_addr, multicast=False):
+    def __init__(
+        self,
+        xknx: "XKNX",
+        local_addr: Tuple[str, int],
+        remote_addr: Tuple[str, int],
+        multicast: bool = False,
+    ):
         """Initialize UDPClient class."""
         # pylint: disable=too-many-arguments
         if not isinstance(local_addr, tuple):
@@ -99,7 +109,7 @@ class UDPClient:
         if not handled:
             logger.debug("UNHANDLED: %s", knxipframe.header.service_type_ident)
 
-    def register_callback(self, callback, service_types=None):
+    def register_callback(self, callback, service_types=None) -> Callback:
         """Register callback."""
         if service_types is None:
             service_types = []
@@ -108,7 +118,7 @@ class UDPClient:
         self.callbacks.append(callb)
         return callb
 
-    def unregister_callback(self, callb):
+    def unregister_callback(self, callb) -> None:
         """Unregister callback."""
         self.callbacks.remove(callb)
 
@@ -147,7 +157,7 @@ class UDPClient:
 
         return sock
 
-    async def connect(self):
+    async def connect(self) -> None:
         """Connect UDP socket. Open UDP port and build mulitcast socket if necessary."""
         udp_client_factory = UDPClient.UDPClientFactory(
             self.local_addr[0],
@@ -170,7 +180,7 @@ class UDPClient:
             )
             self.transport = transport
 
-    def send(self, knxipframe):
+    def send(self, knxipframe) -> None:
         """Send KNXIPFrame to socket."""
         knx_logger.debug("Sending: %s", knxipframe)
         if self.transport is None:
@@ -181,16 +191,16 @@ class UDPClient:
         else:
             self.transport.sendto(bytes(knxipframe.to_knx()))
 
-    def getsockname(self):
+    def getsockname(self) -> Tuple[str, int]:
         """Return sockname."""
         sock = self.transport.get_extra_info("sockname")
         return sock
 
-    def getremote(self):
+    def getremote(self) -> Optional[str]:
         """Return peername."""
         peer = self.transport.get_extra_info("peername")
         return peer
 
-    async def stop(self):
+    async def stop(self) -> None:
         """Stop UDP socket."""
         self.transport.close()
diff --git a/xknx/knxip/__init__.py b/xknx/knxip/__init__.py
index 0b1d000b..009b0a49 100644
--- a/xknx/knxip/__init__.py
+++ b/xknx/knxip/__init__.py
@@ -27,3 +27,34 @@ from .search_request import SearchRequest
 from .search_response import SearchResponse
 from .tunnelling_ack import TunnellingAck
 from .tunnelling_request import TunnellingRequest
+
+__all__ = [
+    "KNXIPBody",
+    "CEMIFrame",
+    "ConnectRequest",
+    "ConnectResponse",
+    "ConnectionStateRequest",
+    "ConnectionStateResponse",
+    "DIB",
+    "DIBDeviceInformation",
+    "DIBGeneric",
+    "DIBSuppSVCFamilies",
+    "DisconnectRequest",
+    "DisconnectResponse",
+    "ErrorCode",
+    "KNXIPHeader",
+    "HPAI",
+    "KNXIPFrame",
+    "CEMIFlags",
+    "CEMIMessageCode",
+    "ConnectRequestType",
+    "DIBServiceFamily",
+    "DIBTypeCode",
+    "KNXIPServiceType",
+    "KNXMedium",
+    "RoutingIndication",
+    "SearchRequest",
+    "SearchResponse",
+    "TunnellingAck",
+    "TunnellingRequest",
+]
diff --git a/xknx/knxip/body.py b/xknx/knxip/body.py
index 1c771f23..aa88bb0b 100644
--- a/xknx/knxip/body.py
+++ b/xknx/knxip/body.py
@@ -1,12 +1,16 @@
 """Basis class for all KNX/IP bodies."""
 import logging
 
+from .knxip_enum import KNXIPServiceType
+
 logger = logging.getLogger("xknx.log")
 
 
 class KNXIPBody:
     """Basis class for all KNX/IP bodies."""
 
+    service_type: KNXIPServiceType = None
+
     def __init__(self, xknx):
         """Initialize KNXIPBody object."""
         self.xknx = xknx
diff --git a/xknx/knxip/hpai.py b/xknx/knxip/hpai.py
index a8740a9a..8d88ed30 100644
--- a/xknx/knxip/hpai.py
+++ b/xknx/knxip/hpai.py
@@ -12,7 +12,7 @@ class HPAI:
     LENGTH = 0x08
     TYPE_UDP = 0x01
 
-    def __init__(self, ip_addr="0.0.0.0", port=0):
+    def __init__(self, ip_addr: str = "0.0.0.0", port: int = 0):
         """Initialize HPAI object."""
         self.ip_addr = ip_addr
         self.port = port


Test patch:
diff --git a/test/core_tests/value_reader_test.py b/test/core_tests/value_reader_test.py
index 5f5ee797..5f78c44f 100644
--- a/test/core_tests/value_reader_test.py
+++ b/test/core_tests/value_reader_test.py
@@ -1,7 +1,7 @@
 """Unit test for value reader."""
 import asyncio
 import unittest
-from unittest.mock import patch
+from unittest.mock import MagicMock, patch
 
 from xknx import XKNX
 from xknx.core import ValueReader
@@ -22,8 +22,7 @@ class TestValueReader(unittest.TestCase):
         """Tear down test class."""
         self.loop.close()
 
-    @patch("xknx.core.ValueReader.timeout")
-    def test_value_reader_read_success(self, timeout_mock):
+    def test_value_reader_read_success(self):
         """Test value reader: successfull read."""
         xknx = XKNX()
         test_group_address = GroupAddress("0/0/0")
@@ -34,23 +33,15 @@ class TestValueReader(unittest.TestCase):
         )
 
         value_reader = ValueReader(xknx, test_group_address)
-        # Create a task for read() (3.5 compatible)
-        read_task = asyncio.ensure_future(value_reader.read())
         # receive the response
         self.loop.run_until_complete(value_reader.telegram_received(response_telegram))
         # and yield the result
-        successfull_read = self.loop.run_until_complete(asyncio.gather(read_task))[0]
+        successfull_read = self.loop.run_until_complete(value_reader.read())
 
         # GroupValueRead telegram is still in the queue because we are not actually processing it
         self.assertEqual(xknx.telegrams.qsize(), 1)
         # Callback was removed again
         self.assertEqual(xknx.telegram_queue.telegram_received_cbs, [])
-        # Timeout handle was cancelled (cancelled method requires Python 3.7)
-        event_has_cancelled = getattr(value_reader.timeout_handle, "cancelled", None)
-        if callable(event_has_cancelled):
-            self.assertTrue(value_reader.timeout_handle.cancelled())
-        # timeout() was never called because there was no timeout
-        timeout_mock.assert_not_called()
         # Telegram was received
         self.assertEqual(value_reader.received_telegram, response_telegram)
         # Successfull read() returns the telegram
@@ -60,7 +51,10 @@ class TestValueReader(unittest.TestCase):
     def test_value_reader_read_timeout(self, logger_warning_mock):
         """Test value reader: read timeout."""
         xknx = XKNX()
-        value_reader = ValueReader(xknx, GroupAddress("0/0/0"), timeout_in_seconds=0)
+        value_reader = ValueReader(xknx, GroupAddress("0/0/0"))
+        value_reader.response_received_or_timeout.wait = MagicMock(
+            side_effect=asyncio.TimeoutError()
+        )
 
         timed_out_read = self.loop.run_until_complete(value_reader.read())
 
@@ -69,20 +63,33 @@ class TestValueReader(unittest.TestCase):
         # Warning was logged
         logger_warning_mock.assert_called_once_with(
             "Error: KNX bus did not respond in time (%s secs) to GroupValueRead request for: %s",
-            0,
+            2.0,
             GroupAddress("0/0/0"),
         )
         # Callback was removed again
         self.assertEqual(xknx.telegram_queue.telegram_received_cbs, [])
-        # Timeout handle was cancelled (cancelled method requires Python 3.7)
-        event_has_cancelled = getattr(value_reader.timeout_handle, "cancelled", None)
-        if callable(event_has_cancelled):
-            self.assertTrue(value_reader.timeout_handle.cancelled())
         # No telegram was received
         self.assertIsNone(value_reader.received_telegram)
         # Unsuccessfull read() returns None
         self.assertIsNone(timed_out_read)
 
+    def test_value_reader_read_cancelled(self):
+        """Test value reader: read cancelled."""
+        xknx = XKNX()
+        value_reader = ValueReader(xknx, GroupAddress("0/0/0"))
+        value_reader.response_received_or_timeout.wait = MagicMock(
+            side_effect=asyncio.CancelledError()
+        )
+        with self.assertRaises(asyncio.CancelledError):
+            self.loop.run_until_complete(value_reader.read())
+
+        # GroupValueRead telegram is still in the queue because we are not actually processing it
+        self.assertEqual(xknx.telegrams.qsize(), 1)
+        # Callback was removed again
+        self.assertEqual(xknx.telegram_queue.telegram_received_cbs, [])
+        # No telegram was received
+        self.assertIsNone(value_reader.received_telegram)
+
     def test_value_reader_send_group_read(self):
         """Test value reader: send_group_read."""
         xknx = XKNX()
diff --git a/test/io_tests/gateway_scanner_test.py b/test/io_tests/gateway_scanner_test.py
index bc8813fe..853f3dc2 100644
--- a/test/io_tests/gateway_scanner_test.py
+++ b/test/io_tests/gateway_scanner_test.py
@@ -1,7 +1,7 @@
 """Unit test for KNX/IP gateway scanner."""
 import asyncio
 import unittest
-from unittest.mock import patch
+from unittest.mock import MagicMock, patch
 
 from xknx import XKNX
 from xknx.io import GatewayScanFilter, GatewayScanner, UDPClient
@@ -128,11 +128,13 @@ class TestGatewayScanner(unittest.TestCase):
         gateway_scanner = GatewayScanner(xknx)
         test_search_response = fake_router_search_response(xknx)
         udp_client_mock = unittest.mock.create_autospec(UDPClient)
-        udp_client_mock.local_addr = ("192.168.42.50", 0, "en1")
+        udp_client_mock.local_addr = ("192.168.42.50", 0)
         udp_client_mock.getsockname.return_value = ("192.168.42.50", 0)
 
         self.assertEqual(gateway_scanner.found_gateways, [])
-        gateway_scanner._response_rec_callback(test_search_response, udp_client_mock)
+        gateway_scanner._response_rec_callback(
+            test_search_response, udp_client_mock, interface="en1"
+        )
         self.assertEqual(
             str(gateway_scanner.found_gateways[0]), str(self.gateway_desc_both)
         )
@@ -145,17 +147,11 @@ class TestGatewayScanner(unittest.TestCase):
         # No interface shall be found
         netifaces_mock.interfaces.return_value = []
 
-        gateway_scanner = GatewayScanner(xknx, timeout_in_seconds=0)
-
-        timed_out_scan = self.loop.run_until_complete(gateway_scanner.scan())
-
-        # Timeout handle was cancelled (cancelled method requires Python 3.7)
-        event_has_cancelled = getattr(
-            gateway_scanner._timeout_handle, "cancelled", None
+        gateway_scanner = GatewayScanner(xknx)
+        gateway_scanner._response_received_or_timeout.wait = MagicMock(
+            side_effect=asyncio.TimeoutError()
         )
-        if callable(event_has_cancelled):
-            self.assertTrue(gateway_scanner._timeout_handle.cancelled())
-        self.assertTrue(gateway_scanner._response_received_or_timeout.is_set())
+        timed_out_scan = self.loop.run_until_complete(gateway_scanner.scan())
         # Unsuccessfull scan() returns None
         self.assertEqual(timed_out_scan, [])
 
diff --git a/test/io_tests/request_response_test.py b/test/io_tests/request_response_test.py
index bd43ebf1..f172dbd9 100644
--- a/test/io_tests/request_response_test.py
+++ b/test/io_tests/request_response_test.py
@@ -1,10 +1,19 @@
 """Unit test for KNX/IP Disconnect Request/Response."""
 import asyncio
 import unittest
+from unittest.mock import MagicMock, patch
 
 from xknx import XKNX
 from xknx.io import RequestResponse, UDPClient
-from xknx.knxip import DisconnectResponse
+from xknx.knxip import DisconnectResponse, KNXIPBody
+
+
+class AsyncMock(MagicMock):
+    """Async Mock."""
+
+    # pylint: disable=invalid-overridden-method
+    async def __call__(self, *args, **kwargs):
+        return super().__call__(*args, **kwargs)
 
 
 class TestConnectResponse(unittest.TestCase):
@@ -28,3 +37,37 @@ class TestConnectResponse(unittest.TestCase):
 
         with self.assertRaises(NotImplementedError):
             self.loop.run_until_complete(request_response.start())
+
+    @patch("logging.Logger.debug")
+    @patch("xknx.io.RequestResponse.send_request", new_callable=AsyncMock)
+    def test_request_response_timeout(self, _send_request_mock, logger_debug_mock):
+        """Test RequestResponse: timeout. No callback shall be left."""
+        xknx = XKNX()
+        udp_client = UDPClient(xknx, ("192.168.1.1", 0), ("192.168.1.2", 1234))
+        requ_resp = RequestResponse(xknx, udp_client, KNXIPBody)
+        requ_resp.response_received_or_timeout.wait = MagicMock(
+            side_effect=asyncio.TimeoutError()
+        )
+        self.loop.run_until_complete(requ_resp.start())
+        # Debug message was logged
+        logger_debug_mock.assert_called_once_with(
+            "Error: KNX bus did not respond in time (%s secs) to request of type '%s'",
+            1.0,
+            "RequestResponse",
+        )
+        # Callback was removed again
+        self.assertEqual(udp_client.callbacks, [])
+
+    @patch("xknx.io.RequestResponse.send_request", new_callable=AsyncMock)
+    def test_request_response_cancelled(self, _send_request_mock):
+        """Test RequestResponse: task cancelled. No callback shall be left."""
+        xknx = XKNX()
+        udp_client = UDPClient(xknx, ("192.168.1.1", 0), ("192.168.1.2", 1234))
+        requ_resp = RequestResponse(xknx, udp_client, KNXIPBody)
+        requ_resp.response_received_or_timeout.wait = MagicMock(
+            side_effect=asyncio.CancelledError()
+        )
+        with self.assertRaises(asyncio.CancelledError):
+            self.loop.run_until_complete(requ_resp.start())
+        # Callback was removed again
+        self.assertEqual(udp_client.callbacks, [])