From b24df30e3f5c3073891e5478e3dd1fe3427ac0c5 Mon Sep 17 00:00:00 2001 From: Denis Shulyaka Date: Sun, 1 Oct 2023 20:26:02 +0300 Subject: [PATCH] Improve coverage --- tests/test_api.py | 16 +++++++++++----- tests/test_application.py | 40 +++++++++++++++++++++++++++++++++------ 2 files changed, 45 insertions(+), 11 deletions(-) diff --git a/tests/test_api.py b/tests/test_api.py index 2ff50cc..44d7862 100644 --- a/tests/test_api.py +++ b/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) diff --git a/tests/test_application.py b/tests/test_application.py index 63de630..1b9241b 100644 --- a/tests/test_application.py +++ b/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