Merge pull request #928 from puddly/puddly/bugfix/quirk-manuf-id-over…

…ride

Allow disabling the automatic manufacturer ID override in quirks

tests/test_quirks.py

@@ -906,3 +906,54 @@ class BadCluster(zigpy.quirks.CustomCluster):
             manufacturer_server_commands = {
                 0x1234: ("foo3", {}, False),
             }
+
+
+async def test_manuf_id_disable(real_device):
+    class TestCluster(ManufacturerSpecificCluster):
+        cluster_id = 0xFF00
+
+    real_device.manufacturer_id_override = 0x1234
+
+    ep = real_device.endpoints[1]
+    ep.add_input_cluster(TestCluster.cluster_id, TestCluster(ep))
+    assert isinstance(ep.just_a_cluster, TestCluster)
+
+    assert ep.manufacturer_id == 0x1234
+
+    # The default behavior for a manufacturer-specific cluster, command, or attribute is
+    # to include the manufacturer ID in the request
+    with patch.object(ep, "request", AsyncMock()) as request_mock:
+        request_mock.return_value = (zcl.foundation.Status.SUCCESS, "done")
+        await ep.just_a_cluster.command(
+            ep.just_a_cluster.commands_by_name["server_cmd0"].id,
+        )
+        await ep.just_a_cluster.read_attributes(["attr0"])
+        await ep.just_a_cluster.write_attributes({"attr0": 1})
+
+    assert len(request_mock.mock_calls) == 3
+
+    for mock_call in request_mock.mock_calls:
+        data = mock_call[1][2]
+        hdr, _ = zcl.foundation.ZCLHeader.deserialize(data)
+        assert hdr.manufacturer == 0x1234
+
+    # But it can be disabled by passing NO_MANUFACTURER_ID
+    with patch.object(ep, "request", AsyncMock()) as request_mock:
+        request_mock.return_value = (zcl.foundation.Status.SUCCESS, "done")
+        await ep.just_a_cluster.command(
+            ep.just_a_cluster.commands_by_name["server_cmd0"].id,
+            manufacturer=zcl.foundation.ZCLHeader.NO_MANUFACTURER_ID,
+        )
+        await ep.just_a_cluster.read_attributes(
+            ["attr0"], manufacturer=zcl.foundation.ZCLHeader.NO_MANUFACTURER_ID
+        )
+        await ep.just_a_cluster.write_attributes(
+            {"attr0": 1}, manufacturer=zcl.foundation.ZCLHeader.NO_MANUFACTURER_ID
+        )
+
+    assert len(request_mock.mock_calls) == 3
+
+    for mock_call in request_mock.mock_calls:
+        data = mock_call[1][2]
+        hdr, _ = zcl.foundation.ZCLHeader.deserialize(data)
+        assert hdr.manufacturer is None

tests/test_zcl_foundation.py

@@ -340,6 +340,23 @@ def test_frame_header_cluster():
     assert hdr.frame_control.is_manufacturer_specific is False
 
 
+def test_frame_header_disable_manufacturer_id():
+    """Test frame header manufacturer ID can be disabled with NO_MANUFACTURER_ID."""
+
+    hdr = foundation.ZCLHeader.cluster(tsn=123, command_id=0x12, manufacturer=None)
+    assert hdr.manufacturer is None
+    hdr.manufacturer = 0x1234
+    assert hdr.manufacturer == 0x1234
+
+    hdr.manufacturer = foundation.ZCLHeader.NO_MANUFACTURER_ID
+    assert hdr.manufacturer is None
+
+    hdr2 = foundation.ZCLHeader.cluster(
+        tsn=123, command_id=0x12, manufacturer=foundation.ZCLHeader.NO_MANUFACTURER_ID
+    )
+    assert hdr2.manufacturer is None
+
+
 def test_data_types():
     """Test data types mappings."""
     assert len(foundation.DATA_TYPES) == len(foundation.DATA_TYPES._idx_by_class)

zigpy/zcl/foundation.py

@@ -2,7 +2,7 @@
 
 import dataclasses
 import keyword
-from typing import Any
+import typing
 import warnings
 
 import zigpy.types as t
@@ -516,6 +516,8 @@ def is_general(self) -> bool:
 
 
 class ZCLHeader(t.Struct):
+    NO_MANUFACTURER_ID = -1  # type: typing.Literal
+
     frame_control: FrameControl
     manufacturer: t.uint16_t = t.StructField(
         requires=lambda hdr: hdr.frame_control.is_manufacturer_specific
@@ -526,6 +528,10 @@ class ZCLHeader(t.Struct):
     def __new__(
         cls, frame_control=None, manufacturer=None, tsn=None, command_id=None
     ) -> ZCLHeader:
+        # Allow "auto manufacturer ID" to be disabled in higher layers
+        if manufacturer is cls.NO_MANUFACTURER_ID:
+            manufacturer = None
+
         if frame_control is not None and manufacturer is not None:
             frame_control.is_manufacturer_specific = True
 
@@ -537,6 +543,9 @@ def is_reply(self) -> bool:
         return self.frame_control.is_reply == 1
 
     def __setattr__(self, name, value) -> None:
+        if name == "manufacturer" and value is self.NO_MANUFACTURER_ID:
+            value = None
+
         super().__setattr__(name, value)
 
         if name == "manufacturer" and self.frame_control is not None:
@@ -820,7 +829,7 @@ class GeneralCommand(t.enum8):
     ).with_compiled_schema()
 
 
-def __getattr__(name: str) -> Any:
+def __getattr__(name: str) -> typing.Any:
     if name == "Command":
         warnings.warn(
             f"`{__name__}.Command` has been renamed to `{__name__}.GeneralCommand",