Use new zigpy initialization (#95)

* Bump up zigpy dependency.
* Initial schema. Add device baudrate.
* Refactor API and Uart to use device schema.
* Update api and uart tests.
* Update tests.
* Fix.

setup.py

@@ -21,6 +21,6 @@
     author_email="rcloran@gmail.com",
     license="GPL-3.0",
     packages=find_packages(exclude=["*.tests"]),
-    install_requires=["pyserial-asyncio", "zigpy-homeassistant >= 0.17.0"],
+    install_requires=["pyserial-asyncio", "zigpy>= 0.20.a1"],
     tests_require=["pytest"],
 )

tests/test_api.py

@@ -7,24 +7,26 @@
 import zigpy.exceptions
 
 from zigpy_xbee import api as xbee_api, types as t, uart
+import zigpy_xbee.config
 from zigpy_xbee.zigbee.application import ControllerApplication
 
+DEVICE_CONFIG = zigpy_xbee.config.SCHEMA_DEVICE(
+    {zigpy_xbee.config.CONF_DEVICE_PATH: "/dev/null"}
+)
+
 
 @pytest.fixture
 def api():
-    api = xbee_api.XBee()
+    api = xbee_api.XBee(DEVICE_CONFIG)
     api._uart = mock.MagicMock()
     return api
 
 
 @pytest.mark.asyncio
 async def test_connect(monkeypatch):
-    api = xbee_api.XBee()
-    dev = mock.MagicMock()
-    monkeypatch.setattr(
-        uart, "connect", mock.MagicMock(side_effect=asyncio.coroutine(mock.MagicMock()))
-    )
-    await api.connect(dev, 115200)
+    api = xbee_api.XBee(DEVICE_CONFIG)
+    monkeypatch.setattr(uart, "connect", CoroutineMock())
+    await api.connect()
 
 
 def test_close(api):
@@ -542,14 +544,13 @@ def test_handle_many_to_one_rri(api):
 
 @pytest.mark.asyncio
 async def test_reconnect_multiple_disconnects(monkeypatch, caplog):
-    api = xbee_api.XBee()
-    dev = mock.sentinel.uart
+    api = xbee_api.XBee(DEVICE_CONFIG)
     connect_mock = CoroutineMock()
     connect_mock.return_value = asyncio.Future()
     connect_mock.return_value.set_result(True)
     monkeypatch.setattr(uart, "connect", connect_mock)
 
-    await api.connect(dev, 115200)
+    await api.connect()
 
     caplog.set_level(logging.DEBUG)
     connected = asyncio.Future()
@@ -568,14 +569,13 @@ async def test_reconnect_multiple_disconnects(monkeypatch, caplog):
 
 @pytest.mark.asyncio
 async def test_reconnect_multiple_attempts(monkeypatch, caplog):
-    api = xbee_api.XBee()
-    dev = mock.sentinel.uart
+    api = xbee_api.XBee(DEVICE_CONFIG)
     connect_mock = CoroutineMock()
     connect_mock.return_value = asyncio.Future()
     connect_mock.return_value.set_result(True)
     monkeypatch.setattr(uart, "connect", connect_mock)
 
-    await api.connect(dev, 115200)
+    await api.connect()
 
     caplog.set_level(logging.DEBUG)
     connected = asyncio.Future()
@@ -597,11 +597,11 @@ async def test_reconnect_multiple_attempts(monkeypatch, caplog):
 async def test_probe_success(mock_connect, mock_at_cmd):
     """Test device probing."""
 
-    res = await xbee_api.XBee.probe(mock.sentinel.uart, mock.sentinel.baud)
+    res = await xbee_api.XBee.probe(DEVICE_CONFIG)
     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_connect.call_args[0][0] == DEVICE_CONFIG
     assert mock_at_cmd.call_count == 1
     assert mock_connect.return_value.close.call_count == 1
 
@@ -613,11 +613,11 @@ async def test_probe_success(mock_connect, mock_at_cmd):
 async def test_probe_success_api_mode(mock_connect, mock_at_cmd, mock_api_mode):
     """Test device probing."""
 
-    res = await xbee_api.XBee.probe(mock.sentinel.uart, mock.sentinel.baud)
+    res = await xbee_api.XBee.probe(DEVICE_CONFIG)
     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_connect.call_args[0][0] == DEVICE_CONFIG
     assert mock_at_cmd.call_count == 1
     assert mock_api_mode.call_count == 1
     assert mock_connect.return_value.close.call_count == 1
@@ -638,11 +638,11 @@ async def test_probe_fail(mock_connect, mock_at_cmd, mock_api_mode, exception):
     mock_api_mode.reset_mock()
     mock_at_cmd.reset_mock()
     mock_connect.reset_mock()
-    res = await xbee_api.XBee.probe(mock.sentinel.uart, mock.sentinel.baud)
+    res = await xbee_api.XBee.probe(DEVICE_CONFIG)
     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_connect.call_args[0][0] == DEVICE_CONFIG
     assert mock_at_cmd.call_count == 1
     assert mock_api_mode.call_count == 1
     assert mock_connect.return_value.close.call_count == 1
@@ -658,11 +658,21 @@ async def test_probe_fail_api_mode(mock_connect, mock_at_cmd, mock_api_mode):
     mock_api_mode.reset_mock()
     mock_at_cmd.reset_mock()
     mock_connect.reset_mock()
-    res = await xbee_api.XBee.probe(mock.sentinel.uart, mock.sentinel.baud)
+    res = await xbee_api.XBee.probe(DEVICE_CONFIG)
     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_connect.call_args[0][0] == DEVICE_CONFIG
     assert mock_at_cmd.call_count == 1
     assert mock_api_mode.call_count == 1
     assert mock_connect.return_value.close.call_count == 1
+
+
+@pytest.mark.asyncio
+@mock.patch.object(xbee_api.XBee, "connect")
+async def test_xbee_new(conn_mck):
+    """Test new class method."""
+    api = await xbee_api.XBee.new(mock.sentinel.application, DEVICE_CONFIG)
+    assert isinstance(api, xbee_api.XBee)
+    assert conn_mck.call_count == 1
+    assert conn_mck.await_count == 1

tests/test_application.py

@@ -1,21 +1,34 @@
 import asyncio
-from unittest import mock
 
+from asynctest import CoroutineMock, mock
 import pytest
 from zigpy import types as t
 from zigpy.zdo.types import ZDOCmd
 
 from zigpy_xbee.api import ModemStatus, XBee
+import zigpy_xbee.config as config
 import zigpy_xbee.types as xbee_t
 from zigpy_xbee.zigbee import application
 
+APP_CONFIG = {
+    config.CONF_DEVICE: {
+        config.CONF_DEVICE_PATH: "/dev/null",
+        config.CONF_DEVICE_BAUDRATE: 115200,
+    },
+    config.CONF_DATABASE: None,
+}
+
 
 @pytest.fixture
-def app(monkeypatch, database_file=None):
+def app(monkeypatch):
     monkeypatch.setattr(application, "TIMEOUT_TX_STATUS", 0.1)
     monkeypatch.setattr(application, "TIMEOUT_REPLY", 0.1)
     monkeypatch.setattr(application, "TIMEOUT_REPLY_EXTENDED", 0.1)
-    return application.ControllerApplication(XBee(), database_file=database_file)
+    app = application.ControllerApplication(APP_CONFIG)
+    api = XBee(APP_CONFIG[config.CONF_DEVICE])
+    monkeypatch.setattr(api, "_command", CoroutineMock())
+    app._api = api
+    return app
 
 
 def test_modem_status(app):
@@ -188,11 +201,7 @@ async def test_broadcast(app):
         b"\x02\x01\x00",
     )
 
-    app._api._command = mock.MagicMock(
-        side_effect=asyncio.coroutine(
-            mock.MagicMock(return_value=xbee_t.TXStatus.SUCCESS)
-        )
-    )
+    app._api._command.return_value = xbee_t.TXStatus.SUCCESS
 
     r = await app.broadcast(profile, cluster, src_ep, dst_ep, grpid, radius, tsn, data)
     assert r[0] == xbee_t.TXStatus.SUCCESS
@@ -202,17 +211,11 @@ async def test_broadcast(app):
     assert app._api._command.call_args[0][4] == dst_ep
     assert app._api._command.call_args[0][9] == data
 
-    app._api._command = mock.MagicMock(
-        side_effect=asyncio.coroutine(
-            mock.MagicMock(return_value=xbee_t.TXStatus.ADDRESS_NOT_FOUND)
-        )
-    )
+    app._api._command.return_value = xbee_t.TXStatus.ADDRESS_NOT_FOUND
     r = await app.broadcast(profile, cluster, src_ep, dst_ep, grpid, radius, tsn, data)
     assert r[0] != xbee_t.TXStatus.SUCCESS
 
-    app._api._command = mock.MagicMock(
-        side_effect=asyncio.coroutine(mock.MagicMock(side_effect=asyncio.TimeoutError))
-    )
+    app._api._command.side_effect = asyncio.TimeoutError
     r = await app.broadcast(profile, cluster, src_ep, dst_ep, grpid, radius, tsn, data)
     assert r[0] != xbee_t.TXStatus.SUCCESS
 
@@ -304,19 +307,17 @@ async def _at_command_mock(cmd, *args):
             "ZS": zs,
         }.get(cmd, None)
 
-    app._api._at_command = mock.MagicMock(
-        spec=XBee._at_command, side_effect=_at_command_mock
-    )
-
     async def init_api_mode_mock():
         nonlocal api_mode
         api_mode = api_config_succeeds
         return api_config_succeeds
 
-    app._api.init_api_mode = mock.MagicMock(side_effect=init_api_mode_mock)
-    app.form_network = mock.MagicMock(side_effect=asyncio.coroutine(mock.MagicMock()))
+    app.form_network = CoroutineMock()
 
-    await app.startup(auto_form=auto_form)
+    with mock.patch.object(XBee, "new") as api:
+        api.return_value._at_command = CoroutineMock(side_effect=_at_command_mock)
+        api.return_value.init_api_mode = CoroutineMock(side_effect=init_api_mode_mock)
+        await app.startup(auto_form=auto_form)
     return app
 
 

tests/test_uart.py

@@ -5,6 +5,11 @@
 import serial_asyncio
 
 from zigpy_xbee import uart
+import zigpy_xbee.config
+
+DEVICE_CONFIG = zigpy_xbee.config.SCHEMA_DEVICE(
+    {zigpy_xbee.config.CONF_DEVICE_PATH: "/dev/null"}
+)
 
 
 @pytest.fixture
@@ -29,7 +34,6 @@ def test_baudrate_fail(gw):
 @pytest.mark.asyncio
 async def test_connect(monkeypatch):
     api = mock.MagicMock()
-    portmock = mock.MagicMock()
 
     async def mock_conn(loop, protocol_factory, **kwargs):
         protocol = protocol_factory()
@@ -38,7 +42,7 @@ async def mock_conn(loop, protocol_factory, **kwargs):
 
     monkeypatch.setattr(serial_asyncio, "create_serial_connection", mock_conn)
 
-    await uart.connect(portmock, 57600, api)
+    await uart.connect(DEVICE_CONFIG, api)
 
 
 def test_command_mode_rsp(gw):