Merge pull request #173 from EasyRiderr/fix_add_gatt

Fix add gatt by adding new **primary** services

Author: kevincar

bless/backends/server.py

@@ -309,7 +309,7 @@ async def add_gatt(self, gatt_tree: Dict):
             uuids and the attributes are the properties
         """
         for service_uuid, service_info in gatt_tree.items():
-            await self.add_new_service(service_uuid)
+            await self.add_new_service(service_uuid, primary=True)
             for char_uuid, char_info in service_info.items():
                 await self.add_new_characteristic(
                     service_uuid,

test/backends/test_server.py

@@ -66,43 +66,7 @@ def hex_to_byte(self, hexstr: str) -> bytearray:
     def byte_to_hex(self, b: bytearray) -> str:
         return "".join([hex(x)[2:] for x in b]).upper()
 
-    @pytest.mark.asyncio
-    async def test_server(self):
-        # Initialize
-        server: BlessServer = BlessServer("Test Server")
-
-        # setup a service
-        service_uuid: str = str(uuid.uuid4())
-        await server.add_new_service(service_uuid)
-
-        assert len(server.services) > 0
-        # print(server.services)
-
-        # setup a characteristic for the service
-        char_uuid: str = str(uuid.uuid4())
-        char_flags: GATTCharacteristicProperties = (
-            GATTCharacteristicProperties.read
-            | GATTCharacteristicProperties.write
-            | GATTCharacteristicProperties.notify
-        )
-        value: Optional[bytearray] = None
-        permissions: GATTAttributePermissions = (
-            GATTAttributePermissions.readable | GATTAttributePermissions.writable
-        )
-
-        if use_encrypted:
-            print("\nEncryption has been enabled, ensure that you are bonded")
-            permissions = (
-                GATTAttributePermissions.read_encryption_required
-                | GATTAttributePermissions.write_encryption_required
-            )
-
-        await server.add_new_characteristic(
-            service_uuid, char_uuid, char_flags, value, permissions
-        )
-
-        assert server.services[service_uuid].get_characteristic(char_uuid)
-
+    async def setup_callbacks_and_start(self, server: BlessServer) -> None:
         # Set up read, write, and subscribe callbacks
         def read(
             characteristic: BlessGATTCharacteristic, request: BlessGATTRequest
@@ -112,7 +76,7 @@ def read(
 
         def write(
             characteristic: BlessGATTCharacteristic,
-            value: bytearray,
+            value: bytes,
             request: BlessGATTRequest,
         ) -> None:
             print(f"Write request: {request}")
@@ -140,6 +104,12 @@ def unsubscribe(
 
         assert await server.is_advertising() is True
 
+    async def read_write_notify_tests_on(
+        self,
+        server: BlessServer,
+        svc_uuid: str,
+        char_uuid: str
+    ) -> None:
         # Subscribe
         assert await server.is_connected() is False
 
@@ -175,16 +145,103 @@ def unsubscribe(
         print("A new value will be notified on the phone")
         await aioconsole.ainput("Press enter to receive the new value...")
 
-        server.update_value(service_uuid, char_uuid)
+        server.update_value(svc_uuid, char_uuid)
 
         new_value: str = await aioconsole.ainput("Enter the new value: ")
         assert new_value == hex_val
 
         # unsubscribe
         print("Unsubscribe from the characteristic")
-        await aioconsole.ainput("Press entery when ready...")
+        await aioconsole.ainput("Press enter when ready...")
         assert await server.is_connected() is False
 
+    @pytest.mark.asyncio
+    async def test_server(self):
+        # Initialize
+        server: BlessServer = BlessServer("Test Server")
+
+        # setup a service
+        service_uuid: str = str(uuid.uuid4())
+        await server.add_new_service(service_uuid)
+
+        assert len(server.services) > 0
+        # print(server.services)
+
+        # setup a characteristic for the service
+        char_uuid: str = str(uuid.uuid4())
+        char_flags: GATTCharacteristicProperties = (
+            GATTCharacteristicProperties.read
+            | GATTCharacteristicProperties.write
+            | GATTCharacteristicProperties.notify
+        )
+        value: Optional[bytearray] = None
+        permissions: GATTAttributePermissions = (
+            GATTAttributePermissions.readable | GATTAttributePermissions.writable
+        )
+
+        if use_encrypted:
+            print("\nEncryption has been enabled, ensure that you are bonded")
+            permissions = (
+                GATTAttributePermissions.read_encryption_required
+                | GATTAttributePermissions.write_encryption_required
+            )
+
+        await server.add_new_characteristic(
+            service_uuid, char_uuid, char_flags, value, permissions
+        )
+
+        assert server.services[service_uuid].get_characteristic(char_uuid)
+
+        await self.setup_callbacks_and_start(server)
+        await self.read_write_notify_tests_on(server, service_uuid, char_uuid)
+
+        # Stop Advertising
+        await server.stop()
+        await asyncio.sleep(2)
+        assert await server.is_advertising() is False
+
+    @pytest.mark.asyncio
+    async def test_server_add_gatt(self):
+        # Initialize
+        server: BlessServer = BlessServer("Test Server with a GATT Tree")
+
+        # setup a gatt tree
+        service1_uuid: str = str(uuid.uuid4())
+        service2_uuid: str = str(uuid.uuid4())
+        characteristic1_uuid: str = str(uuid.uuid4())
+        characteristic2_uuid: str = str(uuid.uuid4())
+        gatt_tree = {
+            service1_uuid: {
+                characteristic1_uuid: {
+                    "Properties": GATTCharacteristicProperties.read
+                    | GATTCharacteristicProperties.write
+                    | GATTCharacteristicProperties.notify,
+                    "Value": None,
+                    "Permissions": GATTAttributePermissions.readable
+                    | GATTAttributePermissions.writable,
+                }
+            },
+            service2_uuid: {
+                characteristic2_uuid: {
+                    "Properties": GATTCharacteristicProperties.read
+                    | GATTCharacteristicProperties.write
+                    | GATTCharacteristicProperties.notify,
+                    "Value": None,
+                    "Permissions": GATTAttributePermissions.readable
+                    | GATTAttributePermissions.writable,
+                }
+            }
+        }
+
+        await server.add_gatt(gatt_tree)
+        assert len(server.services) == 2
+        await self.setup_callbacks_and_start(server)
+        await self.read_write_notify_tests_on(
+            server,
+            service2_uuid,
+            characteristic2_uuid
+        )
+
         # Stop Advertising
         await server.stop()
         await asyncio.sleep(2)