From bbb5e084862e2816d2ee979c9eddb51240b04d68 Mon Sep 17 00:00:00 2001
From: Michael Thamm <mike.thamm@canonical.com>
Date: Fri, 13 Dec 2024 15:34:13 -0500
Subject: [PATCH 01/11] test from file alerts

---
 src/cosl/rules.py          | 69 ++++++++++++++++++++++++++++++++++++++
 tests/test_rules_promql.py | 43 ++++++++++++++++++++++++
 2 files changed, 112 insertions(+)

diff --git a/src/cosl/rules.py b/src/cosl/rules.py
index 820aac0..c421620 100644
--- a/src/cosl/rules.py
+++ b/src/cosl/rules.py
@@ -78,6 +78,8 @@
 import logging
 import os
 import re
+import textwrap
+from pprint import pprint
 from abc import ABC, abstractmethod
 from pathlib import Path
 from typing import Any, Dict, List, Optional, Union, cast
@@ -240,6 +242,59 @@ def _from_dir(self, dir_path: Path, recursive: bool) -> List[Dict[str, Any]]:
 
         return groups
 
+    def _from_str(self, yaml_str: str):
+        yaml.safe_load(textwrap.dedent(yaml_str))
+        rule_str = yaml.safe_load(yaml_str)
+        root_path = Path()
+        file_path = Path()
+        # print(f"HERE: {self.rule_type} && {type(rule_str)} && {type(cast(SingleRuleFormat, rule_str))}")
+        # pprint(rule_str)
+
+        if self._is_official_rule_format(cast(OfficialRuleFileFormat, rule_str)):
+            rule_str = cast(OfficialRuleFileFormat, rule_str)
+            groups = rule_str["groups"]
+        elif self._is_single_rule_format(cast(SingleRuleFormat, rule_str), self.rule_type):
+            # convert to list of groups
+            # group name is made up from the file name
+            rule_str = cast(SingleRuleFormat, rule_str)
+            groups = [{"name": file_path.stem, "rules": [rule_str]}]
+        else:
+            # invalid/unsupported
+            logger.error("Invalid rules file: %s", file_path.name)
+            return []
+
+        # update rules with additional metadata
+        groups = cast(List[OfficialRuleFileItem], groups)
+        for group in groups:
+            if not self._is_already_modified(group["name"]):
+                # update group name with topology and sub-path
+                group["name"] = self._new_group_name(group["name"])
+
+            # add "juju_" topology labels
+            for rule in group["rules"]:
+                if "labels" not in rule:
+                    rule["labels"] = {}
+
+                if self.topology:
+                    # only insert labels that do not already exist
+                    for label, val in self.topology.label_matcher_dict.items():
+                        if label not in rule["labels"]:
+                            rule["labels"][label] = val
+
+                    # insert juju topology filters into a prometheus rule
+                    repl = r'job=~".+"' if self.query_type == "logql" else ""
+                    rule["expr"] = self.tool.inject_label_matchers(  # type: ignore
+                        expression=re.sub(r"%%juju_topology%%,?", repl, rule["expr"]),
+                        topology={
+                            k: rule["labels"][k]
+                            for k in ("juju_model", "juju_model_uuid", "juju_application")
+                            if rule["labels"].get(k) is not None
+                        },
+                        query_type=self.query_type,
+                    )
+
+        return groups
+
     def _from_file(  # noqa: C901
         self, root_path: Path, file_path: Path
     ) -> List[OfficialRuleFileItem]:
@@ -314,6 +369,15 @@ def _from_file(  # noqa: C901
 
             return groups
 
+    def _new_group_name(self, group_name: str) -> str:
+        # Generate group name:
+        #  - name, from juju topology
+        #  - suffix, from the relative path of the rule file;
+        group_name_parts = [self.topology.identifier] if self.topology else []
+        group_name_parts.extend(["generic_rules", group_name, f"{self.rule_type}s"])
+        # filter to remove empty strings
+        return "_".join(filter(None, group_name_parts))
+
     def _group_name(self, root_path: str, file_path: str, group_name: str) -> str:
         """Generate group name from path and topology.
 
@@ -348,6 +412,11 @@ def _is_already_modified(self, name: str) -> bool:
 
     # ---- END STATIC HELPER METHODS --- #
 
+    def add(self, yaml_str: str):
+        temp = self._from_str(yaml_str)
+        pprint(temp)
+        self.groups.extend(temp)
+
     def add_path(self, dir_path: Union[str, Path], *, recursive: bool = False) -> None:
         """Add rules from a dir path.
 
diff --git a/tests/test_rules_promql.py b/tests/test_rules_promql.py
index afef8b8..be4c6a3 100644
--- a/tests/test_rules_promql.py
+++ b/tests/test_rules_promql.py
@@ -3,6 +3,7 @@
 
 import json
 import re
+import textwrap
 import unittest
 import uuid
 from pathlib import Path
@@ -75,6 +76,48 @@ def test_each_alert_expression_is_topology_labeled(self):
                 self.assertIn("juju_model_uuid", labels)
                 self.assertIn("juju_application", labels)
 
+    def test_injecting_generic_alerts(self):
+        ri = AlertRules(
+            query_type="promql",
+            topology=JujuTopology(
+                model="unittest",
+                model_uuid=str(uuid.uuid4()),
+                unit="tester/0",
+                application="tester",
+            ),
+        )
+        ri.add_path(Path(__file__).resolve().parent / "promql_rules" / "prometheus_alert_rules")
+
+        # TODO SingleRuleFormat only require alert, expr, labels. Test others though
+        generic_alert_rules = """
+                groups:
+                - name: GenericRules
+                  rules:
+                  - alert: HostDown
+                    expr: up < 1
+                    for: 5m
+                    labels:
+                      severity: critical
+                  - alert: HostUnavailable
+                    expr: absent(up)
+                    for: 5m
+                    labels:
+                      severity: critical
+                """
+
+        ri.add(generic_alert_rules)
+
+        alerts = ri.as_dict()
+        self.assertIn("groups", alerts)
+        self.assertEqual(len(alerts["groups"]), 5)
+        group = alerts["groups"][0]
+        for rule in group["rules"]:
+            self.assertIn("expr", rule)
+            for labels in expression_labels(rule["expr"]):
+                self.assertIn("juju_model", labels)
+                self.assertIn("juju_model_uuid", labels)
+                self.assertIn("juju_application", labels)
+
 
 def sorted_matchers(matchers) -> str:
     parts = [m.strip() for m in matchers.split(",")]

From 80a56645f239cded7949cd8f5283326b8003fb77 Mon Sep 17 00:00:00 2001
From: sed-i <82407168+sed-i@users.noreply.github.com>
Date: Fri, 13 Dec 2024 16:17:43 -0500
Subject: [PATCH 02/11] Refactor "_from_file"

---
 src/cosl/rules.py | 137 +++++++++++++---------------------------------
 1 file changed, 38 insertions(+), 99 deletions(-)

diff --git a/src/cosl/rules.py b/src/cosl/rules.py
index c421620..64de85e 100644
--- a/src/cosl/rules.py
+++ b/src/cosl/rules.py
@@ -74,7 +74,7 @@
 - `juju_model_uuid`
 - `juju_application`
 """  # noqa: W505
-
+import hashlib
 import logging
 import os
 import re
@@ -242,33 +242,41 @@ def _from_dir(self, dir_path: Path, recursive: bool) -> List[Dict[str, Any]]:
 
         return groups
 
-    def _from_str(self, yaml_str: str):
-        yaml.safe_load(textwrap.dedent(yaml_str))
-        rule_str = yaml.safe_load(yaml_str)
-        root_path = Path()
-        file_path = Path()
-        # print(f"HERE: {self.rule_type} && {type(rule_str)} && {type(cast(SingleRuleFormat, rule_str))}")
-        # pprint(rule_str)
-
-        if self._is_official_rule_format(cast(OfficialRuleFileFormat, rule_str)):
-            rule_str = cast(OfficialRuleFileFormat, rule_str)
-            groups = rule_str["groups"]
-        elif self._is_single_rule_format(cast(SingleRuleFormat, rule_str), self.rule_type):
+    def _from_str(self, yaml_str: str, *, group_name: Optional[str]=None, group_name_prefix: str) -> List[dict]:
+        """Process rules file (TODO: improve desc).
+
+        Raises:
+            ValueError, when invalid rule format given.
+        """
+        if not yaml_str:
+            raise ValueError("Empty")
+        if not isinstance(yaml_str, dict):
+            raise ValueError("Invalid rules (must be a dict)")
+
+        if self._is_official_rule_format(cast(OfficialRuleFileFormat, yaml_str)):
+            yaml_str = cast(OfficialRuleFileFormat, yaml_str)
+            groups = yaml_str["groups"]
+        elif self._is_single_rule_format(cast(SingleRuleFormat, yaml_str), self.rule_type):
             # convert to list of groups
             # group name is made up from the file name
-            rule_str = cast(SingleRuleFormat, rule_str)
-            groups = [{"name": file_path.stem, "rules": [rule_str]}]
+            yaml_str = cast(SingleRuleFormat, yaml_str)
+            if not group_name:
+                # Note: the caller of this function should make sure this never happens:
+                # Either we use the standard format, or we'd pass a group_name.
+                # If/when we drop support for the single-rule-per-file format, this won't
+                # be needed anymore.
+                group_name = hashlib.shake_256(str(yaml_str).encode("utf-8")).hexdigest(20)
+            groups = [{"name": group_name, "rules": [yaml_str]}]
         else:
             # invalid/unsupported
-            logger.error("Invalid rules file: %s", file_path.name)
-            return []
+            raise ValueError("Invalid rule format")
 
         # update rules with additional metadata
         groups = cast(List[OfficialRuleFileItem], groups)
         for group in groups:
             if not self._is_already_modified(group["name"]):
                 # update group name with topology and sub-path
-                group["name"] = self._new_group_name(group["name"])
+                group["name"] = "_".join(filter(None, [group_name_prefix, group_name, f"{self.rule_type}s"]))
 
             # add "juju_" topology labels
             for rule in group["rules"]:
@@ -317,92 +325,23 @@ def _from_file(  # noqa: C901
                 logger.error("Failed to read rules from %s: %s", file_path.name, e)
                 return []
 
-            if not rule_file:
-                logger.warning("Empty rules file: %s", file_path.name)
-                return []
-            if not isinstance(rule_file, dict):
-                logger.error("Invalid rules file (must be a dict): %s", file_path.name)
-                return []
+            # Generate group name prefix
+            #  - name, from juju topology
+            #  - suffix, from the relative path of the rule file;
+            rel_path = os.path.relpath(os.path.dirname(file_path), root_path)
+            rel_path = "" if rel_path == "." else rel_path.replace(os.path.sep, "_")
+            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))
 
-            if self._is_official_rule_format(cast(OfficialRuleFileFormat, rule_file)):
-                rule_file = cast(OfficialRuleFileFormat, rule_file)
-                groups = rule_file["groups"]
-            elif self._is_single_rule_format(cast(SingleRuleFormat, rule_file), self.rule_type):
-                # convert to list of groups
-                # group name is made up from the file name
-                rule_file = cast(SingleRuleFormat, rule_file)
-                groups = [{"name": file_path.stem, "rules": [rule_file]}]
-            else:
-                # invalid/unsupported
-                logger.error("Invalid rules file: %s", file_path.name)
+            try:
+                groups = self._from_str(rule_file, group_name_prefix=group_name_prefix)
+            except ValueError as e:
+                logger.error("Invalid rules file: %s (%s)", file_path.name, e)
                 return []
 
-            # update rules with additional metadata
-            groups = cast(List[OfficialRuleFileItem], groups)
-            for group in groups:
-                if not self._is_already_modified(group["name"]):
-                    # update group name with topology and sub-path
-                    group["name"] = self._group_name(str(root_path), str(file_path), group["name"])
-
-                # add "juju_" topology labels
-                for rule in group["rules"]:
-                    if "labels" not in rule:
-                        rule["labels"] = {}
-
-                    if self.topology:
-                        # only insert labels that do not already exist
-                        for label, val in self.topology.label_matcher_dict.items():
-                            if label not in rule["labels"]:
-                                rule["labels"][label] = val
-
-                        # insert juju topology filters into a prometheus rule
-                        repl = r'job=~".+"' if self.query_type == "logql" else ""
-                        rule["expr"] = self.tool.inject_label_matchers(  # type: ignore
-                            expression=re.sub(r"%%juju_topology%%,?", repl, rule["expr"]),
-                            topology={
-                                k: rule["labels"][k]
-                                for k in ("juju_model", "juju_model_uuid", "juju_application")
-                                if rule["labels"].get(k) is not None
-                            },
-                            query_type=self.query_type,
-                        )
-
             return groups
 
-    def _new_group_name(self, group_name: str) -> str:
-        # Generate group name:
-        #  - name, from juju topology
-        #  - suffix, from the relative path of the rule file;
-        group_name_parts = [self.topology.identifier] if self.topology else []
-        group_name_parts.extend(["generic_rules", group_name, f"{self.rule_type}s"])
-        # filter to remove empty strings
-        return "_".join(filter(None, group_name_parts))
-
-    def _group_name(self, root_path: str, file_path: str, group_name: str) -> str:
-        """Generate group name from path and topology.
-
-        The group name is made up of the relative path between the root dir_path, the file path,
-        and topology identifier.
-
-        Args:
-            root_path: path to the root rules dir.
-            file_path: path to rule file.
-            group_name: original group name to keep as part of the new augmented group name
-
-        Returns:
-            New group name, augmented by juju topology and relative path.
-        """
-        rel_path = os.path.relpath(os.path.dirname(file_path), root_path)
-        rel_path = "" if rel_path == "." else rel_path.replace(os.path.sep, "_")
-
-        # Generate group name:
-        #  - name, from juju topology
-        #  - suffix, from the relative path of the rule file;
-        group_name_parts = [self.topology.identifier] if self.topology else []
-        group_name_parts.extend([rel_path, group_name, f"{self.rule_type}s"])
-        # filter to remove empty strings
-        return "_".join(filter(None, group_name_parts))
-
     def _is_already_modified(self, name: str) -> bool:
         """Detect whether a group name has already been modified with juju topology."""
         modified_matcher = re.compile(r"^.*?_[\da-f]{8}_.*?alerts$")

From dc03e07de928af1814556a218cc484a3c5111c24 Mon Sep 17 00:00:00 2001
From: Michael Thamm <mike.thamm@canonical.com>
Date: Tue, 17 Dec 2024 13:01:54 -0500
Subject: [PATCH 03/11] feature to add alerts from string to support generic
 alert rules

---
 src/cosl/rules.py          | 142 +++++++++++++---------
 tests/test_rules_promql.py | 243 +++++++++++++++++++++++++++++++------
 2 files changed, 290 insertions(+), 95 deletions(-)

diff --git a/src/cosl/rules.py b/src/cosl/rules.py
index 64de85e..691419f 100644
--- a/src/cosl/rules.py
+++ b/src/cosl/rules.py
@@ -74,12 +74,11 @@
 - `juju_model_uuid`
 - `juju_application`
 """  # noqa: W505
+
 import hashlib
 import logging
 import os
 import re
-import textwrap
-from pprint import pprint
 from abc import ABC, abstractmethod
 from pathlib import Path
 from typing import Any, Dict, List, Optional, Union, cast
@@ -242,8 +241,60 @@ def _from_dir(self, dir_path: Path, recursive: bool) -> List[Dict[str, Any]]:
 
         return groups
 
-    def _from_str(self, yaml_str: str, *, group_name: Optional[str]=None, group_name_prefix: str) -> List[dict]:
-        """Process rules file (TODO: improve desc).
+    def _from_file(  # noqa: C901
+        self, root_path: Path, file_path: Path
+    ) -> List[OfficialRuleFileItem]:
+        """Read a rules file from path.
+
+        Args:
+            root_path: full path to the root rules folder (used only for generating group name)
+            file_path: full path to a *.rule file.
+
+        Returns:
+            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 = os.path.relpath(os.path.dirname(file_path), root_path)
+            rel_path = "" if rel_path == "." else rel_path.replace(os.path.sep, "_")
+            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_str(
+                    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_str(
+        self,
+        yaml_str: str,
+        *,
+        group_name: Optional[str] = None,
+        group_name_prefix: Optional[str] = None,
+    ) -> List[OfficialRuleFileItem]:
+        """Process rules from string, injecting juju topology. If a single-rule format is provided, a hash of the yaml file is injected into the group name to ensure uniqueness.
+
+        Args:
+            yaml_str: rules content in single-rule or official-rule format as a string
+            group_name: a custom identifier for the rule name to include in the group name
+            group_name_prefix: a custom group identifier to prefix the resulting group name, likely Juju topology and relative path context
 
         Raises:
             ValueError, when invalid rule format given.
@@ -257,15 +308,17 @@ def _from_str(self, yaml_str: str, *, group_name: Optional[str]=None, group_name
             yaml_str = cast(OfficialRuleFileFormat, yaml_str)
             groups = yaml_str["groups"]
         elif self._is_single_rule_format(cast(SingleRuleFormat, yaml_str), self.rule_type):
-            # convert to list of groups
-            # group name is made up from the file name
             yaml_str = cast(SingleRuleFormat, yaml_str)
             if not group_name:
-                # Note: the caller of this function should make sure this never happens:
+                # Note: the caller of this function should ensure this never happens:
                 # Either we use the standard format, or we'd pass a group_name.
                 # If/when we drop support for the single-rule-per-file format, this won't
                 # be needed anymore.
-                group_name = hashlib.shake_256(str(yaml_str).encode("utf-8")).hexdigest(20)
+                group_name = hashlib.shake_256(str(yaml_str).encode("utf-8")).hexdigest(10)
+            else:
+                group_name = self._sanitize_metric_name(group_name)
+
+            # convert to list of groups to match official rule format
             groups = [{"name": group_name, "rules": [yaml_str]}]
         else:
             # invalid/unsupported
@@ -276,7 +329,9 @@ def _from_str(self, yaml_str: str, *, group_name: Optional[str]=None, group_name
         for group in groups:
             if not self._is_already_modified(group["name"]):
                 # update group name with topology and sub-path
-                group["name"] = "_".join(filter(None, [group_name_prefix, group_name, f"{self.rule_type}s"]))
+                group["name"] = "_".join(
+                    filter(None, [group_name_prefix, group["name"], f"{self.rule_type}s"])
+                )
 
             # add "juju_" topology labels
             for rule in group["rules"]:
@@ -303,45 +358,6 @@ def _from_str(self, yaml_str: str, *, group_name: Optional[str]=None, group_name
 
         return groups
 
-    def _from_file(  # noqa: C901
-        self, root_path: Path, file_path: Path
-    ) -> List[OfficialRuleFileItem]:
-        """Read a rules file from path, injecting juju topology.
-
-        Args:
-            root_path: full path to the root rules folder (used only for generating group name)
-            file_path: full path to a *.rule file.
-
-        Returns:
-            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 = os.path.relpath(os.path.dirname(file_path), root_path)
-            rel_path = "" if rel_path == "." else rel_path.replace(os.path.sep, "_")
-            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_str(rule_file, 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 _is_already_modified(self, name: str) -> bool:
         """Detect whether a group name has already been modified with juju topology."""
         modified_matcher = re.compile(r"^.*?_[\da-f]{8}_.*?alerts$")
@@ -349,12 +365,31 @@ def _is_already_modified(self, name: str) -> bool:
             return False
         return True
 
+    def _sanitize_metric_name(self, metric_name: str) -> str:
+        """Sanitize a metric name according to https://prometheus.io/docs/concepts/data_model/#metric-names-and-labels."""
+        return "".join(char if re.match(r"[a-zA-Z0-9_:]", char) else "_" for char in metric_name)
+
     # ---- END STATIC HELPER METHODS --- #
 
-    def add(self, yaml_str: str):
-        temp = self._from_str(yaml_str)
-        pprint(temp)
-        self.groups.extend(temp)
+    def add(
+        self,
+        yaml_str: str,
+        group_name: Optional[str] = None,
+        group_name_prefix: Optional[str] = None,
+    ) -> None:
+        """Add rules from a string to the existing ruleset.
+
+        Args:
+            yaml_str: a single-rule or official-rule YAML string
+            group_name: a custom group name, used only if the new rule is of single-rule format
+            group_name_prefix: a custom group name prefix, used only if the new rule is of single-rule format
+        """
+        kwargs = {}
+        if group_name is not None:
+            kwargs["group_name"] = group_name
+        if group_name_prefix is not None:
+            kwargs["group_name_prefix"] = group_name_prefix
+        self.groups.extend(self._from_str(yaml_str, **kwargs))
 
     def add_path(self, dir_path: Union[str, Path], *, recursive: bool = False) -> None:
         """Add rules from a dir path.
@@ -365,9 +400,6 @@ def add_path(self, dir_path: Union[str, Path], *, recursive: bool = False) -> No
         Args:
             dir_path: either a rules file or a dir of rules files.
             recursive: whether to read files recursively or not (no impact if `path` is a file).
-
-        Returns:
-            True if path was added else False.
         """
         path = Path(dir_path) if isinstance(dir_path, str) else dir_path
         if path.is_dir():
diff --git a/tests/test_rules_promql.py b/tests/test_rules_promql.py
index be4c6a3..034f0b0 100644
--- a/tests/test_rules_promql.py
+++ b/tests/test_rules_promql.py
@@ -41,6 +41,7 @@ def test_each_alert_rule_is_topology_labeled(self):
                     self.assertIn("juju_application", labels)
                     self.assertIn("juju_model_uuid", labels)
                     # alerts should not have unit information if not already present
+                    # https://discourse.charmhub.io/t/juju-topology-labels/8874
                     self.assertNotIn("juju_unit", rule["labels"])
                     self.assertNotIn("juju_unit=", rule["expr"])
                 else:
@@ -76,47 +77,188 @@ def test_each_alert_expression_is_topology_labeled(self):
                 self.assertIn("juju_model_uuid", labels)
                 self.assertIn("juju_application", labels)
 
-    def test_injecting_generic_alerts(self):
-        ri = AlertRules(
-            query_type="promql",
-            topology=JujuTopology(
-                model="unittest",
-                model_uuid=str(uuid.uuid4()),
-                unit="tester/0",
-                application="tester",
-            ),
-        )
-        ri.add_path(Path(__file__).resolve().parent / "promql_rules" / "prometheus_alert_rules")
-
-        # TODO SingleRuleFormat only require alert, expr, labels. Test others though
-        generic_alert_rules = """
-                groups:
-                - name: GenericRules
-                  rules:
-                  - alert: HostDown
-                    expr: up < 1
-                    for: 5m
-                    labels:
-                      severity: critical
-                  - alert: HostUnavailable
-                    expr: absent(up)
-                    for: 5m
-                    labels:
-                      severity: critical
-                """
-
-        ri.add(generic_alert_rules)
 
-        alerts = ri.as_dict()
-        self.assertIn("groups", alerts)
-        self.assertEqual(len(alerts["groups"]), 5)
-        group = alerts["groups"][0]
-        for rule in group["rules"]:
-            self.assertIn("expr", rule)
-            for labels in expression_labels(rule["expr"]):
-                self.assertIn("juju_model", labels)
-                self.assertIn("juju_model_uuid", labels)
-                self.assertIn("juju_application", labels)
+class TestAddRuleFromStr(unittest.TestCase):
+    def setUp(self):
+        rule = """
+            alert: SomeRuleName
+            expr: up
+            labels:
+              severity: critical
+        """
+        self.rule = yaml.safe_load(textwrap.dedent(rule))
+
+    def test_official_rule_add_alerts_from_string(self):
+        official_rule = """
+            groups:
+              - name: SomeGroupName
+                rules:
+                - alert: FooRule
+                  expr: up < 1
+                  labels:
+                    severity: critical
+                - alert: BarRule
+                  expr: absent(up)
+                  labels:
+                    severity: critical
+        """
+        added_group = yaml.safe_load(textwrap.dedent(official_rule))["groups"][0]
+        expected_rules = {
+            "groups": [
+                {
+                    "name": "initial_alerts",
+                    "rules": [self.rule],
+                },
+                {"name": "SomeGroupName_alerts", "rules": added_group["rules"]},
+            ]
+        }
+        # GIVEN an alert rule
+        rules = AlertRules(query_type="promql")
+        rules.groups = rules._from_str(self.rule, group_name="initial")
+        # WHEN a rule is added from string in official-rule format
+        rules.add(yaml.safe_load(textwrap.dedent(official_rule)))
+        rules_dict = rules.as_dict()
+        # THEN the new rule is a combination of all
+        self.assertEqual({}, DeepDiff(expected_rules, rules_dict))
+
+    def test_single_rule_add_alerts_from_string(self):
+        single_rule = """
+            alert: BarRule
+            expr: up < 1
+            labels:
+              severity: critical
+        """
+        rules = yaml.safe_load(textwrap.dedent(single_rule))
+        expected_rules = {
+            "groups": [
+                {
+                    "name": "initial_alerts",
+                    "rules": [self.rule],
+                },
+                {"name": "some_new_alerts", "rules": [rules]},
+            ]
+        }
+        # GIVEN an alert rule
+        rules = AlertRules(query_type="promql")
+        rules.groups = rules._from_str(self.rule, group_name="initial")
+        # WHEN a rule is added from string in single-rule format with a custom group name and prefix
+        rules.add(
+            yaml.safe_load(textwrap.dedent(single_rule)),
+            group_name="new",
+            group_name_prefix="some",
+        )
+        rules_dict = rules.as_dict()
+        # THEN the new rule is a combination of all
+        # AND the added rule group name has the custom group name and prefix
+        self.assertEqual({}, DeepDiff(expected_rules, rules_dict))
+
+
+class TestFromStrGroupName(unittest.TestCase):
+    def setUp(self):
+        official_rule = """
+            groups:
+              - name: SomeGroupName
+                rules:
+                - alert: SomeRuleName
+                  expr: up < 1
+                  for: 5m
+                  labels:
+                    severity: critical
+        """
+        single_rule = """
+            alert: SomeRuleName
+            expr: up < 1
+            for: 5m
+            labels:
+              severity: critical
+        """
+        self.official_rule = yaml.safe_load(textwrap.dedent(official_rule))
+        self.single_rule = yaml.safe_load(textwrap.dedent(single_rule))
+
+    def test_single_rule_from_string_group_sanitized(self):
+        # GIVEN an alert rule in single-rule format
+        rules = AlertRules(query_type="promql")
+        # WHEN processed as string and provided an illegal custom group name
+        groups = rules._from_str(self.single_rule, group_name="Foo$123/Hello:World(go_od)bye")
+        for group in groups:
+            # THEN the group name only contains characters in [a-zA-Z0-9_:]
+            # AND specials characters are replaced with "_"
+            self.assertEqual(group["name"], "Foo_123_Hello:World_go_od_bye_alerts")
+
+    def test_single_rule_from_string(self):
+        # GIVEN an alert rule in single-rule format
+        rules = AlertRules(query_type="promql")
+        # WHEN processed as string
+        groups = rules._from_str(self.single_rule)
+        for group in groups:
+            # THEN group name contains hash
+            self.assertNotEqual(get_hash(group["name"]), "")
+
+    def test_single_rule_from_string_custom_group(self):
+        # GIVEN an alert rule in single-rule format
+        rules = AlertRules(query_type="promql")
+        # WHEN processed as string and provided custom group name
+        groups = rules._from_str(self.single_rule, group_name="foo")
+        for group in groups:
+            # THEN group name does not contain hash
+            self.assertEqual(get_hash(group["name"]), "")
+            # AND group name contains the custom group name
+            self.assertIn("foo", group["name"])
+
+    def test_single_rule_from_string_custom_group_prefix(self):
+        # GIVEN an alert rule in single-rule format
+        rules = AlertRules(query_type="promql")
+        # WHEN processed as string and provided custom group name prefix
+        groups = rules._from_str(self.single_rule, group_name_prefix="foo")
+        for group in groups:
+            # THEN group name does not include the original group name
+            self.assertNotIn("SomeGroupName", group["name"])
+            # AND group name is prefixed with custom value
+            self.assertTrue(group["name"].startswith("foo"))
+
+    def test_official_rule_from_string(self):
+        # GIVEN an alert rule in official-rule format
+        rules = AlertRules(query_type="promql")
+        # WHEN processed as string
+        groups = rules._from_str(self.official_rule)
+        for group in groups:
+            # THEN group name matches the group name in the alert
+            self.assertIn("SomeGroupName", group["name"])
+
+    def test_official_rule_from_string_custom_group_prefix(self):
+        # GIVEN an alert rule in official-rule format
+        rules = AlertRules(query_type="promql")
+        # WHEN processed as string and provided custom group name prefix
+        groups = rules._from_str(self.official_rule, group_name_prefix="foo")
+        for group in groups:
+            # THEN group name includes the original group name
+            self.assertIn("SomeGroupName", group["name"])
+            # AND group name is prefixed with custom value
+            self.assertTrue(group["name"].startswith("foo"))
+
+    def test_raises_value_error_empty_str(self):
+        rules = AlertRules(query_type="promql")
+        with self.assertRaises(ValueError) as ctx:
+            rules._from_str("")
+        self.assertEqual(str(ctx.exception), "Empty")
+
+    def test_raises_value_error_not_dict(self):
+        rules = AlertRules(query_type="promql")
+        with self.assertRaises(ValueError) as ctx:
+            rules._from_str("[]")
+        self.assertEqual(str(ctx.exception), "Invalid rules (must be a dict)")
+
+    def test_raises_value_error_invalid_rule_format(self):
+        invalid_rule = """
+            expr: up < 1
+            for: 5m
+            labels:
+              severity: critical
+        """
+        rules = AlertRules(query_type="promql")
+        with self.assertRaises(ValueError) as ctx:
+            rules._from_str(yaml.safe_load(textwrap.dedent(invalid_rule)))
+        self.assertEqual(str(ctx.exception), "Invalid rule format")
 
 
 def sorted_matchers(matchers) -> str:
@@ -142,6 +284,27 @@ def expression_labels(expr):
         yield labels
 
 
+def get_hash(group_name: str) -> str:
+    """Extract the hash of the group name when a group name was not provided to _from_str method in the Rules class.
+
+    This occurs when the single-rule format is provided rather than official-rule format.
+
+    Args:
+        group_name: a string representing the group name of the rules.
+
+    Returns:
+        the matching hexdigest of the group name if a match is found, otherwise an empty string.
+    """
+    import re
+
+    pattern = r"([a-f0-9]{20})_"
+    match = re.search(pattern, group_name)
+    if match:
+        return match.group(1)
+    else:
+        return ""
+
+
 class TestAlertRulesWithOneRulePerFile(unittest.TestCase):
     def setUp(self) -> None:
         free_standing_rule = {

From caf4c8477a8e610652f357f5faa0368c40fbcbe6 Mon Sep 17 00:00:00 2001
From: Michael Thamm <mike.thamm@canonical.com>
Date: Tue, 17 Dec 2024 17:16:41 -0500
Subject: [PATCH 04/11] update static checks and remove type casting for single
 and official rule formats

---
 src/cosl/rules.py          | 34 ++++++++++++++++------------------
 tests/test_rules_promql.py |  6 ------
 2 files changed, 16 insertions(+), 24 deletions(-)

diff --git a/src/cosl/rules.py b/src/cosl/rules.py
index 691419f..3e30d2a 100644
--- a/src/cosl/rules.py
+++ b/src/cosl/rules.py
@@ -87,11 +87,9 @@
 
 from . import CosTool, JujuTopology
 from .types import (
-    OfficialRuleFileFormat,
     OfficialRuleFileItem,
     QueryType,
     RuleType,
-    SingleRuleFormat,
 )
 
 logger = logging.getLogger(__name__)
@@ -150,7 +148,7 @@ def __init__(self, query_type: QueryType, topology: Optional[JujuTopology] = Non
         self.query_type = query_type
         self.topology = topology
         self.tool = CosTool(default_query_type=query_type)
-        self.groups = []  # type: List[Dict[str, Any]]
+        self.groups: List[OfficialRuleFileItem] = []
 
     @property
     @abstractmethod
@@ -161,7 +159,7 @@ def rule_type(self) -> RuleType:
     # --- HELPER METHODS FOR READING FILES, SHOULD BE STATIC --- #
 
     @staticmethod
-    def _is_official_rule_format(rules_dict: OfficialRuleFileFormat) -> bool:
+    def _is_official_rule_format(rules_dict: Dict[str, Any]) -> bool:
         """Are rules in the upstream format as supported by Prometheus or Loki.
 
         Rules in dictionary format are in "official" form if they
@@ -177,7 +175,7 @@ def _is_official_rule_format(rules_dict: OfficialRuleFileFormat) -> bool:
         return "groups" in rules_dict
 
     @staticmethod
-    def _is_single_rule_format(rules_dict: SingleRuleFormat, rule_type: RuleType) -> bool:
+    def _is_single_rule_format(rules_dict: Dict[str, Any], rule_type: RuleType) -> bool:
         """Are alert rules in single rule format.
 
         This library supports reading of rules in a custom format that
@@ -213,7 +211,7 @@ def _multi_suffix_glob(
         all_files_in_dir = dir_path.glob("**/*" if recursive else "*")
         return list(filter(lambda f: f.is_file() and f.suffix in suffixes, all_files_in_dir))
 
-    def _from_dir(self, dir_path: Path, recursive: bool) -> List[Dict[str, Any]]:
+    def _from_dir(self, dir_path: Path, recursive: bool) -> List[OfficialRuleFileItem]:
         """Read all rule files in a directory.
 
         All rules from files for the same directory are loaded into a single
@@ -228,7 +226,7 @@ def _from_dir(self, dir_path: Path, recursive: bool) -> List[Dict[str, Any]]:
             a list of dictionaries representing prometheus rule groups, each dictionary
             representing a group (structure determined by `yaml.safe_load`).
         """
-        groups = []  # type: List[Dict[str, Any]]
+        groups: List[OfficialRuleFileItem] = []
 
         # Gather all records into a list of groups
         for file_path in Rules._multi_suffix_glob(
@@ -284,7 +282,7 @@ def _from_file(  # noqa: C901
 
     def _from_str(
         self,
-        yaml_str: str,
+        yaml_str: Dict[str, Any],
         *,
         group_name: Optional[str] = None,
         group_name_prefix: Optional[str] = None,
@@ -292,7 +290,7 @@ def _from_str(
         """Process rules from string, injecting juju topology. If a single-rule format is provided, a hash of the yaml file is injected into the group name to ensure uniqueness.
 
         Args:
-            yaml_str: rules content in single-rule or official-rule format as a string
+            yaml_str: rules content in single-rule or official-rule format as a YAML dict
             group_name: a custom identifier for the rule name to include in the group name
             group_name_prefix: a custom group identifier to prefix the resulting group name, likely Juju topology and relative path context
 
@@ -301,14 +299,13 @@ def _from_str(
         """
         if not yaml_str:
             raise ValueError("Empty")
-        if not isinstance(yaml_str, dict):
-            raise ValueError("Invalid rules (must be a dict)")
 
-        if self._is_official_rule_format(cast(OfficialRuleFileFormat, yaml_str)):
-            yaml_str = cast(OfficialRuleFileFormat, yaml_str)
+        if self._is_official_rule_format(yaml_str):
+            # TODO DO not cast since pyright knows its a str, we still need to
+            yaml_str = yaml_str
             groups = yaml_str["groups"]
-        elif self._is_single_rule_format(cast(SingleRuleFormat, yaml_str), self.rule_type):
-            yaml_str = cast(SingleRuleFormat, yaml_str)
+        elif self._is_single_rule_format(yaml_str, self.rule_type):
+            yaml_str = yaml_str
             if not group_name:
                 # Note: the caller of this function should ensure this never happens:
                 # Either we use the standard format, or we'd pass a group_name.
@@ -367,24 +364,25 @@ def _is_already_modified(self, name: str) -> bool:
 
     def _sanitize_metric_name(self, metric_name: str) -> str:
         """Sanitize a metric name according to https://prometheus.io/docs/concepts/data_model/#metric-names-and-labels."""
+        # TODO re.sub() instead of match. Check other examples in code base
         return "".join(char if re.match(r"[a-zA-Z0-9_:]", char) else "_" for char in metric_name)
 
     # ---- END STATIC HELPER METHODS --- #
 
     def add(
         self,
-        yaml_str: str,
+        yaml_str: Dict[str, Any],
         group_name: Optional[str] = None,
         group_name_prefix: Optional[str] = None,
     ) -> None:
         """Add rules from a string to the existing ruleset.
 
         Args:
-            yaml_str: a single-rule or official-rule YAML string
+            yaml_str: a single-rule or official-rule YAML dict
             group_name: a custom group name, used only if the new rule is of single-rule format
             group_name_prefix: a custom group name prefix, used only if the new rule is of single-rule format
         """
-        kwargs = {}
+        kwargs: Dict[str, str] = {}
         if group_name is not None:
             kwargs["group_name"] = group_name
         if group_name_prefix is not None:
diff --git a/tests/test_rules_promql.py b/tests/test_rules_promql.py
index 034f0b0..9243359 100644
--- a/tests/test_rules_promql.py
+++ b/tests/test_rules_promql.py
@@ -242,12 +242,6 @@ def test_raises_value_error_empty_str(self):
             rules._from_str("")
         self.assertEqual(str(ctx.exception), "Empty")
 
-    def test_raises_value_error_not_dict(self):
-        rules = AlertRules(query_type="promql")
-        with self.assertRaises(ValueError) as ctx:
-            rules._from_str("[]")
-        self.assertEqual(str(ctx.exception), "Invalid rules (must be a dict)")
-
     def test_raises_value_error_invalid_rule_format(self):
         invalid_rule = """
             expr: up < 1

From 15b88ab9e3831ae58e227690a20af985ea84ad25 Mon Sep 17 00:00:00 2001
From: Michael Thamm <mike.thamm@canonical.com>
Date: Wed, 18 Dec 2024 09:32:41 -0500
Subject: [PATCH 05/11] chore: remove TODOs

---
 src/cosl/rules.py | 2 --
 1 file changed, 2 deletions(-)

diff --git a/src/cosl/rules.py b/src/cosl/rules.py
index 3e30d2a..cfabc04 100644
--- a/src/cosl/rules.py
+++ b/src/cosl/rules.py
@@ -301,7 +301,6 @@ def _from_str(
             raise ValueError("Empty")
 
         if self._is_official_rule_format(yaml_str):
-            # TODO DO not cast since pyright knows its a str, we still need to
             yaml_str = yaml_str
             groups = yaml_str["groups"]
         elif self._is_single_rule_format(yaml_str, self.rule_type):
@@ -364,7 +363,6 @@ def _is_already_modified(self, name: str) -> bool:
 
     def _sanitize_metric_name(self, metric_name: str) -> str:
         """Sanitize a metric name according to https://prometheus.io/docs/concepts/data_model/#metric-names-and-labels."""
-        # TODO re.sub() instead of match. Check other examples in code base
         return "".join(char if re.match(r"[a-zA-Z0-9_:]", char) else "_" for char in metric_name)
 
     # ---- END STATIC HELPER METHODS --- #

From f8a3d53d65f39048f66c79d09061b753654f0170 Mon Sep 17 00:00:00 2001
From: Michael Thamm <mike.thamm@canonical.com>
Date: Wed, 18 Dec 2024 10:38:46 -0500
Subject: [PATCH 06/11] chore: remove os.path with pathlib

---
 src/cosl/rules.py | 5 ++---
 1 file changed, 2 insertions(+), 3 deletions(-)

diff --git a/src/cosl/rules.py b/src/cosl/rules.py
index cfabc04..8526e06 100644
--- a/src/cosl/rules.py
+++ b/src/cosl/rules.py
@@ -77,7 +77,6 @@
 
 import hashlib
 import logging
-import os
 import re
 from abc import ABC, abstractmethod
 from pathlib import Path
@@ -264,8 +263,8 @@ def _from_file(  # noqa: C901
             # Generate group name prefix
             #  - name, from juju topology
             #  - suffix, from the relative path of the rule file;
-            rel_path = os.path.relpath(os.path.dirname(file_path), root_path)
-            rel_path = "" if rel_path == "." else rel_path.replace(os.path.sep, "_")
+            rel_path = file_path.parent.relative_to(root_path)
+            rel_path = "" if rel_path == Path(".") else rel_path.as_posix().replace("/", "_")
             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))

From 300298a7f15b0c86c4856eccdfba280d49cda04f Mon Sep 17 00:00:00 2001
From: Michael Thamm <mike.thamm@canonical.com>
Date: Thu, 19 Dec 2024 10:39:33 -0500
Subject: [PATCH 07/11] chore: PR cleanup from suggestions

---
 src/cosl/rules.py          | 17 ++++++-----------
 tests/test_rules_promql.py | 12 ++++++++----
 2 files changed, 14 insertions(+), 15 deletions(-)

diff --git a/src/cosl/rules.py b/src/cosl/rules.py
index 8526e06..e7547bb 100644
--- a/src/cosl/rules.py
+++ b/src/cosl/rules.py
@@ -264,7 +264,7 @@ def _from_file(  # noqa: C901
             #  - 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 rel_path.as_posix().replace("/", "_")
+            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))
@@ -300,18 +300,14 @@ def _from_str(
             raise ValueError("Empty")
 
         if self._is_official_rule_format(yaml_str):
-            yaml_str = yaml_str
             groups = yaml_str["groups"]
         elif self._is_single_rule_format(yaml_str, self.rule_type):
-            yaml_str = yaml_str
             if not group_name:
                 # Note: the caller of this function should ensure this never happens:
                 # Either we use the standard format, or we'd pass a group_name.
                 # If/when we drop support for the single-rule-per-file format, this won't
                 # be needed anymore.
                 group_name = hashlib.shake_256(str(yaml_str).encode("utf-8")).hexdigest(10)
-            else:
-                group_name = self._sanitize_metric_name(group_name)
 
             # convert to list of groups to match official rule format
             groups = [{"name": group_name, "rules": [yaml_str]}]
@@ -327,6 +323,8 @@ def _from_str(
                 group["name"] = "_".join(
                     filter(None, [group_name_prefix, group["name"], f"{self.rule_type}s"])
                 )
+            # after sanitizing we should not modify group["name"] anymore
+            group["name"] = self._sanitize_metric_name(group["name"])
 
             # add "juju_" topology labels
             for rule in group["rules"]:
@@ -379,12 +377,9 @@ def add(
             group_name: a custom group name, used only if the new rule is of single-rule format
             group_name_prefix: a custom group name prefix, used only if the new rule is of single-rule format
         """
-        kwargs: Dict[str, str] = {}
-        if group_name is not None:
-            kwargs["group_name"] = group_name
-        if group_name_prefix is not None:
-            kwargs["group_name_prefix"] = group_name_prefix
-        self.groups.extend(self._from_str(yaml_str, **kwargs))
+        self.groups.extend(
+            self._from_str(yaml_str, group_name=group_name, group_name_prefix=group_name_prefix)
+        )
 
     def add_path(self, dir_path: Union[str, Path], *, recursive: bool = False) -> None:
         """Add rules from a dir path.
diff --git a/tests/test_rules_promql.py b/tests/test_rules_promql.py
index 9243359..675a175 100644
--- a/tests/test_rules_promql.py
+++ b/tests/test_rules_promql.py
@@ -147,8 +147,8 @@ def test_single_rule_add_alerts_from_string(self):
             group_name="new",
             group_name_prefix="some",
         )
-        rules_dict = rules.as_dict()
         # THEN the new rule is a combination of all
+        rules_dict = rules.as_dict()
         # AND the added rule group name has the custom group name and prefix
         self.assertEqual({}, DeepDiff(expected_rules, rules_dict))
 
@@ -179,11 +179,15 @@ def test_single_rule_from_string_group_sanitized(self):
         # GIVEN an alert rule in single-rule format
         rules = AlertRules(query_type="promql")
         # WHEN processed as string and provided an illegal custom group name
-        groups = rules._from_str(self.single_rule, group_name="Foo$123/Hello:World(go_od)bye")
+        groups = rules._from_str(
+            self.single_rule, group_name="Foo$123/Hello:World(go_od)bye!@#^&*()[]{}|;:,.<>?`~_"
+        )
         for group in groups:
             # THEN the group name only contains characters in [a-zA-Z0-9_:]
-            # AND specials characters are replaced with "_"
-            self.assertEqual(group["name"], "Foo_123_Hello:World_go_od_bye_alerts")
+            # AND special characters are replaced with "_"
+            self.assertEqual(
+                group["name"], "Foo_123_Hello:World_go_od_bye______________:_________alerts"
+            )
 
     def test_single_rule_from_string(self):
         # GIVEN an alert rule in single-rule format

From 7b458eab1abcb19d0c6b06dfe73e64590aff5e4b Mon Sep 17 00:00:00 2001
From: Michael Thamm <mike.thamm@canonical.com>
Date: Thu, 19 Dec 2024 16:22:31 -0500
Subject: [PATCH 08/11] update naming from str to dict

---
 src/cosl/rules.py          | 32 ++++++++++++++-------------
 tests/test_rules_promql.py | 44 +++++++++++++++++++-------------------
 2 files changed, 39 insertions(+), 37 deletions(-)

diff --git a/src/cosl/rules.py b/src/cosl/rules.py
index e7547bb..2a3aef1 100644
--- a/src/cosl/rules.py
+++ b/src/cosl/rules.py
@@ -270,7 +270,7 @@ def _from_file(  # noqa: C901
             group_name_prefix = "_".join(filter(None, group_name_parts))
 
             try:
-                groups = self._from_str(
+                groups = self._from_dict(
                     rule_file, group_name=file_path.stem, group_name_prefix=group_name_prefix
                 )
             except ValueError as e:
@@ -279,38 +279,40 @@ def _from_file(  # noqa: C901
 
             return groups
 
-    def _from_str(
+    def _from_dict(
         self,
-        yaml_str: Dict[str, Any],
+        yaml_dict: Dict[str, Any],
         *,
         group_name: Optional[str] = None,
         group_name_prefix: Optional[str] = None,
     ) -> List[OfficialRuleFileItem]:
-        """Process rules from string, injecting juju topology. If a single-rule format is provided, a hash of the yaml file is injected into the group name to ensure uniqueness.
+        """Process rules from dict, injecting juju topology. If a single-rule format is provided, a hash of the yaml file is injected into the group name to ensure uniqueness.
 
         Args:
-            yaml_str: rules content in single-rule or official-rule format as a YAML dict
+            yaml_dict: rules content in single-rule or official-rule format as a YAML dict
             group_name: a custom identifier for the rule name to include in the group name
             group_name_prefix: a custom group identifier to prefix the resulting group name, likely Juju topology and relative path context
 
         Raises:
             ValueError, when invalid rule format given.
         """
-        if not yaml_str:
+        # TODO _alerts_alerts is leftover code from elsewhere
+        # TODO rename this to be yaml_dict
+        if not yaml_dict:
             raise ValueError("Empty")
 
-        if self._is_official_rule_format(yaml_str):
-            groups = yaml_str["groups"]
-        elif self._is_single_rule_format(yaml_str, self.rule_type):
+        if self._is_official_rule_format(yaml_dict):
+            groups = yaml_dict["groups"]
+        elif self._is_single_rule_format(yaml_dict, self.rule_type):
             if not group_name:
                 # Note: the caller of this function should ensure this never happens:
                 # Either we use the standard format, or we'd pass a group_name.
                 # If/when we drop support for the single-rule-per-file format, this won't
                 # be needed anymore.
-                group_name = hashlib.shake_256(str(yaml_str).encode("utf-8")).hexdigest(10)
+                group_name = hashlib.shake_256(str(yaml_dict).encode("utf-8")).hexdigest(10)
 
             # convert to list of groups to match official rule format
-            groups = [{"name": group_name, "rules": [yaml_str]}]
+            groups = [{"name": group_name, "rules": [yaml_dict]}]
         else:
             # invalid/unsupported
             raise ValueError("Invalid rule format")
@@ -366,19 +368,19 @@ def _sanitize_metric_name(self, metric_name: str) -> str:
 
     def add(
         self,
-        yaml_str: Dict[str, Any],
+        yaml_dict: Dict[str, Any],
         group_name: Optional[str] = None,
         group_name_prefix: Optional[str] = None,
     ) -> None:
-        """Add rules from a string to the existing ruleset.
+        """Add rules from dict to the existing ruleset.
 
         Args:
-            yaml_str: a single-rule or official-rule YAML dict
+            yaml_dict: a single-rule or official-rule YAML dict
             group_name: a custom group name, used only if the new rule is of single-rule format
             group_name_prefix: a custom group name prefix, used only if the new rule is of single-rule format
         """
         self.groups.extend(
-            self._from_str(yaml_str, group_name=group_name, group_name_prefix=group_name_prefix)
+            self._from_dict(yaml_dict, group_name=group_name, group_name_prefix=group_name_prefix)
         )
 
     def add_path(self, dir_path: Union[str, Path], *, recursive: bool = False) -> None:
diff --git a/tests/test_rules_promql.py b/tests/test_rules_promql.py
index 675a175..a018370 100644
--- a/tests/test_rules_promql.py
+++ b/tests/test_rules_promql.py
@@ -88,7 +88,7 @@ def setUp(self):
         """
         self.rule = yaml.safe_load(textwrap.dedent(rule))
 
-    def test_official_rule_add_alerts_from_string(self):
+    def test_official_rule_add_alerts_from_dict(self):
         official_rule = """
             groups:
               - name: SomeGroupName
@@ -114,14 +114,14 @@ def test_official_rule_add_alerts_from_string(self):
         }
         # GIVEN an alert rule
         rules = AlertRules(query_type="promql")
-        rules.groups = rules._from_str(self.rule, group_name="initial")
+        rules.groups = rules._from_dict(self.rule, group_name="initial")
         # WHEN a rule is added from string in official-rule format
         rules.add(yaml.safe_load(textwrap.dedent(official_rule)))
         rules_dict = rules.as_dict()
         # THEN the new rule is a combination of all
         self.assertEqual({}, DeepDiff(expected_rules, rules_dict))
 
-    def test_single_rule_add_alerts_from_string(self):
+    def test_single_rule_add_alerts_from_dict(self):
         single_rule = """
             alert: BarRule
             expr: up < 1
@@ -140,7 +140,7 @@ def test_single_rule_add_alerts_from_string(self):
         }
         # GIVEN an alert rule
         rules = AlertRules(query_type="promql")
-        rules.groups = rules._from_str(self.rule, group_name="initial")
+        rules.groups = rules._from_dict(self.rule, group_name="initial")
         # WHEN a rule is added from string in single-rule format with a custom group name and prefix
         rules.add(
             yaml.safe_load(textwrap.dedent(single_rule)),
@@ -153,7 +153,7 @@ def test_single_rule_add_alerts_from_string(self):
         self.assertEqual({}, DeepDiff(expected_rules, rules_dict))
 
 
-class TestFromStrGroupName(unittest.TestCase):
+class TestFromDictGroupName(unittest.TestCase):
     def setUp(self):
         official_rule = """
             groups:
@@ -175,11 +175,11 @@ def setUp(self):
         self.official_rule = yaml.safe_load(textwrap.dedent(official_rule))
         self.single_rule = yaml.safe_load(textwrap.dedent(single_rule))
 
-    def test_single_rule_from_string_group_sanitized(self):
+    def test_single_rule_from_dict_group_sanitized(self):
         # GIVEN an alert rule in single-rule format
         rules = AlertRules(query_type="promql")
         # WHEN processed as string and provided an illegal custom group name
-        groups = rules._from_str(
+        groups = rules._from_dict(
             self.single_rule, group_name="Foo$123/Hello:World(go_od)bye!@#^&*()[]{}|;:,.<>?`~_"
         )
         for group in groups:
@@ -189,61 +189,61 @@ def test_single_rule_from_string_group_sanitized(self):
                 group["name"], "Foo_123_Hello:World_go_od_bye______________:_________alerts"
             )
 
-    def test_single_rule_from_string(self):
+    def test_single_rule_from_dict(self):
         # GIVEN an alert rule in single-rule format
         rules = AlertRules(query_type="promql")
         # WHEN processed as string
-        groups = rules._from_str(self.single_rule)
+        groups = rules._from_dict(self.single_rule)
         for group in groups:
             # THEN group name contains hash
             self.assertNotEqual(get_hash(group["name"]), "")
 
-    def test_single_rule_from_string_custom_group(self):
+    def test_single_rule_from_dict_custom_group(self):
         # GIVEN an alert rule in single-rule format
         rules = AlertRules(query_type="promql")
         # WHEN processed as string and provided custom group name
-        groups = rules._from_str(self.single_rule, group_name="foo")
+        groups = rules._from_dict(self.single_rule, group_name="foo")
         for group in groups:
             # THEN group name does not contain hash
             self.assertEqual(get_hash(group["name"]), "")
             # AND group name contains the custom group name
             self.assertIn("foo", group["name"])
 
-    def test_single_rule_from_string_custom_group_prefix(self):
+    def test_single_rule_from_dict_custom_group_prefix(self):
         # GIVEN an alert rule in single-rule format
         rules = AlertRules(query_type="promql")
         # WHEN processed as string and provided custom group name prefix
-        groups = rules._from_str(self.single_rule, group_name_prefix="foo")
+        groups = rules._from_dict(self.single_rule, group_name_prefix="foo")
         for group in groups:
             # THEN group name does not include the original group name
             self.assertNotIn("SomeGroupName", group["name"])
             # AND group name is prefixed with custom value
             self.assertTrue(group["name"].startswith("foo"))
 
-    def test_official_rule_from_string(self):
+    def test_official_rule_from_dict(self):
         # GIVEN an alert rule in official-rule format
         rules = AlertRules(query_type="promql")
-        # WHEN processed as string
-        groups = rules._from_str(self.official_rule)
+        # WHEN processed as dict
+        groups = rules._from_dict(self.official_rule)
         for group in groups:
             # THEN group name matches the group name in the alert
             self.assertIn("SomeGroupName", group["name"])
 
-    def test_official_rule_from_string_custom_group_prefix(self):
+    def test_official_rule_from_dict_custom_group_prefix(self):
         # GIVEN an alert rule in official-rule format
         rules = AlertRules(query_type="promql")
         # WHEN processed as string and provided custom group name prefix
-        groups = rules._from_str(self.official_rule, group_name_prefix="foo")
+        groups = rules._from_dict(self.official_rule, group_name_prefix="foo")
         for group in groups:
             # THEN group name includes the original group name
             self.assertIn("SomeGroupName", group["name"])
             # AND group name is prefixed with custom value
             self.assertTrue(group["name"].startswith("foo"))
 
-    def test_raises_value_error_empty_str(self):
+    def test_raises_value_error_empty_dict(self):
         rules = AlertRules(query_type="promql")
         with self.assertRaises(ValueError) as ctx:
-            rules._from_str("")
+            rules._from_dict("")
         self.assertEqual(str(ctx.exception), "Empty")
 
     def test_raises_value_error_invalid_rule_format(self):
@@ -255,7 +255,7 @@ def test_raises_value_error_invalid_rule_format(self):
         """
         rules = AlertRules(query_type="promql")
         with self.assertRaises(ValueError) as ctx:
-            rules._from_str(yaml.safe_load(textwrap.dedent(invalid_rule)))
+            rules._from_dict(yaml.safe_load(textwrap.dedent(invalid_rule)))
         self.assertEqual(str(ctx.exception), "Invalid rule format")
 
 
@@ -283,7 +283,7 @@ def expression_labels(expr):
 
 
 def get_hash(group_name: str) -> str:
-    """Extract the hash of the group name when a group name was not provided to _from_str method in the Rules class.
+    """Extract the hash of the group name when a group name was not provided to _from_dict method in the Rules class.
 
     This occurs when the single-rule format is provided rather than official-rule format.
 

From 502a0a919f7529b741de1f0e3146b84c1671505a Mon Sep 17 00:00:00 2001
From: Michael Thamm <mike.thamm@canonical.com>
Date: Fri, 20 Dec 2024 11:31:48 -0500
Subject: [PATCH 09/11] chore: switch yaml_dict to rule_dict

---
 src/cosl/rules.py | 24 +++++++++++-------------
 1 file changed, 11 insertions(+), 13 deletions(-)

diff --git a/src/cosl/rules.py b/src/cosl/rules.py
index 2a3aef1..b12adc9 100644
--- a/src/cosl/rules.py
+++ b/src/cosl/rules.py
@@ -281,7 +281,7 @@ def _from_file(  # noqa: C901
 
     def _from_dict(
         self,
-        yaml_dict: Dict[str, Any],
+        rule_dict: Dict[str, Any],
         *,
         group_name: Optional[str] = None,
         group_name_prefix: Optional[str] = None,
@@ -289,30 +289,28 @@ def _from_dict(
         """Process rules from dict, injecting juju topology. If a single-rule format is provided, a hash of the yaml file is injected into the group name to ensure uniqueness.
 
         Args:
-            yaml_dict: rules content in single-rule or official-rule format as a YAML dict
+            rule_dict: rules content in single-rule or official-rule format as a YAML dict
             group_name: a custom identifier for the rule name to include in the group name
             group_name_prefix: a custom group identifier to prefix the resulting group name, likely Juju topology and relative path context
 
         Raises:
             ValueError, when invalid rule format given.
         """
-        # TODO _alerts_alerts is leftover code from elsewhere
-        # TODO rename this to be yaml_dict
-        if not yaml_dict:
+        if not rule_dict:
             raise ValueError("Empty")
 
-        if self._is_official_rule_format(yaml_dict):
-            groups = yaml_dict["groups"]
-        elif self._is_single_rule_format(yaml_dict, self.rule_type):
+        if self._is_official_rule_format(rule_dict):
+            groups = rule_dict["groups"]
+        elif self._is_single_rule_format(rule_dict, self.rule_type):
             if not group_name:
                 # Note: the caller of this function should ensure this never happens:
                 # Either we use the standard format, or we'd pass a group_name.
                 # If/when we drop support for the single-rule-per-file format, this won't
                 # be needed anymore.
-                group_name = hashlib.shake_256(str(yaml_dict).encode("utf-8")).hexdigest(10)
+                group_name = hashlib.shake_256(str(rule_dict).encode("utf-8")).hexdigest(10)
 
             # convert to list of groups to match official rule format
-            groups = [{"name": group_name, "rules": [yaml_dict]}]
+            groups = [{"name": group_name, "rules": [rule_dict]}]
         else:
             # invalid/unsupported
             raise ValueError("Invalid rule format")
@@ -368,19 +366,19 @@ def _sanitize_metric_name(self, metric_name: str) -> str:
 
     def add(
         self,
-        yaml_dict: Dict[str, Any],
+        rule_dict: Dict[str, Any],
         group_name: Optional[str] = None,
         group_name_prefix: Optional[str] = None,
     ) -> None:
         """Add rules from dict to the existing ruleset.
 
         Args:
-            yaml_dict: a single-rule or official-rule YAML dict
+            rule_dict: a single-rule or official-rule YAML dict
             group_name: a custom group name, used only if the new rule is of single-rule format
             group_name_prefix: a custom group name prefix, used only if the new rule is of single-rule format
         """
         self.groups.extend(
-            self._from_dict(yaml_dict, group_name=group_name, group_name_prefix=group_name_prefix)
+            self._from_dict(rule_dict, group_name=group_name, group_name_prefix=group_name_prefix)
         )
 
     def add_path(self, dir_path: Union[str, Path], *, recursive: bool = False) -> None:

From a3932fe0d7485539757fa4bdb312b4f03cdbbc23 Mon Sep 17 00:00:00 2001
From: Michael Thamm <mike.thamm@canonical.com>
Date: Fri, 20 Dec 2024 13:05:57 -0500
Subject: [PATCH 10/11] remove textwrap and accept dict as input to _from_str
 and add methods

---
 tests/test_rules_promql.py | 116 ++++++++++++++++++-------------------
 1 file changed, 57 insertions(+), 59 deletions(-)

diff --git a/tests/test_rules_promql.py b/tests/test_rules_promql.py
index a018370..38662e4 100644
--- a/tests/test_rules_promql.py
+++ b/tests/test_rules_promql.py
@@ -3,7 +3,6 @@
 
 import json
 import re
-import textwrap
 import unittest
 import uuid
 from pathlib import Path
@@ -80,62 +79,63 @@ def test_each_alert_expression_is_topology_labeled(self):
 
 class TestAddRuleFromStr(unittest.TestCase):
     def setUp(self):
-        rule = """
-            alert: SomeRuleName
-            expr: up
-            labels:
-              severity: critical
-        """
-        self.rule = yaml.safe_load(textwrap.dedent(rule))
+        self.rule = {
+            "alert": "SomeRuleName",
+            "expr": "up < 1",
+            "labels": {"severity": "critical"},
+        }
 
     def test_official_rule_add_alerts_from_dict(self):
-        official_rule = """
-            groups:
-              - name: SomeGroupName
-                rules:
-                - alert: FooRule
-                  expr: up < 1
-                  labels:
-                    severity: critical
-                - alert: BarRule
-                  expr: absent(up)
-                  labels:
-                    severity: critical
-        """
-        added_group = yaml.safe_load(textwrap.dedent(official_rule))["groups"][0]
+        official_rule = {
+            "groups": [
+                {
+                    "name": "SomeGroupName",
+                    "rules": [
+                        {
+                            "alert": "FooRule",
+                            "expr": "up < 1",
+                            "labels": {"severity": "critical"},
+                        },
+                        {
+                            "alert": "BarRule",
+                            "expr": "absent(up)",
+                            "labels": {"severity": "critical"},
+                        },
+                    ],
+                }
+            ]
+        }
         expected_rules = {
             "groups": [
                 {
                     "name": "initial_alerts",
                     "rules": [self.rule],
                 },
-                {"name": "SomeGroupName_alerts", "rules": added_group["rules"]},
+                {"name": "SomeGroupName_alerts", "rules": official_rule["groups"][0]["rules"]},
             ]
         }
         # GIVEN an alert rule
         rules = AlertRules(query_type="promql")
         rules.groups = rules._from_dict(self.rule, group_name="initial")
         # WHEN a rule is added from string in official-rule format
-        rules.add(yaml.safe_load(textwrap.dedent(official_rule)))
+        rules.add(official_rule)
         rules_dict = rules.as_dict()
         # THEN the new rule is a combination of all
         self.assertEqual({}, DeepDiff(expected_rules, rules_dict))
 
     def test_single_rule_add_alerts_from_dict(self):
-        single_rule = """
-            alert: BarRule
-            expr: up < 1
-            labels:
-              severity: critical
-        """
-        rules = yaml.safe_load(textwrap.dedent(single_rule))
+        single_rule = {
+            "alert": "BarRule",
+            "expr": "up < 1",
+            "labels": {"severity": "critical"},
+        }
         expected_rules = {
             "groups": [
                 {
                     "name": "initial_alerts",
                     "rules": [self.rule],
                 },
-                {"name": "some_new_alerts", "rules": [rules]},
+                {"name": "some_new_alerts", "rules": [single_rule]},
             ]
         }
         # GIVEN an alert rule
@@ -143,7 +143,7 @@ def test_single_rule_add_alerts_from_dict(self):
         rules.groups = rules._from_dict(self.rule, group_name="initial")
         # WHEN a rule is added from string in single-rule format with a custom group name and prefix
         rules.add(
-            yaml.safe_load(textwrap.dedent(single_rule)),
+            single_rule,
             group_name="new",
             group_name_prefix="some",
         )
@@ -155,25 +155,25 @@ def test_single_rule_add_alerts_from_dict(self):
 
 class TestFromDictGroupName(unittest.TestCase):
     def setUp(self):
-        official_rule = """
-            groups:
-              - name: SomeGroupName
-                rules:
-                - alert: SomeRuleName
-                  expr: up < 1
-                  for: 5m
-                  labels:
-                    severity: critical
-        """
-        single_rule = """
-            alert: SomeRuleName
-            expr: up < 1
-            for: 5m
-            labels:
-              severity: critical
-        """
-        self.official_rule = yaml.safe_load(textwrap.dedent(official_rule))
-        self.single_rule = yaml.safe_load(textwrap.dedent(single_rule))
+        self.official_rule = {
+            "groups": [
+                {
+                    "name": "SomeGroupName",
+                    "rules": [
+                        {
+                            "alert": "SomeRuleName",
+                            "expr": "up < 1",
+                            "labels": {"severity": "critical"},
+                        }
+                    ],
+                }
+            ]
+        }
+        self.single_rule = {
+            "alert": "SomeRuleName",
+            "expr": "up < 1",
+            "labels": {"severity": "critical"},
+        }
 
     def test_single_rule_from_dict_group_sanitized(self):
         # GIVEN an alert rule in single-rule format
@@ -247,15 +247,13 @@ def test_raises_value_error_empty_dict(self):
         self.assertEqual(str(ctx.exception), "Empty")
 
     def test_raises_value_error_invalid_rule_format(self):
-        invalid_rule = """
-            expr: up < 1
-            for: 5m
-            labels:
-              severity: critical
-        """
+        invalid_rule = {
+            "expr": "up < 1",
+            "labels": {"severity": "critical"},
+        }
         rules = AlertRules(query_type="promql")
         with self.assertRaises(ValueError) as ctx:
-            rules._from_dict(yaml.safe_load(textwrap.dedent(invalid_rule)))
+            rules._from_dict(invalid_rule)
         self.assertEqual(str(ctx.exception), "Invalid rule format")
 
 

From fc755eca84e71c2eade967cae9dc6ed3ec23340a Mon Sep 17 00:00:00 2001
From: Michael Thamm <mike.thamm@canonical.com>
Date: Fri, 20 Dec 2024 14:02:15 -0500
Subject: [PATCH 11/11] bump cos-lib version to 0.0.49

---
 pyproject.toml | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/pyproject.toml b/pyproject.toml
index ecddda0..fc2e25b 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "cosl"
-version = "0.0.48"
+version = "0.0.49"
 authors = [
     { name = "sed-i", email = "82407168+sed-i@users.noreply.github.com" },
 ]