#873 Add write_attributes_raw method (#887)

* Add write_attributes_raw method
Enables writing attributes that are not predefined.
Similar to read_attributes_raw

* Alphabetical order for radios

* Better argument name for write_attributes_raw

* Need await on write_attributes_raw call

* Renamed other occurences of 'args'

* Try multiple mock patch to catch uppermost exception

* Adjusted test to detect exception, also test other method

* Removed dual mock test

* Renamed args to attrs in write_attributes

README.md

@@ -60,8 +60,8 @@ Older packages of tagged versions are still available on the "zigpy-homeassistan
 Packages of tagged versions of the radio libraries are released via separate projects on PyPI
 - https://pypi.org/project/zigpy/
   - https://pypi.org/project/bellows/
-  - https://pypi.org/project/zigpy-deconz/
   - https://pypi.org/project/zigpy-cc/
+  - https://pypi.org/project/zigpy-deconz/
   - https://pypi.org/project/zigpy-xbee/
   - https://pypi.org/project/zigpy-zigate/
   - https://pypi.org/project/zigpy-znp/

tests/test_zcl.py

@@ -388,12 +388,28 @@ async def test_write_wrong_attribute(cluster):
         assert cluster._write_attributes.call_count == 1
 
 
+async def test_write_unknown_attribute(cluster):
+    with patch.object(cluster, "_write_attributes", new=AsyncMock()):
+        with pytest.raises(KeyError):
+            # Using an invalid attribute name, the call should fail
+            await cluster.write_attributes({"dummy_attribute": 5})
+        assert cluster._write_attributes.call_count == 0
+
+
 async def test_write_attributes_wrong_type(cluster):
     with patch.object(cluster, "_write_attributes", new=AsyncMock()):
         await cluster.write_attributes({18: 0x2222})
         assert cluster._write_attributes.call_count == 1
 
 
+async def test_write_attributes_raw(cluster):
+    with patch.object(cluster, "_write_attributes", new=AsyncMock()):
+        # write_attributes_raw does not check the attributes,
+        # send to unknown attribute in cluster, the write should be effective
+        await cluster.write_attributes_raw({0: 5, 0x3000: 5})
+        assert cluster._write_attributes.call_count == 1
+
+
 @pytest.mark.parametrize(
     "cluster_id, attr, value, serialized",
     (

zigpy/zcl/__init__.py

@@ -374,18 +374,25 @@ def _write_attr_records(
     async def write_attributes(
         self, attributes: Dict[Union[str, int], Any], manufacturer: Optional[int] = None
     ) -> List:
-        args = self._write_attr_records(attributes)
-        result = await self._write_attributes(args, manufacturer=manufacturer)
+        """Write attributes to device with internal 'attributes' validation"""
+        attrs = self._write_attr_records(attributes)
+        return await self.write_attributes_raw(attrs, manufacturer)
+
+    async def write_attributes_raw(
+        self, attrs: List[foundation.Attribute], manufacturer: Optional[int] = None
+    ) -> List:
+        """Write attributes to device without internal 'attributes' validation"""
+        result = await self._write_attributes(attrs, manufacturer=manufacturer)
         if not isinstance(result[0], list):
             return result
 
         records = result[0]
         if len(records) == 1 and records[0].status == foundation.Status.SUCCESS:
-            for attr_rec in args:
+            for attr_rec in attrs:
                 self._attr_cache[attr_rec.attrid] = attr_rec.value.value
         else:
             failed = [rec.attrid for rec in records]
-            for attr_rec in args:
+            for attr_rec in attrs:
                 if attr_rec.attrid not in failed:
                     self._attr_cache[attr_rec.attrid] = attr_rec.value.value