[RFC] - Leverage device diagnostics files for tests (#181)

* initial crack at device from diagnostics file

* device diag file

* junk test to test device from diag file

* update fixtures

* add devices test

* device files

* assert entity counts and test identify

* cleanup and prime unsupported attributes

* update device files

* remove junk test

* .

* tweak test and add more devices

* att more devices and extend KoF tweak

* Migrate to `pathlib`

* Exclude the only device causing test failures

* Remove device changed due to quirk

---------

Co-authored-by: puddly <[email protected]>

tests/conftest.py

@@ -1,9 +1,10 @@
 """Test configuration for the ZHA component."""
 
 import asyncio
-from collections.abc import Awaitable, Callable, Generator
-from contextlib import contextmanager
+from collections.abc import Awaitable, Callable, Coroutine, Generator
+from contextlib import contextmanager, suppress
 import itertools
+import json
 import logging
 import os
 import reprlib
@@ -474,3 +475,161 @@ def _mock_dev(
         return device
 
     return _mock_dev
+
+
+@pytest.fixture
+def zigpy_device_from_json(
+    zigpy_app_controller: ControllerApplication,
+) -> Callable[..., zigpy.device.Device]:
+    """Make a fake device using the specified cluster classes."""
+
+    def _mock_dev_from_json(
+        json_file: str,
+        patch_cluster: bool = True,
+        quirk: Optional[Callable] = None,
+    ) -> zigpy.device.Device:
+        """Make a fake device using the specified cluster classes."""
+        with open(json_file, encoding="utf-8") as file:
+            device_data = json.load(file)
+
+        ieee = zigpy.types.EUI64.convert(device_data["ieee"])
+        nwk = device_data["nwk"]
+        manufacturer = device_data["manufacturer"]
+        model = device_data["model"]
+        node_descriptor = device_data["signature"]["node_descriptor"]
+        endpoints = device_data["signature"]["endpoints"]
+        cluster_data = device_data["cluster_details"]
+
+        device = zigpy.device.Device(zigpy_app_controller, ieee, nwk)
+        device.manufacturer = manufacturer
+        device.model = model
+
+        node_desc = zdo_t.NodeDescriptor(
+            logical_type=node_descriptor["logical_type"],
+            complex_descriptor_available=node_descriptor[
+                "complex_descriptor_available"
+            ],
+            user_descriptor_available=node_descriptor["user_descriptor_available"],
+            reserved=node_descriptor["reserved"],
+            aps_flags=node_descriptor["aps_flags"],
+            frequency_band=node_descriptor["frequency_band"],
+            mac_capability_flags=node_descriptor["mac_capability_flags"],
+            manufacturer_code=node_descriptor["manufacturer_code"],
+            maximum_buffer_size=node_descriptor["maximum_buffer_size"],
+            maximum_incoming_transfer_size=node_descriptor[
+                "maximum_incoming_transfer_size"
+            ],
+            server_mask=node_descriptor["server_mask"],
+            maximum_outgoing_transfer_size=node_descriptor[
+                "maximum_outgoing_transfer_size"
+            ],
+            descriptor_capability_field=node_descriptor["descriptor_capability_field"],
+        )
+        device.node_desc = node_desc
+        device.last_seen = time.time()
+
+        orig_endpoints = (
+            device_data["original_signature"]["endpoints"]
+            if "original_signature" in device_data
+            else endpoints
+        )
+        for epid, ep in orig_endpoints.items():
+            endpoint = device.add_endpoint(int(epid))
+            profile = None
+            with suppress(Exception):
+                profile = zigpy.profiles.PROFILES[int(ep["profile_id"], 16)]
+
+            endpoint.device_type = (
+                profile.DeviceType(int(ep["device_type"], 16))
+                if profile
+                else int(ep["device_type"], 16)
+            )
+            endpoint.profile_id = (
+                profile.PROFILE_ID if profile else int(ep["profile_id"], 16)
+            )
+            endpoint.request = AsyncMock(return_value=[0])
+
+            for cluster_id in ep["input_clusters"]:
+                endpoint.add_input_cluster(int(cluster_id, 16))
+
+            for cluster_id in ep["output_clusters"]:
+                endpoint.add_output_cluster(int(cluster_id, 16))
+
+        if quirk:
+            device = quirk(zigpy_app_controller, device.ieee, device.nwk, device)
+        else:
+            device = get_device(device)
+
+        for epid, ep in cluster_data.items():
+            endpoint.request = AsyncMock(return_value=[0])
+            for cluster_id, cluster in ep["in_clusters"].items():
+                real_cluster = device.endpoints[int(epid)].in_clusters[
+                    int(cluster_id, 16)
+                ]
+                if patch_cluster:
+                    common.patch_cluster(real_cluster)
+                for attr_id, attr in cluster["attributes"].items():
+                    if (
+                        attr["value"] is None
+                        or attr_id in cluster["unsupported_attributes"]
+                    ):
+                        continue
+                    real_cluster._attr_cache[int(attr_id, 16)] = attr["value"]
+                    real_cluster.PLUGGED_ATTR_READS[int(attr_id, 16)] = attr["value"]
+                for unsupported_attr in cluster["unsupported_attributes"]:
+                    if isinstance(
+                        unsupported_attr, str
+                    ) and unsupported_attr.startswith("0x"):
+                        attrid = int(unsupported_attr, 16)
+                        real_cluster.unsupported_attributes.add(attrid)
+                        if attrid in real_cluster.attributes:
+                            real_cluster.unsupported_attributes.add(
+                                real_cluster.attributes[attrid].name
+                            )
+                    else:
+                        real_cluster.unsupported_attributes.add(unsupported_attr)
+
+            for cluster_id, cluster in ep["out_clusters"].items():
+                real_cluster = device.endpoints[int(epid)].out_clusters[
+                    int(cluster_id, 16)
+                ]
+                if patch_cluster:
+                    common.patch_cluster(real_cluster)
+                for attr_id, attr in cluster["attributes"].items():
+                    if (
+                        attr["value"] is None
+                        or attr_id in cluster["unsupported_attributes"]
+                    ):
+                        continue
+                    real_cluster._attr_cache[int(attr_id, 16)] = attr["value"]
+                    real_cluster.PLUGGED_ATTR_READS[int(attr_id, 16)] = attr["value"]
+                for unsupported_attr in cluster["unsupported_attributes"]:
+                    if isinstance(
+                        unsupported_attr, str
+                    ) and unsupported_attr.startswith("0x"):
+                        attrid = int(unsupported_attr, 16)
+                        real_cluster.unsupported_attributes.add(attrid)
+                        if attrid in real_cluster.attributes:
+                            real_cluster.unsupported_attributes.add(
+                                real_cluster.attributes[attrid].name
+                            )
+                    else:
+                        real_cluster.unsupported_attributes.add(unsupported_attr)
+
+        return device
+
+    return _mock_dev_from_json
+
+
+@pytest.fixture
+def zha_device_from_file(
+    zigpy_device_from_json: Callable[..., zigpy.device.Device],
+    device_joined: Callable[[zigpy.device.Device], Awaitable[Device]],
+) -> Callable[[str], Coroutine[Any, Any, Device]]:
+    """Return a ZHA device from a JSON file."""
+
+    async def _zha_device_from_file(file_path: str) -> Device:
+        zigpy_dev = zigpy_device_from_json(file_path)
+        return await device_joined(zigpy_dev)
+
+    return _zha_device_from_file