Added tests for GATT services

Author: aefarrell

.github/workflows/python-package.yml

@@ -27,7 +27,7 @@ jobs:
     - name: Install dependencies
       run: |
         python -m pip install --upgrade pip
-        python -m pip install flake8 pytest pytest-cov
+        python -m pip install flake8 pytest pytest-cov pytest-asyncio
         python -m pip install -e .
     - name: Lint with flake8
       run: |

atmotube/__init__.py

@@ -1,6 +1,7 @@
-from .gatt_notify import (InvalidAtmoTubeService,
-                          start_gatt_notifications,
-                          get_available_characteristics)
+from .gatt import (InvalidAtmoTubeService,
+                   gatt_notify,
+                   start_gatt_notifications,
+                   get_available_characteristics)
 from .packets import (InvalidByteData,
                       AtmoTubePacket,
                       StatusPacket,

atmotube/gatt_notify.py atmotube/gatt.pyatmotube/gatt_notify.py renamed to atmotube/gatt.py

@@ -1,64 +1,102 @@
-from unittest.mock import Mock, patch
+import pytest
+from unittest.mock import ANY, AsyncMock, Mock
+from bleak import BleakClient
 
 from atmotube import (
     AtmoTube_Service_UUID,
     AtmoTube_PRO_UUID,
+    StatusPacket,
+    SPS30Packet,
+    BME280Packet,
+    SGPC3Packet,
     InvalidAtmoTubeService,
-    get_available_characteristics)
+    get_available_characteristics,
+    gatt_notify)
 
-from atmotube.gatt_notify import ALL_PACKETS
+ALL_PACKETS = [(AtmoTube_PRO_UUID.STATUS, StatusPacket),
+               (AtmoTube_PRO_UUID.SPS30, SPS30Packet),
+               (AtmoTube_PRO_UUID.BME280, BME280Packet),
+               (AtmoTube_PRO_UUID.SGPC3, SGPC3Packet)]
 
 
 def test_get_available_characteristics_all():
-    with patch("bleak.BleakClient", new_callable=Mock) as MockClient:
-        mock_client = MockClient.return_value
-        mock_client.services.get_service.return_value = Mock(
-            uuid=AtmoTube_Service_UUID.PRO,
-            characteristics=[
-                Mock(uuid=AtmoTube_PRO_UUID.SGPC3),
-                Mock(uuid=AtmoTube_PRO_UUID.BME280),
-                Mock(uuid=AtmoTube_PRO_UUID.STATUS),
-                Mock(uuid=AtmoTube_PRO_UUID.SPS30),
-            ]
-        )
+    client = Mock(spec=BleakClient)
+    client.services.get_service.return_value = Mock(
+        uuid=AtmoTube_Service_UUID.PRO,
+        characteristics=[
+            Mock(uuid=AtmoTube_PRO_UUID.SGPC3),
+            Mock(uuid=AtmoTube_PRO_UUID.BME280),
+            Mock(uuid=AtmoTube_PRO_UUID.STATUS),
+            Mock(uuid=AtmoTube_PRO_UUID.SPS30),
+        ]
+    )
 
-        # Call the function
-        packet_list = get_available_characteristics(mock_client)
+    # Call the function
+    packet_list = get_available_characteristics(client)
 
-        # Assertions
-        assert packet_list == ALL_PACKETS
+    # Assertions
+    assert packet_list == ALL_PACKETS
 
 
 def test_get_available_characteristics_no_pm():
-    with patch("bleak.BleakClient", new_callable=Mock) as MockClient:
-        mock_client = MockClient.return_value
-        mock_client.services.get_service.return_value = Mock(
-            uuid=AtmoTube_Service_UUID.PRO,
-            characteristics=[
-                Mock(uuid=AtmoTube_PRO_UUID.SGPC3),
-                Mock(uuid=AtmoTube_PRO_UUID.BME280),
-                Mock(uuid=AtmoTube_PRO_UUID.STATUS),
-            ]
-        )
+    client = Mock(spec=BleakClient)
+    client.services.get_service.return_value = Mock(
+        uuid=AtmoTube_Service_UUID.PRO,
+        characteristics=[
+            Mock(uuid=AtmoTube_PRO_UUID.SGPC3),
+            Mock(uuid=AtmoTube_PRO_UUID.BME280),
+            Mock(uuid=AtmoTube_PRO_UUID.STATUS),
+        ]
+    )
 
-        # Call the function
-        packet_list = get_available_characteristics(mock_client)
+    # Call the function
+    packet_list = get_available_characteristics(client)
 
-        # Assertions
-        assert packet_list == [packet for packet in ALL_PACKETS
-                               if packet[0] != AtmoTube_PRO_UUID.SPS30]
+    # Assertions
+    assert packet_list == [packet for packet in ALL_PACKETS
+                           if packet[0] != AtmoTube_PRO_UUID.SPS30]
 
 
 def test_get_available_characteristics_invalid_service():
-    with patch("bleak.BleakClient", new_callable=Mock) as MockClient:
-        mock_client = MockClient.return_value
-        mock_client.services.get_service.return_value = None
-
-        # Call the function and expect an exception
-        try:
-            get_available_characteristics(mock_client)
-        except Exception as e:
-            assert isinstance(e, InvalidAtmoTubeService)
-            assert str(e) == "AtmoTube Pro service not found"
-        else:
-            assert False, "Expected exception was not raised"
+    client = Mock(spec=BleakClient)
+    client.services.get_service.return_value = None
+
+    # Call the function and expect an exception
+    try:
+        get_available_characteristics(client)
+    except Exception as e:
+        assert isinstance(e, InvalidAtmoTubeService)
+        assert str(e) == "AtmoTube Pro service not found"
+    else:
+        assert False, "Expected exception was not raised"
+
+
+class MockPacket:
+    def __init__(self, data):
+        self.data = data
+
+
+@pytest.mark.asyncio
+async def test_gatt_notify_async_callback():
+    uuid = "example-uuid"
+    client = AsyncMock(spec=BleakClient)
+    callback = AsyncMock()
+
+    # Call gatt_notify, and await the returned task
+    task = gatt_notify(client, uuid, MockPacket, callback)
+    await task
+
+    # Assert that start_notify was called with the uuid and something tbd
+    client.start_notify.assert_called_once_with(uuid, ANY)
+
+    # Dummy notification being sent
+    packet_callback = client.start_notify.call_args[0][1]
+    notification_data = bytearray([0, 1, 2, 3])
+    await packet_callback(None, notification_data)
+
+    # Assert that the original callback is called with one argument
+    # and that argument is of the correct instance and holds the data
+    callback.assert_awaited_once()
+    callback.assert_called_once_with(ANY)
+    assert isinstance(callback.call_args.args[0], MockPacket)
+    assert callback.call_args.args[0].data == notification_data