Handle tx_status notification frames. (#40)

* Add TXStatus and DiscoveryStatus enums.
* Use non keyword arguments for _handle_* methods.
* Properly handle tx_status notifications.
* Update test units.

Author: Adminiuga

tests/test_api.py

@@ -176,15 +176,27 @@ def test_api_frame(api):
     for cmd_name, cmd_opts in xbee_api.COMMAND_REQUESTS.items():
         cmd_id, schema, repl = cmd_opts
         if schema:
-            args = [ieee if isinstance(a(), t.EUI64) else a() for a in schema]
+            args = [ieee if issubclass(a, t.EUI64) else a() for a in schema]
             frame, repl = api._api_frame(cmd_name, *args)
         else:
             frame, repl = api._api_frame(cmd_name)
 
 
 def test_frame_received(api, monkeypatch):
     monkeypatch.setattr(t, 'deserialize', mock.MagicMock(
-        return_value=(mock.sentinel.deserialize_data, b'')))
+        return_value=(
+            [
+                mock.sentinel.arg_0,
+                mock.sentinel.arg_1,
+                mock.sentinel.arg_2,
+                mock.sentinel.arg_3,
+                mock.sentinel.arg_4,
+                mock.sentinel.arg_5,
+                mock.sentinel.arg_6,
+                mock.sentinel.arg_7,
+                mock.sentinel.arg_8,
+            ], b'')
+    ))
     my_handler = mock.MagicMock()
 
     for cmd, cmd_opts in xbee_api.COMMAND_RESPONSES.items():
@@ -196,7 +208,10 @@ def test_frame_received(api, monkeypatch):
         assert t.deserialize.call_count == 1
         assert t.deserialize.call_args[0][0] == payload
         assert my_handler.call_count == 1
-        assert my_handler.call_args[0][0] == mock.sentinel.deserialize_data
+        assert my_handler.call_args[0][0] == mock.sentinel.arg_0
+        assert my_handler.call_args[0][1] == mock.sentinel.arg_1
+        assert my_handler.call_args[0][2] == mock.sentinel.arg_2
+        assert my_handler.call_args[0][3] == mock.sentinel.arg_3
         t.deserialize.reset_mock()
         my_handler.reset_mock()
 
@@ -224,7 +239,7 @@ def _handle_at_response(api, tsn, status, at_response=b''):
     data = (tsn, 'AI'.encode('ascii'), status, at_response)
     response = asyncio.Future()
     api._awaiting[tsn] = (response, )
-    api._handle_at_response(data)
+    api._handle_at_response(*data)
     return response
 
 
@@ -264,18 +279,18 @@ def test_handle_at_response_undef_error(api):
 def test_handle_remote_at_rsp(api):
     api._handle_at_response = mock.MagicMock()
     s = mock.sentinel
-    api._handle_remote_at_response([s.frame_id, s.ieee, s.nwk, s.cmd,
-                                    s.status, s.data])
+    api._handle_remote_at_response(s.frame_id, s.ieee, s.nwk, s.cmd,
+                                   s.status, s.data)
     assert api._handle_at_response.call_count == 1
-    assert api._handle_at_response.call_args[0][0][0] == s.frame_id
-    assert api._handle_at_response.call_args[0][0][1] == s.cmd
-    assert api._handle_at_response.call_args[0][0][2] == s.status
-    assert api._handle_at_response.call_args[0][0][3] == s.data
+    assert api._handle_at_response.call_args[0][0] == s.frame_id
+    assert api._handle_at_response.call_args[0][1] == s.cmd
+    assert api._handle_at_response.call_args[0][2] == s.status
+    assert api._handle_at_response.call_args[0][3] == s.data
 
 
 def _send_modem_event(api, event):
     api._app = mock.MagicMock(spec=ControllerApplication)
-    api._handle_modem_status([event])
+    api._handle_modem_status(event)
     assert api._app.handle_modem_status.call_count == 1
     assert api._app.handle_modem_status.call_args[0][0] == event
 
@@ -301,15 +316,60 @@ def test_handle_modem_status(api):
 
 
 def test_handle_explicit_rx_indicator(api):
-    data = b'\x00\x01\x02\x03\x04\x05\x06\x07'
+    s = mock.sentinel
+    data = [s.src_ieee, s.src_nwk, s.src_ep, s.dst_ep, s.cluster_id, s.profile,
+            s.opts, b'abcdef']
     api._app = mock.MagicMock()
     api._app.handle_rx = mock.MagicMock()
-    api._handle_explicit_rx_indicator(data)
+    api._handle_explicit_rx_indicator(*data)
     assert api._app.handle_rx.call_count == 1
 
 
-def test_handle_tx_status(api):
-    api._handle_tx_status(b'\x01\x02\x03\x04')
+def _handle_tx_status(api, status, wrong_frame_id=False):
+    status = t.TXStatus(status)
+    frame_id = 0x12
+    send_fut = mock.MagicMock(spec=asyncio.Future)
+    api._awaiting[frame_id] = (send_fut, )
+    s = mock.sentinel
+    if wrong_frame_id:
+        frame_id += 1
+    api._handle_tx_status(frame_id, s.dst_nwk, s.retries, status,
+                          t.DiscoveryStatus())
+    return send_fut
+
+
+def test_handle_tx_status_success(api):
+    fut = _handle_tx_status(api, t.TXStatus.SUCCESS)
+    assert len(api._awaiting) == 0
+    assert fut.set_result.call_count == 1
+    assert fut.set_exception.call_count == 0
+
+
+def test_handle_tx_status_except(api):
+    fut = _handle_tx_status(api, t.TXStatus.ADDRESS_NOT_FOUND)
+    assert len(api._awaiting) == 0
+    assert fut.set_result.call_count == 0
+    assert fut.set_exception.call_count == 1
+
+
+def test_handle_tx_status_unexpected(api):
+    fut = _handle_tx_status(api, 1, wrong_frame_id=True)
+    assert len(api._awaiting) == 1
+    assert fut.set_result.call_count == 0
+    assert fut.set_exception.call_count == 0
+
+
+def test_handle_tx_status_duplicate(api):
+    status = t.TXStatus.SUCCESS
+    frame_id = 0x12
+    send_fut = mock.MagicMock(spec=asyncio.Future)
+    send_fut.set_result.side_effect = asyncio.InvalidStateError
+    api._awaiting[frame_id] = (send_fut, )
+    s = mock.sentinel
+    api._handle_tx_status(frame_id, s.dst_nwk, s.retries, status, s.disc)
+    assert len(api._awaiting) == 0
+    assert send_fut.set_result.call_count == 1
+    assert send_fut.set_exception.call_count == 0
 
 
 @pytest.mark.asyncio
@@ -422,10 +482,11 @@ def test_set_application(api):
 
 
 def test_handle_route_record_indicator(api):
-    api._handle_route_record_indicator(mock.sentinel.ri)
+    s = mock.sentinel
+    api._handle_route_record_indicator(s.ieee, s.src, s.rx_opts, s.hops)
 
 
 def test_handle_many_to_one_rri(api):
     ieee = t.EUI64([t.uint8_t(a) for a in range(0, 8)])
     nwk = 0x1234
-    api._handle_many_to_one_rri([ieee, nwk, 0])
+    api._handle_many_to_one_rri(ieee, nwk, 0)

tests/test_application.py

@@ -3,14 +3,17 @@
 
 import pytest
 
+from zigpy.exceptions import DeliveryError
 from zigpy.types import EUI64, uint16_t
 from zigpy_xbee.api import ModemStatus, XBee
-from zigpy_xbee.zigbee.application import ControllerApplication
+from zigpy_xbee.zigbee import application
 
 
 @pytest.fixture
-def app(database_file=None):
-    return ControllerApplication(XBee(), database_file=database_file)
+def app(monkeypatch, database_file=None):
+    monkeypatch.setattr(application, 'TIMEOUT_TX_STATUS', 0.1)
+    return application.ControllerApplication(XBee(),
+                                             database_file=database_file)
 
 
 def test_modem_status(app):
@@ -293,7 +296,7 @@ async def mock_at_command(cmd, *args):
     app._api._queued_at = mock.MagicMock(spec=XBee._at_command,
                                          side_effect=mock_at_command)
     app._get_association_state = mock.MagicMock(
-        spec=ControllerApplication._get_association_state,
+        spec=application.ControllerApplication._get_association_state,
         side_effect=asyncio.coroutine(mock.MagicMock(return_value=0x00))
     )
 
@@ -434,17 +437,26 @@ async def test_permit(app):
     assert app._api._at_command.call_args_list[0][0][1] == time_s
 
 
-async def _test_request(app, do_reply=True, expect_reply=True, **kwargs):
+async def _test_request(app, do_reply=True, expect_reply=True,
+                        send_success=True, send_timeout=False, **kwargs):
     seq = 123
     nwk = 0x2345
     ieee = EUI64(b'\x01\x02\x03\x04\x05\x06\x07\x08')
     app.add_device(ieee, nwk)
 
     def _mock_command(cmdname, ieee, nwk, src_ep, dst_ep, cluster,
                       profile, radius, options, data):
+        send_fut = asyncio.Future()
+        if not send_timeout:
+            if send_success:
+                send_fut.set_result(True)
+            else:
+                send_fut.set_exception(DeliveryError())
+
         if expect_reply:
             if do_reply:
                 app._pending[seq].set_result(mock.sentinel.reply_result)
+        return send_fut
 
     app._api._command = mock.MagicMock(side_effect=_mock_command)
     return await app.request(nwk, 0x0260, 1, 2, 3, seq, [4, 5, 6], expect_reply=expect_reply, **kwargs)
@@ -457,15 +469,27 @@ async def test_request_with_reply(app):
 
 @pytest.mark.asyncio
 async def test_request_expect_no_reply(app):
-    assert await _test_request(app, False, False, tries=2, timeout=0.1) is None
+    assert await _test_request(app, False, False, tries=2, timeout=0.1) is True
 
 
 @pytest.mark.asyncio
 async def test_request_no_reply(app):
-    with pytest.raises(asyncio.TimeoutError):
+    with pytest.raises(DeliveryError):
         await _test_request(app, False, True, tries=2, timeout=0.1)
 
 
+@pytest.mark.asyncio
+async def test_request_send_timeout(app):
+    with pytest.raises(DeliveryError):
+        await _test_request(app, False, True, send_timeout=True, tries=2, timeout=0.1)
+
+
+@pytest.mark.asyncio
+async def test_request_send_fail(app):
+    with pytest.raises(DeliveryError):
+        await _test_request(app, False, True, send_success=False, tries=2, timeout=0.1)
+
+
 def _handle_reply(app, tsn):
     app.handle_message = mock.MagicMock()
     return app._handle_reply(

zigpy_xbee/api.py

@@ -4,6 +4,7 @@
 import functools
 import logging
 
+from zigpy.exceptions import DeliveryError
 from zigpy.types import LVList
 
 from . import uart
@@ -35,14 +36,14 @@ class ModemStatus(t.uint8_t, t.UndefinedEnum):
     'queued_at': (0x09, (t.FrameId, t.ATCommand, t.Bytes), 0x88),
     'remote_at': (0x17, (t.FrameId, t.EUI64, t.NWK, t.uint8_t, t.ATCommand, t.Bytes), 0x97),
     'tx': (0x10, (), None),
-    'tx_explicit': (0x11, (t.FrameId, t.EUI64, t.NWK, t.uint8_t, t.uint8_t, t.uint16_t, t.uint16_t, t.uint8_t, t.uint8_t, t.Bytes), None),
+    'tx_explicit': (0x11, (t.FrameId, t.EUI64, t.NWK, t.uint8_t, t.uint8_t, t.uint16_t, t.uint16_t, t.uint8_t, t.uint8_t, t.Bytes), 0x8b),
     'create_source_route': (0x21, (t.FrameId, t.EUI64, t.NWK, t.uint8_t, LVList(t.NWK)), None),
     'register_joining_device': (0x24, (), None),
 }
 COMMAND_RESPONSES = {
     'at_response': (0x88, (t.FrameId, t.ATCommand, t.uint8_t, t.Bytes), None),
     'modem_status': (0x8A, (ModemStatus, ), None),
-    'tx_status': (0x8B, (t.FrameId, t.NWK, t.uint8_t, t.uint8_t, t.uint8_t), None),
+    'tx_status': (0x8B, (t.FrameId, t.NWK, t.uint8_t, t.TXStatus, t.DiscoveryStatus), None),
     'route_information': (0x8D, (), None),
     'rx': (0x90, (), None),
     'explicit_rx_indicator': (0x91, (t.EUI64, t.NWK, t.uint8_t, t.uint8_t, t.uint16_t, t.uint16_t, t.uint8_t, t.Bytes), None),
@@ -269,15 +270,15 @@ def frame_received(self, data):
         LOGGER.debug("Frame received: %s", command)
         data, rest = t.deserialize(data[1:], COMMAND_RESPONSES[command][1])
         try:
-            getattr(self, '_handle_%s' % (command, ))(data)
+            getattr(self, '_handle_%s' % (command, ))(*data)
         except AttributeError:
             LOGGER.error("No '%s' handler. Data: %s", command,
                          binascii.hexlify(data))
 
-    def _handle_at_response(self, data):
-        fut, = self._awaiting.pop(data[0])
+    def _handle_at_response(self, frame_id, cmd, status, value):
+        fut, = self._awaiting.pop(frame_id)
         try:
-            status = ATCommandResult(data[2])
+            status = ATCommandResult(status)
         except ValueError:
             status = ATCommandResult.ERROR
 
@@ -286,25 +287,26 @@ def _handle_at_response(self, data):
                 RuntimeError("AT Command response: {}".format(status.name)))
             return
 
-        response_type = AT_COMMANDS[data[1].decode('ascii')]
-        if response_type is None or len(data[3]) == 0:
+        response_type = AT_COMMANDS[cmd.decode('ascii')]
+        if response_type is None or len(value) == 0:
             fut.set_result(None)
             return
 
-        response, remains = response_type.deserialize(data[3])
+        response, remains = response_type.deserialize(value)
         fut.set_result(response)
 
-    def _handle_remote_at_response(self, data):
+    def _handle_remote_at_response(self, frame_id, ieee, nwk, cmd, status, value):
         """Remote AT command response."""
-        LOGGER.debug("Remote AT command response: %s", data)
-        return self._handle_at_response((data[0], data[3], data[4], data[5]))
+        LOGGER.debug("Remote AT command response from: %s",
+                     (frame_id, ieee, nwk, cmd, status, value))
+        return self._handle_at_response(frame_id, cmd, status, value)
 
-    def _handle_many_to_one_rri(self, data):
-        LOGGER.debug("_handle_many_to_one_rri: %s", data)
+    def _handle_many_to_one_rri(self, ieee, nwk, reserved):
+        LOGGER.debug("_handle_many_to_one_rri: %s", (ieee, nwk, reserved))
 
-    def _handle_modem_status(self, data):
-        LOGGER.debug("Handle modem status frame: %s", data)
-        status = data[0]
+    def _handle_modem_status(self, status):
+        LOGGER.debug("Handle modem status frame: %s", status)
+        status = status
         if status == ModemStatus.COORDINATOR_STARTED:
             self.coordinator_started_event.set()
         elif status in (ModemStatus.HARDWARE_RESET, ModemStatus.WATCHDOG_TIMER_RESET):
@@ -316,16 +318,38 @@ def _handle_modem_status(self, data):
         if self._app:
             self._app.handle_modem_status(status)
 
-    def _handle_explicit_rx_indicator(self, data):
-        LOGGER.debug("_handle_explicit_rx: opts=%s", data[6])
-        self._app.handle_rx(*data)
+    def _handle_explicit_rx_indicator(self, ieee, nwk, src_ep,
+                                      dst_ep, cluster, profile, rx_opts, data):
+        LOGGER.debug("_handle_explicit_rx: %s",
+                     (ieee, nwk, dst_ep, cluster, rx_opts,
+                      binascii.hexlify(data)))
+        self._app.handle_rx(ieee, nwk, src_ep, dst_ep, cluster,
+                            profile, rx_opts, data)
 
-    def _handle_route_record_indicator(self, data):
+    def _handle_route_record_indicator(self, ieee, src, rx_opts, hops):
         """Handle Route Record indicator from a device."""
-        LOGGER.debug("_handle_route_record_indicator: %s", data)
+        LOGGER.debug("_handle_route_record_indicator: %s",
+                     (ieee, src, rx_opts, hops))
 
-    def _handle_tx_status(self, data):
-        LOGGER.debug("tx_status: %s", data)
+    def _handle_tx_status(self, frame_id, nwk, tries, tx_status, dsc_status):
+        LOGGER.debug(
+            ("tx_explicit to 0x%04x: %s after %i tries. Discovery Status: %s,"
+             " Frame #%i"), nwk, tx_status, tries, dsc_status, frame_id)
+        try:
+            fut, = self._awaiting.pop(frame_id)
+        except KeyError:
+            LOGGER.debug("unexpected tx_status report received")
+            return
+
+        try:
+            if tx_status in (t.TXStatus.SUCCESS,
+                             t.TXStatus.BROADCAST_APS_TX_ATTEMPT):
+                fut.set_result(tx_status)
+            else:
+                fut.set_exception(
+                    DeliveryError('%s' % (tx_status, )))
+        except asyncio.InvalidStateError as ex:
+            LOGGER.debug("duplicate tx_status for %s nwk? State: %s", nwk, ex)
 
     def set_application(self, app):
         self._app = app