Improve coverage (#157)

Author: Shulyaka

tests/test_api.py

@@ -31,7 +31,13 @@ async def test_connect(monkeypatch):
 
 def test_close(api):
     uart = api._uart
+    conn_lost_task = mock.MagicMock()
+    api._conn_lost_task = conn_lost_task
+
     api.close()
+
+    assert api._conn_lost_task is None
+    assert conn_lost_task.cancel.call_count == 1
     assert api._uart is None
     assert uart.close.call_count == 1
 
@@ -564,7 +570,7 @@ async def test_reconnect_multiple_attempts(monkeypatch, caplog):
 
 
 @mock.patch.object(xbee_api.XBee, "_at_command", new_callable=mock.AsyncMock)
-@mock.patch.object(uart, "connect")
+@mock.patch.object(uart, "connect", return_value=mock.MagicMock())
 async def test_probe_success(mock_connect, mock_at_cmd):
     """Test device probing."""
 
@@ -579,7 +585,7 @@ async def test_probe_success(mock_connect, mock_at_cmd):
 
 @mock.patch.object(xbee_api.XBee, "init_api_mode", return_value=True)
 @mock.patch.object(xbee_api.XBee, "_at_command", side_effect=asyncio.TimeoutError)
-@mock.patch.object(uart, "connect")
+@mock.patch.object(uart, "connect", return_value=mock.MagicMock())
 async def test_probe_success_api_mode(mock_connect, mock_at_cmd, mock_api_mode):
     """Test device probing."""
 
@@ -595,7 +601,7 @@ async def test_probe_success_api_mode(mock_connect, mock_at_cmd, mock_api_mode):
 
 @mock.patch.object(xbee_api.XBee, "init_api_mode")
 @mock.patch.object(xbee_api.XBee, "_at_command", side_effect=asyncio.TimeoutError)
-@mock.patch.object(uart, "connect")
+@mock.patch.object(uart, "connect", return_value=mock.MagicMock())
 @pytest.mark.parametrize(
     "exception",
     (asyncio.TimeoutError, serial.SerialException, zigpy.exceptions.APIException),
@@ -619,7 +625,7 @@ async def test_probe_fail(mock_connect, mock_at_cmd, mock_api_mode, exception):
 
 @mock.patch.object(xbee_api.XBee, "init_api_mode", return_value=False)
 @mock.patch.object(xbee_api.XBee, "_at_command", side_effect=asyncio.TimeoutError)
-@mock.patch.object(uart, "connect")
+@mock.patch.object(uart, "connect", return_value=mock.MagicMock())
 async def test_probe_fail_api_mode(mock_connect, mock_at_cmd, mock_api_mode):
     """Test device probing fails."""
 
@@ -636,7 +642,7 @@ async def test_probe_fail_api_mode(mock_connect, mock_at_cmd, mock_api_mode):
     assert mock_connect.return_value.close.call_count == 1
 
 
-@mock.patch.object(xbee_api.XBee, "connect")
+@mock.patch.object(xbee_api.XBee, "connect", return_value=mock.MagicMock())
 async def test_xbee_new(conn_mck):
     """Test new class method."""
     api = await xbee_api.XBee.new(mock.sentinel.application, DEVICE_CONFIG)

tests/test_application.py

@@ -279,8 +279,16 @@ async def test_get_association_state(app):
     assert ai is mock.sentinel.ai
 
 
-async def test_write_network_info(app, node_info, network_info):
-    app._api._queued_at = mock.AsyncMock(spec=XBee._queued_at)
+@pytest.mark.parametrize("legacy_module", (False, True))
+async def test_write_network_info(app, node_info, network_info, legacy_module):
+    def _mock_queued_at(name, *args):
+        if legacy_module and name == "CE":
+            raise RuntimeError("Legacy module")
+        return "OK"
+
+    app._api._queued_at = mock.AsyncMock(
+        spec=XBee._queued_at, side_effect=_mock_queued_at
+    )
     app._api._at_command = mock.AsyncMock(spec=XBee._at_command)
     app._api._running = mock.AsyncMock(spec=app._api._running)
 
@@ -308,7 +316,7 @@ async def _test_start_network(
     legacy_module=False,
 ):
     ai_tries = 5
-    app.state.node_info.nwk = mock.sentinel.nwk
+    app.state.node_info = zigpy.state.NodeInfo()
 
     def _at_command_mock(cmd, *args):
         nonlocal ai_tries
@@ -348,7 +356,6 @@ def init_api_mode_mock():
         await app.connect()
 
     app.form_network = mock.AsyncMock()
-    await app.load_network_info()
     await app.start_network()
     return app
 
@@ -459,6 +466,25 @@ async def test_request_send_fail(app):
         await _test_request(app, send_success=False)
 
 
+async def test_request_unknown_device(app):
+    dev = zigpy.device.Device(
+        application=app, ieee=xbee_t.UNKNOWN_IEEE, nwk=xbee_t.UNKNOWN_NWK
+    )
+    with pytest.raises(
+        zigpy.exceptions.DeliveryError,
+        match="Cannot send a packet to a device without a known IEEE address",
+    ):
+        await app.request(
+            dev,
+            0x0260,
+            1,
+            2,
+            3,
+            123,
+            b"\xaa\x55\xbe\xef",
+        )
+
+
 async def test_request_extended_timeout(app):
     r = await _test_request(app, True, True, extended_timeout=False)
     assert r[0] == xbee_t.TXStatus.SUCCESS
@@ -485,12 +511,14 @@ async def test_shutdown(app):
     assert mack_close.call_count == 1
 
 
-def test_remote_at_cmd(app, device):
+async def test_remote_at_cmd(app, device):
     dev = device()
     app.get_device = mock.MagicMock(return_value=dev)
     app._api = mock.MagicMock(spec=XBee)
     s = mock.sentinel
-    app.remote_at_command(s.nwk, s.cmd, s.data, apply_changes=True, encryption=True)
+    await app.remote_at_command(
+        s.nwk, s.cmd, s.data, apply_changes=True, encryption=True
+    )
     assert app._api._remote_at_command.call_count == 1
     assert app._api._remote_at_command.call_args[0][0] is dev.ieee
     assert app._api._remote_at_command.call_args[0][1] == s.nwk