Repository: XKNX/xknx
Pull request #935 (2022-04-28 22:34:20)

Try multiple gateways in Automatic mode
<!--
  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.
-->
- Connect to next found interface in case of unsuccessful initial connection when using "automatic" mode
- Connect to first found interface immediately after it was discovered - while scanning for others in background until connection is successful

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)
- [x] 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
- [ ] 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


codecov[bot]: # [Codecov](https://codecov.io/gh/XKNX/xknx/pull/935?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX) Report
> Merging [#935](https://codecov.io/gh/XKNX/xknx/pull/935?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX) (664e7b0) into [main](https://codecov.io/gh/XKNX/xknx/commit/8b1431850a20caafaebddc039f41a6ccae8583ef?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX) (8b14318) will **decrease** coverage by `0.03%`.
> The diff coverage is `7.14%`.

[![Impacted file tree graph](https://codecov.io/gh/XKNX/xknx/pull/935/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://codecov.io/gh/XKNX/xknx/pull/935?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX)

```diff
@@            Coverage Diff             @@
##             main     #935      +/-   ##
==========================================
- Coverage   95.98%   95.94%   -0.04%     
==========================================
  Files         130      130              
  Lines        7763     7766       +3     
==========================================
  Hits         7451     7451              
- Misses        312      315       +3     
```


| [Impacted Files](https://codecov.io/gh/XKNX/xknx/pull/935?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX) | Coverage Δ | |
|---|---|---|
| [xknx/io/knxip\_interface.py](https://codecov.io/gh/XKNX/xknx/pull/935/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX#diff-eGtueC9pby9rbnhpcF9pbnRlcmZhY2UucHk=) | `75.47% <7.14%> (-1.46%)` | :arrow_down: |

farmio: Dropped coverage in xknx.io.transport package comes from a fixed test that did open a socket - we should not do that unless we have a mock-server 😬

Patch:
diff --git a/changelog.md b/changelog.md
index 9e4d6fa1..96116bee 100644
--- a/changelog.md
+++ b/changelog.md
@@ -9,6 +9,7 @@
 - Skip SearchResponse results for Core-V2 devices - wait for SearchResponseExtended
 - Identify interfaces having KNX IP Secure Tunneling required and skip if using Automatic connection mode
 - Only send SearchRequests from one interface for each `scan()` call
+- Connect to next found interface in case of unsuccessful initial connection when using "automatic" mode
 
 ### Internals
 
diff --git a/xknx/io/gateway_scanner.py b/xknx/io/gateway_scanner.py
index 61afc268..843a4da7 100644
--- a/xknx/io/gateway_scanner.py
+++ b/xknx/io/gateway_scanner.py
@@ -9,7 +9,7 @@ from __future__ import annotations
 import asyncio
 from functools import partial
 import logging
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, AsyncGenerator
 
 from xknx.exceptions import XKNXException
 from xknx.io import util
@@ -182,27 +182,51 @@ class GatewayScanner:
     def __init__(
         self,
         xknx: XKNX,
-        timeout_in_seconds: float = 4.0,
-        stop_on_found: int | None = 1,
+        local_ip: str | None = None,
+        timeout_in_seconds: float = 3.0,
+        stop_on_found: int | None = None,
         scan_filter: GatewayScanFilter = GatewayScanFilter(),
     ):
         """Initialize GatewayScanner class."""
         self.xknx = xknx
+        self.local_ip = local_ip
         self.timeout_in_seconds = timeout_in_seconds
-        self.stop_on_found = stop_on_found or 0
+        self.stop_on_found = stop_on_found
         self.scan_filter = scan_filter
         self.found_gateways: dict[HPAI, GatewayDescriptor] = {}
         self._response_received_event = asyncio.Event()
 
-    async def scan(
-        self,
-        local_ip: str | None = None,
-    ) -> list[GatewayDescriptor]:
+    async def scan(self) -> list[GatewayDescriptor]:
         """Scan and return a list of GatewayDescriptors on success."""
-        local_ip = local_ip or await util.get_default_local_ip(
+        await self._scan()
+        return list(self.found_gateways.values())
+
+    async def async_scan(self) -> AsyncGenerator[GatewayDescriptor, None]:
+        """Search and yield found gateways."""
+        queue: asyncio.Queue[GatewayDescriptor | None] = asyncio.Queue()
+        scan_task = asyncio.create_task(self._scan(queue=queue))
+        try:
+            while True:
+                gateway = await queue.get()
+                if gateway is None:
+                    return
+                yield gateway
+        finally:
+            # cleanup after GeneratorExit or XKNXExceptions
+            if not scan_task.done():
+                scan_task.cancel()
+            await scan_task  # to bubble up exceptions
+
+    async def _scan(
+        self, queue: asyncio.Queue[GatewayDescriptor | None] | None = None
+    ) -> None:
+        """Scan for gateways."""
+        local_ip = self.local_ip or await util.get_default_local_ip(
             remote_ip=self.xknx.multicast_group
         )
         if local_ip is None:
+            if queue is not None:
+                queue.put_nowait(None)
             raise XKNXException("No usable network interface found.")
         interface_name = util.get_local_interface_name(local_ip=local_ip)
         logger.debug("Searching on %s / %s", interface_name, local_ip)
@@ -212,7 +236,7 @@ class GatewayScanner:
             remote_addr=(self.xknx.multicast_group, self.xknx.multicast_port),
         )
         udp_transport.register_callback(
-            partial(self._response_rec_callback, interface=interface_name),
+            partial(self._response_rec_callback, interface=interface_name, queue=queue),
             [
                 KNXIPServiceType.SEARCH_RESPONSE,
                 KNXIPServiceType.SEARCH_RESPONSE_EXTENDED,
@@ -226,10 +250,12 @@ class GatewayScanner:
             )
         except asyncio.TimeoutError:
             pass
+        except asyncio.CancelledError:
+            pass
         finally:
             udp_transport.stop()
-
-        return list(self.found_gateways.values())
+            if queue is not None:
+                queue.put_nowait(None)
 
     @staticmethod
     async def _send_search_requests(udp_transport: UDPTransport) -> None:
@@ -261,6 +287,7 @@ class GatewayScanner:
         source: HPAI,
         udp_transport: UDPTransport,
         interface: str = "",
+        queue: asyncio.Queue[GatewayDescriptor | None] | None = None,
     ) -> None:
         """Verify and handle knxipframe. Callback from internal udp_transport."""
         if not isinstance(knx_ip_frame.body, (SearchResponse, SearchResponseExtended)):
@@ -292,6 +319,7 @@ class GatewayScanner:
         logger.debug("Found KNX/IP device at %s: %s", source, repr(gateway))
         if self.scan_filter.match(gateway):
             self.found_gateways[knx_ip_frame.body.control_endpoint] = gateway
-
-            if len(self.found_gateways) >= self.stop_on_found:
+            if queue is not None:
+                queue.put_nowait(gateway)
+            if self.stop_on_found and len(self.found_gateways) >= self.stop_on_found:
                 self._response_received_event.set()
diff --git a/xknx/io/knxip_interface.py b/xknx/io/knxip_interface.py
index 67138ede..9888338b 100644
--- a/xknx/io/knxip_interface.py
+++ b/xknx/io/knxip_interface.py
@@ -23,7 +23,7 @@ from xknx.io import util
 from xknx.secure import Keyring, load_key_ring
 
 from .connection import ConnectionConfig, ConnectionType
-from .gateway_scanner import GatewayDescriptor, GatewayScanFilter, GatewayScanner
+from .gateway_scanner import GatewayDescriptor, GatewayScanner
 from .routing import Routing
 from .tunnel import SecureTunnel, TCPTunnel, UDPTunnel, _Tunnel
 
@@ -139,24 +139,34 @@ class KNXIPInterface:
 
     async def _start_automatic(self) -> None:
         """Start GatewayScanner and connect to the found device."""
-        gateway = await self.find_gateway(
-            scan_filter=self.connection_config.scan_filter,  # secure disabled by default
+        async for gateway in GatewayScanner(
+            self.xknx,
             local_ip=self.connection_config.local_ip,
-        )
-        self._gateway_info = gateway
-
-        if gateway.supports_tunnelling_tcp:
-            await self._start_tunnelling_tcp(
-                gateway_ip=gateway.ip_addr,
-                gateway_port=gateway.port,
-            )
-        elif gateway.supports_tunnelling:
-            await self._start_tunnelling_udp(
-                gateway_ip=gateway.ip_addr,
-                gateway_port=gateway.port,
-            )
-        elif gateway.supports_routing:
-            await self._start_routing(local_ip=self.connection_config.local_ip)
+            scan_filter=self.connection_config.scan_filter,  # secure disabled by default
+        ).async_scan():
+            try:
+                if gateway.supports_tunnelling_tcp:
+                    await self._start_tunnelling_tcp(
+                        gateway_ip=gateway.ip_addr,
+                        gateway_port=gateway.port,
+                    )
+                elif gateway.supports_tunnelling:
+                    await self._start_tunnelling_udp(
+                        gateway_ip=gateway.ip_addr,
+                        gateway_port=gateway.port,
+                    )
+                elif gateway.supports_routing:
+                    await self._start_routing(
+                        local_ip=self.connection_config.local_ip,
+                    )
+            except CommunicationError as ex:
+                logger.debug("Could not connect to %s: %s", gateway, ex)
+                continue
+            else:
+                self._gateway_info = gateway
+                break
+        else:
+            raise CommunicationError("No usable KNX/IP device found.")
 
     async def _start_tunnelling_tcp(
         self,
@@ -283,18 +293,6 @@ class KNXIPInterface:
             return await self._interface.request_description()
         return None
 
-    async def find_gateway(
-        self,
-        scan_filter: GatewayScanFilter,
-        local_ip: str | None = None,
-    ) -> GatewayDescriptor:
-        """Find KNX/IP Gateway."""
-        gatewayscanner = GatewayScanner(self.xknx, scan_filter=scan_filter)
-        gateways = await gatewayscanner.scan(local_ip=local_ip)
-        if not gateways:
-            raise XKNXException("No Gateways found")
-        return gateways[0]
-
 
 class KNXIPInterfaceThreaded(KNXIPInterface):
     """Class for managing KNX/IP Tunneling or Routing connections."""


Test patch:
diff --git a/test/io_tests/gateway_scanner_test.py b/test/io_tests/gateway_scanner_test.py
index dc51b5df..e98f3ec8 100644
--- a/test/io_tests/gateway_scanner_test.py
+++ b/test/io_tests/gateway_scanner_test.py
@@ -1,8 +1,11 @@
 """Unit test for KNX/IP gateway scanner."""
 import asyncio
-from unittest.mock import create_autospec, patch
+from unittest.mock import Mock, create_autospec, patch
+
+import pytest
 
 from xknx import XKNX
+from xknx.exceptions import XKNXException
 from xknx.io import GatewayScanFilter, GatewayScanner
 from xknx.io.gateway_scanner import GatewayDescriptor
 from xknx.io.transport import UDPTransport
@@ -170,7 +173,19 @@ class TestGatewayScanner:
             gateway_scanner.found_gateways[test_search_response.body.control_endpoint]
         ) == str(self.gateway_desc_both)
 
-    async def test_scan_timeout(self, time_travel):
+    @patch("xknx.io.gateway_scanner.UDPTransport.connect")
+    @patch("xknx.io.gateway_scanner.UDPTransport.send")
+    @patch(
+        "xknx.io.gateway_scanner.UDPTransport.getsockname",
+        return_value=("10.1.1.2", 56789),
+    )
+    async def test_scan_timeout(
+        self,
+        getsockname_mock,
+        udp_transport_send_mock,
+        udp_transport_connect_mock,
+        time_travel,
+    ):
         """Test gateway scanner timeout."""
         xknx = XKNX()
         gateway_scanner = GatewayScanner(xknx)
@@ -179,6 +194,83 @@ class TestGatewayScanner:
         # Unsuccessfull scan() returns empty list
         assert await timed_out_scan_task == []
 
+    @patch("xknx.io.gateway_scanner.UDPTransport.connect")
+    @patch("xknx.io.gateway_scanner.UDPTransport.send")
+    async def test_async_scan_timeout(
+        self,
+        udp_transport_send_mock,
+        udp_transport_connect_mock,
+        time_travel,
+    ):
+        """Test gateway scanner timeout for async generator."""
+
+        async def test():
+            xknx = XKNX()
+            async for _ in GatewayScanner(xknx).async_scan():
+                break
+            else:
+                return True
+
+        # timeout
+        with patch(
+            "xknx.io.util.get_default_local_ip",
+            return_value="10.1.1.2",
+        ), patch(
+            "xknx.io.gateway_scanner.UDPTransport.getsockname",
+            return_value=("10.1.1.2", 56789),
+        ):
+            timed_out_scan_task = asyncio.create_task(test())
+            await time_travel(3)
+            assert await timed_out_scan_task
+        # no matching interface found
+        with patch(
+            "xknx.io.util.get_default_local_ip",
+            return_value=None,
+        ):
+            timed_out_scan_task = asyncio.create_task(test())
+            await time_travel(3)
+            with pytest.raises(XKNXException):
+                await timed_out_scan_task
+
+    @patch("xknx.io.gateway_scanner.UDPTransport.connect")
+    @patch("xknx.io.gateway_scanner.UDPTransport.send")
+    async def test_async_scan_exit(
+        self,
+        udp_transport_send_mock,
+        udp_transport_connect_mock,
+        time_travel,
+    ):
+        """Test gateway scanner timeout for async generator."""
+        xknx = XKNX()
+        test_search_response = fake_router_search_response()
+        udp_transport_mock = Mock()
+        udp_transport_mock.local_addr = ("10.1.1.2", 56789)
+
+        gateway_scanner = GatewayScanner(xknx, local_ip="10.1.1.2")
+
+        async def test():
+            async for gateway in gateway_scanner.async_scan():
+                assert isinstance(gateway, GatewayDescriptor)
+                return True
+            return False
+
+        with patch(
+            "xknx.io.gateway_scanner.UDPTransport.getsockname",
+            return_value=("10.1.1.2", 56789),
+        ), patch(
+            "xknx.io.gateway_scanner.UDPTransport.register_callback"
+        ) as register_callback_mock:
+            scan_task = asyncio.create_task(test())
+            await time_travel(0)
+            _fished_response_rec_callback = register_callback_mock.call_args.args[0]
+            _fished_response_rec_callback(
+                test_search_response,
+                HPAI("192.168.42.50", 0),
+                udp_transport_mock,
+            )
+            assert await scan_task
+            await time_travel(0)  # for task cleanup
+
     @patch("xknx.io.gateway_scanner.UDPTransport.connect")
     @patch("xknx.io.gateway_scanner.UDPTransport.send")
     async def test_send_search_requests(
diff --git a/test/io_tests/knxip_interface_test.py b/test/io_tests/knxip_interface_test.py
index a50a074e..0af39050 100644
--- a/test/io_tests/knxip_interface_test.py
+++ b/test/io_tests/knxip_interface_test.py
@@ -7,12 +7,14 @@ import pytest
 
 from xknx import XKNX
 from xknx.exceptions.exception import (
+    CommunicationError,
     InterfaceWithUserIdNotFound,
     InvalidSecureConfiguration,
 )
 from xknx.io import (
     ConnectionConfig,
     ConnectionType,
+    GatewayDescriptor,
     SecureConfig,
     knx_interface_factory,
 )
@@ -32,12 +34,42 @@ class TestKNXIPInterface:
         """Test starting automatic connection."""
         connection_config = ConnectionConfig()
         assert connection_config.connection_type == ConnectionType.AUTOMATIC
+        interface = knx_interface_factory(self.xknx, connection_config)
         with patch("xknx.io.KNXIPInterface._start_automatic") as start_automatic_mock:
-            interface = knx_interface_factory(self.xknx, connection_config)
             await interface.start()
             start_automatic_mock.assert_called_once_with()
             assert threading.active_count() == 1
 
+        async def gateway_generator_mock(_):
+            yield GatewayDescriptor(
+                ip_addr="10.1.2.3", port=3671, supports_tunnelling_tcp=True
+            )
+            yield GatewayDescriptor(
+                ip_addr="10.1.2.3", port=3671, supports_tunnelling=True
+            )
+            yield GatewayDescriptor(
+                ip_addr="10.1.2.3", port=3671, supports_routing=True
+            )
+
+        with patch(
+            "xknx.io.knxip_interface.GatewayScanner.async_scan",
+            new=gateway_generator_mock,
+        ), patch(
+            "xknx.io.KNXIPInterface._start_tunnelling_tcp",
+            side_effect=CommunicationError("Error"),
+        ) as start_tunnelling_tcp_mock, patch(
+            "xknx.io.KNXIPInterface._start_tunnelling_udp",
+            side_effect=CommunicationError("Error"),
+        ) as start_tunnelling_udp_mock, patch(
+            "xknx.io.KNXIPInterface._start_routing",
+            side_effect=CommunicationError("Error"),
+        ) as start_routing_mock:
+            with pytest.raises(CommunicationError):
+                await interface.start()
+            start_tunnelling_tcp_mock.assert_called_once()
+            start_tunnelling_udp_mock.assert_called_once()
+            start_routing_mock.assert_called_once()
+
     async def test_start_udp_tunnel_connection(self):
         """Test starting UDP tunnel connection."""
         # without gateway_ip automatic is called