Updated spec_parsing.py to now use the data model directory from python package

src/python_testing/matter_testing_infrastructure/chip/testing/spec_parsing.py

@@ -17,15 +17,16 @@
 
 import glob
 import logging
-import os
+import pathlib
 import typing
 import xml.etree.ElementTree as ElementTree
 from copy import deepcopy
 from dataclasses import dataclass
 from enum import Enum, auto
-from typing import Callable, Optional
+from typing import Callable, Optional, Union
 
 import chip.clusters as Clusters
+import chip.testing
 import chip.testing.conformance as conformance_support
 from chip.testing.conformance import (OPTIONAL_CONFORM, TOP_LEVEL_CONFORMANCE_TAGS, ConformanceDecision, ConformanceException,
                                       ConformanceParseParameters, feature, is_disallowed, mandatory, optional, or_operation,
@@ -518,50 +519,75 @@ class DataModelLevel(str, Enum):
     kDeviceType = 'device_types'
 
 
-def _get_data_model_root() -> str:
-    """Attempts to find ${CHIP_ROOT}/data_model or equivalent."""
+def _get_data_model_root() -> pathlib.PosixPath:
+    """Attempts to find the 'data_model' directory inside the 'chip.testing' package."""
+    try:
+        # Locate the directory where the 'chip.testing' module is located
+        package_dir = pathlib.Path(chip.testing.__file__).parent
 
-    # Since this class is generally in a module, we have to rely on being bootstrapped or
-    # we use CWD if we cannot
-    choices = [os.getcwd()]
+        # Construct the path to the 'data_model' directory (assuming this structure)
+        data_model_root = package_dir / 'data_model'
 
-    if 'PW_PROJECT_ROOT' in os.environ:
-        choices.insert(0, os.environ['PW_PROJECT_ROOT'])
+        if not data_model_root.exists():
+            raise FileNotFoundError(f"Data model directory not found in the package at {data_model_root}.")
 
-    for c in choices:
-        data_model_path = os.path.join(c, 'data_model')
-        if os.path.exists(os.path.join(data_model_path, 'master', 'scraper_version')):
-            return data_model_path
-    raise FileNotFoundError('Cannot find a CHIP_ROOT/data_model path. Tried %r as prefixes.' % choices)
+        return data_model_root
 
+    except Exception as e:
+        raise FileNotFoundError(f"Failed to find the data model root: {e}")
 
-def get_data_model_directory(data_model_directory: typing.Union[PrebuiltDataModelDirectory, str], data_model_level: DataModelLevel) -> str:
-    if data_model_directory == PrebuiltDataModelDirectory.k1_3:
-        return os.path.join(_get_data_model_root(), '1.3', data_model_level)
-    elif data_model_directory == PrebuiltDataModelDirectory.k1_4:
-        return os.path.join(_get_data_model_root(), '1.4', data_model_level)
-    elif data_model_directory == PrebuiltDataModelDirectory.kMaster:
-        return os.path.join(_get_data_model_root(), 'master', data_model_level)
+
+def get_data_model_directory(data_model_directory: Union[PrebuiltDataModelDirectory, str], data_model_level: str) -> pathlib.PosixPath:
+    """
+    Get the directory of the data model for a specific version and level from the installed package.
+    """
+    data_model_root = _get_data_model_root()
+
+    # Build path based on the version and data model level
+    if isinstance(data_model_directory, PrebuiltDataModelDirectory):
+        version_map = {
+            PrebuiltDataModelDirectory.k1_3: '1.3',
+            PrebuiltDataModelDirectory.k1_4: '1.4',
+            PrebuiltDataModelDirectory.kMaster: 'master',
+        }
+
+        version = version_map.get(data_model_directory)
+        if not version:
+            raise ValueError(f"Unsupported data model directory: {data_model_directory}")
+
+        return data_model_root / version / data_model_level
     else:
-        return data_model_directory
+        # If it's a custom directory, return it directly
+        return pathlib.Path(data_model_directory)
 
 
-def build_xml_clusters(data_model_directory: typing.Union[PrebuiltDataModelDirectory, str] = PrebuiltDataModelDirectory.k1_4) -> tuple[dict[uint, XmlCluster], list[ProblemNotice]]:
-    dir = get_data_model_directory(data_model_directory, DataModelLevel.kCluster)
+def build_xml_clusters(data_model_directory: Union[PrebuiltDataModelDirectory, str] = PrebuiltDataModelDirectory.k1_4) -> tuple[dict[int, XmlCluster], list[ProblemNotice]]:
+    # Get the data model directory path inside the package
+    dir = get_data_model_directory(data_model_directory, 'clusters')
 
     clusters: dict[int, XmlCluster] = {}
     pure_base_clusters: dict[str, XmlCluster] = {}
     ids_by_name: dict[str, int] = {}
     problems: list[ProblemNotice] = []
-    files = glob.glob(f'{dir}/*.xml')
-    if not files:
-        raise SpecParsingException(f'No data model files found in specified directory {dir}')
 
-    for xml in files:
-        logging.info(f'Parsing file {xml}')
-        tree = ElementTree.parse(f'{xml}')
-        root = tree.getroot()
-        add_cluster_data_from_xml(root, clusters, pure_base_clusters, ids_by_name, problems)
+    try:
+        # Use pathlib.Path to list all XML files in the data model directory
+        xml_files = [f for f in dir.glob('**/*.xml')]  # Recursively find all .xml files
+
+        if not xml_files:
+            raise SpecParsingException(f'No XML files found in the specified package directory {dir}')
+
+        # Parse each XML file found inside the package
+        for xml in xml_files:
+            logging.info(f'Parsing file {xml}')
+            # Open and parse each XML file from the directory
+            with xml.open('r') as file:
+                tree = ElementTree.parse(file)
+                root = tree.getroot()
+                add_cluster_data_from_xml(root, clusters, pure_base_clusters, ids_by_name, problems)
+
+    except Exception as e:
+        raise SpecParsingException(f"Error reading XML files from the package: {e}")
 
     # There are a few clusters where the conformance columns are listed as desc. These clusters need specific, targeted tests
     # to properly assess conformance. Here, we list them as Optional to allow these for the general test. Targeted tests are described below.
@@ -578,6 +604,7 @@ def remove_problem(location: typing.Union[CommandPathLocation, FeaturePathLocati
     mask = clusters[descriptor_id].feature_map[code]
     clusters[descriptor_id].features[mask].conformance = optional()
     remove_problem(FeaturePathLocation(endpoint_id=0, cluster_id=descriptor_id, feature_code=code))
+
     action_id = Clusters.Actions.id
     for c in Clusters.ClusterObjects.ALL_ACCEPTED_COMMANDS[action_id]:
         clusters[action_id].accepted_commands[c].conformance = optional()