diff --git a/pyproject.toml b/pyproject.toml index 7ac3103..e649fea 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "cosl" -version = "1.9.1" +version = "1.10.0" authors = [ { name = "sed-i", email = "82407168+sed-i@users.noreply.github.com" }, ] diff --git a/src/cosl/backends/__init__.py b/src/cosl/backends/__init__.py new file mode 100644 index 0000000..a2f44f9 --- /dev/null +++ b/src/cosl/backends/__init__.py @@ -0,0 +1,4 @@ +# Copyright 2026 Canonical Ltd. +# See LICENSE file for licensing details. + +"""Implementations of the RuleBackend class.""" diff --git a/src/cosl/backends/grouped_rules.py b/src/cosl/backends/grouped_rules.py new file mode 100644 index 0000000..f6bc424 --- /dev/null +++ b/src/cosl/backends/grouped_rules.py @@ -0,0 +1,186 @@ +# Copyright 2026 Canonical Ltd. +# See LICENSE file for licensing details. +"""Shared base for groups-based rule backends (Prometheus & Loki). + +This module provides :class:`_GroupedRuleBackend`, the common base class for +:class:`~cosl.prometheus.PrometheusRuleBackend` and +:class:`~cosl.loki.LokiRuleBackend`. +""" + +import copy +import hashlib +import logging +import re +from pathlib import Path +from typing import Any, Dict, List, Mapping, Optional, Tuple, cast + +import yaml + +from ..cos_tool import CosTool +from ..juju_topology import JujuTopology +from ..rules import RuleBackend +from ..types import ( + RULE_TYPES, + OfficialRuleFileFormat, + OfficialRuleFileItem, + QueryType, + SingleRuleFormat, +) + +logger = logging.getLogger(__name__) + + +class _GroupedRuleBackend(RuleBackend[OfficialRuleFileItem]): # type: ignore + """Shared base for Prometheus and Loki rule backends. + + Handles the groups-based rule format, topology injection into group names, + rule labels, and expressions. Subclasses only need to set :attr:`query_type`. + """ + + query_type: QueryType + + def __init__(self, topology: Optional[JujuTopology] = None) -> None: + super().__init__(topology=topology) + self.tool = CosTool(default_query_type=self.query_type) + + @property + def file_suffixes(self) -> List[str]: + return [".rule", ".rules", ".yml", ".yaml"] + + def from_dict( + self, + rule_dict: Mapping[str, Any], + **kwargs: Any, + ) -> List[OfficialRuleFileItem]: + """Parse a Prometheus/Loki rule dict, normalise, and inject topology. + + Args: + rule_dict: Raw rule content as a YAML-loaded dict. + **kwargs: Accepts ``group_name`` (str) — an identifier for the + group (typically the file stem), and ``group_name_prefix`` + (str) — a prefix for the group name (typically topology + identifier + relative path). + """ + group_name: Optional[str] = kwargs.get("group_name") + group_name_prefix: Optional[str] = kwargs.get("group_name_prefix") + + if not rule_dict: + raise ValueError("Empty") + + rule_copy = copy.deepcopy(rule_dict) + if self._is_official_format(rule_copy): + groups = [OfficialRuleFileItem(**g) for g in rule_copy.get("groups", [])] + elif self._is_single_rule_format(rule_copy): + single_rule = cast(SingleRuleFormat, rule_copy) + 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(single_rule).encode("utf-8")).hexdigest(10) + + # convert to list of groups to match official rule format + groups = [OfficialRuleFileItem(name=group_name, rules=[single_rule])] + else: + # invalid/unsupported + raise ValueError("Invalid rule format") + + # update rules with additional metadata + for group in groups: + if not self._is_already_modified(group["name"]): + # update group name with topology and sub-path + new_name = "_".join(filter(None, [group_name_prefix, group["name"]])) + if not new_name.endswith("_rules"): + new_name += "_rules" + group["name"] = new_name + # 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"]: + 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 + + # Inject topology matchers into the expression + repl = r'job=~".+"' if self.query_type == "logql" else "" + rule["expr"] = self.tool.inject_label_matchers( + 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( + self, + file_path: Path, + **kwargs: Any, + ) -> List[OfficialRuleFileItem]: + """Read a rule file, using file context for group naming. + + Args: + file_path: Absolute path to the rule file. + **kwargs: Must include ``root_path`` (:class:`~pathlib.Path`) — + the root rules directory used for computing relative paths + for group name prefixes. + """ + root_path: Path = kwargs.get("root_path", file_path.parent) + with file_path.open() as f: + try: + rule_file = yaml.safe_load(f) + except Exception as e: + logger.error("Failed to read rules from %s: %s", file_path.name, e) + return [] + + # Compute group name context from topology + relative path + rel_path = file_path.parent.relative_to(root_path) + rel_path_str = "" if rel_path == Path(".") else str(rel_path) + prefix_parts = [self.topology.identifier] if self.topology else [] + prefix_parts.append(rel_path_str) + group_name_prefix = "_".join(filter(None, prefix_parts)) + + try: + return 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 [] + + def validate(self, rules: Dict[str, List[OfficialRuleFileItem]]) -> Tuple[bool, str]: + """Validate rules using ``cos-tool``.""" + return self.tool.validate_alert_rules(OfficialRuleFileFormat(**rules)) + + def as_dict(self, items: List[OfficialRuleFileItem]) -> Dict[str, List[OfficialRuleFileItem]]: + """Serialise as ``{"groups": [...]}``.""" + return {"groups": items} if items else {} + + @staticmethod + def _is_official_format(rules_dict: Mapping[str, Any]) -> bool: + return "groups" in rules_dict + + @staticmethod + def _is_single_rule_format(rules_dict: Mapping[str, Any]) -> bool: + return "expr" in rules_dict and not RULE_TYPES.isdisjoint(rules_dict) + + @staticmethod + def _is_already_modified(name: str) -> bool: + """Detect whether a group name already contains a topology UUID hash.""" + return re.match(r"^.*?_[\da-f]{8}_.*?rules$", name) is not None + + @staticmethod + def _sanitize_metric_name(metric_name: str) -> str: + """Sanitize a metric name per the Prometheus data model.""" + return "".join(char if re.match(r"[a-zA-Z0-9_:]", char) else "_" for char in metric_name) diff --git a/src/cosl/backends/loki.py b/src/cosl/backends/loki.py new file mode 100644 index 0000000..3fafd2b --- /dev/null +++ b/src/cosl/backends/loki.py @@ -0,0 +1,40 @@ +# Copyright 2026 Canonical Ltd. +# See LICENSE file for licensing details. +"""Loki rule backend. + +This module provides :class:`LokiRuleBackend`, the Loki-specific +implementation of :class:`~cosl.rules.RuleBackend`. It is used by +:class:`~cosl.rules.AbstractRules` to load, validate, and manage +alert and recording rules written in LogQL. +""" + +from ..types import QueryType +from .grouped_rules import _GroupedRuleBackend # type: ignore + + +class LokiRuleBackend(_GroupedRuleBackend): + """Backend for Loki alerting / recording rules (LogQL). + + Inherits all behaviour from :class:`~cosl.backends.grouped_rules._GroupedRuleBackend` + and sets :attr:`query_type` to ``"logql"``. + + Responsibilities: + + * Parse rule files/dicts in the official Loki rule format or the + single-rule-per-file shorthand. + * Inject Juju topology labels into rule labels and LogQL expressions + via ``cos-tool``. + * Validate the resulting rules through ``cos-tool`` LogQL validation. + * Serialise the rules into the ``{"groups": [...]}`` format expected by + Loki. + + Usage:: + + from cosl.backends.loki import LokiRuleBackend + from cosl.rules import AbstractRules + + rules = AbstractRules(backend=LokiRuleBackend(topology=my_topology)) + rules.add_path("./loki_alert_rules") + """ + + query_type: QueryType = "logql" diff --git a/src/cosl/backends/prometheus.py b/src/cosl/backends/prometheus.py new file mode 100644 index 0000000..cd03259 --- /dev/null +++ b/src/cosl/backends/prometheus.py @@ -0,0 +1,40 @@ +# Copyright 2026 Canonical Ltd. +# See LICENSE file for licensing details. +"""Prometheus rule backend. + +This module provides :class:`PrometheusRuleBackend`, the Prometheus-specific +implementation of :class:`~cosl.rules.RuleBackend`. It is used by +:class:`~cosl.rules.AbstractRules` to load, validate, and manage +alert and recording rules written in PromQL. +""" + +from ..types import QueryType +from .grouped_rules import _GroupedRuleBackend # type: ignore + + +class PrometheusRuleBackend(_GroupedRuleBackend): + """Backend for Prometheus alerting / recording rules (PromQL). + + Inherits all behaviour from :class:`~cosl.backends.grouped_rules._GroupedRuleBackend` + and sets :attr:`query_type` to ``"promql"``. + + Responsibilities: + + * Parse rule files/dicts in the official Prometheus rule format or the + single-rule-per-file shorthand. + * Inject Juju topology labels into rule labels and PromQL expressions + via ``cos-tool``. + * Validate the resulting rules through ``cos-tool`` PromQL validation. + * Serialise the rules into the ``{"groups": [...]}`` format expected by + Prometheus. + + Usage:: + + from cosl.backends.prometheus import PrometheusRuleBackend + from cosl.rules import AbstractRules + + rules = AbstractRules(backend=PrometheusRuleBackend(topology=my_topology)) + rules.add_path("./prometheus_alert_rules") + """ + + query_type: QueryType = "promql" diff --git a/src/cosl/rules.py b/src/cosl/rules.py index 6e5230b..7a99418 100644 --- a/src/cosl/rules.py +++ b/src/cosl/rules.py @@ -1,10 +1,10 @@ # Copyright 2026 Canonical Ltd. # See LICENSE file for licensing details. -"""Alerting and Recording Rules. +"""The rules module. ## Overview -## Rules +## Rules class (Legacy) This library also supports gathering alerting and recording rules from all related charms and enabling corresponding alerting/recording rules within the @@ -73,13 +73,45 @@ - `juju_model` - `juju_model_uuid` - `juju_application` -""" # noqa: W505 + +## AbstractRules (Latest) + +The ``AbstractRules`` class is a format-agnostic aggregator that collects alerting, +recording, or detection rules from files, directories and dicts. All format-specific +logic — parsing, topology injection, validation, and serialization — is +delegated to a :class:`RuleBackend` implementation. + +Usage:: + + from cosl.juju_topology import JujuTopology + from cosl.backends.prometheus import PrometheusRuleBackend + from cosl.backends.loki import LokiRuleBackend + from cosl.sigma import SigmaRuleBackend + + self._topology = JujuTopology.from_charm(charm) + + # Prometheus + prom_rules = AbstractRules(backend=PrometheusRuleBackend(topology=self._topology)) + prom_rules.add_path("src/prometheus_alert_rules") + print(prom_rules.as_dict()) # {"groups": [...]} + + # Loki + loki_rules = AbstractRules(backend=LokiRuleBackend(topology=self._topology)) + loki_rules.add_path("src/loki_alert_rules") + print(loki_rules.as_dict()) # {"groups": [...]} + + # Sigma + sigma_rules = AbstractRules(backend=SigmaRuleBackend(topology=self._topology)) + sigma_rules.add_path("src/sigma_rules") + print(sigma_rules.as_dict()) # {"rules": [...]} +""" import contextlib import copy import hashlib import logging import re +from abc import ABC, abstractmethod from dataclasses import dataclass from pathlib import Path from types import SimpleNamespace @@ -88,9 +120,12 @@ ClassVar, Dict, Final, + Generic, List, Mapping, Optional, + Tuple, + TypeVar, Union, cast, ) @@ -193,6 +228,8 @@ def aggregator_rules(self) -> OfficialRuleFileFormat: generic_alert_groups: Final = _GenericAlertGroups() +T = TypeVar("T") + class InvalidRulePathError(Exception): """Raised if the rules folder cannot be found or is otherwise invalid.""" @@ -208,10 +245,232 @@ def __init__( super().__init__(self.message) +@dataclass +class Result(Generic[T]): + """Result of rule validation. + + Attributes: + rules: The rules dictionary. + errmsg: Optional error message produced during validation. + """ + + rules: Dict[str, List[T]] + errmsg: Optional[str] + + +class RuleBackend(ABC, Generic[T]): + """Abstract base for format-specific rule handling. + + Type parameter *T* is the internal representation of a single rule item. + + Subclasses must implement four methods: + + * :meth:`file_suffixes` — which file extensions this backend reads. + * :meth:`from_dict` — parse a raw dict into normalised rule items, + injecting Juju topology where appropriate. + * :meth:`validate` — check the serialised output for correctness. + * :meth:`as_dict` — convert internal rule items into the backend's output format. + """ + + def __init__(self, topology: Optional[JujuTopology] = None) -> None: + self.topology = topology + + @property + @abstractmethod + def file_suffixes(self) -> List[str]: + """File extensions this backend supports (e.g. ``['.rule', '.yml']``).""" + ... + + @abstractmethod + def from_dict( + self, + rule_dict: Mapping[str, Any], + **kwargs: Any, + ) -> List[T]: + """Parse a rule dict, normalise it, and inject topology. + + Args: + rule_dict: Raw rule content as a YAML-loaded dict. + **kwargs: Backend-specific keyword arguments. + + Returns: + A list of normalised rule items. + + Raises: + ValueError: If *rule_dict* is empty or in an invalid format. + """ + ... + + def from_file( + self, + file_path: Path, + **kwargs: Any, + ) -> List[T]: + """Read a single rule file and parse it. + + The default implementation loads YAML and delegates to :meth:`from_dict`. + Backends that need file-level context (e.g. Prometheus uses the file + stem as a group name) should override this method. + + Args: + file_path: Absolute path to the rule file. + **kwargs: Backend-specific keyword arguments. + + Returns: + A list of normalised rule items, or an empty list on error. + """ + with file_path.open() as f: + try: + rule_file = yaml.safe_load(f) + except Exception as e: + logger.error("Failed to read rules from %s: %s", file_path.name, e) + return [] + + try: + return self.from_dict(rule_file, **kwargs) + except ValueError as e: + logger.error("Invalid rules file: %s (%s)", file_path.name, e) + return [] + + @abstractmethod + def validate(self, rules: Dict[str, List[T]]) -> Tuple[bool, str]: + """Validate rules in their serialised dict form. + + Args: + rules: The output of :meth:`as_dict`. + + Returns: + A ``(is_valid, error_message)`` tuple. + """ + ... + + @abstractmethod + def as_dict(self, items: List[T]) -> Dict[str, List[T]]: + """Serialise rule items into the backend's output format.""" + ... + + +class AbstractRules(Generic[T]): + """Format-agnostic rule aggregator. + + Collects rules from files and dicts, delegating format-specific parsing, + topology injection, and validation to a pluggable :class:`RuleBackend`. + """ + + def __init__(self, backend: RuleBackend[T]): + """Build a Rules instance. + + Args: + backend: A :class:`RuleBackend` implementation that handles all + format-specific logic. Pass topology directly to the backend + constructor (e.g. ``PrometheusRuleBackend(topology=topo)``). + """ + self.backend = backend + self._items: List[T] = [] + + def add( + self, + rule_dict: Mapping[str, Any], + **kwargs: Any, + ) -> None: + """Add rules from a dict to the existing ruleset. + + Args: + rule_dict: A rule mapping in whatever format the backend accepts. + **kwargs: Backend-specific keyword arguments forwarded to + :meth:`RuleBackend.from_dict`. For example, Prometheus/Loki + backends accept ``group_name`` and ``group_name_prefix``. + """ + self._items.extend( + self.backend.from_dict( + rule_dict, + **kwargs, + ) + ) + + def add_path(self, dir_path: Union[str, Path], *, recursive: bool = False) -> None: + """Add rules from a file or directory. + + All rules from files are aggregated into the internal ruleset. + Group names (where applicable) are augmented with Juju topology. + + Args: + dir_path: A rules file or a directory of rule files. + recursive: Whether to read files recursively (no impact if + *dir_path* is a single file). + """ + path = Path(dir_path) if isinstance(dir_path, str) else dir_path + if path.is_dir(): + self._items.extend(self._from_dir(path, recursive)) + elif path.is_file(): + self._items.extend(self.backend.from_file(path, root_path=path.parent)) + else: + raise InvalidRulePathError( + rules_absolute_path=path, message=f"Invalid rules path: {path}" + ) + + def as_dict(self) -> Dict[str, List[T]]: + """Return the accumulated rules in the backend's output format. + + Returns: + A dictionary whose structure depends on the backend + (e.g. ``{"groups": [...]}`` for Prometheus). + """ + return self.backend.as_dict(self._items) if self._items else {} + + def validate( + self, + rules: Mapping[str, List[T]], + ) -> Result[T]: + """Validate rules. + + This is a **standalone** operation — it validates the given *rules* + without modifying the internal ruleset. + + Args: + rules: A rule mapping in the backend's output format. + + Returns: + A :class:`Result` with the rules and an optional + error message. + """ + if not rules: + return Result(rules=self.backend.as_dict([]), errmsg=None) + + output = dict(rules) + valid, errmsg = self.backend.validate(output) + if not valid: + return Result(rules=output, errmsg=errmsg) + return Result(rules=output, errmsg=None) + + def _from_dir(self, dir_path: Path, recursive: bool) -> List[T]: + """Read all matching rule files in a directory.""" + all_files = dir_path.glob("**/*" if recursive else "*") + matched = sorted( + f for f in all_files if f.is_file() and f.suffix in self.backend.file_suffixes + ) + items: List[T] = [] + for file_path in matched: + from_file = self.backend.from_file(file_path, root_path=dir_path) + if from_file: + logger.debug("Reading rule from %s", file_path) + items.extend(from_file) + return items + + +# --------------------------------------------------------------------------- +# OLDER RULES CLASS +# --------------------------------------------------------------------------- + + @dataclass class InjectResult: """Typed result for rule injection and validation. + .. deprecated:: + Use :class:`Result` when switching over from Rules to AbstractRules class. This class + will be removed in a future release. + Attributes: rules: The (possibly injected) rules dictionary. errmsg: Optional error message produced during validation. @@ -224,6 +483,10 @@ class InjectResult: class Rules: """Utility class for amalgamating alerting/recording rule files and injecting juju topology. + .. deprecated:: + Use :class:`AbstractRules` with Prometheus and Loki backends. This class will be + removed in a future release. + A `Rules` object supports aggregating rules from files and directories in both official and single rule file formats using the `add_path()` method. All the rules read are annotated with Juju topology labels and amalgamated into a single data structure @@ -560,7 +823,8 @@ class AlertRules(Rules): """Utility class for amalgamating alerting files and injecting juju topology. .. deprecated:: - Use :class:`Rules` directly. This class will be removed in a future release. + Use :class:`AbstractRules` with Prometheus and Loki backends. This class will be + removed in a future release. The official format is a YAML file conforming to the Prometheus/Cortex documentation (https://prometheus.io/docs/prometheus/latest/configuration/alerting_rules/). @@ -575,7 +839,8 @@ class RecordingRules(Rules): """Utility class for amalgamating recording files and injecting juju topology. .. deprecated:: - Use :class:`Rules` directly. This class will be removed in a future release. + Use :class:`AbstractRules` with Prometheus and Loki backends. This class will be + removed in a future release. The official format is a YAML file conforming to the Prometheus/Cortex documentation (https://prometheus.io/docs/prometheus/latest/configuration/recording_rules/). diff --git a/src/cosl/types.py b/src/cosl/types.py index 6123a03..30aa00b 100644 --- a/src/cosl/types.py +++ b/src/cosl/types.py @@ -54,14 +54,14 @@ class OfficialRuleFileItem(TypedDict): - """Typing for a single node of the official rule file format.""" + """A single group in the official Prometheus/Loki rule file format.""" name: str rules: List[SingleRuleFormat] class OfficialRuleFileFormat(TypedDict, total=False): - """Typing for the official rule file format. + """The official Prometheus/Loki rule file format. References: - https://prometheus.io/docs/prometheus/latest/configuration/alerting_rules/ diff --git a/tests/helpers.py b/tests/helpers.py new file mode 100644 index 0000000..b5d87ff --- /dev/null +++ b/tests/helpers.py @@ -0,0 +1,52 @@ +# Copyright 2026 Canonical Ltd. +# See LICENSE file for licensing details. + +"""Shared test helpers and fixtures for AbstractRules tests.""" + +from pathlib import Path + +import yaml + +from cosl.juju_topology import JujuTopology + +FIXTURE_DIR = Path(__file__).resolve().parent / "promql_rules" +PROMETHEUS_RULES_DIR = FIXTURE_DIR / "prometheus_alert_rules" +BAD_YAML_RULE_PATH = FIXTURE_DIR / "bad_alert_rules" / "bad_yaml.rule" + + +def make_topology(**overrides): + defaults = { + "model": "mymodel", + "model_uuid": "12de4fae-06cc-4ceb-9089-567be09fec78", + "application": "myapp", + "unit": "myapp/0", + "charm_name": "mycharm", + } + defaults.update(overrides) + return JujuTopology(**defaults) + + +def load_rule(path: Path) -> dict: + """Load a YAML rule file into a dict.""" + with path.open() as f: + return yaml.safe_load(f) + + +# Single-rule format loaded from fixture +SINGLE_ALERT_RULE = load_rule(PROMETHEUS_RULES_DIR / "cpu_overuse.rule") + +# Official groups format (no fixture file exists for this format) +OFFICIAL_RULE = { + "groups": [ + { + "name": "TestGroup", + "rules": [ + { + "alert": "TestAlert", + "expr": "up < 1", + "labels": {"severity": "warning"}, + } + ], + } + ] +} diff --git a/tests/test_backend.py b/tests/test_backend.py new file mode 100644 index 0000000..2627c50 --- /dev/null +++ b/tests/test_backend.py @@ -0,0 +1,187 @@ +# Copyright 2026 Canonical Ltd. +# See LICENSE file for licensing details. + +"""Tests for _GroupedRuleBackend (from_dict, from_file, as_dict, file_suffixes). + +These tests exercise the shared backend base class via PrometheusRuleBackend, +since _GroupedRuleBackend is private and requires a concrete query_type. +""" + +import re +import unittest + +from helpers import ( + BAD_YAML_RULE_PATH, + OFFICIAL_RULE, + PROMETHEUS_RULES_DIR, + SINGLE_ALERT_RULE, + load_rule, + make_topology, +) + +from cosl.backends.prometheus import PrometheusRuleBackend + +# =================================================================== +# GroupedRules – from_dict +# =================================================================== + + +class TestGroupedRulesFromDict(unittest.TestCase): + """Tests for _GroupedRuleBackend.from_dict (via PrometheusRuleBackend).""" + + def test_official_format_parsed(self): + """Official groups format is parsed correctly.""" + backend = PrometheusRuleBackend() + groups = backend.from_dict(OFFICIAL_RULE) + self.assertEqual(len(groups), 1) + self.assertIn("TestGroup", groups[0]["name"]) + + def test_single_rule_format_parsed(self): + """Single-rule format is wrapped into a group.""" + backend = PrometheusRuleBackend() + groups = backend.from_dict(SINGLE_ALERT_RULE, group_name="my_group") + self.assertEqual(len(groups), 1) + self.assertIn("my_group", groups[0]["name"]) + self.assertEqual(len(groups[0]["rules"]), 1) + self.assertEqual(groups[0]["rules"][0]["alert"], "CPUOverUse") + + def test_single_rule_gets_hash_name_when_no_group_name(self): + """When no group_name is provided, a hash-based name is generated.""" + backend = PrometheusRuleBackend() + groups = backend.from_dict(SINGLE_ALERT_RULE) + # Group name should end with _rules + self.assertTrue(groups[0]["name"].endswith("_rules")) + + def test_group_name_prefix_applied(self): + """group_name_prefix is prepended to the group name.""" + backend = PrometheusRuleBackend() + groups = backend.from_dict( + SINGLE_ALERT_RULE, + group_name="mygroup", + group_name_prefix="prefix", + ) + self.assertTrue(groups[0]["name"].startswith("prefix_")) + + def test_official_format_group_name_prefix(self): + """For official format, the existing group name is prefixed.""" + backend = PrometheusRuleBackend() + groups = backend.from_dict(OFFICIAL_RULE, group_name_prefix="topo") + self.assertTrue(groups[0]["name"].startswith("topo_")) + self.assertIn("TestGroup", groups[0]["name"]) + + def test_empty_dict_raises_value_error(self): + """An empty dict raises ValueError.""" + backend = PrometheusRuleBackend() + with self.assertRaises(ValueError) as ctx: + backend.from_dict({}) + self.assertEqual(str(ctx.exception), "Empty") + + def test_topology_labels_injected(self): + """Topology labels are injected into rule labels.""" + topo = make_topology() + backend = PrometheusRuleBackend(topology=topo) + groups = backend.from_dict(SINGLE_ALERT_RULE, group_name="test") + rule = groups[0]["rules"][0] + self.assertIn("juju_model", rule["labels"]) + self.assertIn("juju_application", rule["labels"]) + self.assertIn("juju_model_uuid", rule["labels"]) + self.assertEqual(rule["labels"]["juju_model"], "mymodel") + self.assertEqual(rule["labels"]["juju_application"], "myapp") + + def test_topology_labels_not_overwritten(self): + """Pre-existing topology labels in a rule are not overwritten.""" + topo = make_topology() + backend = PrometheusRuleBackend(topology=topo) + rule_with_labels = { + "alert": "Test", + "expr": "up < 1", + "labels": {"severity": "critical", "juju_model": "existing_model"}, + } + groups = backend.from_dict(rule_with_labels, group_name="test") + self.assertEqual(groups[0]["rules"][0]["labels"]["juju_model"], "existing_model") + + def test_no_topology_no_labels_injected(self): + """Without topology, no juju labels are added.""" + backend = PrometheusRuleBackend() + groups = backend.from_dict(SINGLE_ALERT_RULE, group_name="test") + rule = groups[0]["rules"][0] + self.assertNotIn("juju_model", rule["labels"]) + + def test_group_name_sanitized(self): + """Special characters in group names are sanitized to underscores.""" + backend = PrometheusRuleBackend() + groups = backend.from_dict(SINGLE_ALERT_RULE, group_name="Foo$Bar/Baz") + name = groups[0]["name"] + # Only [a-zA-Z0-9_:] should remain + self.assertIsNotNone(re.match(r"^[a-zA-Z0-9_:]+$", name)) + + def test_juju_topology_placeholder_replaced_promql(self): + """The %%juju_topology%% placeholder is replaced in PromQL expressions.""" + topo = make_topology() + backend = PrometheusRuleBackend(topology=topo) + rule = load_rule(PROMETHEUS_RULES_DIR / "with_template_string.rule") + groups = backend.from_dict(rule, group_name="test") + expr = groups[0]["rules"][0]["expr"] + self.assertNotIn("%%juju_topology%%", expr) + + +# =================================================================== +# GroupedRules – from_file +# =================================================================== + + +class TestGroupedRulesFromFile(unittest.TestCase): + """Tests for from_file using existing fixture files using Prometheus backend.""" + + def test_from_file_single_rule(self): + """A single-rule fixture file is parsed into one group.""" + backend = PrometheusRuleBackend() + path = PROMETHEUS_RULES_DIR / "cpu_overuse.rule" + groups = backend.from_file(path, root_path=path.parent) + self.assertEqual(len(groups), 1) + self.assertEqual(groups[0]["rules"][0]["alert"], "CPUOverUse") + + def test_from_file_group_name_from_stem(self): + """Group name should be derived from the file stem.""" + backend = PrometheusRuleBackend() + path = PROMETHEUS_RULES_DIR / "cpu_overuse.rule" + groups = backend.from_file(path, root_path=path.parent) + self.assertIn("cpu_overuse", groups[0]["name"]) + + def test_from_file_invalid_yaml_returns_empty(self): + """Invalid YAML files return an empty list instead of raising.""" + backend = PrometheusRuleBackend() + groups = backend.from_file(BAD_YAML_RULE_PATH, root_path=BAD_YAML_RULE_PATH.parent) + self.assertEqual(groups, []) + + +# =================================================================== +# GroupedRules – as_dict, validate, file_suffixes +# =================================================================== + + +class TestGroupedRulesOther(unittest.TestCase): + """Tests for as_dict, validate, and file_suffixes using Prometheus backend.""" + + def test_prometheus_as_dict(self): + """as_dict wraps items under a 'groups' key.""" + backend = PrometheusRuleBackend() + items = backend.from_dict(OFFICIAL_RULE) + result = backend.as_dict(items) + self.assertIn("groups", result) + self.assertEqual(len(result["groups"]), 1) + + def test_prometheus_as_dict_empty(self): + """as_dict returns an empty dict when given no items.""" + backend = PrometheusRuleBackend() + result = backend.as_dict([]) + self.assertEqual(result, {}) + + def test_file_suffixes(self): + """file_suffixes returns the expected rule file extensions.""" + backend = PrometheusRuleBackend() + self.assertEqual(backend.file_suffixes, [".rule", ".rules", ".yml", ".yaml"]) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_generic_rules.py b/tests/test_generic_rules.py new file mode 100644 index 0000000..94644f3 --- /dev/null +++ b/tests/test_generic_rules.py @@ -0,0 +1,157 @@ +# Copyright 2026 Canonical Ltd. +# See LICENSE file for licensing details. + +"""Tests for AbstractRules (add, add_path, validate, backward compatibility).""" + +import unittest + +from deepdiff import DeepDiff +from helpers import OFFICIAL_RULE, PROMETHEUS_RULES_DIR, SINGLE_ALERT_RULE, make_topology + +from cosl.backends.prometheus import PrometheusRuleBackend +from cosl.rules import AbstractRules + +# =================================================================== +# AbstractRules – add +# =================================================================== + + +class TestAbstractRulesAdd(unittest.TestCase): + """Tests for AbstractRules.add method.""" + + def test_add_single_rule(self): + """Adding a single rule creates one group.""" + rules = AbstractRules(backend=PrometheusRuleBackend()) + rules.add(SINGLE_ALERT_RULE, group_name="mygroup") + result = rules.as_dict() + self.assertIn("groups", result) + self.assertEqual(len(result["groups"]), 1) + + def test_add_official_rule(self): + """Adding an official-format rule creates one group.""" + rules = AbstractRules(backend=PrometheusRuleBackend()) + rules.add(OFFICIAL_RULE) + result = rules.as_dict() + self.assertEqual(len(result["groups"]), 1) + + def test_add_multiple_rules_accumulate(self): + """Multiple add calls accumulate groups.""" + rules = AbstractRules(backend=PrometheusRuleBackend()) + rules.add(SINGLE_ALERT_RULE, group_name="group1") + rules.add(SINGLE_ALERT_RULE, group_name="group2") + result = rules.as_dict() + self.assertEqual(len(result["groups"]), 2) + + def test_add_with_topology(self): + """Rules added with topology get juju labels injected.""" + topo = make_topology() + rules = AbstractRules(backend=PrometheusRuleBackend(topology=topo)) + rules.add(SINGLE_ALERT_RULE, group_name="test") + result = rules.as_dict() + rule = result["groups"][0]["rules"][0] + self.assertEqual(rule["labels"]["juju_model"], "mymodel") + + def test_as_dict_empty_when_no_rules_added(self): + """as_dict returns empty dict when no rules have been added.""" + rules = AbstractRules(backend=PrometheusRuleBackend()) + self.assertEqual(rules.as_dict(), {}) + + def test_topology_set_on_backend(self): + """Topology passed to the backend is available on AbstractRules.backend.""" + topo = make_topology() + backend = PrometheusRuleBackend(topology=topo) + rules = AbstractRules(backend=backend) + self.assertEqual(rules.backend.topology, topo) + + +# =================================================================== +# AbstractRules – add_path +# =================================================================== + + +class TestAbstractRulesAddPath(unittest.TestCase): + """Tests for AbstractRules.add_path with file and directory loading.""" + + def setUp(self): + self.rules_dir = PROMETHEUS_RULES_DIR + self.topology = make_topology() + + def test_add_path_single_file(self): + """Loading a single file creates one group.""" + rules = AbstractRules(backend=PrometheusRuleBackend(topology=self.topology)) + rules.add_path(self.rules_dir / "cpu_overuse.rule") + result = rules.as_dict() + self.assertIn("groups", result) + self.assertEqual(len(result["groups"]), 1) + + def test_add_path_directory_non_recursive(self): + """Non-recursive directory scan finds only top-level files.""" + rules = AbstractRules(backend=PrometheusRuleBackend(topology=self.topology)) + rules.add_path(self.rules_dir) + result = rules.as_dict() + # Should find top-level .rule files but not nested/ + top_level_rules = [ + f for f in self.rules_dir.iterdir() if f.is_file() and f.suffix == ".rule" + ] + self.assertEqual(len(result["groups"]), len(top_level_rules)) + + def test_add_path_directory_recursive(self): + """Recursive directory scan finds nested files.""" + rules = AbstractRules(backend=PrometheusRuleBackend(topology=self.topology)) + rules.add_path(self.rules_dir, recursive=True) + result = rules.as_dict() + all_rules = list(self.rules_dir.rglob("*.rule")) + self.assertEqual(len(result["groups"]), len(all_rules)) + + def test_add_path_nonexistent_path_raises(self): + """A nonexistent path raises InvalidRulePathError.""" + from cosl.rules import InvalidRulePathError + + rules = AbstractRules(backend=PrometheusRuleBackend()) + with self.assertRaises(InvalidRulePathError): + rules.add_path("/nonexistent/path") + + def test_add_path_topology_in_group_name(self): + """Group names include topology identifier.""" + rules = AbstractRules(backend=PrometheusRuleBackend(topology=self.topology)) + rules.add_path(self.rules_dir / "cpu_overuse.rule") + result = rules.as_dict() + group_name = result["groups"][0]["name"] + self.assertIn(self.topology.identifier, group_name) + + def test_add_path_nested_includes_relative_path(self): + """Nested files include relative directory path in group name prefix.""" + rules = AbstractRules(backend=PrometheusRuleBackend(topology=self.topology)) + rules.add_path(self.rules_dir, recursive=True) + result = rules.as_dict() + nested_groups = [g for g in result["groups"] if "nested" in g["name"]] + self.assertTrue(len(nested_groups) > 0) + + +# =================================================================== +# AbstractRules – produces same output as legacy Rules +# =================================================================== + + +class TestAbstractRulesBackwardCompat(unittest.TestCase): + """Verify AbstractRules with PrometheusRuleBackend matches legacy Rules output.""" + + def setUp(self): + self.topology = make_topology() + + def test_add_produces_same_as_legacy(self): + """AbstractRules.add with group_name produces same structure as legacy Rules._from_dict.""" + from cosl.rules import Rules + + legacy = Rules(query_type="promql") + legacy_groups = legacy._from_dict(SINGLE_ALERT_RULE, group_name="test") + + generic = AbstractRules(backend=PrometheusRuleBackend()) + generic.add(SINGLE_ALERT_RULE, group_name="test") + generic_result = generic.as_dict() + + self.assertEqual({}, DeepDiff({"groups": legacy_groups}, generic_result)) + + +if __name__ == "__main__": + unittest.main()