PICS checker test implementation (#30970)

* PICS checker test implementation

* Address review comments

* Address review comments

* Remove tests for case-insensitive pics - we don't want this

* Fix case-insensitive pics and add test

* Fix pics case in test

src/python_testing/TC_TIMESYNC_2_1.py

@@ -164,15 +164,15 @@ async def test_TC_TIMESYNC_2_1(self):
  asserts.assert_true(False, "NTPServerAvailable is mandatory if the NTPS (TIMESYNC.S.F02) feature is supported")
 
  self.print_step(12, "Read TimeZoneListMaxSize")
- if self.check_pics("TIMESYNC.S.A000A"):
+ if self.check_pics("TIMESYNC.S.A000a"):
  size = await self.read_ts_attribute_expect_success(endpoint=endpoint, attribute=attributes.TimeZoneListMaxSize)
  asserts.assert_greater_equal(size, 1, "TimeZoneListMaxSize must be at least 1")
  asserts.assert_less_equal(size, 2, "TimeZoneListMaxSize must be max 2")
  elif self.check_pics("TIMESYNC.S.F00"):
  asserts.assert_true(False, "TimeZoneListMaxSize is mandatory if the TZ (TIMESYNC.S.F00) feature is supported")
 
  self.print_step(13, "Read DSTOffsetListMaxSize")
- if self.check_pics("TIMESYNC.S.A000B"):
+ if self.check_pics("TIMESYNC.S.A000b"):
  size = await self.read_ts_attribute_expect_success(endpoint=endpoint, attribute=attributes.DSTOffsetListMaxSize)
  asserts.assert_greater_equal(size, 1, "DSTOffsetListMaxSize must be at least 1")
  elif self.check_pics("TIMESYNC.S.F00"):

src/python_testing/TC_pics_checker.py

@@ -0,0 +1,160 @@
+#
+# Copyright (c) 2023 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 math
+
+import chip.clusters as Clusters
+from basic_composition_support import BasicCompositionTests
+from global_attribute_ids import GlobalAttributeIds
+from matter_testing_support import (AttributePathLocation, ClusterPathLocation, CommandPathLocation, FeaturePathLocation,
+ MatterBaseTest, async_test_body, default_matter_test_main)
+from mobly import asserts
+from spec_parsing_support import build_xml_clusters
+
+
+def attribute_pics(pics_base: str, id: int) -> str:
+ return f'{pics_base}.S.A{id:04x}'
+
+
+def accepted_cmd_pics(pics_base: str, id: int) -> str:
+ return f'{pics_base}.S.C{id:02x}.Rsp'
+
+
+def generated_cmd_pics(pics_base: str, id: int) -> str:
+ return f'{pics_base}.S.C{id:02x}.Tx'
+
+
+def feature_pics(pics_base: str, bit: int) -> str:
+ return f'{pics_base}.S.F{bit:02x}'
+
+
+class TC_PICS_Checker(MatterBaseTest, BasicCompositionTests):
+ @async_test_body
+ async def setup_class(self):
+ super().setup_class()
+ await self.setup_class_helper(False)
+ # build_xml_cluster returns a list of issues found when paring the XML
+ # Problems in the XML shouldn't cause test failure, but we want them recorded
+ # so they are added to the list of problems that get output when the test set completes.
+ self.xml_clusters, self.problems = build_xml_clusters()
+
+ def _check_and_record_errors(self, location, required, pics):
+ if required and not self.check_pics(pics):
+ self.record_error("PICS check", location=location,
+ problem=f"An element found on the device, but the corresponding PICS {pics} was not found in pics list")
+ self.success = False
+ elif not required and self.check_pics(pics):
+ self.record_error("PICS check", location=location, problem=f"PICS {pics} found in PICS list, but not on device")
+ self.success = False
+
+ def _add_pics_for_lists(self, cluster_id: int, attribute_id_of_element_list: GlobalAttributeIds) -> None:
+ try:
+ if attribute_id_of_element_list == GlobalAttributeIds.ATTRIBUTE_LIST_ID:
+ all_spec_elements_to_check = Clusters.ClusterObjects.ALL_ATTRIBUTES[cluster_id]
+ pics_mapper = attribute_pics
+ elif attribute_id_of_element_list == GlobalAttributeIds.ACCEPTED_COMMAND_LIST_ID:
+ all_spec_elements_to_check = Clusters.ClusterObjects.ALL_ACCEPTED_COMMANDS[cluster_id]
+ pics_mapper = accepted_cmd_pics
+
+ elif attribute_id_of_element_list == GlobalAttributeIds.GENERATED_COMMAND_LIST_ID:
+ all_spec_elements_to_check = Clusters.ClusterObjects.ALL_GENERATED_COMMANDS[cluster_id]
+ pics_mapper = generated_cmd_pics
+ else:
+ asserts.fail("add_pics_for_list function called for non-list attribute")
+ except KeyError:
+ # This cluster does not have any of this element type
+ return
+
+ for element_id in all_spec_elements_to_check:
+ if element_id > 0xF000:
+ # No pics for global elements
+ continue
+ pics = pics_mapper(self.xml_clusters[cluster_id].pics, element_id)
+
+ if cluster_id not in self.endpoint.keys():
+ # This cluster is not on this endpoint
+ required = False
+ elif element_id in self.endpoint[cluster_id][attribute_id_of_element_list]:
+ # Cluster and element are on the endpoint
+ required = True
+ else:
+ # Cluster is on the endpoint but the element is not in the list
+ required = False
+
+ if attribute_id_of_element_list == GlobalAttributeIds.ATTRIBUTE_LIST_ID:
+ location = AttributePathLocation(endpoint_id=self.endpoint_id, cluster_id=cluster_id, attribute_id=element_id)
+ else:
+ location = CommandPathLocation(endpoint_id=self.endpoint_id, cluster_id=cluster_id, command_id=element_id)
+
+ self._check_and_record_errors(location, required, pics)
+
+ def test_TC_pics_checker(self):
+ self.endpoint_id = self.matter_test_config.endpoint
+ self.endpoint = self.endpoints_tlv[self.endpoint_id]
+ self.success = True
+
+ for cluster_id, cluster in Clusters.ClusterObjects.ALL_CLUSTERS.items():
+ # Data model XML is used to get the PICS code for this cluster. If we don't know the PICS
+ # code, we can't evaluate the PICS list. Clusters that are present on the device but are
+ # not present in the spec are checked in the IDM tests.
+ if cluster_id not in self.xml_clusters or self.xml_clusters[cluster_id].pics is None:
+ continue
+
+ # Ensure the PICS.S code is correctly marked
+ pics_cluster = f'{self.xml_clusters[cluster_id].pics}.S'
+ location = ClusterPathLocation(endpoint_id=self.endpoint_id, cluster_id=cluster_id)
+ self._check_and_record_errors(location, cluster_id in self.endpoint, pics_cluster)
+
+ self._add_pics_for_lists(cluster_id, GlobalAttributeIds.ATTRIBUTE_LIST_ID)
+ self._add_pics_for_lists(cluster_id, GlobalAttributeIds.ACCEPTED_COMMAND_LIST_ID)
+ self._add_pics_for_lists(cluster_id, GlobalAttributeIds.GENERATED_COMMAND_LIST_ID)
+
+ try:
+ cluster_features = cluster.Bitmaps.Feature
+ except AttributeError:
+ # cluster has no features
+ continue
+
+ pics_base = self.xml_clusters[cluster_id].pics
+ try:
+ feature_map = self.endpoint[cluster_id][GlobalAttributeIds.FEATURE_MAP_ID]
+ except KeyError:
+ feature_map = 0
+
+ for feature_mask in cluster_features:
+ # Codegen in python uses feature masks (0x01, 0x02, 0x04 etc.)
+ # PICS uses the mask bit number (1, 2, 3)
+ # Convert the mask to a bit number so we can check the PICS.
+ feature_bit = int(math.log2(feature_mask))
+ pics = feature_pics(pics_base, feature_bit)
+ if feature_mask & feature_map:
+ required = True
+ else:
+ required = False
+
+ try:
+ location = FeaturePathLocation(endpoint_id=self.endpoint_id, cluster_id=cluster_id,
+ feature_code=self.xml_clusters[cluster_id].features[feature_mask].code)
+ except KeyError:
+ location = ClusterPathLocation(endpoint_id=self.endpoint_id, cluster_id=cluster_id)
+ self._check_and_record_errors(location, required, pics)
+
+ if not self.success:
+ self.fail_current_test("At least one PICS error was found for this endpoint")
+
+
+if __name__ == "__main__":
+ default_matter_test_main()

src/python_testing/TestMatterTestingSupport.py

@@ -136,7 +136,7 @@ async def test_type_checking(self):
  async def test_pics_support(self):
  pics_list = ['TEST.S.A0000=1',
  'TEST.S.A0001=0',
- 'lower.s.a0000=1',
+ 'TEST.S.A000a=1'
  '',
  ' ',
  '# comment',
@@ -148,10 +148,9 @@ async def test_pics_support(self):
 
  asserts.assert_true(self.check_pics("TEST.S.A0000"), "PICS parsed incorrectly for TEST.S.A0000")
  asserts.assert_false(self.check_pics("TEST.S.A0001"), "PICS parsed incorrectly for TEST.S.A0001")
- asserts.assert_true(self.check_pics("LOWER.S.A0000"), "PICS pased incorrectly for LOWER.S.A0000")
+ asserts.assert_true(self.check_pics("TEST.S.A000a"), "PICS parsed incorrectly for TEST.S.A000a")
  asserts.assert_true(self.check_pics("SPACE.S.A0000"), "PICS parsed incorrectly for SPACE.S.A0000")
  asserts.assert_false(self.check_pics("NOT.S.A0000"), "PICS parsed incorrectly for NOT.S.A0000")
- asserts.assert_true(self.check_pics(" test.s.a0000"), "PICS checker lowercase handled incorrectly")
 
  # invalid pics file should throw a value error
  pics_list.append("BAD.S.A000=5")

src/python_testing/basic_composition_support.py

@@ -97,11 +97,11 @@ def ConvertValue(value) -> Any:
 
 
 class BasicCompositionTests:
- async def setup_class_helper(self):
+ async def setup_class_helper(self, default_to_pase: bool = True):
  dev_ctrl = self.default_controller
  self.problems = []
 
- do_test_over_pase = self.user_params.get("use_pase_only", True)
+ do_test_over_pase = self.user_params.get("use_pase_only", default_to_pase)
  dump_device_composition_path: Optional[str] = self.user_params.get("dump_device_composition_path", None)
 
  if do_test_over_pase:

src/python_testing/drlk_2_x_common.py

@@ -149,8 +149,10 @@ async def run_drlk_test_common(self, lockUnlockCommand, lockUnlockCmdRspPICS, lo
  self.print_step("1", "TH writes the RequirePINforRemoteOperation attribute value as false on the DUT")
  attribute = attributes.RequirePINforRemoteOperation(False)
  if self.check_pics("DRLK.S.M.RequirePINForRemoteOperationAttributeWritable"):
+ print("---------------------- PICS is true")
  await self.write_drlk_attribute_expect_success(attribute=attribute)
  else:
+ print("---------------------- PICS is false")
  await self.write_drlk_attribute_expect_error(attribute=attribute, error=Status.UnsupportedWrite)
 
  if self.check_pics("DRLK.S.A0033"):

src/python_testing/matter_testing_support.py

@@ -150,7 +150,7 @@ def parse_pics(lines=typing.List[str]) -> dict[str, bool]:
  if val not in ["1", "0"]:
  raise ValueError('PICS {} must have a value of 0 or 1'.format(key))
 
- pics[key.strip().upper()] = (val == "1")
+ pics[key.strip()] = (val == "1")
  return pics
 
 
@@ -725,7 +725,7 @@ def teardown_class(self):
 
  def check_pics(self, pics_key: str) -> bool:
  picsd = self.matter_test_config.pics
- pics_key = pics_key.strip().upper()
+ pics_key = pics_key.strip()
  return pics_key in picsd and picsd[pics_key]
 
  async def read_single_attribute(

src/python_testing/spec_parsing_support.py

@@ -100,6 +100,7 @@ class XmlCluster:
  accepted_commands: dict[uint, XmlCommand]
  generated_commands: dict[uint, XmlCommand]
  events: dict[uint, XmlEvent]
+ pics: str
 
 
 class CommandType(Enum):
@@ -124,6 +125,12 @@ def __init__(self, cluster, cluster_id, name, is_alias):
  except (KeyError, StopIteration):
  self._derived = None
 
+ try:
+ classification = next(cluster.iter('classification'))
+ self._pics = classification.attrib['picsCode']
+ except (KeyError, StopIteration):
+ self._pics = None
+
  self.feature_elements = self.get_all_feature_elements()
  self.attribute_elements = self.get_all_attribute_elements()
  self.command_elements = self.get_all_command_elements()
@@ -348,31 +355,29 @@ def create_cluster(self) -> XmlCluster:
  attributes=self.parse_attributes(),
  accepted_commands=self.parse_commands(CommandType.ACCEPTED),
  generated_commands=self.parse_commands(CommandType.GENERATED),
- events=self.parse_events())
+ events=self.parse_events(), pics=self._pics)
 
  def get_problems(self) -> list[ProblemNotice]:
  return self._problems
 
 
 def build_xml_clusters() -> tuple[list[XmlCluster], list[ProblemNotice]]:
  # workaround for aliased clusters not appearing in the xml. Remove this once https://github.com/csa-data-model/projects/issues/373 is addressed
- conc_clusters = {0x040C: 'Carbon Monoxide Concentration Measurement',
- 0x040D: 'Carbon Dioxide Concentration Measurement',
- 0x0413: 'Nitrogen Dioxide Concentration Measurement',
- 0x0415: 'Ozone Concentration Measurement',
- 0x042A: 'PM2.5 Concentration Measurement',
- 0x042B: 'Formaldehyde Concentration Measurement',
- 0x042C: 'PM1 Concentration Measurement',
- 0x042D: 'PM10 Concentration Measurement',
- 0x042E: 'Total Volatile Organic Compounds Concentration Measurement',
- 0x042F: 'Radon Concentration Measurement'}
+ conc_clusters = {0x040C: ('Carbon Monoxide Concentration Measurement', 'CMOCONC'),
+ 0x040D: ('Carbon Dioxide Concentration Measurement', 'CDOCONC'),
+ 0x0413: ('Nitrogen Dioxide Concentration Measurement', 'NDOCONC'),
+ 0x0415: ('Ozone Concentration Measurement', 'OZCONC'),
+ 0x042A: ('PM2.5 Concentration Measurement', 'PMICONC'),
+ 0x042B: ('Formaldehyde Concentration Measurement', 'FLDCONC'),
+ 0x042C: ('PM1 Concentration Measurement', 'PMHCONC'),
+ 0x042D: ('PM10 Concentration Measurement', 'PMKCONC'),
+ 0x042E: ('Total Volatile Organic Compounds Concentration Measurement', 'TVOCCONC'),
+ 0x042F: ('Radon Concentration Measurement', 'RNCONC')}
  conc_base_name = 'Concentration Measurement Clusters'
- resource_clusters = {0x0071: 'HEPA Filter Monitoring',
- 0x0072: 'Activated Carbon Filter Monitoring'}
+ resource_clusters = {0x0071: ('HEPA Filter Monitoring', 'HEPAFREMON'),
+ 0x0072: ('Activated Carbon Filter Monitoring', 'ACFREMON')}
  resource_base_name = 'Resource Monitoring Clusters'
- water_clusters = {0x0405: 'Relative Humidity Measurement',
- 0x0407: 'Leaf Wetness Measurement',
- 0x0408: 'Soil Moisture Measurement'}
+ water_clusters = {0x0405: ('Relative Humidity Measurement', 'RH')}
  water_base_name = 'Water Content Measurement Clusters'
  aliases = {conc_base_name: conc_clusters, resource_base_name: resource_clusters, water_base_name: water_clusters}
 
@@ -482,15 +487,16 @@ def remove_problem(location: typing.Union[CommandPathLocation, FeaturePathLocati
  new = XmlCluster(revision=c.revision, derived=c.derived, name=c.name,
  feature_map=feature_map, attribute_map=attribute_map, command_map=command_map,
  features=features, attributes=attributes, accepted_commands=accepted_commands,
- generated_commands=generated_commands, events=events)
+ generated_commands=generated_commands, events=events, pics=c.pics)
  clusters[id] = new
 
  for alias_base_name, aliased_clusters in aliases.items():
- for id, alias_name in aliased_clusters.items():
+ for id, (alias_name, pics) in aliased_clusters.items():
  base = derived_clusters[alias_base_name]
  new = deepcopy(base)
  new.derived = alias_base_name
  new.name = alias_name
+ new.pics = pics
  clusters[id] = new
 
  # TODO: All these fixups should be removed BEFORE SVE if at all possible