refactor: Rules class helper methods as free functions

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "cosl"
-version = "1.9.2"
+version = "1.10.0"
 authors = [
     { name = "sed-i", email = "82407168+sed-i@users.noreply.github.com" },
 ]

src/cosl/rules.py

@@ -221,6 +221,53 @@ class InjectResult:
     errmsg: Optional[str]
 
 
+_RULE_FILE_SUFFIXES = [".rule", ".rules", ".yml", ".yaml"]
+
+
+def _multi_suffix_glob(dir_path: Path, suffixes: List[str], recursive: bool = True) -> List[Path]:
+    """Get all files in a directory that have a matching suffix.
+
+    The result is sorted to avoid unnecessary relation-get calls.
+
+    Args:
+        dir_path: path to the directory to glob from.
+        suffixes: list of suffixes to include in the glob (items should begin with a period).
+        recursive: a flag indicating whether a glob is recursive (nested) or not.
+
+    Returns:
+        List of files in ``dir_path`` that have one of the suffixes specified in ``suffixes``.
+    """
+    all_files_in_dir = dir_path.glob("**/*" if recursive else "*")
+    all_files = {p for p in all_files_in_dir if p.is_file()}
+    matched = {p for p in all_files if p.suffix in suffixes}
+    ignored = all_files - matched
+    if ignored:
+        logger.info(
+            "Ignoring files with unrecognized suffix (expected one of %s): %s",
+            suffixes,
+            ", ".join(str(p) for p in sorted(ignored)),
+        )
+    return sorted(matched)
+
+
+def _read_rule_file(file_path: Path) -> Optional[Any]:
+    """Read and parse a YAML rule file.
+
+    Args:
+        file_path: full path to a rule file.
+
+    Returns:
+        The parsed YAML content (typically a dict, but may be a list or scalar for
+        non-mapping YAML, and ``None`` for an empty file), or ``None`` if parsing failed.
+    """
+    with file_path.open() as rf:
+        try:
+            return yaml.safe_load(rf)
+        except Exception as e:
+            logger.error("Failed to read rules from %s: %s", file_path.name, e)
+            return None
+
+
 class Rules:
     """Utility class for amalgamating alerting/recording rule  files and injecting juju topology.
 
@@ -304,34 +351,6 @@ def _is_single_rule_format(rules_dict: Mapping[str, Any]) -> bool:
         # one rule per file
         return "expr" in rules_dict and not RULE_TYPES.isdisjoint(rules_dict)
 
-    @staticmethod
-    def _multi_suffix_glob(
-        dir_path: Path, suffixes: List[str], recursive: bool = True
-    ) -> List[Path]:
-        """Helper function for getting all files in a directory that have a matching suffix.
-
-        The result is sorted to avoid unnecessary relation-get calls.
-
-        Args:
-            dir_path: path to the directory to glob from.
-            suffixes: list of suffixes to include in the glob (items should begin with a period).
-            recursive: a flag indicating whether a glob is recursive (nested) or not.
-
-        Returns:
-            List of files in `dir_path` that have one of the suffixes specified in `suffixes`.
-        """
-        all_files_in_dir = dir_path.glob("**/*" if recursive else "*")
-        all_files = {p for p in all_files_in_dir if p.is_file()}
-        matched = {p for p in all_files if p.suffix in suffixes}
-        ignored = all_files - matched
-        if ignored:
-            logger.info(
-                "Ignoring files with unrecognized suffix (expected one of %s): %s",
-                suffixes,
-                ", ".join(str(p) for p in sorted(ignored)),
-            )
-        return sorted(matched)
-
     def _from_dir(self, dir_path: Path, recursive: bool) -> List[OfficialRuleFileItem]:
         """Read all rule files in a directory.
 
@@ -350,9 +369,7 @@ def _from_dir(self, dir_path: Path, recursive: bool) -> List[OfficialRuleFileIte
         groups: List[OfficialRuleFileItem] = []
 
         # Gather all records into a list of groups
-        for file_path in Rules._multi_suffix_glob(
-            dir_path, [".rule", ".rules", ".yml", ".yaml"], recursive
-        ):
+        for file_path in _multi_suffix_glob(dir_path, _RULE_FILE_SUFFIXES, recursive):
             groups_from_file = self._from_file(dir_path, file_path)
             if groups_from_file:
                 logger.debug("Reading rule from %s", file_path)
@@ -373,33 +390,28 @@ def _from_file(  # noqa: C901
             A list of dictionaries representing the rules file, if file is valid (the structure is
             formed by `yaml.safe_load` of the file); an empty list otherwise.
         """
-        with file_path.open() as rf:
-            # Load a list of rules from file then add labels and filters
-            try:
-                rule_file = yaml.safe_load(rf)
-
-            except Exception as e:
-                logger.error("Failed to read rules from %s: %s", file_path.name, e)
-                return []
-
-            # Generate group name prefix
-            #  - name, from juju topology
-            #  - suffix, from the relative path of the rule file;
-            rel_path = file_path.parent.relative_to(root_path)
-            rel_path = "" if rel_path == Path(".") else str(rel_path)
-            group_name_parts = [self.topology.identifier] if self.topology else []
-            group_name_parts.append(rel_path)
-            group_name_prefix = "_".join(filter(None, group_name_parts))
-
-            try:
-                groups = self._from_dict(
-                    rule_file, group_name=file_path.stem, group_name_prefix=group_name_prefix
-                )
-            except ValueError as e:
-                logger.error("Invalid rules file: %s (%s)", file_path.name, e)
-                return []
-
-            return groups
+        rule_file = _read_rule_file(file_path)
+        if rule_file is None:
+            return []
+
+        # Generate group name prefix
+        #  - name, from juju topology
+        #  - suffix, from the relative path of the rule file;
+        rel_path = file_path.parent.relative_to(root_path)
+        rel_path = "" if rel_path == Path(".") else str(rel_path)
+        group_name_parts = [self.topology.identifier] if self.topology else []
+        group_name_parts.append(rel_path)
+        group_name_prefix = "_".join(filter(None, group_name_parts))
+
+        try:
+            groups = self._from_dict(
+                rule_file, group_name=file_path.stem, group_name_prefix=group_name_prefix
+            )
+        except ValueError as e:
+            logger.error("Invalid rules file: %s (%s)", file_path.name, e)
+            return []
+
+        return groups
 
     def _from_dict(
         self,

tests/test_rules_promql.py

@@ -13,7 +13,7 @@
 from fs.tempfs import TempFS
 
 from cosl.juju_topology import JujuTopology
-from cosl.rules import AlertRules, Rules
+from cosl.rules import AlertRules, Rules, _multi_suffix_glob
 
 
 class TestAddRulesFromPath(unittest.TestCase):
@@ -847,9 +847,7 @@ def test_multi_suffix_glob_logs_ignored_files(caplog):
 
     suffixes = [".rules", ".yml", ".yaml", ".rule"]
     with caplog.at_level("INFO", logger="cosl.rules"):
-        matched = Rules._multi_suffix_glob(
-            Path(sandbox.getsyspath("/")), suffixes, recursive=False
-        )
+        matched = _multi_suffix_glob(Path(sandbox.getsyspath("/")), suffixes, recursive=False)
 
     matched_names = {p.name for p in matched}
     assert matched_names == {"valid.rules", "also_valid.yml"}