Implement radio probe (#101)

Author: Adminiuga

tests/test_api.py

@@ -1,9 +1,9 @@
 import asyncio
 import logging
-from unittest import mock
 
-import asynctest
+from asynctest import CoroutineMock, mock
 import pytest
+import serial
 
 from zigpy_deconz import api as deconz_api, types as t, uart
 import zigpy_deconz.exception
@@ -32,9 +32,10 @@ async def test_connect(monkeypatch):
 
 
 def test_close(api):
-    api._uart.close = mock.MagicMock()
+    uart = api._uart
     api.close()
-    assert api._uart.close.call_count == 1
+    assert api._uart is None
+    assert uart.close.call_count == 1
 
 
 def test_commands():
@@ -438,7 +439,7 @@ def test_device_state_network_state(data, network_state):
 async def test_reconnect_multiple_disconnects(monkeypatch, caplog):
     api = deconz_api.Deconz()
     dev = mock.sentinel.uart
-    connect_mock = asynctest.CoroutineMock()
+    connect_mock = CoroutineMock()
     connect_mock.return_value = asyncio.Future()
     connect_mock.return_value.set_result(True)
     monkeypatch.setattr(uart, "connect", connect_mock)
@@ -464,7 +465,7 @@ async def test_reconnect_multiple_disconnects(monkeypatch, caplog):
 async def test_reconnect_multiple_attempts(monkeypatch, caplog):
     api = deconz_api.Deconz()
     dev = mock.sentinel.uart
-    connect_mock = asynctest.CoroutineMock()
+    connect_mock = CoroutineMock()
     connect_mock.return_value = asyncio.Future()
     connect_mock.return_value.set_result(True)
     monkeypatch.setattr(uart, "connect", connect_mock)
@@ -477,9 +478,57 @@ async def test_reconnect_multiple_attempts(monkeypatch, caplog):
     connect_mock.reset_mock()
     connect_mock.side_effect = [asyncio.TimeoutError, OSError, connected]
 
-    with asynctest.mock.patch("asyncio.sleep"):
+    with mock.patch("asyncio.sleep"):
         api.connection_lost("connection lost")
         await api._conn_lost_task
 
     assert api._uart is mock.sentinel.uart_reconnect
     assert connect_mock.call_count == 3
+
+
+@pytest.mark.asyncio
+@mock.patch.object(deconz_api.Deconz, "device_state", new_callable=CoroutineMock)
+@mock.patch.object(uart, "connect")
+async def test_probe_success(mock_connect, mock_device_state):
+    """Test device probing."""
+
+    res = await deconz_api.Deconz.probe(mock.sentinel.uart, mock.sentinel.baud)
+    assert res is True
+    assert mock_connect.call_count == 1
+    assert mock_connect.await_count == 1
+    assert mock_connect.call_args[0][0] is mock.sentinel.uart
+    assert mock_device_state.call_count == 1
+    assert mock_connect.return_value.close.call_count == 1
+
+    mock_connect.reset_mock()
+    mock_device_state.reset_mock()
+    mock_connect.reset_mock()
+    res = await deconz_api.Deconz.probe(mock.sentinel.uart, mock.sentinel.baud)
+    assert res is True
+    assert mock_connect.call_count == 1
+    assert mock_connect.await_count == 1
+    assert mock_connect.call_args[0][0] is mock.sentinel.uart
+    assert mock_device_state.call_count == 1
+    assert mock_connect.return_value.close.call_count == 1
+
+
+@pytest.mark.asyncio
+@mock.patch.object(deconz_api.Deconz, "device_state", new_callable=CoroutineMock)
+@mock.patch.object(uart, "connect")
+@pytest.mark.parametrize(
+    "exception",
+    (asyncio.TimeoutError, serial.SerialException, zigpy_deconz.exception.CommandError),
+)
+async def test_probe_fail(mock_connect, mock_device_state, exception):
+    """Test device probing fails."""
+
+    mock_device_state.side_effect = exception
+    mock_device_state.reset_mock()
+    mock_connect.reset_mock()
+    res = await deconz_api.Deconz.probe(mock.sentinel.uart, mock.sentinel.baud)
+    assert res is False
+    assert mock_connect.call_count == 1
+    assert mock_connect.await_count == 1
+    assert mock_connect.call_args[0][0] is mock.sentinel.uart
+    assert mock_device_state.call_count == 1
+    assert mock_connect.return_value.close.call_count == 1

zigpy_deconz/api.py

@@ -4,14 +4,17 @@
 import logging
 import typing
 
-from zigpy_deconz.exception import CommandError
+import serial
+
+from zigpy_deconz.exception import APIException, CommandError
 
 from . import types as t, uart
 
 LOGGER = logging.getLogger(__name__)
 
 COMMAND_TIMEOUT = 2
 DECONZ_BAUDRATE = 38400
+PROBE_TIMEOUT = 3
 MIN_PROTO_VERSION = 0x010B
 
 
@@ -255,7 +258,9 @@ async def _reconnect_till_done(self) -> None:
         )
 
     def close(self):
-        return self._uart.close()
+        if self._uart:
+            self._uart.close()
+            self._uart = None
 
     async def _command(self, cmd, *args):
         LOGGER.debug("Command %s %s", cmd, args)
@@ -327,6 +332,26 @@ def change_network_state(self, state):
     def _handle_change_network_state(self, data):
         LOGGER.debug("Change network state response: %s", NetworkState(data[0]).name)
 
+    @classmethod
+    async def probe(cls, device: str, baudrate: int = DECONZ_BAUDRATE) -> bool:
+        """Probe port for the device presence."""
+        api = cls()
+        try:
+            await asyncio.wait_for(api._probe(device, baudrate), timeout=PROBE_TIMEOUT)
+            return True
+        except (asyncio.TimeoutError, serial.SerialException, APIException) as exc:
+            LOGGER.debug("Unsuccessful radio probe of '%s' port", exc_info=exc)
+        finally:
+            api.close()
+
+        return False
+
+    async def _probe(self, device: str, baudrate: int = DECONZ_BAUDRATE) -> None:
+        """Open port and try sending a command"""
+        await self.connect(device, baudrate)
+        await self.device_state()
+        self.close()
+
     async def read_parameter(self, id_, *args):
         try:
             if isinstance(id_, str):

zigpy_deconz/exception.py

@@ -2,7 +2,7 @@
 
 
 class CommandError(APIException):
-    def __init__(self, status, *args, **kwargs):
+    def __init__(self, status=1, *args, **kwargs):
         self._status = status
         super().__init__(*args, **kwargs)