| # | |
| # Copyright (c) 2025 Project CHIP Authors | |
| # All rights reserved. | |
| # | |
| # Licensed under the Apache License, Version 2.0 (the "License"); | |
| # you may not use this file except in compliance with the License. | |
| # You may obtain a copy of the License at | |
| # | |
| # http://www.apache.org/licenses/LICENSE-2.0 | |
| # | |
| # Unless required by applicable law or agreed to in writing, software | |
| # distributed under the License is distributed on an "AS IS" BASIS, | |
| # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | |
| # See the License for the specific language governing permissions and | |
| # limitations under the License. | |
| # | |
| import os | |
| import tempfile | |
| from mobly import asserts | |
| import matter.clusters as Clusters | |
| from matter.clusters.Attribute import AsyncReadTransaction | |
| from matter.testing.global_attribute_ids import GlobalAttributeIds | |
| from matter.testing.matter_testing import CertificationUnitTestNoDevice | |
| from matter.testing.pics import (BASE_PICS_CODES_DERIVED, BasePicsFacts, base_pics_facts_to_pics_codes, | |
| derive_base_pics_facts_from_device_wildcard, generate_device_element_pics_from_device_wildcard, | |
| read_pics_from_file) | |
| from matter.testing.runner import default_matter_test_main | |
| from matter.testing.spec_parsing import PrebuiltDataModelDirectory, build_xml_clusters | |
| class TestPicsHelpers(CertificationUnitTestNoDevice): | |
| def test_pics_generation(self): | |
| xml_cluster, _ = build_xml_clusters(PrebuiltDataModelDirectory.k1_4_1) | |
| # 0: opcreds with an attribute, an accepted command, a generated command | |
| # 0: cadmin with a feature, an attribute, an accepted command | |
| # 1: Door lock cluster with two features, two attributes, two accepted commands, two generated commands | |
| wildcard = AsyncReadTransaction.ReadResponse(attributes={}, events=[], tlvAttributes={}) | |
| desc = Clusters.Descriptor | |
| opcreds = Clusters.OperationalCredentials | |
| cadmin = Clusters.AdministratorCommissioning | |
| lock = Clusters.DoorLock | |
| wildcard.tlvAttributes[0] = {desc.id: {}, opcreds.id: {}, cadmin.id: {}} | |
| wildcard.tlvAttributes[1] = {lock.id: {}} | |
| wildcard.tlvAttributes[0][desc.id][desc.Attributes.AttributeList.attribute_id] = [ | |
| desc.Attributes.DeviceTypeList.attribute_id] | |
| wildcard.tlvAttributes[0][desc.id][desc.Attributes.AcceptedCommandList.attribute_id] = [] | |
| wildcard.tlvAttributes[0][desc.id][desc.Attributes.GeneratedCommandList.attribute_id] = [] | |
| wildcard.tlvAttributes[0][desc.id][desc.Attributes.FeatureMap.attribute_id] = 0 | |
| desc_expected = ['DESC.S', 'DESC.S.A0000'] | |
| # The rest of the processing is done through tlvAttributes, but this one needs to be parsed, so it comes from here. | |
| wildcard.attributes = {0: {desc: {desc.Attributes.DeviceTypeList: [ | |
| desc.Structs.DeviceTypeStruct(deviceType=0x16, revision=1)]}}} | |
| wildcard.tlvAttributes[0][opcreds.id][opcreds.Attributes.AttributeList.attribute_id] = [ | |
| opcreds.Attributes.NOCs.attribute_id] | |
| wildcard.tlvAttributes[0][opcreds.id][opcreds.Attributes.AcceptedCommandList.attribute_id] = [ | |
| opcreds.Commands.AttestationRequest.command_id, opcreds.Commands.AddTrustedRootCertificate.command_id] | |
| wildcard.tlvAttributes[0][opcreds.id][opcreds.Attributes.GeneratedCommandList.attribute_id] = [ | |
| opcreds.Commands.AttestationResponse.command_id] | |
| wildcard.tlvAttributes[0][opcreds.id][opcreds.Attributes.FeatureMap.attribute_id] = 0 | |
| opcreds_expected_suffix = ['A0000', 'C00.Rsp', 'C0b.Rsp', 'C01.Tx'] | |
| opcreds_expected = [f'OPCREDS.S.{s}' for s in opcreds_expected_suffix] | |
| opcreds_expected.append('OPCREDS.S') | |
| wildcard.tlvAttributes[0][cadmin.id][cadmin.Attributes.AttributeList.attribute_id] = [ | |
| cadmin.Attributes.AdminFabricIndex.attribute_id] | |
| wildcard.tlvAttributes[0][cadmin.id][cadmin.Attributes.AcceptedCommandList.attribute_id] = [ | |
| cadmin.Commands.OpenCommissioningWindow.command_id] | |
| wildcard.tlvAttributes[0][cadmin.id][cadmin.Attributes.GeneratedCommandList.attribute_id] = [] | |
| wildcard.tlvAttributes[0][cadmin.id][cadmin.Attributes.FeatureMap.attribute_id] = 1 | |
| cadmin_expected_suffix = ['A0001', 'C00.Rsp', 'F00'] | |
| cadmin_expected = [f'CADMIN.S.{s}' for s in cadmin_expected_suffix] | |
| cadmin_expected.append('CADMIN.S') | |
| wildcard.tlvAttributes[1][lock.id][lock.Attributes.AttributeList.attribute_id] = [ | |
| lock.Attributes.DoorState.attribute_id, lock.Attributes.LockType.attribute_id] | |
| wildcard.tlvAttributes[1][lock.id][lock.Attributes.AcceptedCommandList.attribute_id] = [ | |
| lock.Commands.GetUser.command_id, lock.Commands.GetCredentialStatus.command_id] | |
| wildcard.tlvAttributes[1][lock.id][lock.Attributes.GeneratedCommandList.attribute_id] = [ | |
| lock.Commands.GetUserResponse.command_id, lock.Commands.GetCredentialStatusResponse.command_id] | |
| wildcard.tlvAttributes[1][lock.id][lock.Attributes.FeatureMap.attribute_id] = 3 | |
| lock_expected_suffix = ['A0003', 'A0001', 'C1b.Rsp', 'C24.Rsp', 'C1c.Tx', 'C25.Tx', 'F00', 'F01'] | |
| lock_expected = [f'DRLK.S.{s}' for s in lock_expected_suffix] | |
| lock_expected.append('DRLK.S') | |
| def check_expected_pics(pics_list: dict[int, list[str]]): | |
| asserts.assert_equal(set(pics_list.keys()), {0, 1}, "Unexpected endpoints in PICS list") | |
| asserts.assert_equal(set(pics_list[0]), set(desc_expected + opcreds_expected + cadmin_expected + | |
| ['IDM.S', 'MCORE.ROLE.COMMISSIONEE']), "Incorrect PICS list on EP0") | |
| asserts.assert_equal(set(pics_list[1]), set(lock_expected + ['IDM.S']), "Incorrect PICS list on EP1") | |
| pics_list, problems = generate_device_element_pics_from_device_wildcard(wildcard, xml_cluster) | |
| asserts.assert_equal(len(problems), 0, "Unexpected problems found generating PICS list") | |
| check_expected_pics(pics_list) | |
| # Add globals, should be no errors, should not appear | |
| wildcard.tlvAttributes[1][lock.id][lock.Attributes.AttributeList.attribute_id].extend( | |
| [lock.Attributes.AttributeList.attribute_id, lock.Attributes.AcceptedCommandList.attribute_id, lock.Attributes.GeneratedCommandList.attribute_id, lock.Attributes.FeatureMap.attribute_id, lock.Attributes.ClusterRevision.attribute_id]) | |
| pics_list, problems = generate_device_element_pics_from_device_wildcard(wildcard, xml_cluster) | |
| asserts.assert_equal(len(problems), 0, "Unexpected problems found generating PICS list") | |
| check_expected_pics(pics_list) | |
| # Add MEI cluster, MEI attribute, MEI accepted command, MEI generated command - should be no errors, should not appear | |
| unit_testing = Clusters.UnitTesting | |
| wildcard.tlvAttributes[1][unit_testing.id] = {} | |
| # MEI cluster | |
| wildcard.tlvAttributes[1][unit_testing.id][unit_testing.Attributes.AttributeList.attribute_id] = [ | |
| unit_testing.Attributes.Bitmap16.attribute_id] | |
| wildcard.tlvAttributes[1][lock.id][lock.Attributes.AttributeList.attribute_id].append(0x60060000) | |
| wildcard.tlvAttributes[1][lock.id][lock.Attributes.AcceptedCommandList.attribute_id].append(0x60060000) | |
| wildcard.tlvAttributes[1][lock.id][lock.Attributes.GeneratedCommandList.attribute_id].append(0x60060001) | |
| pics_list, problems = generate_device_element_pics_from_device_wildcard(wildcard, xml_cluster) | |
| asserts.assert_equal(len(problems), 0, "Unexpected problems found generating PICS list") | |
| check_expected_pics(pics_list) | |
| # Add a standard cluster that's not part of the standard cluster set - this should cause an error because we don't know the PICS | |
| unknown_standard_cluster = max(xml_cluster.keys()) + 1 | |
| wildcard.tlvAttributes[1][unknown_standard_cluster] = {} | |
| wildcard.tlvAttributes[1][unknown_standard_cluster][unit_testing.Attributes.AttributeList.attribute_id] = [0] | |
| pics_list, problems = generate_device_element_pics_from_device_wildcard(wildcard, xml_cluster) | |
| asserts.assert_equal(len(problems), 1, "Unexpected problems found generating PICS list") | |
| check_expected_pics(pics_list) | |
| def test_base_pics_facts_to_pics_codes(self): | |
| # Empty facts must produce an empty code set so a non-commissioned, | |
| # non-server fact bundle does not accidentally mark anything. | |
| asserts.assert_equal(base_pics_facts_to_pics_codes(BasePicsFacts()), set(), | |
| "Empty BasePicsFacts must produce no PICS codes") | |
| # Every individual flag, mapped one at a time, must appear in the | |
| # tracked-codes set so callers can rely on BASE_PICS_CODES_DERIVED. | |
| cases = [ | |
| (BasePicsFacts(is_commissionee=True), "MCORE.ROLE.COMMISSIONEE"), | |
| (BasePicsFacts(is_server=True), "MCORE.IDM.S"), | |
| (BasePicsFacts(is_bridge=True), "MCORE.BRIDGE"), | |
| (BasePicsFacts(is_ota_requestor=True), "MCORE.OTA.Requestor"), | |
| (BasePicsFacts(is_ota_provider=True), "MCORE.OTA.Provider"), | |
| (BasePicsFacts(has_groups_on_multiple_endpoints=True), "MCORE.G.MULTIENDPOINT"), | |
| ] | |
| for facts, expected_code in cases: | |
| codes = base_pics_facts_to_pics_codes(facts) | |
| asserts.assert_equal(codes, {expected_code}, f"Expected only {expected_code}, got {codes}") | |
| asserts.assert_in(expected_code, BASE_PICS_CODES_DERIVED, | |
| f"{expected_code} must be in BASE_PICS_CODES_DERIVED so the test step iterates it") | |
| # All-on must produce all codes. | |
| all_on = BasePicsFacts( | |
| is_commissionee=True, is_server=True, is_bridge=True, | |
| is_ota_requestor=True, is_ota_provider=True, | |
| has_groups_on_multiple_endpoints=True) | |
| asserts.assert_equal(base_pics_facts_to_pics_codes(all_on), set(BASE_PICS_CODES_DERIVED), | |
| "All-on facts must produce the full tracked-codes set") | |
| def test_derive_base_pics_facts_from_device_wildcard(self): | |
| xml_cluster, _ = build_xml_clusters(PrebuiltDataModelDirectory.k1_4_1) | |
| desc = Clusters.Descriptor | |
| opcreds = Clusters.OperationalCredentials | |
| groups = Clusters.Groups | |
| ota_req = Clusters.OtaSoftwareUpdateRequestor | |
| # Minimal "root node plus bridged child with Groups on EP1 and EP2" wildcard. | |
| # EP0: root node device type, OpCreds (so endpoint has a server), | |
| # OTA Requestor (server-side). | |
| # EP1: Aggregator device type, Groups server. | |
| # EP2: Groups server. | |
| wildcard = AsyncReadTransaction.ReadResponse(attributes={}, events=[], tlvAttributes={}) | |
| wildcard.attributes[0] = { | |
| desc: {desc.Attributes.DeviceTypeList: [ | |
| desc.Structs.DeviceTypeStruct(deviceType=0x16, revision=1)]}, | |
| } | |
| wildcard.attributes[1] = { | |
| desc: {desc.Attributes.DeviceTypeList: [ | |
| desc.Structs.DeviceTypeStruct(deviceType=0x000E, revision=1)]}, | |
| } | |
| wildcard.attributes[2] = { | |
| desc: {desc.Attributes.DeviceTypeList: []}, | |
| } | |
| # Build minimal global-attribute fills for each cluster the helper | |
| # will visit. Empty attribute/command lists are fine for the | |
| # rules covered here; event conformance is covered in its own test. | |
| def _empty_globals(feature_map=0): | |
| return { | |
| GlobalAttributeIds.ATTRIBUTE_LIST_ID: [], | |
| GlobalAttributeIds.ACCEPTED_COMMAND_LIST_ID: [], | |
| GlobalAttributeIds.GENERATED_COMMAND_LIST_ID: [], | |
| GlobalAttributeIds.FEATURE_MAP_ID: feature_map, | |
| GlobalAttributeIds.CLUSTER_REVISION_ID: 1, | |
| } | |
| wildcard.tlvAttributes[0] = { | |
| desc.id: _empty_globals(), | |
| opcreds.id: _empty_globals(), | |
| ota_req.id: _empty_globals(), | |
| } | |
| wildcard.tlvAttributes[1] = { | |
| desc.id: _empty_globals(), | |
| groups.id: _empty_globals(), | |
| } | |
| wildcard.tlvAttributes[2] = { | |
| desc.id: _empty_globals(), | |
| groups.id: _empty_globals(), | |
| } | |
| facts, problems = derive_base_pics_facts_from_device_wildcard(wildcard, xml_cluster) | |
| asserts.assert_equal(problems, [], "Unexpected derivation problems") | |
| asserts.assert_true(facts.is_commissionee, "Root node device type on EP0 must derive is_commissionee") | |
| asserts.assert_true(facts.is_server, "Standard server clusters on EP0 must derive is_server") | |
| asserts.assert_true(facts.is_bridge, "Aggregator device type on EP1 must derive is_bridge") | |
| asserts.assert_true(facts.is_ota_requestor, "OTA Requestor cluster on EP0 must derive is_ota_requestor") | |
| asserts.assert_false(facts.is_ota_provider, "OTA Provider not present, must not derive is_ota_provider") | |
| asserts.assert_true(facts.has_groups_on_multiple_endpoints, | |
| "Groups on EP1 + EP2 must derive has_groups_on_multiple_endpoints") | |
| # Translator should map this exactly to the expected MCORE set, | |
| # excluding the one it doesn't cover (OTA Provider). | |
| expected_codes = { | |
| "MCORE.ROLE.COMMISSIONEE", "MCORE.IDM.S", "MCORE.BRIDGE", | |
| "MCORE.OTA.Requestor", "MCORE.G.MULTIENDPOINT", | |
| } | |
| asserts.assert_equal(base_pics_facts_to_pics_codes(facts), expected_codes, | |
| "Translator output drifted from the per-rule facts") | |
| def test_derive_base_pics_facts_groups_on_single_endpoint(self): | |
| # Groups cluster on a single endpoint must NOT derive | |
| # has_groups_on_multiple_endpoints. The >= 2 threshold is the | |
| # whole point of the MCORE.G.MULTIENDPOINT PICS. | |
| xml_cluster, _ = build_xml_clusters(PrebuiltDataModelDirectory.k1_4_1) | |
| desc = Clusters.Descriptor | |
| groups = Clusters.Groups | |
| wildcard = AsyncReadTransaction.ReadResponse(attributes={}, events=[], tlvAttributes={}) | |
| wildcard.attributes[0] = {desc: {desc.Attributes.DeviceTypeList: []}} | |
| wildcard.attributes[1] = {desc: {desc.Attributes.DeviceTypeList: []}} | |
| wildcard.tlvAttributes[0] = { | |
| desc.id: { | |
| GlobalAttributeIds.ATTRIBUTE_LIST_ID: [], | |
| GlobalAttributeIds.ACCEPTED_COMMAND_LIST_ID: [], | |
| GlobalAttributeIds.GENERATED_COMMAND_LIST_ID: [], | |
| GlobalAttributeIds.FEATURE_MAP_ID: 0, | |
| GlobalAttributeIds.CLUSTER_REVISION_ID: 1, | |
| }, | |
| } | |
| wildcard.tlvAttributes[1] = { | |
| desc.id: { | |
| GlobalAttributeIds.ATTRIBUTE_LIST_ID: [], | |
| GlobalAttributeIds.ACCEPTED_COMMAND_LIST_ID: [], | |
| GlobalAttributeIds.GENERATED_COMMAND_LIST_ID: [], | |
| GlobalAttributeIds.FEATURE_MAP_ID: 0, | |
| GlobalAttributeIds.CLUSTER_REVISION_ID: 1, | |
| }, | |
| groups.id: { | |
| GlobalAttributeIds.ATTRIBUTE_LIST_ID: [], | |
| GlobalAttributeIds.ACCEPTED_COMMAND_LIST_ID: [], | |
| GlobalAttributeIds.GENERATED_COMMAND_LIST_ID: [], | |
| GlobalAttributeIds.FEATURE_MAP_ID: 0, | |
| GlobalAttributeIds.CLUSTER_REVISION_ID: 1, | |
| }, | |
| } | |
| facts, problems = derive_base_pics_facts_from_device_wildcard(wildcard, xml_cluster) | |
| asserts.assert_equal(problems, [], "Unexpected derivation problems") | |
| asserts.assert_false(facts.has_groups_on_multiple_endpoints, | |
| "Groups on a single endpoint must not derive MCORE.G.MULTIENDPOINT") | |
| def test_derive_base_pics_facts_mandatory_events(self): | |
| # AccessControl on EP0 has spec-MANDATORY events (AccessControlEntryChanged, | |
| # AccessControlExtensionChanged). Their conformance evaluates to | |
| # mandatory with no feature-map dependency, so they should land in | |
| # facts.mandatory_events_by_cluster regardless of feature bits. | |
| xml_cluster, _ = build_xml_clusters(PrebuiltDataModelDirectory.k1_4_1) | |
| acl = Clusters.AccessControl | |
| desc = Clusters.Descriptor | |
| # Only run this rule if the chosen DM XML actually has events for | |
| # AccessControl; otherwise the test would be tautological. | |
| asserts.assert_true( | |
| len(xml_cluster[acl.id].events) > 0, | |
| "Test fixture assumption broken: AccessControl 1.4.1 must have events in DM XML") | |
| wildcard = AsyncReadTransaction.ReadResponse(attributes={}, events=[], tlvAttributes={}) | |
| wildcard.attributes[0] = {desc: {desc.Attributes.DeviceTypeList: []}} | |
| wildcard.tlvAttributes[0] = { | |
| desc.id: { | |
| GlobalAttributeIds.ATTRIBUTE_LIST_ID: [], | |
| GlobalAttributeIds.ACCEPTED_COMMAND_LIST_ID: [], | |
| GlobalAttributeIds.GENERATED_COMMAND_LIST_ID: [], | |
| GlobalAttributeIds.FEATURE_MAP_ID: 0, | |
| GlobalAttributeIds.CLUSTER_REVISION_ID: 1, | |
| }, | |
| acl.id: { | |
| GlobalAttributeIds.ATTRIBUTE_LIST_ID: [], | |
| GlobalAttributeIds.ACCEPTED_COMMAND_LIST_ID: [], | |
| GlobalAttributeIds.GENERATED_COMMAND_LIST_ID: [], | |
| GlobalAttributeIds.FEATURE_MAP_ID: 0, | |
| GlobalAttributeIds.CLUSTER_REVISION_ID: 1, | |
| }, | |
| } | |
| facts, _ = derive_base_pics_facts_from_device_wildcard(wildcard, xml_cluster) | |
| ep_events = facts.mandatory_events_by_cluster.get(0, {}) | |
| acl_mandatory = ep_events.get(acl.id, set()) | |
| asserts.assert_true( | |
| len(acl_mandatory) > 0, | |
| "AccessControl spec-mandatory events must be derived for any DUT with AccessControl on EP0") | |
| def test_read_pics_from_file_endpoint_naming_variants(self): | |
| """Endpoint subdir matching tolerates common naming conventions.""" | |
| test_xml = """<?xml version='1.0' encoding='utf-8'?> | |
| <clusterPICS> | |
| <picsItem> | |
| <itemNumber>TEST.S</itemNumber> | |
| <support>true</support> | |
| </picsItem> | |
| </clusterPICS> | |
| """ | |
| for subdir_name in ['endpoint0', 'EP0', 'ep_0', 'Endpoint 0', '0']: | |
| with tempfile.TemporaryDirectory() as d: | |
| ep_dir = os.path.join(d, subdir_name) | |
| os.makedirs(ep_dir) | |
| with open(os.path.join(ep_dir, 'cluster.xml'), 'w') as f: | |
| f.write(test_xml) | |
| pics = read_pics_from_file(d, endpoint=0) | |
| asserts.assert_true(pics.get('TEST.S'), f'Failed for subdir name: {subdir_name}') | |
| if __name__ == "__main__": | |
| default_matter_test_main() |