From f7d0cbba87c14a7697b104992db632171d0afe01 Mon Sep 17 00:00:00 2001 From: Sina Date: Mon, 24 Aug 2026 11:17:57 -0400 Subject: [PATCH 1/6] feat: Alert Knob --- src/cosl/__init__.py | 3 + src/cosl/rules_customization.py | 447 ++++++++++++++++++++++ tests/test_rules_customization.py | 605 ++++++++++++++++++++++++++++++ 3 files changed, 1055 insertions(+) create mode 100644 src/cosl/rules_customization.py create mode 100644 tests/test_rules_customization.py diff --git a/src/cosl/__init__.py b/src/cosl/__init__.py index 8539b2fa..ac0718fd 100644 --- a/src/cosl/__init__.py +++ b/src/cosl/__init__.py @@ -8,6 +8,7 @@ from .juju_topology import JujuTopology from .mandatory_relation_pairs import MandatoryRelationPairs from .rules import AlertRules, RecordingRules +from .rules_customization import AlertRulesCustomization, AlertRulesCustomizationError from .types import type_convert_stored __all__ = [ @@ -18,6 +19,8 @@ "DashboardPath40UID", "AlertRules", "RecordingRules", + "AlertRulesCustomization", + "AlertRulesCustomizationError", "MandatoryRelationPairs", "type_convert_stored", ] diff --git a/src/cosl/rules_customization.py b/src/cosl/rules_customization.py new file mode 100644 index 00000000..e1fb600b --- /dev/null +++ b/src/cosl/rules_customization.py @@ -0,0 +1,447 @@ +# Copyright 2026 Canonical Ltd. +# See LICENSE file for licensing details. +"""Admin-facing alert rule customization. + +## Overview + +This module provides :class:`AlertRulesCustomization`, a pure transformation helper that +takes relation-derived alert rule files (the same dict that relation libraries such as +``MetricsConsumer.alerts`` produce) and an admin-provided YAML customization config, and +returns the modified rules in the same format. + +The customization config supports three top-level keys: + +- ``remove``: drop matching alerting rules (or entire groups, when ``group`` is the only + selector). +- ``patch``: modify matching alerting rules by merging a ``set`` block into them. +- ``add``: insert admin-authored rule groups into the output under the fixed key + ``custom_alert_rules``. + +Matching is performed via a ``where`` block, supporting exact equality on ``alert``, +``group``, ``labels`` and ``annotations``. All fields within one ``where`` are ANDed; +multiple entries in the ``remove``/``patch`` lists provide OR semantics. + +Recording rules are never removed or patched unless an entire group is dropped via a +group-only ``where`` selector. + +Rules added via the ``add`` block do NOT receive automatic Juju topology injection: the +admin must include topology matchers in their expressions and labels manually. + +This class is a pure transformation helper. It does not call CosTool, Pebble, +Prometheus, Loki or Mimir APIs, does not write files and does not set statuses. +Validation of the resulting rules is the charm's responsibility after calling +:meth:`AlertRulesCustomization.apply`. +""" + +import collections.abc +import copy +import logging +from typing import Any, Dict, List, Mapping, Optional, cast + +import yaml + +from .types import OfficialRuleFileFormat + +logger = logging.getLogger(__name__) + +CUSTOM_ALERT_RULES_KEY = "custom_alert_rules" +"""Fixed output key under which rules from the ``add`` block are inserted.""" + +_VALID_TOP_LEVEL_KEYS = frozenset({"remove", "patch", "add"}) +_VALID_WHERE_KEYS = frozenset({"alert", "group", "labels", "annotations"}) +_VALID_SET_KEYS = frozenset({"alert", "expr", "for", "labels", "annotations"}) + + +class AlertRulesCustomizationError(Exception): + """Raised when the alert rules customization configuration is invalid.""" + + +def _validate_where(where: Any, context: str) -> Dict[str, Any]: + """Validate a ``where`` selector block and return it as a plain dict. + + Raises: + AlertRulesCustomizationError: if the where block is missing, not a mapping, + empty, contains unknown keys, or has wrongly-typed values. + """ + if not isinstance(where, collections.abc.Mapping): + raise AlertRulesCustomizationError(f"{context}: 'where' must be a mapping") + validated_where: Dict[str, Any] = dict(cast(Mapping[Any, Any], where)) + if not validated_where: + raise AlertRulesCustomizationError(f"{context}: 'where' must not be empty") + unknown_keys = set(validated_where.keys()) - _VALID_WHERE_KEYS + if unknown_keys: + raise AlertRulesCustomizationError( + f"{context}: unknown 'where' keys {sorted(unknown_keys)}; " + f"expected a subset of {sorted(_VALID_WHERE_KEYS)}" + ) + for key in ("alert", "group"): + if key in validated_where and not isinstance(validated_where[key], str): + raise AlertRulesCustomizationError(f"{context}: 'where.{key}' must be a string") + for key in ("labels", "annotations"): + if key in validated_where and not isinstance( + validated_where[key], collections.abc.Mapping + ): + raise AlertRulesCustomizationError( + f"{context}: 'where.{key}' must be a mapping of key-value pairs" + ) + return validated_where + + +def _validate_set(set_block: Any, context: str) -> Dict[str, Any]: + """Validate a patch ``set`` block and return it as a plain dict. + + Raises: + AlertRulesCustomizationError: if the set block is missing, not a mapping, + contains unknown keys, or has wrongly-typed values. + """ + if not isinstance(set_block, collections.abc.Mapping): + raise AlertRulesCustomizationError(f"{context}: 'set' must be a mapping") + validated_set: Dict[str, Any] = dict(cast(Mapping[Any, Any], set_block)) + unknown_keys = set(validated_set.keys()) - _VALID_SET_KEYS + if unknown_keys: + raise AlertRulesCustomizationError( + f"{context}: unknown 'set' keys {sorted(unknown_keys)}; " + f"expected a subset of {sorted(_VALID_SET_KEYS)}" + ) + for key in ("alert", "expr", "for"): + if key in validated_set and not isinstance(validated_set[key], str): + raise AlertRulesCustomizationError(f"{context}: 'set.{key}' must be a string") + for key in ("labels", "annotations"): + if key in validated_set and not isinstance(validated_set[key], collections.abc.Mapping): + raise AlertRulesCustomizationError( + f"{context}: 'set.{key}' must be a mapping of key-value pairs" + ) + return validated_set + + +def _validate_remove(entries: Any) -> List[Dict[str, Any]]: + """Validate the ``remove`` operation list and return it. + + Raises: + AlertRulesCustomizationError: if the operation list is malformed. + """ + if entries is None: + return [] + if not isinstance(entries, list): + raise AlertRulesCustomizationError("'remove' must be a list of operations") + validated: List[Dict[str, Any]] = [] + for index, item in enumerate(cast(List[Any], entries)): + context = f"remove[{index}]" + if not isinstance(item, collections.abc.Mapping): + raise AlertRulesCustomizationError(f"{context} must be a mapping with a 'where' key") + entry: Dict[str, Any] = dict(cast(Mapping[Any, Any], item)) + if "where" not in entry: + raise AlertRulesCustomizationError(f"{context}: missing required key 'where'") + validated.append({"where": _validate_where(entry["where"], context)}) + return validated + + +def _validate_patch(entries: Any) -> List[Dict[str, Any]]: + """Validate the ``patch`` operation list and return it. + + Raises: + AlertRulesCustomizationError: if the operation list is malformed. + """ + if entries is None: + return [] + if not isinstance(entries, list): + raise AlertRulesCustomizationError("'patch' must be a list of operations") + validated: List[Dict[str, Any]] = [] + for index, item in enumerate(cast(List[Any], entries)): + context = f"patch[{index}]" + if not isinstance(item, collections.abc.Mapping): + raise AlertRulesCustomizationError( + f"{context} must be a mapping with 'where' and 'set' keys" + ) + entry: Dict[str, Any] = dict(cast(Mapping[Any, Any], item)) + if "where" not in entry: + raise AlertRulesCustomizationError(f"{context}: missing required key 'where'") + if "set" not in entry: + raise AlertRulesCustomizationError(f"{context}: missing required key 'set'") + validated.append( + { + "where": _validate_where(entry["where"], context), + "set": _validate_set(entry["set"], context), + } + ) + return validated + + +def _validate_add(add_block: Any) -> Optional[OfficialRuleFileFormat]: + """Validate the ``add`` block and return it. + + Raises: + AlertRulesCustomizationError: if the add block does not conform to the + official rule file format shape. + """ + if add_block is None: + return None + if not isinstance(add_block, collections.abc.Mapping): + raise AlertRulesCustomizationError("'add' must be a mapping with a 'groups' key") + validated_add: Dict[str, Any] = dict(cast(Mapping[Any, Any], add_block)) + if "groups" not in validated_add: + raise AlertRulesCustomizationError("'add': missing required key 'groups'") + if not isinstance(validated_add["groups"], list): + raise AlertRulesCustomizationError("'add.groups' must be a list") + groups: List[Any] = cast(List[Any], validated_add["groups"]) + for index, item in enumerate(groups): + if not isinstance(item, collections.abc.Mapping): + raise AlertRulesCustomizationError(f"'add.groups[{index}]' must be a mapping") + group: Dict[str, Any] = dict(cast(Mapping[Any, Any], item)) + name = group.get("name") + if name is None: + raise AlertRulesCustomizationError( + f"'add.groups[{index}]': missing required key 'name'" + ) + if not isinstance(name, str): + raise AlertRulesCustomizationError(f"'add.groups[{index}].name' must be a string") + if "rules" not in group: + raise AlertRulesCustomizationError( + f"'add.groups[{index}]': missing required key 'rules'" + ) + if not isinstance(group["rules"], list): + raise AlertRulesCustomizationError(f"'add.groups[{index}].rules' must be a list") + return cast(OfficialRuleFileFormat, validated_add) + + +class AlertRulesCustomization: + """Apply admin-defined remove/patch/add operations to relation-derived alert rules. + + Build an instance with :meth:`from_yaml`, then call :meth:`apply` on the alerts dict + (e.g. ``self.metrics_consumer.alerts``). The instance is reusable: ``apply()`` can be + called multiple times on different inputs. + """ + + def __init__( + self, + remove: Optional[List[Dict[str, Any]]] = None, + patch: Optional[List[Dict[str, Any]]] = None, + add: Optional[OfficialRuleFileFormat] = None, + ): + r"""Build a customization object from pre-validated operation blocks. + + Prefer :meth:`from_yaml` for parsing and validating user input. + """ + self._remove: List[Dict[str, Any]] = remove or [] + self._patch: List[Dict[str, Any]] = patch or [] + self._add: Optional[OfficialRuleFileFormat] = copy.deepcopy(add) + + @classmethod + def from_yaml(cls, config_string: str) -> "AlertRulesCustomization": + """Parse and validate the customization YAML. + + Args: + config_string: raw YAML string, e.g. from a charm config option. + + Returns: + An ``AlertRulesCustomization`` instance. If the config string is empty, + whitespace-only or parses to ``None``, the returned instance is a no-op. + + Raises: + AlertRulesCustomizationError: on invalid YAML, unknown top-level keys + (only ``remove``, ``patch``, ``add`` are allowed), invalid operation + shape (missing ``where``, unknown selector keys, unknown set keys), + empty ``where`` selectors, or an ``add`` block not conforming to the + official rule file format shape. + """ + if not config_string or not config_string.strip(): + # Empty or whitespace-only config: no-op. + return cls() + + try: + parsed = yaml.safe_load(config_string) + except yaml.YAMLError as e: + raise AlertRulesCustomizationError(f"invalid YAML: {e}") from e + + if parsed is None: + # Config parsing to null: no-op. + return cls() + + if not isinstance(parsed, collections.abc.Mapping): + raise AlertRulesCustomizationError( + f"configuration must be a mapping with keys {sorted(_VALID_TOP_LEVEL_KEYS)}; " + f"got {type(parsed).__name__}" + ) + config: Dict[str, Any] = dict(cast(Mapping[Any, Any], parsed)) + + unknown_keys = set(config.keys()) - _VALID_TOP_LEVEL_KEYS + if unknown_keys: + raise AlertRulesCustomizationError( + f"unknown top-level keys {sorted(unknown_keys)}; " + f"expected a subset of {sorted(_VALID_TOP_LEVEL_KEYS)}" + ) + + return cls( + remove=_validate_remove(config.get("remove")), + patch=_validate_patch(config.get("patch")), + add=_validate_add(config.get("add")), + ) + + def apply( + self, relation_alerts: Mapping[str, OfficialRuleFileFormat] + ) -> Dict[str, OfficialRuleFileFormat]: + """Apply remove, patch and add operations to the input rules. + + Operations run in this order: remove, patch, add. The input is never mutated; + the transformations operate on a deep copy. + + Args: + relation_alerts: mapping of identifier to rule file, e.g. + ``self.metrics_consumer.alerts``. + + Returns: + The transformed rules, in the same format as the input. Identifiers whose + ``groups`` list becomes empty after removal are dropped. If ``add`` is + configured, its groups are inserted under ``custom_alert_rules``. + """ + output: Dict[str, OfficialRuleFileFormat] = copy.deepcopy(dict(relation_alerts)) + + self._apply_remove(output) + self._apply_patch(output) + self._apply_add(output) + + return output + + def _matches(self, where: Mapping[str, Any], group_name: str, rule: Mapping[str, Any]) -> bool: + """Does this rule (in this group) satisfy all fields of this where block? + + All fields present in the where block must match (AND semantics). Exact equality + is used for ``alert`` and ``group``; ``labels``/``annotations`` require every + key-value pair in the where block to exist in the rule's corresponding mapping. + """ + if "group" in where and where["group"] != group_name: + return False + if "alert" in where and rule.get("alert") != where["alert"]: + return False + if "labels" in where: + where_labels: Dict[Any, Any] = where["labels"] + labels: Dict[Any, Any] = rule.get("labels") or {} + if any(labels.get(key) != value for key, value in where_labels.items()): + return False + if "annotations" in where: + where_annotations: Dict[Any, Any] = where["annotations"] + annotations: Dict[Any, Any] = rule.get("annotations") or {} + if any(annotations.get(key) != value for key, value in where_annotations.items()): + return False + return True + + @staticmethod + def _is_group_only_selector(where: Mapping[str, Any]) -> bool: + """Is ``group`` the only key of this where block?""" + return set(where.keys()) == {"group"} + + def _apply_remove(self, output: Dict[str, OfficialRuleFileFormat]) -> None: + """Drop matching alerting rules, prune empty groups and empty identifiers.""" + if not self._remove: + return + + where_blocks: List[Mapping[str, Any]] = [entry["where"] for entry in self._remove] + + def matches_any_remove(group_name: str, rule: Mapping[str, Any]) -> bool: + # OR semantics across remove entries. + return any(self._matches(where, group_name, rule) for where in where_blocks) + + for identifier in list(output): + rule_file = output[identifier] + kept_groups: List[Any] = [] + for group in cast(List[Any], rule_file.get("groups", [])): + group_name = str(group.get("name", "")) + if any( + self._is_group_only_selector(where) and where["group"] == group_name + for where in where_blocks + ): + # A group-only selector drops the entire group, recording rules included. + logger.debug("Removed entire group '%s' from '%s'", group_name, identifier) + continue + + kept_rules: List[Any] = [] + for rule in cast(List[Any], group.get("rules", [])): + if "alert" not in rule: + # Recording rules are never removed unless the whole group is dropped. + kept_rules.append(rule) + continue + if matches_any_remove(group_name, rule): + logger.debug( + "Removed rule '%s' from group '%s' ('%s')", + rule.get("alert"), + group_name, + identifier, + ) + continue + kept_rules.append(rule) + + if kept_rules: + group["rules"] = kept_rules + kept_groups.append(group) + else: + logger.debug("Pruned empty group '%s' from '%s'", group_name, identifier) + + if kept_groups: + rule_file["groups"] = kept_groups + else: + logger.debug("Dropped identifier '%s': no groups left", identifier) + del output[identifier] + + def _apply_patch(self, output: Dict[str, OfficialRuleFileFormat]) -> None: + """Merge each patch's ``set`` block into every matching alerting rule.""" + if not self._patch: + return + + for identifier, rule_file in output.items(): + for group in cast(List[Any], rule_file.get("groups", [])): + group_name = str(group.get("name", "")) + for rule in cast(List[Any], group.get("rules", [])): + if "alert" not in rule: + # Recording rules are never patched. + continue + for entry in self._patch: + if self._matches(entry["where"], group_name, rule): + self._patch_rule( + cast(Dict[str, Any], rule), entry["set"], identifier, group_name + ) + + @staticmethod + def _patch_rule( + rule: Dict[str, Any], + set_block: Mapping[str, Any], + identifier: str, + group_name: str, + ) -> None: + """Merge a single ``set`` block into a rule, logging what changed.""" + changes: List[str] = [] + if "alert" in set_block: + changes.append(f"alert={set_block['alert']}") + rule["alert"] = set_block["alert"] + if "expr" in set_block: + changes.append("expr") + rule["expr"] = set_block["expr"] + if "for" in set_block: + changes.append(f"for={set_block['for']}") + rule["for"] = set_block["for"] + if "labels" in set_block: + changes.append(f"labels={set_block['labels']}") + labels: Dict[Any, Any] = rule.setdefault("labels", {}) + labels.update(set_block["labels"]) + if "annotations" in set_block: + changes.append(f"annotations={set_block['annotations']}") + annotations: Dict[Any, Any] = rule.setdefault("annotations", {}) + annotations.update(set_block["annotations"]) + logger.debug( + "Patched rule '%s' in group '%s' ('%s'): %s", + rule.get("alert"), + group_name, + identifier, + ", ".join(changes), + ) + + def _apply_add(self, output: Dict[str, OfficialRuleFileFormat]) -> None: + """Insert the ``add`` block's groups under the fixed custom alert rules key.""" + if self._add is None: + return + output[CUSTOM_ALERT_RULES_KEY] = copy.deepcopy(self._add) + added_groups = cast(List[Any], self._add.get("groups", [])) + logger.debug( + "Added %d group(s) under '%s'", + len(added_groups), + CUSTOM_ALERT_RULES_KEY, + ) diff --git a/tests/test_rules_customization.py b/tests/test_rules_customization.py new file mode 100644 index 00000000..1bad8831 --- /dev/null +++ b/tests/test_rules_customization.py @@ -0,0 +1,605 @@ +# Copyright 2026 Canonical Ltd. +# See LICENSE file for licensing details. + +import copy +import unittest + +from cosl.rules_customization import ( + CUSTOM_ALERT_RULES_KEY, + AlertRulesCustomization, + AlertRulesCustomizationError, +) + + +def _sample_alerts(): + """Relation alerts dict in the same shape as MetricsConsumer.alerts.""" + return { + "app-1": { + "groups": [ + { + "name": "group_a", + "rules": [ + { + "alert": "HighLatency", + "expr": "latency > 100", + "for": "10m", + "labels": {"severity": "critical", "juju_application": "app-1"}, + "annotations": {"summary": "latency is high"}, + }, + { + "alert": "LowThroughput", + "expr": "throughput < 10", + "for": "5m", + "labels": {"severity": "warning"}, + }, + { + "record": "job:latency:mean5m", + "expr": "avg(latency)", + "labels": {"severity": "warning"}, + }, + ], + }, + { + "name": "group_b", + "rules": [ + {"alert": "HostDown", "expr": "up < 1"}, + ], + }, + ] + }, + "app-2": { + "groups": [ + { + "name": "group_c", + "rules": [ + {"alert": "OtherAlert", "expr": "x > 0"}, + ], + } + ] + }, + } + + +def _find_rule(alerts, identifier, group_name, rule_name, *, by_record=False): + """Fetch a single rule from an alerts dict for assertions.""" + key = "record" if by_record else "alert" + groups = alerts[identifier]["groups"] + group = next(g for g in groups if g["name"] == group_name) + return next(rule for rule in group["rules"] if rule.get(key) == rule_name) + + +class TestFromYamlValidation(unittest.TestCase): + def test_invalid_yaml_raises(self): + with self.assertRaises(AlertRulesCustomizationError): + AlertRulesCustomization.from_yaml("remove: [unclosed") + + def test_non_mapping_top_level_raises(self): + for config in ("- a\n- b", "42", '"just a string"'): + with self.assertRaises(AlertRulesCustomizationError): + AlertRulesCustomization.from_yaml(config) + + def test_unknown_top_level_key_raises(self): + with self.assertRaisesRegex(AlertRulesCustomizationError, "top-level"): + AlertRulesCustomization.from_yaml(""" + remove: + - where: + alert: Foo + destroy: + - where: + alert: Foo + """) + + def test_remove_missing_where_raises(self): + with self.assertRaisesRegex(AlertRulesCustomizationError, "remove"): + AlertRulesCustomization.from_yaml("remove:\n - alert: Foo") + + def test_remove_empty_where_raises(self): + with self.assertRaisesRegex(AlertRulesCustomizationError, "'where' must not be empty"): + AlertRulesCustomization.from_yaml(""" + remove: + - where: {} + """) + + def test_remove_unknown_where_key_raises(self): + with self.assertRaisesRegex(AlertRulesCustomizationError, "'where' keys"): + AlertRulesCustomization.from_yaml(""" + remove: + - where: + expr: up < 1 + """) + + def test_patch_missing_where_raises(self): + with self.assertRaisesRegex(AlertRulesCustomizationError, "patch"): + AlertRulesCustomization.from_yaml(""" + patch: + - set: + for: 5m + """) + + def test_patch_missing_set_raises(self): + with self.assertRaisesRegex(AlertRulesCustomizationError, "patch"): + AlertRulesCustomization.from_yaml(""" + patch: + - where: + alert: Foo + """) + + def test_patch_empty_where_raises(self): + with self.assertRaisesRegex(AlertRulesCustomizationError, "'where' must not be empty"): + AlertRulesCustomization.from_yaml(""" + patch: + - where: {} + set: + for: 5m + """) + + def test_patch_unknown_where_key_raises(self): + with self.assertRaisesRegex(AlertRulesCustomizationError, "'where' keys"): + AlertRulesCustomization.from_yaml(""" + patch: + - where: + record: some:record + set: + expr: up + """) + + def test_patch_unknown_set_key_raises(self): + with self.assertRaisesRegex(AlertRulesCustomizationError, "'set' keys"): + AlertRulesCustomization.from_yaml(""" + patch: + - where: + alert: Foo + set: + duration: 5m + """) + + def test_add_without_groups_raises(self): + with self.assertRaisesRegex(AlertRulesCustomizationError, "'add'"): + AlertRulesCustomization.from_yaml(""" + add: + rules: + - alert: Foo + expr: up + """) + + def test_add_groups_not_a_list_raises(self): + with self.assertRaisesRegex(AlertRulesCustomizationError, "'add.groups' must be a list"): + AlertRulesCustomization.from_yaml(""" + add: + groups: + name: my-group + rules: [] + """) + + def test_add_group_missing_name_or_rules_raises(self): + for group in ("rules: []", "name: my-group"): + with self.assertRaisesRegex(AlertRulesCustomizationError, "add.groups"): + AlertRulesCustomization.from_yaml(f"add:\n groups:\n - {group}") + + def test_add_malformed_values_raise(self): + cases = { + "add not a mapping": "add: groups", + "group not a mapping": "add:\n groups:\n - just-a-string", + "name not a string": ("add:\n groups:\n - name: [1]\n rules: []"), + "rules not a list": ("add:\n groups:\n - name: my-group\n rules: nope"), + } + for case, config in cases.items(): + with self.subTest(case): + with self.assertRaises(AlertRulesCustomizationError): + AlertRulesCustomization.from_yaml(config) + + def test_operations_not_a_list_raises(self): + for key in ("remove", "patch"): + with self.assertRaisesRegex(AlertRulesCustomizationError, f"'{key}'"): + AlertRulesCustomization.from_yaml(f"{key}: not-a-list") + + def test_malformed_operation_entries_raise(self): + cases = { + "where not a mapping": "remove:\n - where: nope", + "where.alert not a string": ("remove:\n - where:\n alert: [1, 2]"), + "where.labels not a mapping": ("remove:\n - where:\n labels: severity"), + "set not a mapping": "patch:\n - where:\n alert: Foo\n set: nope", + "set.expr not a string": ( + "patch:\n - where:\n alert: Foo\n set:\n expr: {a: b}" + ), + "set.labels not a mapping": ( + "patch:\n - where:\n alert: Foo\n set:\n labels: x" + ), + "remove entry not a mapping": "remove:\n - just-a-string", + "patch entry not a mapping": "patch:\n - just-a-string", + } + for case, config in cases.items(): + with self.subTest(case): + with self.assertRaises(AlertRulesCustomizationError): + AlertRulesCustomization.from_yaml(config) + + +class TestNoOpConfigs(unittest.TestCase): + def _assert_noop(self, config_string): + sample = _sample_alerts() + result = AlertRulesCustomization.from_yaml(config_string).apply(sample) + self.assertEqual(result, sample) + + def test_empty_config_is_noop(self): + self._assert_noop("") + + def test_whitespace_config_is_noop(self): + self._assert_noop(" \n\t ") + + def test_none_parsing_config_is_noop(self): + # yaml.safe_load of these strings returns None or empty structures. + self._assert_noop("~") + self._assert_noop("# just a comment") + self._assert_noop("{}") + + def test_zero_match_remove_is_noop(self): + self._assert_noop(""" + remove: + - where: + alert: NoSuchAlert + - where: + labels: + nope: nothing + """) + + def test_zero_match_patch_is_noop(self): + self._assert_noop(""" + patch: + - where: + alert: NoSuchAlert + set: + for: 1m + """) + + +class TestRemove(unittest.TestCase): + def test_remove_by_alert_name(self): + result = AlertRulesCustomization.from_yaml(""" + remove: + - where: + alert: LowThroughput + """).apply(_sample_alerts()) + + group_names = [g["name"] for g in result["app-1"]["groups"]] + self.assertEqual(group_names, ["group_a", "group_b"]) + rule_names = [r.get("alert") for r in result["app-1"]["groups"][0]["rules"]] + self.assertEqual(rule_names, ["HighLatency", None]) # record remains + + def test_remove_by_group_only_drops_entire_group_including_recording_rules(self): + result = AlertRulesCustomization.from_yaml(""" + remove: + - where: + group: group_a + """).apply(_sample_alerts()) + + group_names = [g["name"] for g in result["app-1"]["groups"]] + self.assertEqual(group_names, ["group_b"]) + + def test_remove_group_with_other_selector_keeps_recording_rules(self): + result = AlertRulesCustomization.from_yaml(""" + remove: + - where: + group: group_a + alert: LowThroughput + """).apply(_sample_alerts()) + + group = next(g for g in result["app-1"]["groups"] if g["name"] == "group_a") + rule_names = [r.get("alert") or r.get("record") for r in group["rules"]] + # Only the matching alerting rule is removed; the rest survives. + self.assertEqual(rule_names, ["HighLatency", "job:latency:mean5m"]) + + def test_remove_by_alert_and_labels_combined(self): + result = AlertRulesCustomization.from_yaml(""" + remove: + - where: + alert: HighLatency + labels: + severity: critical + """).apply(_sample_alerts()) + self.assertNotIn("HighLatency", str(result)) + + # Same alert but non-matching label value: nothing removed. + result = AlertRulesCustomization.from_yaml(""" + remove: + - where: + alert: HighLatency + labels: + severity: warning + """).apply(_sample_alerts()) + self.assertIn("HighLatency", str(result)) + + def test_remove_by_group_and_labels_combined(self): + result = AlertRulesCustomization.from_yaml(""" + remove: + - where: + group: group_a + labels: + severity: warning + """).apply(_sample_alerts()) + + group = next(g for g in result["app-1"]["groups"] if g["name"] == "group_a") + rule_names = [r.get("alert") or r.get("record") for r in group["rules"]] + # Both the alerting rule and the recording rule carry severity=warning, + # but only the alerting rule may be removed (group selector is combined). + self.assertEqual(rule_names, ["HighLatency", "job:latency:mean5m"]) + + def test_remove_by_annotations(self): + result = AlertRulesCustomization.from_yaml(""" + remove: + - where: + annotations: + summary: latency is high + """).apply(_sample_alerts()) + self.assertNotIn("HighLatency", str(result)) + self.assertIn("LowThroughput", str(result)) + + def test_remove_multiple_entries_are_ored(self): + result = AlertRulesCustomization.from_yaml(""" + remove: + - where: + alert: HighLatency + - where: + alert: OtherAlert + """).apply(_sample_alerts()) + self.assertNotIn("HighLatency", str(result)) + self.assertNotIn("OtherAlert", str(result)) + self.assertIn("LowThroughput", str(result)) + self.assertIn("HostDown", str(result)) + + def test_remove_prunes_empty_groups_and_drops_empty_identifiers(self): + result = AlertRulesCustomization.from_yaml(""" + remove: + - where: + alert: HostDown + """).apply(_sample_alerts()) + # group_b became empty and was pruned; app-1 keeps group_a only. + self.assertEqual([g["name"] for g in result["app-1"]["groups"]], ["group_a"]) + self.assertIn("app-2", result) + + result = AlertRulesCustomization.from_yaml(""" + remove: + - where: + alert: OtherAlert + """).apply(_sample_alerts()) + # app-2's only group became empty, so app-2 was dropped entirely. + self.assertNotIn("app-2", result) + self.assertIn("app-1", result) + + def test_remove_preserves_recording_rules_when_group_not_sole_selector(self): + result = AlertRulesCustomization.from_yaml(""" + remove: + - where: + labels: + severity: warning + """).apply(_sample_alerts()) + record = _find_rule(result, "app-1", "group_a", "job:latency:mean5m", by_record=True) + self.assertEqual(record["expr"], "avg(latency)") + + +class TestPatch(unittest.TestCase): + def test_patch_updates_for(self): + result = AlertRulesCustomization.from_yaml(""" + patch: + - where: + alert: HighLatency + set: + for: 30m + """).apply(_sample_alerts()) + self.assertEqual(_find_rule(result, "app-1", "group_a", "HighLatency")["for"], "30m") + # Untouched rule keeps its original value. + self.assertEqual(_find_rule(result, "app-1", "group_a", "LowThroughput")["for"], "5m") + + def test_patch_replaces_alert_name(self): + result = AlertRulesCustomization.from_yaml(""" + patch: + - where: + alert: HighLatency + set: + alert: RenamedLatency + """).apply(_sample_alerts()) + renamed = _find_rule(result, "app-1", "group_a", "RenamedLatency") + self.assertEqual(renamed["expr"], "latency > 100") + + def test_patch_replaces_expr(self): + result = AlertRulesCustomization.from_yaml(""" + patch: + - where: + alert: HostDown + set: + expr: up == 0 + """).apply(_sample_alerts()) + self.assertEqual(_find_rule(result, "app-1", "group_b", "HostDown")["expr"], "up == 0") + + def test_patch_merges_labels(self): + result = AlertRulesCustomization.from_yaml(""" + patch: + - where: + alert: HighLatency + set: + labels: + severity: page + extra: added + """).apply(_sample_alerts()) + labels = _find_rule(result, "app-1", "group_a", "HighLatency")["labels"] + # existing key overwritten, new key added, other keys untouched + self.assertEqual(labels["severity"], "page") + self.assertEqual(labels["extra"], "added") + self.assertEqual(labels["juju_application"], "app-1") + + def test_patch_merges_annotations(self): + result = AlertRulesCustomization.from_yaml(""" + patch: + - where: + alert: HighLatency + set: + annotations: + summary: new summary + description: new description + """).apply(_sample_alerts()) + annotations = _find_rule(result, "app-1", "group_a", "HighLatency")["annotations"] + self.assertEqual(annotations["summary"], "new summary") + self.assertEqual(annotations["description"], "new description") + + def test_patch_skips_recording_rules(self): + result = AlertRulesCustomization.from_yaml(""" + patch: + - where: + group: group_a + set: + expr: hacked + """).apply(_sample_alerts()) + record = _find_rule(result, "app-1", "group_a", "job:latency:mean5m", by_record=True) + self.assertEqual(record["expr"], "avg(latency)") + # Alerting rules in the same group were patched. + self.assertEqual(_find_rule(result, "app-1", "group_a", "HighLatency")["expr"], "hacked") + + def test_patch_matches_on_labels(self): + result = AlertRulesCustomization.from_yaml(""" + patch: + - where: + labels: + severity: warning + set: + labels: + severity: critical + """).apply(_sample_alerts()) + self.assertEqual( + _find_rule(result, "app-1", "group_a", "LowThroughput")["labels"]["severity"], + "critical", + ) + # The recording rule also has severity=warning but must not be patched. + record = _find_rule(result, "app-1", "group_a", "job:latency:mean5m", by_record=True) + self.assertEqual(record["labels"]["severity"], "warning") + # The alert in the other app is unaffected. + self.assertEqual(_find_rule(result, "app-2", "group_c", "OtherAlert")["expr"], "x > 0") + + +class TestAdd(unittest.TestCase): + _config = """ + add: + groups: + - name: my-custom-alerts + rules: + - alert: MyAlert + expr: up{juju_model="prod"} == 0 + for: 5m + """ + + def test_add_inserts_groups_under_fixed_key(self): + sample = _sample_alerts() + result = AlertRulesCustomization.from_yaml(self._config).apply(sample) + + custom = result[CUSTOM_ALERT_RULES_KEY] + self.assertEqual(custom["groups"][0]["name"], "my-custom-alerts") + self.assertEqual(custom["groups"][0]["rules"][0]["alert"], "MyAlert") + # Existing identifiers are left untouched. + self.assertEqual(set(result), set(sample) | {CUSTOM_ALERT_RULES_KEY}) + + def test_added_rules_receive_no_topology_injection(self): + result = AlertRulesCustomization.from_yaml(self._config).apply(_sample_alerts()) + rule = result[CUSTOM_ALERT_RULES_KEY]["groups"][0]["rules"][0] + self.assertEqual(rule["expr"], 'up{juju_model="prod"} == 0') + self.assertNotIn("labels", rule) + + def test_added_rules_are_deep_copied(self): + customization = AlertRulesCustomization.from_yaml(self._config) + result = customization.apply({}) + result[CUSTOM_ALERT_RULES_KEY]["groups"][0]["rules"][0]["alert"] = "Mangled" + # A subsequent apply() must not be affected by mutations of a previous output. + fresh = customization.apply({})[CUSTOM_ALERT_RULES_KEY]["groups"][0]["rules"][0] + self.assertEqual(fresh["alert"], "MyAlert") + + +class TestApplySemantics(unittest.TestCase): + def test_input_is_not_mutated(self): + sample = _sample_alerts() + snapshot = copy.deepcopy(sample) + AlertRulesCustomization.from_yaml(""" + remove: + - where: + alert: LowThroughput + patch: + - where: + alert: HighLatency + set: + for: 1h + labels: + severity: page + add: + groups: + - name: added + rules: + - alert: Added + expr: up + """).apply(sample) + self.assertEqual(sample, snapshot) + + def test_order_of_operations_is_remove_then_patch_then_add(self): + result = AlertRulesCustomization.from_yaml(""" + remove: + - where: + alert: GoneForever + patch: + - where: + alert: GoneForever + set: + for: 1m + - where: + alert: Survivor + set: + for: 2m + add: + groups: + - name: added + rules: + - alert: GoneForever + expr: up + """).apply( + { + "app": { + "groups": [ + { + "name": "g", + "rules": [ + {"alert": "GoneForever", "expr": "x", "for": "10m"}, + {"alert": "Survivor", "expr": "y", "for": "10m"}, + ], + } + ] + } + } + ) + rules = result["app"]["groups"][0]["rules"] + # Removed despite a patch entry targeting it; patch applied to the survivor; + # the added rule lands under the fixed key, untouched by remove/patch. + self.assertEqual([r["alert"] for r in rules], ["Survivor"]) + self.assertEqual(rules[0]["for"], "2m") + added = result[CUSTOM_ALERT_RULES_KEY]["groups"][0]["rules"][0] + self.assertEqual(added["alert"], "GoneForever") + self.assertNotIn("for", added) # the patch entry did not leak into the added rule + + def test_apply_is_reusable_across_inputs(self): + customization = AlertRulesCustomization.from_yaml(""" + remove: + - where: + alert: HostDown + """) + result_1 = customization.apply(_sample_alerts()) + self.assertNotIn("HostDown", str(result_1["app-1"])) + self.assertIn("app-2", result_1) + + result_2 = customization.apply( + { + "other": { + "groups": [{"name": "g", "rules": [{"alert": "HostDown", "expr": "up < 1"}]}] + } + } + ) + self.assertEqual(result_2, {}) + + # And the first input's result is unchanged by the second call. + self.assertIn("app-1", result_1) + + +if __name__ == "__main__": + unittest.main() From 566648856dcca61022a346583ec1189c48fb4062 Mon Sep 17 00:00:00 2001 From: Sina Date: Mon, 24 Aug 2026 12:56:23 -0400 Subject: [PATCH 2/6] fix: make test YAML configs formatter-agnostic across black versions CI resolves black 24.8.0 (Python 3.8) while local dev on newer interpreters resolves black 26.x; the two styles disagree on multiline string literals passed directly as call arguments. Assign such configs to local variables before passing them to from_yaml(), which both styles format identically. --- tests/test_rules_customization.py | 262 +++++++++++++++--------------- 1 file changed, 132 insertions(+), 130 deletions(-) diff --git a/tests/test_rules_customization.py b/tests/test_rules_customization.py index 1bad8831..9f0882c0 100644 --- a/tests/test_rules_customization.py +++ b/tests/test_rules_customization.py @@ -79,15 +79,16 @@ def test_non_mapping_top_level_raises(self): AlertRulesCustomization.from_yaml(config) def test_unknown_top_level_key_raises(self): + config = """ + remove: + - where: + alert: Foo + destroy: + - where: + alert: Foo + """ with self.assertRaisesRegex(AlertRulesCustomizationError, "top-level"): - AlertRulesCustomization.from_yaml(""" - remove: - - where: - alert: Foo - destroy: - - where: - alert: Foo - """) + AlertRulesCustomization.from_yaml(config) def test_remove_missing_where_raises(self): with self.assertRaisesRegex(AlertRulesCustomizationError, "remove"): @@ -95,81 +96,43 @@ def test_remove_missing_where_raises(self): def test_remove_empty_where_raises(self): with self.assertRaisesRegex(AlertRulesCustomizationError, "'where' must not be empty"): - AlertRulesCustomization.from_yaml(""" - remove: - - where: {} - """) + AlertRulesCustomization.from_yaml("remove:\n - where: {}") def test_remove_unknown_where_key_raises(self): with self.assertRaisesRegex(AlertRulesCustomizationError, "'where' keys"): - AlertRulesCustomization.from_yaml(""" - remove: - - where: - expr: up < 1 - """) + AlertRulesCustomization.from_yaml("remove:\n - where:\n expr: up < 1") def test_patch_missing_where_raises(self): with self.assertRaisesRegex(AlertRulesCustomizationError, "patch"): - AlertRulesCustomization.from_yaml(""" - patch: - - set: - for: 5m - """) + AlertRulesCustomization.from_yaml("patch:\n - set:\n for: 5m") def test_patch_missing_set_raises(self): with self.assertRaisesRegex(AlertRulesCustomizationError, "patch"): - AlertRulesCustomization.from_yaml(""" - patch: - - where: - alert: Foo - """) + AlertRulesCustomization.from_yaml("patch:\n - where:\n alert: Foo") def test_patch_empty_where_raises(self): with self.assertRaisesRegex(AlertRulesCustomizationError, "'where' must not be empty"): - AlertRulesCustomization.from_yaml(""" - patch: - - where: {} - set: - for: 5m - """) + AlertRulesCustomization.from_yaml("patch:\n - where: {}\n set:\n for: 5m") def test_patch_unknown_where_key_raises(self): with self.assertRaisesRegex(AlertRulesCustomizationError, "'where' keys"): - AlertRulesCustomization.from_yaml(""" - patch: - - where: - record: some:record - set: - expr: up - """) + AlertRulesCustomization.from_yaml( + "patch:\n - where:\n record: some:record\n set:\n expr: up" + ) def test_patch_unknown_set_key_raises(self): with self.assertRaisesRegex(AlertRulesCustomizationError, "'set' keys"): - AlertRulesCustomization.from_yaml(""" - patch: - - where: - alert: Foo - set: - duration: 5m - """) + AlertRulesCustomization.from_yaml( + "patch:\n - where:\n alert: Foo\n set:\n duration: 5m" + ) def test_add_without_groups_raises(self): with self.assertRaisesRegex(AlertRulesCustomizationError, "'add'"): - AlertRulesCustomization.from_yaml(""" - add: - rules: - - alert: Foo - expr: up - """) + AlertRulesCustomization.from_yaml("add:\n rules:\n - alert: Foo\n expr: up") def test_add_groups_not_a_list_raises(self): with self.assertRaisesRegex(AlertRulesCustomizationError, "'add.groups' must be a list"): - AlertRulesCustomization.from_yaml(""" - add: - groups: - name: my-group - rules: [] - """) + AlertRulesCustomization.from_yaml("add:\n groups:\n name: my-group\n rules: []") def test_add_group_missing_name_or_rules_raises(self): for group in ("rules: []", "name: my-group"): @@ -180,8 +143,8 @@ def test_add_malformed_values_raise(self): cases = { "add not a mapping": "add: groups", "group not a mapping": "add:\n groups:\n - just-a-string", - "name not a string": ("add:\n groups:\n - name: [1]\n rules: []"), - "rules not a list": ("add:\n groups:\n - name: my-group\n rules: nope"), + "name not a string": "add:\n groups:\n - name: [1]\n rules: []", + "rules not a list": "add:\n groups:\n - name: my-group\n rules: nope", } for case, config in cases.items(): with self.subTest(case): @@ -196,8 +159,8 @@ def test_operations_not_a_list_raises(self): def test_malformed_operation_entries_raise(self): cases = { "where not a mapping": "remove:\n - where: nope", - "where.alert not a string": ("remove:\n - where:\n alert: [1, 2]"), - "where.labels not a mapping": ("remove:\n - where:\n labels: severity"), + "where.alert not a string": "remove:\n - where:\n alert: [1, 2]", + "where.labels not a mapping": "remove:\n - where:\n labels: severity", "set not a mapping": "patch:\n - where:\n alert: Foo\n set: nope", "set.expr not a string": ( "patch:\n - where:\n alert: Foo\n set:\n expr: {a: b}" @@ -233,32 +196,38 @@ def test_none_parsing_config_is_noop(self): self._assert_noop("{}") def test_zero_match_remove_is_noop(self): - self._assert_noop(""" + config = """ remove: - where: alert: NoSuchAlert - where: labels: nope: nothing - """) + """ + self._assert_noop(config) def test_zero_match_patch_is_noop(self): - self._assert_noop(""" + config = """ patch: - where: alert: NoSuchAlert set: for: 1m - """) + """ + self._assert_noop(config) class TestRemove(unittest.TestCase): + def _apply(self, config): + return AlertRulesCustomization.from_yaml(config).apply(_sample_alerts()) + def test_remove_by_alert_name(self): - result = AlertRulesCustomization.from_yaml(""" + config = """ remove: - where: alert: LowThroughput - """).apply(_sample_alerts()) + """ + result = self._apply(config) group_names = [g["name"] for g in result["app-1"]["groups"]] self.assertEqual(group_names, ["group_a", "group_b"]) @@ -266,22 +235,24 @@ def test_remove_by_alert_name(self): self.assertEqual(rule_names, ["HighLatency", None]) # record remains def test_remove_by_group_only_drops_entire_group_including_recording_rules(self): - result = AlertRulesCustomization.from_yaml(""" + config = """ remove: - where: group: group_a - """).apply(_sample_alerts()) + """ + result = self._apply(config) group_names = [g["name"] for g in result["app-1"]["groups"]] self.assertEqual(group_names, ["group_b"]) def test_remove_group_with_other_selector_keeps_recording_rules(self): - result = AlertRulesCustomization.from_yaml(""" + config = """ remove: - where: group: group_a alert: LowThroughput - """).apply(_sample_alerts()) + """ + result = self._apply(config) group = next(g for g in result["app-1"]["groups"] if g["name"] == "group_a") rule_names = [r.get("alert") or r.get("record") for r in group["rules"]] @@ -289,33 +260,36 @@ def test_remove_group_with_other_selector_keeps_recording_rules(self): self.assertEqual(rule_names, ["HighLatency", "job:latency:mean5m"]) def test_remove_by_alert_and_labels_combined(self): - result = AlertRulesCustomization.from_yaml(""" + config_matching = """ remove: - where: alert: HighLatency labels: severity: critical - """).apply(_sample_alerts()) + """ + result = self._apply(config_matching) self.assertNotIn("HighLatency", str(result)) # Same alert but non-matching label value: nothing removed. - result = AlertRulesCustomization.from_yaml(""" + config_not_matching = """ remove: - where: alert: HighLatency labels: severity: warning - """).apply(_sample_alerts()) + """ + result = self._apply(config_not_matching) self.assertIn("HighLatency", str(result)) def test_remove_by_group_and_labels_combined(self): - result = AlertRulesCustomization.from_yaml(""" + config = """ remove: - where: group: group_a labels: severity: warning - """).apply(_sample_alerts()) + """ + result = self._apply(config) group = next(g for g in result["app-1"]["groups"] if g["name"] == "group_a") rule_names = [r.get("alert") or r.get("record") for r in group["rules"]] @@ -324,94 +298,111 @@ def test_remove_by_group_and_labels_combined(self): self.assertEqual(rule_names, ["HighLatency", "job:latency:mean5m"]) def test_remove_by_annotations(self): - result = AlertRulesCustomization.from_yaml(""" + config = """ remove: - where: annotations: summary: latency is high - """).apply(_sample_alerts()) + """ + result = self._apply(config) + self.assertNotIn("HighLatency", str(result)) self.assertIn("LowThroughput", str(result)) def test_remove_multiple_entries_are_ored(self): - result = AlertRulesCustomization.from_yaml(""" + config = """ remove: - where: alert: HighLatency - where: alert: OtherAlert - """).apply(_sample_alerts()) + """ + result = self._apply(config) + self.assertNotIn("HighLatency", str(result)) self.assertNotIn("OtherAlert", str(result)) self.assertIn("LowThroughput", str(result)) self.assertIn("HostDown", str(result)) def test_remove_prunes_empty_groups_and_drops_empty_identifiers(self): - result = AlertRulesCustomization.from_yaml(""" + config_pruning_group = """ remove: - where: alert: HostDown - """).apply(_sample_alerts()) + """ + result = self._apply(config_pruning_group) # group_b became empty and was pruned; app-1 keeps group_a only. self.assertEqual([g["name"] for g in result["app-1"]["groups"]], ["group_a"]) self.assertIn("app-2", result) - result = AlertRulesCustomization.from_yaml(""" + config_dropping_identifier = """ remove: - where: alert: OtherAlert - """).apply(_sample_alerts()) + """ + result = self._apply(config_dropping_identifier) # app-2's only group became empty, so app-2 was dropped entirely. self.assertNotIn("app-2", result) self.assertIn("app-1", result) def test_remove_preserves_recording_rules_when_group_not_sole_selector(self): - result = AlertRulesCustomization.from_yaml(""" + config = """ remove: - where: labels: severity: warning - """).apply(_sample_alerts()) + """ + result = self._apply(config) + record = _find_rule(result, "app-1", "group_a", "job:latency:mean5m", by_record=True) self.assertEqual(record["expr"], "avg(latency)") class TestPatch(unittest.TestCase): + def _apply(self, config): + return AlertRulesCustomization.from_yaml(config).apply(_sample_alerts()) + def test_patch_updates_for(self): - result = AlertRulesCustomization.from_yaml(""" + config = """ patch: - where: alert: HighLatency set: for: 30m - """).apply(_sample_alerts()) + """ + result = self._apply(config) + self.assertEqual(_find_rule(result, "app-1", "group_a", "HighLatency")["for"], "30m") # Untouched rule keeps its original value. self.assertEqual(_find_rule(result, "app-1", "group_a", "LowThroughput")["for"], "5m") def test_patch_replaces_alert_name(self): - result = AlertRulesCustomization.from_yaml(""" + config = """ patch: - where: alert: HighLatency set: alert: RenamedLatency - """).apply(_sample_alerts()) + """ + result = self._apply(config) + renamed = _find_rule(result, "app-1", "group_a", "RenamedLatency") self.assertEqual(renamed["expr"], "latency > 100") def test_patch_replaces_expr(self): - result = AlertRulesCustomization.from_yaml(""" + config = """ patch: - where: alert: HostDown set: expr: up == 0 - """).apply(_sample_alerts()) + """ + result = self._apply(config) + self.assertEqual(_find_rule(result, "app-1", "group_b", "HostDown")["expr"], "up == 0") def test_patch_merges_labels(self): - result = AlertRulesCustomization.from_yaml(""" + config = """ patch: - where: alert: HighLatency @@ -419,7 +410,9 @@ def test_patch_merges_labels(self): labels: severity: page extra: added - """).apply(_sample_alerts()) + """ + result = self._apply(config) + labels = _find_rule(result, "app-1", "group_a", "HighLatency")["labels"] # existing key overwritten, new key added, other keys untouched self.assertEqual(labels["severity"], "page") @@ -427,7 +420,7 @@ def test_patch_merges_labels(self): self.assertEqual(labels["juju_application"], "app-1") def test_patch_merges_annotations(self): - result = AlertRulesCustomization.from_yaml(""" + config = """ patch: - where: alert: HighLatency @@ -435,26 +428,30 @@ def test_patch_merges_annotations(self): annotations: summary: new summary description: new description - """).apply(_sample_alerts()) + """ + result = self._apply(config) + annotations = _find_rule(result, "app-1", "group_a", "HighLatency")["annotations"] self.assertEqual(annotations["summary"], "new summary") self.assertEqual(annotations["description"], "new description") def test_patch_skips_recording_rules(self): - result = AlertRulesCustomization.from_yaml(""" + config = """ patch: - where: group: group_a set: expr: hacked - """).apply(_sample_alerts()) + """ + result = self._apply(config) + record = _find_rule(result, "app-1", "group_a", "job:latency:mean5m", by_record=True) self.assertEqual(record["expr"], "avg(latency)") # Alerting rules in the same group were patched. self.assertEqual(_find_rule(result, "app-1", "group_a", "HighLatency")["expr"], "hacked") def test_patch_matches_on_labels(self): - result = AlertRulesCustomization.from_yaml(""" + config = """ patch: - where: labels: @@ -462,7 +459,9 @@ def test_patch_matches_on_labels(self): set: labels: severity: critical - """).apply(_sample_alerts()) + """ + result = self._apply(config) + self.assertEqual( _find_rule(result, "app-1", "group_a", "LowThroughput")["labels"]["severity"], "critical", @@ -512,9 +511,7 @@ def test_added_rules_are_deep_copied(self): class TestApplySemantics(unittest.TestCase): def test_input_is_not_mutated(self): - sample = _sample_alerts() - snapshot = copy.deepcopy(sample) - AlertRulesCustomization.from_yaml(""" + config = """ remove: - where: alert: LowThroughput @@ -531,11 +528,14 @@ def test_input_is_not_mutated(self): rules: - alert: Added expr: up - """).apply(sample) + """ + sample = _sample_alerts() + snapshot = copy.deepcopy(sample) + AlertRulesCustomization.from_yaml(config).apply(sample) self.assertEqual(sample, snapshot) def test_order_of_operations_is_remove_then_patch_then_add(self): - result = AlertRulesCustomization.from_yaml(""" + config = """ remove: - where: alert: GoneForever @@ -554,21 +554,22 @@ def test_order_of_operations_is_remove_then_patch_then_add(self): rules: - alert: GoneForever expr: up - """).apply( - { - "app": { - "groups": [ - { - "name": "g", - "rules": [ - {"alert": "GoneForever", "expr": "x", "for": "10m"}, - {"alert": "Survivor", "expr": "y", "for": "10m"}, - ], - } - ] - } + """ + relation_alerts = { + "app": { + "groups": [ + { + "name": "g", + "rules": [ + {"alert": "GoneForever", "expr": "x", "for": "10m"}, + {"alert": "Survivor", "expr": "y", "for": "10m"}, + ], + } + ] } - ) + } + result = AlertRulesCustomization.from_yaml(config).apply(relation_alerts) + rules = result["app"]["groups"][0]["rules"] # Removed despite a patch entry targeting it; patch applied to the survivor; # the added rule lands under the fixed key, untouched by remove/patch. @@ -579,22 +580,23 @@ def test_order_of_operations_is_remove_then_patch_then_add(self): self.assertNotIn("for", added) # the patch entry did not leak into the added rule def test_apply_is_reusable_across_inputs(self): - customization = AlertRulesCustomization.from_yaml(""" + config = """ remove: - where: alert: HostDown - """) + """ + customization = AlertRulesCustomization.from_yaml(config) + result_1 = customization.apply(_sample_alerts()) self.assertNotIn("HostDown", str(result_1["app-1"])) self.assertIn("app-2", result_1) - result_2 = customization.apply( - { - "other": { - "groups": [{"name": "g", "rules": [{"alert": "HostDown", "expr": "up < 1"}]}] - } + other_relation_alerts = { + "other": { + "groups": [{"name": "g", "rules": [{"alert": "HostDown", "expr": "up < 1"}]}] } - ) + } + result_2 = customization.apply(other_relation_alerts) self.assertEqual(result_2, {}) # And the first input's result is unchanged by the second call. From 58f74d5c5a0d566f324067d29e79840a3a8c37de Mon Sep 17 00:00:00 2001 From: Sina Date: Mon, 24 Aug 2026 14:09:21 -0400 Subject: [PATCH 3/6] feat: bump version --- pyproject.toml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pyproject.toml b/pyproject.toml index fff72bd8..8d4dc40d 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "cosl" -version = "1.10.3" +version = "1.11.3" authors = [ { name = "sed-i", email = "82407168+sed-i@users.noreply.github.com" }, ] From 637ee1d1463fa03a4f2bf1102ab2a67b7979314f Mon Sep 17 00:00:00 2001 From: Sina Date: Tue, 25 Aug 2026 16:19:20 -0400 Subject: [PATCH 4/6] fix: remove all addition related logic --- pyproject.toml | 5 + src/cosl/rules_customization.py | 79 ++----------- tests/test_rules_customization.py | 83 +------------ uv.lock | 188 ++++++++++++++++-------------- 4 files changed, 114 insertions(+), 241 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index 8d4dc40d..e2a95e3c 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -123,3 +123,8 @@ ignore-words-list = "assertIn" [tool.hatch.metadata] # allow git+ dependencies in pyproject allow-direct-references = true + +[dependency-groups] +test = [ + "pytest>=8.3.5", +] diff --git a/src/cosl/rules_customization.py b/src/cosl/rules_customization.py index e1fb600b..80dd17a7 100644 --- a/src/cosl/rules_customization.py +++ b/src/cosl/rules_customization.py @@ -9,13 +9,11 @@ ``MetricsConsumer.alerts`` produce) and an admin-provided YAML customization config, and returns the modified rules in the same format. -The customization config supports three top-level keys: +The customization config supports two top-level keys: - ``remove``: drop matching alerting rules (or entire groups, when ``group`` is the only selector). - ``patch``: modify matching alerting rules by merging a ``set`` block into them. -- ``add``: insert admin-authored rule groups into the output under the fixed key - ``custom_alert_rules``. Matching is performed via a ``where`` block, supporting exact equality on ``alert``, ``group``, ``labels`` and ``annotations``. All fields within one ``where`` are ANDed; @@ -24,9 +22,6 @@ Recording rules are never removed or patched unless an entire group is dropped via a group-only ``where`` selector. -Rules added via the ``add`` block do NOT receive automatic Juju topology injection: the -admin must include topology matchers in their expressions and labels manually. - This class is a pure transformation helper. It does not call CosTool, Pebble, Prometheus, Loki or Mimir APIs, does not write files and does not set statuses. Validation of the resulting rules is the charm's responsibility after calling @@ -44,10 +39,7 @@ logger = logging.getLogger(__name__) -CUSTOM_ALERT_RULES_KEY = "custom_alert_rules" -"""Fixed output key under which rules from the ``add`` block are inserted.""" - -_VALID_TOP_LEVEL_KEYS = frozenset({"remove", "patch", "add"}) +_VALID_TOP_LEVEL_KEYS = frozenset({"remove", "patch"}) _VALID_WHERE_KEYS = frozenset({"alert", "group", "labels", "annotations"}) _VALID_SET_KEYS = frozenset({"alert", "expr", "for", "labels", "annotations"}) @@ -167,45 +159,8 @@ def _validate_patch(entries: Any) -> List[Dict[str, Any]]: return validated -def _validate_add(add_block: Any) -> Optional[OfficialRuleFileFormat]: - """Validate the ``add`` block and return it. - - Raises: - AlertRulesCustomizationError: if the add block does not conform to the - official rule file format shape. - """ - if add_block is None: - return None - if not isinstance(add_block, collections.abc.Mapping): - raise AlertRulesCustomizationError("'add' must be a mapping with a 'groups' key") - validated_add: Dict[str, Any] = dict(cast(Mapping[Any, Any], add_block)) - if "groups" not in validated_add: - raise AlertRulesCustomizationError("'add': missing required key 'groups'") - if not isinstance(validated_add["groups"], list): - raise AlertRulesCustomizationError("'add.groups' must be a list") - groups: List[Any] = cast(List[Any], validated_add["groups"]) - for index, item in enumerate(groups): - if not isinstance(item, collections.abc.Mapping): - raise AlertRulesCustomizationError(f"'add.groups[{index}]' must be a mapping") - group: Dict[str, Any] = dict(cast(Mapping[Any, Any], item)) - name = group.get("name") - if name is None: - raise AlertRulesCustomizationError( - f"'add.groups[{index}]': missing required key 'name'" - ) - if not isinstance(name, str): - raise AlertRulesCustomizationError(f"'add.groups[{index}].name' must be a string") - if "rules" not in group: - raise AlertRulesCustomizationError( - f"'add.groups[{index}]': missing required key 'rules'" - ) - if not isinstance(group["rules"], list): - raise AlertRulesCustomizationError(f"'add.groups[{index}].rules' must be a list") - return cast(OfficialRuleFileFormat, validated_add) - - class AlertRulesCustomization: - """Apply admin-defined remove/patch/add operations to relation-derived alert rules. + """Apply admin-defined remove/patch operations to relation-derived alert rules. Build an instance with :meth:`from_yaml`, then call :meth:`apply` on the alerts dict (e.g. ``self.metrics_consumer.alerts``). The instance is reusable: ``apply()`` can be @@ -216,7 +171,6 @@ def __init__( self, remove: Optional[List[Dict[str, Any]]] = None, patch: Optional[List[Dict[str, Any]]] = None, - add: Optional[OfficialRuleFileFormat] = None, ): r"""Build a customization object from pre-validated operation blocks. @@ -224,7 +178,6 @@ def __init__( """ self._remove: List[Dict[str, Any]] = remove or [] self._patch: List[Dict[str, Any]] = patch or [] - self._add: Optional[OfficialRuleFileFormat] = copy.deepcopy(add) @classmethod def from_yaml(cls, config_string: str) -> "AlertRulesCustomization": @@ -239,10 +192,9 @@ def from_yaml(cls, config_string: str) -> "AlertRulesCustomization": Raises: AlertRulesCustomizationError: on invalid YAML, unknown top-level keys - (only ``remove``, ``patch``, ``add`` are allowed), invalid operation + (only ``remove``, ``patch`` are allowed), invalid operation shape (missing ``where``, unknown selector keys, unknown set keys), - empty ``where`` selectors, or an ``add`` block not conforming to the - official rule file format shape. + empty ``where`` selectors. """ if not config_string or not config_string.strip(): # Empty or whitespace-only config: no-op. @@ -274,15 +226,14 @@ def from_yaml(cls, config_string: str) -> "AlertRulesCustomization": return cls( remove=_validate_remove(config.get("remove")), patch=_validate_patch(config.get("patch")), - add=_validate_add(config.get("add")), ) def apply( self, relation_alerts: Mapping[str, OfficialRuleFileFormat] ) -> Dict[str, OfficialRuleFileFormat]: - """Apply remove, patch and add operations to the input rules. + """Apply remove and patch operations to the input rules. - Operations run in this order: remove, patch, add. The input is never mutated; + Operations run in this order: remove, patch. The input is never mutated; the transformations operate on a deep copy. Args: @@ -291,14 +242,12 @@ def apply( Returns: The transformed rules, in the same format as the input. Identifiers whose - ``groups`` list becomes empty after removal are dropped. If ``add`` is - configured, its groups are inserted under ``custom_alert_rules``. + ``groups`` list becomes empty after removal are dropped. """ output: Dict[str, OfficialRuleFileFormat] = copy.deepcopy(dict(relation_alerts)) self._apply_remove(output) self._apply_patch(output) - self._apply_add(output) return output @@ -433,15 +382,3 @@ def _patch_rule( identifier, ", ".join(changes), ) - - def _apply_add(self, output: Dict[str, OfficialRuleFileFormat]) -> None: - """Insert the ``add`` block's groups under the fixed custom alert rules key.""" - if self._add is None: - return - output[CUSTOM_ALERT_RULES_KEY] = copy.deepcopy(self._add) - added_groups = cast(List[Any], self._add.get("groups", [])) - logger.debug( - "Added %d group(s) under '%s'", - len(added_groups), - CUSTOM_ALERT_RULES_KEY, - ) diff --git a/tests/test_rules_customization.py b/tests/test_rules_customization.py index 9f0882c0..b928dbd7 100644 --- a/tests/test_rules_customization.py +++ b/tests/test_rules_customization.py @@ -5,7 +5,6 @@ import unittest from cosl.rules_customization import ( - CUSTOM_ALERT_RULES_KEY, AlertRulesCustomization, AlertRulesCustomizationError, ) @@ -126,31 +125,6 @@ def test_patch_unknown_set_key_raises(self): "patch:\n - where:\n alert: Foo\n set:\n duration: 5m" ) - def test_add_without_groups_raises(self): - with self.assertRaisesRegex(AlertRulesCustomizationError, "'add'"): - AlertRulesCustomization.from_yaml("add:\n rules:\n - alert: Foo\n expr: up") - - def test_add_groups_not_a_list_raises(self): - with self.assertRaisesRegex(AlertRulesCustomizationError, "'add.groups' must be a list"): - AlertRulesCustomization.from_yaml("add:\n groups:\n name: my-group\n rules: []") - - def test_add_group_missing_name_or_rules_raises(self): - for group in ("rules: []", "name: my-group"): - with self.assertRaisesRegex(AlertRulesCustomizationError, "add.groups"): - AlertRulesCustomization.from_yaml(f"add:\n groups:\n - {group}") - - def test_add_malformed_values_raise(self): - cases = { - "add not a mapping": "add: groups", - "group not a mapping": "add:\n groups:\n - just-a-string", - "name not a string": "add:\n groups:\n - name: [1]\n rules: []", - "rules not a list": "add:\n groups:\n - name: my-group\n rules: nope", - } - for case, config in cases.items(): - with self.subTest(case): - with self.assertRaises(AlertRulesCustomizationError): - AlertRulesCustomization.from_yaml(config) - def test_operations_not_a_list_raises(self): for key in ("remove", "patch"): with self.assertRaisesRegex(AlertRulesCustomizationError, f"'{key}'"): @@ -472,43 +446,6 @@ def test_patch_matches_on_labels(self): # The alert in the other app is unaffected. self.assertEqual(_find_rule(result, "app-2", "group_c", "OtherAlert")["expr"], "x > 0") - -class TestAdd(unittest.TestCase): - _config = """ - add: - groups: - - name: my-custom-alerts - rules: - - alert: MyAlert - expr: up{juju_model="prod"} == 0 - for: 5m - """ - - def test_add_inserts_groups_under_fixed_key(self): - sample = _sample_alerts() - result = AlertRulesCustomization.from_yaml(self._config).apply(sample) - - custom = result[CUSTOM_ALERT_RULES_KEY] - self.assertEqual(custom["groups"][0]["name"], "my-custom-alerts") - self.assertEqual(custom["groups"][0]["rules"][0]["alert"], "MyAlert") - # Existing identifiers are left untouched. - self.assertEqual(set(result), set(sample) | {CUSTOM_ALERT_RULES_KEY}) - - def test_added_rules_receive_no_topology_injection(self): - result = AlertRulesCustomization.from_yaml(self._config).apply(_sample_alerts()) - rule = result[CUSTOM_ALERT_RULES_KEY]["groups"][0]["rules"][0] - self.assertEqual(rule["expr"], 'up{juju_model="prod"} == 0') - self.assertNotIn("labels", rule) - - def test_added_rules_are_deep_copied(self): - customization = AlertRulesCustomization.from_yaml(self._config) - result = customization.apply({}) - result[CUSTOM_ALERT_RULES_KEY]["groups"][0]["rules"][0]["alert"] = "Mangled" - # A subsequent apply() must not be affected by mutations of a previous output. - fresh = customization.apply({})[CUSTOM_ALERT_RULES_KEY]["groups"][0]["rules"][0] - self.assertEqual(fresh["alert"], "MyAlert") - - class TestApplySemantics(unittest.TestCase): def test_input_is_not_mutated(self): config = """ @@ -522,19 +459,13 @@ def test_input_is_not_mutated(self): for: 1h labels: severity: page - add: - groups: - - name: added - rules: - - alert: Added - expr: up """ sample = _sample_alerts() snapshot = copy.deepcopy(sample) AlertRulesCustomization.from_yaml(config).apply(sample) self.assertEqual(sample, snapshot) - def test_order_of_operations_is_remove_then_patch_then_add(self): + def test_order_of_operations_is_remove_then_patch(self): config = """ remove: - where: @@ -548,12 +479,6 @@ def test_order_of_operations_is_remove_then_patch_then_add(self): alert: Survivor set: for: 2m - add: - groups: - - name: added - rules: - - alert: GoneForever - expr: up """ relation_alerts = { "app": { @@ -571,13 +496,9 @@ def test_order_of_operations_is_remove_then_patch_then_add(self): result = AlertRulesCustomization.from_yaml(config).apply(relation_alerts) rules = result["app"]["groups"][0]["rules"] - # Removed despite a patch entry targeting it; patch applied to the survivor; - # the added rule lands under the fixed key, untouched by remove/patch. + # Removed despite a patch entry targeting it; patch applied to the survivor. self.assertEqual([r["alert"] for r in rules], ["Survivor"]) self.assertEqual(rules[0]["for"], "2m") - added = result[CUSTOM_ALERT_RULES_KEY]["groups"][0]["rules"][0] - self.assertEqual(added["alert"], "GoneForever") - self.assertNotIn("for", added) # the patch entry did not leak into the added rule def test_apply_is_reusable_across_inputs(self): config = """ diff --git a/uv.lock b/uv.lock index e20c5c5b..377eb0f4 100644 --- a/uv.lock +++ b/uv.lock @@ -1,5 +1,5 @@ version = 1 -revision = 2 +revision = 3 requires-python = ">=3.8" resolution-markers = [ "python_full_version >= '3.10'", @@ -36,13 +36,13 @@ resolution-markers = [ "python_full_version < '3.9'", ] dependencies = [ - { name = "click", version = "8.1.8", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.9'" }, - { name = "mypy-extensions", marker = "python_full_version < '3.9'" }, - { name = "packaging", marker = "python_full_version < '3.9'" }, - { name = "pathspec", version = "0.12.1", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.9'" }, - { name = "platformdirs", version = "4.3.6", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.9'" }, - { name = "tomli", marker = "python_full_version < '3.9'" }, - { name = "typing-extensions", version = "4.13.2", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.9'" }, + { name = "click", version = "8.1.8", source = { registry = "https://pypi.org/simple" } }, + { name = "mypy-extensions" }, + { name = "packaging" }, + { name = "pathspec", version = "0.12.1", source = { registry = "https://pypi.org/simple" } }, + { name = "platformdirs", version = "4.3.6", source = { registry = "https://pypi.org/simple" } }, + { name = "tomli" }, + { name = "typing-extensions", version = "4.13.2", source = { registry = "https://pypi.org/simple" } }, ] sdist = { url = "https://files.pythonhosted.org/packages/04/b0/46fb0d4e00372f4a86a6f8efa3cb193c9f64863615e39010b1477e010578/black-24.8.0.tar.gz", hash = "sha256:2500945420b6784c38b9ee885af039f5e7471ef284ab03fa35ecdde4688cd83f", size = 644810, upload-time = "2024-08-02T17:43:18.405Z" } wheels = [ @@ -77,14 +77,14 @@ resolution-markers = [ "python_full_version == '3.9.*'", ] dependencies = [ - { name = "click", version = "8.1.8", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version == '3.9.*'" }, - { name = "mypy-extensions", marker = "python_full_version == '3.9.*'" }, - { name = "packaging", marker = "python_full_version == '3.9.*'" }, - { name = "pathspec", version = "1.1.1", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version == '3.9.*'" }, - { name = "platformdirs", version = "4.4.0", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version == '3.9.*'" }, - { name = "pytokens", marker = "python_full_version == '3.9.*'" }, - { name = "tomli", marker = "python_full_version == '3.9.*'" }, - { name = "typing-extensions", version = "4.15.0", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version == '3.9.*'" }, + { name = "click", version = "8.1.8", source = { registry = "https://pypi.org/simple" } }, + { name = "mypy-extensions" }, + { name = "packaging" }, + { name = "pathspec", version = "1.1.1", source = { registry = "https://pypi.org/simple" } }, + { name = "platformdirs", version = "4.4.0", source = { registry = "https://pypi.org/simple" } }, + { name = "pytokens" }, + { name = "tomli" }, + { name = "typing-extensions", version = "4.15.0", source = { registry = "https://pypi.org/simple" } }, ] sdist = { url = "https://files.pythonhosted.org/packages/8c/ad/33adf4708633d047950ff2dfdea2e215d84ac50ef95aff14a614e4b6e9b2/black-25.11.0.tar.gz", hash = "sha256:9a323ac32f5dc75ce7470501b887250be5005a01602e931a15e45593f70f6e08", size = 655669, upload-time = "2025-11-10T01:53:50.558Z" } wheels = [ @@ -123,14 +123,14 @@ resolution-markers = [ "python_full_version >= '3.10'", ] dependencies = [ - { name = "click", version = "8.4.1", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.10'" }, - { name = "mypy-extensions", marker = "python_full_version >= '3.10'" }, - { name = "packaging", marker = "python_full_version >= '3.10'" }, - { name = "pathspec", version = "1.1.1", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.10'" }, - { name = "platformdirs", version = "4.10.0", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.10'" }, - { name = "pytokens", marker = "python_full_version >= '3.10'" }, - { name = "tomli", marker = "python_full_version == '3.10.*'" }, - { name = "typing-extensions", version = "4.15.0", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version == '3.10.*'" }, + { name = "click", version = "8.4.1", source = { registry = "https://pypi.org/simple" } }, + { name = "mypy-extensions" }, + { name = "packaging" }, + { name = "pathspec", version = "1.1.1", source = { registry = "https://pypi.org/simple" } }, + { name = "platformdirs", version = "4.10.0", source = { registry = "https://pypi.org/simple" } }, + { name = "pytokens" }, + { name = "tomli", marker = "python_full_version < '3.11'" }, + { name = "typing-extensions", version = "4.15.0", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.11'" }, ] sdist = { url = "https://files.pythonhosted.org/packages/c0/37/5628dd55bf2b34257fc7603f0fe97c40e3aaf24265f416a9c85c95ca1436/black-26.5.1.tar.gz", hash = "sha256:dd321f668053961824bcc1be1cc1df748b2d7e4fa28086b08331e577b0100a73", size = 679439, upload-time = "2026-05-18T16:53:36.107Z" } wheels = [ @@ -171,7 +171,7 @@ resolution-markers = [ "python_full_version < '3.9'", ] dependencies = [ - { name = "colorama", marker = "python_full_version < '3.10' and sys_platform == 'win32'" }, + { name = "colorama", marker = "sys_platform == 'win32'" }, ] sdist = { url = "https://files.pythonhosted.org/packages/b9/2e/0090cbf739cee7d23781ad4b89a9894a41538e4fcf4c31dcdd705b78eb8b/click-8.1.8.tar.gz", hash = "sha256:ed53c9d8990d83c2a27deae68e4ee337473f6330c040a31d4225c9574d16096a", size = 226593, upload-time = "2024-12-21T18:38:44.339Z" } wheels = [ @@ -186,7 +186,7 @@ resolution-markers = [ "python_full_version >= '3.10'", ] dependencies = [ - { name = "colorama", marker = "python_full_version >= '3.10' and sys_platform == 'win32'" }, + { name = "colorama", marker = "sys_platform == 'win32'" }, ] sdist = { url = "https://files.pythonhosted.org/packages/9b/98/518d8e5081007684232226f475082b30087d0f585e8457db087298259f49/click-8.4.1.tar.gz", hash = "sha256:918b5633eddf6b41c32d4f454bf0de810065c74e3f7dbf8ee5452f8be88d3e96", size = 353007, upload-time = "2026-05-22T04:08:37.769Z" } wheels = [ @@ -229,7 +229,7 @@ wheels = [ [[package]] name = "cosl" -version = "1.10.2" +version = "1.11.3" source = { editable = "." } dependencies = [ { name = "diskcache" }, @@ -268,6 +268,13 @@ dev = [ { name = "setuptools", version = "81.0.0", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.9'" }, ] +[package.dev-dependencies] +test = [ + { name = "pytest", version = "8.3.5", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.9'" }, + { name = "pytest", version = "8.4.2", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version == '3.9.*'" }, + { name = "pytest", version = "9.1.0", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.10'" }, +] + [package.metadata] requires-dist = [ { name = "black", marker = "extra == 'dev'" }, @@ -290,6 +297,9 @@ requires-dist = [ ] provides-extras = ["dev"] +[package.metadata.requires-dev] +test = [{ name = "pytest", specifier = ">=8.3.5" }] + [[package]] name = "coverage" version = "7.6.1" @@ -374,7 +384,7 @@ wheels = [ [package.optional-dependencies] toml = [ - { name = "tomli", marker = "python_full_version < '3.9'" }, + { name = "tomli" }, ] [[package]] @@ -493,7 +503,7 @@ wheels = [ [package.optional-dependencies] toml = [ - { name = "tomli", marker = "python_full_version == '3.9.*'" }, + { name = "tomli" }, ] [[package]] @@ -614,7 +624,7 @@ wheels = [ [package.optional-dependencies] toml = [ - { name = "tomli", marker = "python_full_version >= '3.10' and python_full_version <= '3.11'" }, + { name = "tomli", marker = "python_full_version <= '3.11'" }, ] [[package]] @@ -634,7 +644,7 @@ name = "deprecated" version = "1.3.1" source = { registry = "https://pypi.org/simple" } dependencies = [ - { name = "wrapt", marker = "python_full_version < '3.9'" }, + { name = "wrapt" }, ] sdist = { url = "https://files.pythonhosted.org/packages/49/85/12f0a49a7c4ffb70572b6c2ef13c90c88fd190debda93b23f026b25f9634/deprecated-1.3.1.tar.gz", hash = "sha256:b1b50e0ff0c1fddaa5708a2c6b0a6588bb09b892825ab2b214ac9ea9d92a5223", size = 2932523, upload-time = "2025-10-30T08:19:02.757Z" } wheels = [ @@ -655,8 +665,8 @@ name = "exceptiongroup" version = "1.3.1" source = { registry = "https://pypi.org/simple" } dependencies = [ - { name = "typing-extensions", version = "4.13.2", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.9'" }, - { name = "typing-extensions", version = "4.15.0", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.9' and python_full_version < '3.13'" }, + { name = "typing-extensions", version = "4.13.2", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.9' or python_full_version >= '3.11'" }, + { name = "typing-extensions", version = "4.15.0", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.9' and python_full_version < '3.11'" }, ] sdist = { url = "https://files.pythonhosted.org/packages/50/79/66800aadf48771f6b62f7eb014e352e5d06856655206165d775e675a02c9/exceptiongroup-1.3.1.tar.gz", hash = "sha256:8b412432c6055b0b7d14c310000ae93352ed6754f70fa8f7c34141f91c4e3219", size = 30371, upload-time = "2025-11-21T23:01:54.787Z" } wheels = [ @@ -686,7 +696,7 @@ resolution-markers = [ "python_full_version < '3.9'", ] dependencies = [ - { name = "zipp", version = "3.20.2", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.9'" }, + { name = "zipp", version = "3.20.2", source = { registry = "https://pypi.org/simple" } }, ] sdist = { url = "https://files.pythonhosted.org/packages/cd/12/33e59336dca5be0c398a7482335911a33aa0e20776128f038019f1a95f1b/importlib_metadata-8.5.0.tar.gz", hash = "sha256:71522656f0abace1d072b9e5481a48f07c138e00f079c38c8f883823f9c26bd7", size = 55304, upload-time = "2024-09-11T14:56:08.937Z" } wheels = [ @@ -701,7 +711,7 @@ resolution-markers = [ "python_full_version == '3.9.*'", ] dependencies = [ - { name = "zipp", version = "3.23.1", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version == '3.9.*'" }, + { name = "zipp", version = "3.23.1", source = { registry = "https://pypi.org/simple" } }, ] sdist = { url = "https://files.pythonhosted.org/packages/f3/49/3b30cad09e7771a4982d9975a8cbf64f00d4a1ececb53297f1d9a7be1b10/importlib_metadata-8.7.1.tar.gz", hash = "sha256:49fef1ae6440c182052f407c8d34a68f72efc36db9ca90dc0113398f2fdde8bb", size = 57107, upload-time = "2025-12-21T10:00:19.278Z" } wheels = [ @@ -759,8 +769,8 @@ resolution-markers = [ "python_full_version < '3.9'", ] dependencies = [ - { name = "deprecated", marker = "python_full_version < '3.9'" }, - { name = "importlib-metadata", version = "8.5.0", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.9'" }, + { name = "deprecated" }, + { name = "importlib-metadata", version = "8.5.0", source = { registry = "https://pypi.org/simple" } }, ] sdist = { url = "https://files.pythonhosted.org/packages/9a/8d/1f5a45fbcb9a7d87809d460f09dc3399e3fbd31d7f3e14888345e9d29951/opentelemetry_api-1.33.1.tar.gz", hash = "sha256:1c6055fc0a2d3f23a50c7e17e16ef75ad489345fd3df1f8b8af7c0bbf8a109e8", size = 65002, upload-time = "2025-05-16T18:52:41.146Z" } wheels = [ @@ -775,8 +785,8 @@ resolution-markers = [ "python_full_version == '3.9.*'", ] dependencies = [ - { name = "importlib-metadata", version = "8.7.1", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version == '3.9.*'" }, - { name = "typing-extensions", version = "4.15.0", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version == '3.9.*'" }, + { name = "importlib-metadata", version = "8.7.1", source = { registry = "https://pypi.org/simple" } }, + { name = "typing-extensions", version = "4.15.0", source = { registry = "https://pypi.org/simple" } }, ] sdist = { url = "https://files.pythonhosted.org/packages/fa/fc/b7564cbef36601aef0d6c9bc01f7badb64be8e862c2e1c3c5c3b43b53e4f/opentelemetry_api-1.41.1.tar.gz", hash = "sha256:0ad1814d73b875f84494387dae86ce0b12c68556331ce6ce8fe789197c949621", size = 71416, upload-time = "2026-04-24T13:15:38.262Z" } wheels = [ @@ -791,7 +801,7 @@ resolution-markers = [ "python_full_version >= '3.10'", ] dependencies = [ - { name = "typing-extensions", version = "4.15.0", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.10'" }, + { name = "typing-extensions", version = "4.15.0", source = { registry = "https://pypi.org/simple" } }, ] sdist = { url = "https://files.pythonhosted.org/packages/b4/1c/125e1c936c0873796771b7f04f6c93b9f1bf5d424cea90fda94a99f61da8/opentelemetry_api-1.42.1.tar.gz", hash = "sha256:56c63bea9f77b62856be8c47600474acad853b2924b99b1687c4cb6297166716", size = 72296, upload-time = "2026-05-21T16:32:49.335Z" } wheels = [ @@ -807,12 +817,12 @@ resolution-markers = [ "python_full_version < '3.9'", ] dependencies = [ - { name = "importlib-metadata", version = "8.5.0", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.9'" }, + { name = "importlib-metadata", version = "8.5.0", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version != '3.9.*'" }, { name = "importlib-metadata", version = "8.7.1", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version == '3.9.*'" }, - { name = "opentelemetry-api", version = "1.33.1", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.9'" }, + { name = "opentelemetry-api", version = "1.33.1", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version != '3.9.*'" }, { name = "opentelemetry-api", version = "1.41.1", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version == '3.9.*'" }, - { name = "pyyaml", marker = "python_full_version < '3.10'" }, - { name = "websocket-client", version = "1.8.0", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.9'" }, + { name = "pyyaml" }, + { name = "websocket-client", version = "1.8.0", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version != '3.9.*'" }, { name = "websocket-client", version = "1.9.0", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version == '3.9.*'" }, ] sdist = { url = "https://files.pythonhosted.org/packages/93/98/84789a5e15ad76e043301bbd70b4d39cffaa717ae6eecb662fd0dd0cc7af/ops-2.23.2.tar.gz", hash = "sha256:a69b0c5bc65ebd91720fc96459e81b969df851f3a6c9c855ee73de7004205987", size = 528074, upload-time = "2026-02-11T03:58:15.565Z" } @@ -822,7 +832,7 @@ wheels = [ [package.optional-dependencies] testing = [ - { name = "ops-scenario", version = "7.23.2", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.10'" }, + { name = "ops-scenario", version = "7.23.2", source = { registry = "https://pypi.org/simple" } }, ] [[package]] @@ -833,9 +843,9 @@ resolution-markers = [ "python_full_version >= '3.10'", ] dependencies = [ - { name = "opentelemetry-api", version = "1.42.1", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.10'" }, - { name = "pyyaml", marker = "python_full_version >= '3.10'" }, - { name = "websocket-client", version = "1.9.0", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.10'" }, + { name = "opentelemetry-api", version = "1.42.1", source = { registry = "https://pypi.org/simple" } }, + { name = "pyyaml" }, + { name = "websocket-client", version = "1.9.0", source = { registry = "https://pypi.org/simple" } }, ] sdist = { url = "https://files.pythonhosted.org/packages/42/60/ad398d889fd03b1b4f950fad54ad4d0cf7a81fde21f9866a8139c8f03684/ops-3.7.1.tar.gz", hash = "sha256:1765bf6d1cff376ea27608542e183b055c89f2c5f54bca602072bcc817195abc", size = 582424, upload-time = "2026-05-28T04:13:43.906Z" } wheels = [ @@ -844,7 +854,7 @@ wheels = [ [package.optional-dependencies] testing = [ - { name = "ops-scenario", version = "8.7.1", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.10'" }, + { name = "ops-scenario", version = "8.7.1", source = { registry = "https://pypi.org/simple" } }, ] [[package]] @@ -856,8 +866,8 @@ resolution-markers = [ "python_full_version < '3.9'", ] dependencies = [ - { name = "ops", version = "2.23.2", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.10'" }, - { name = "pyyaml", marker = "python_full_version < '3.10'" }, + { name = "ops", version = "2.23.2", source = { registry = "https://pypi.org/simple" } }, + { name = "pyyaml" }, ] sdist = { url = "https://files.pythonhosted.org/packages/ee/e7/1514b9c27b9364ec1d04791b0fb76d1a629c8cc73f90fbf6dd867457c914/ops_scenario-7.23.2.tar.gz", hash = "sha256:327dda8b3c871ccf16246ed4f8c7e093526af42093549976596cede58cbc8488", size = 70017, upload-time = "2026-02-11T03:58:16.553Z" } wheels = [ @@ -872,9 +882,9 @@ resolution-markers = [ "python_full_version >= '3.10'", ] dependencies = [ - { name = "ops", version = "3.7.1", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.10'" }, - { name = "pyyaml", marker = "python_full_version >= '3.10'" }, - { name = "typing-extensions", version = "4.15.0", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.10'" }, + { name = "ops", version = "3.7.1", source = { registry = "https://pypi.org/simple" } }, + { name = "pyyaml" }, + { name = "typing-extensions", version = "4.15.0", source = { registry = "https://pypi.org/simple" } }, ] sdist = { url = "https://files.pythonhosted.org/packages/4f/06/705b452b9145107978eb0400a1e751839ed1ec59dc152fa45a2e5fba0ea3/ops_scenario-8.7.1.tar.gz", hash = "sha256:25b24c7c612b8e089ad05f24a47c138999d1cc0a2683e0f7c74e0520c7bb6043", size = 78576, upload-time = "2026-05-28T04:13:45.292Z" } wheels = [ @@ -993,9 +1003,9 @@ resolution-markers = [ "python_full_version < '3.9'", ] dependencies = [ - { name = "annotated-types", marker = "python_full_version < '3.9'" }, - { name = "pydantic-core", version = "2.27.2", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.9'" }, - { name = "typing-extensions", version = "4.13.2", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.9'" }, + { name = "annotated-types" }, + { name = "pydantic-core", version = "2.27.2", source = { registry = "https://pypi.org/simple" } }, + { name = "typing-extensions", version = "4.13.2", source = { registry = "https://pypi.org/simple" } }, ] sdist = { url = "https://files.pythonhosted.org/packages/b7/ae/d5220c5c52b158b1de7ca89fc5edb72f304a70a4c540c84c8844bf4008de/pydantic-2.10.6.tar.gz", hash = "sha256:ca5daa827cce33de7a42be142548b0096bf05a7e7b365aebfa5f8eeec7128236", size = 761681, upload-time = "2025-01-24T01:42:12.693Z" } wheels = [ @@ -1011,10 +1021,10 @@ resolution-markers = [ "python_full_version == '3.9.*'", ] dependencies = [ - { name = "annotated-types", marker = "python_full_version >= '3.9'" }, - { name = "pydantic-core", version = "2.46.4", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.9'" }, - { name = "typing-extensions", version = "4.15.0", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.9'" }, - { name = "typing-inspection", marker = "python_full_version >= '3.9'" }, + { name = "annotated-types" }, + { name = "pydantic-core", version = "2.46.4", source = { registry = "https://pypi.org/simple" } }, + { name = "typing-extensions", version = "4.15.0", source = { registry = "https://pypi.org/simple" } }, + { name = "typing-inspection" }, ] sdist = { url = "https://files.pythonhosted.org/packages/18/a5/b60d21ac674192f8ab0ba4e9fd860690f9b4a6e51ca5df118733b487d8d6/pydantic-2.13.4.tar.gz", hash = "sha256:c40756b57adaa8b1efeeced5c196f3f3b7c435f90e84ea7f443901bec8099ef6", size = 844775, upload-time = "2026-05-06T13:43:05.343Z" } wheels = [ @@ -1029,7 +1039,7 @@ resolution-markers = [ "python_full_version < '3.9'", ] dependencies = [ - { name = "typing-extensions", version = "4.13.2", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.9'" }, + { name = "typing-extensions", version = "4.13.2", source = { registry = "https://pypi.org/simple" } }, ] sdist = { url = "https://files.pythonhosted.org/packages/fc/01/f3e5ac5e7c25833db5eb555f7b7ab24cd6f8c322d3a3ad2d67a952dc0abc/pydantic_core-2.27.2.tar.gz", hash = "sha256:eb026e5a4c1fee05726072337ff51d1efb6f59090b7da90d30ea58625b1ffb39", size = 413443, upload-time = "2024-12-18T11:31:54.917Z" } wheels = [ @@ -1143,7 +1153,7 @@ resolution-markers = [ "python_full_version == '3.9.*'", ] dependencies = [ - { name = "typing-extensions", version = "4.15.0", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.9'" }, + { name = "typing-extensions", version = "4.15.0", source = { registry = "https://pypi.org/simple" } }, ] sdist = { url = "https://files.pythonhosted.org/packages/9d/56/921726b776ace8d8f5db44c4ef961006580d91dc52b803c489fafd1aa249/pydantic_core-2.46.4.tar.gz", hash = "sha256:62f875393d7f270851f20523dd2e29f082bcc82292d66db2b64ea71f64b6e1c1", size = 471464, upload-time = "2026-05-06T13:37:06.98Z" } wheels = [ @@ -1299,12 +1309,12 @@ resolution-markers = [ "python_full_version < '3.9'", ] dependencies = [ - { name = "colorama", marker = "python_full_version < '3.9' and sys_platform == 'win32'" }, - { name = "exceptiongroup", marker = "python_full_version < '3.9'" }, - { name = "iniconfig", version = "2.1.0", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.9'" }, - { name = "packaging", marker = "python_full_version < '3.9'" }, - { name = "pluggy", version = "1.5.0", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.9'" }, - { name = "tomli", marker = "python_full_version < '3.9'" }, + { name = "colorama", marker = "sys_platform == 'win32'" }, + { name = "exceptiongroup" }, + { name = "iniconfig", version = "2.1.0", source = { registry = "https://pypi.org/simple" } }, + { name = "packaging" }, + { name = "pluggy", version = "1.5.0", source = { registry = "https://pypi.org/simple" } }, + { name = "tomli" }, ] sdist = { url = "https://files.pythonhosted.org/packages/ae/3c/c9d525a414d506893f0cd8a8d0de7706446213181570cdbd766691164e40/pytest-8.3.5.tar.gz", hash = "sha256:f4efe70cc14e511565ac476b57c279e12a855b11f48f212af1080ef2263d3845", size = 1450891, upload-time = "2025-03-02T12:54:54.503Z" } wheels = [ @@ -1319,13 +1329,13 @@ resolution-markers = [ "python_full_version == '3.9.*'", ] dependencies = [ - { name = "colorama", marker = "python_full_version == '3.9.*' and sys_platform == 'win32'" }, - { name = "exceptiongroup", marker = "python_full_version == '3.9.*'" }, - { name = "iniconfig", version = "2.1.0", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version == '3.9.*'" }, - { name = "packaging", marker = "python_full_version == '3.9.*'" }, - { name = "pluggy", version = "1.6.0", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version == '3.9.*'" }, - { name = "pygments", marker = "python_full_version == '3.9.*'" }, - { name = "tomli", marker = "python_full_version == '3.9.*'" }, + { name = "colorama", marker = "sys_platform == 'win32'" }, + { name = "exceptiongroup" }, + { name = "iniconfig", version = "2.1.0", source = { registry = "https://pypi.org/simple" } }, + { name = "packaging" }, + { name = "pluggy", version = "1.6.0", source = { registry = "https://pypi.org/simple" } }, + { name = "pygments" }, + { name = "tomli" }, ] sdist = { url = "https://files.pythonhosted.org/packages/a3/5c/00a0e072241553e1a7496d638deababa67c5058571567b92a7eaa258397c/pytest-8.4.2.tar.gz", hash = "sha256:86c0d0b93306b961d58d62a4db4879f27fe25513d4b969df351abdddb3c30e01", size = 1519618, upload-time = "2025-09-04T14:34:22.711Z" } wheels = [ @@ -1340,13 +1350,13 @@ resolution-markers = [ "python_full_version >= '3.10'", ] dependencies = [ - { name = "colorama", marker = "python_full_version >= '3.10' and sys_platform == 'win32'" }, - { name = "exceptiongroup", marker = "python_full_version == '3.10.*'" }, - { name = "iniconfig", version = "2.3.0", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.10'" }, - { name = "packaging", marker = "python_full_version >= '3.10'" }, - { name = "pluggy", version = "1.6.0", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.10'" }, - { name = "pygments", marker = "python_full_version >= '3.10'" }, - { name = "tomli", marker = "python_full_version == '3.10.*'" }, + { name = "colorama", marker = "sys_platform == 'win32'" }, + { name = "exceptiongroup", marker = "python_full_version < '3.11'" }, + { name = "iniconfig", version = "2.3.0", source = { registry = "https://pypi.org/simple" } }, + { name = "packaging" }, + { name = "pluggy", version = "1.6.0", source = { registry = "https://pypi.org/simple" } }, + { name = "pygments" }, + { name = "tomli", marker = "python_full_version < '3.11'" }, ] sdist = { url = "https://files.pythonhosted.org/packages/84/0e/b5858858d74958632c49b72cb25a3976ff9f632397626715be71c89d3971/pytest-9.1.0.tar.gz", hash = "sha256:41dd9148c08072446394cefd3d79701701335a9f4cae69ba92e39f6c7f5c061c", size = 1634181, upload-time = "2026-06-13T18:52:45.983Z" } wheels = [ @@ -1361,8 +1371,8 @@ resolution-markers = [ "python_full_version < '3.9'", ] dependencies = [ - { name = "coverage", version = "7.6.1", source = { registry = "https://pypi.org/simple" }, extra = ["toml"], marker = "python_full_version < '3.9'" }, - { name = "pytest", version = "8.3.5", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.9'" }, + { name = "coverage", version = "7.6.1", source = { registry = "https://pypi.org/simple" }, extra = ["toml"] }, + { name = "pytest", version = "8.3.5", source = { registry = "https://pypi.org/simple" } }, ] sdist = { url = "https://files.pythonhosted.org/packages/74/67/00efc8d11b630c56f15f4ad9c7f9223f1e5ec275aaae3fa9118c6a223ad2/pytest-cov-5.0.0.tar.gz", hash = "sha256:5837b58e9f6ebd335b0f8060eecce69b662415b16dc503883a02f45dfeb14857", size = 63042, upload-time = "2024-03-24T20:16:34.856Z" } wheels = [ @@ -1378,10 +1388,10 @@ resolution-markers = [ "python_full_version == '3.9.*'", ] dependencies = [ - { name = "coverage", version = "7.10.7", source = { registry = "https://pypi.org/simple" }, extra = ["toml"], marker = "python_full_version == '3.9.*'" }, + { name = "coverage", version = "7.10.7", source = { registry = "https://pypi.org/simple" }, extra = ["toml"], marker = "python_full_version < '3.10'" }, { name = "coverage", version = "7.14.1", source = { registry = "https://pypi.org/simple" }, extra = ["toml"], marker = "python_full_version >= '3.10'" }, - { name = "pluggy", version = "1.6.0", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.9'" }, - { name = "pytest", version = "8.4.2", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version == '3.9.*'" }, + { name = "pluggy", version = "1.6.0", source = { registry = "https://pypi.org/simple" } }, + { name = "pytest", version = "8.4.2", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.10'" }, { name = "pytest", version = "9.1.0", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.10'" }, ] sdist = { url = "https://files.pythonhosted.org/packages/b1/51/a849f96e117386044471c8ec2bd6cfebacda285da9525c9106aeb28da671/pytest_cov-7.1.0.tar.gz", hash = "sha256:30674f2b5f6351aa09702a9c8c364f6a01c27aae0c1366ae8016160d1efc56b2", size = 55592, upload-time = "2026-03-21T20:11:16.284Z" } @@ -1697,7 +1707,7 @@ name = "typing-inspection" version = "0.4.2" source = { registry = "https://pypi.org/simple" } dependencies = [ - { name = "typing-extensions", version = "4.15.0", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.9'" }, + { name = "typing-extensions", version = "4.15.0", source = { registry = "https://pypi.org/simple" } }, ] sdist = { url = "https://files.pythonhosted.org/packages/55/e3/70399cb7dd41c10ac53367ae42139cf4b1ca5f36bb3dc6c9d33acdb43655/typing_inspection-0.4.2.tar.gz", hash = "sha256:ba561c48a67c5958007083d386c3295464928b01faa735ab8547c5692e87f464", size = 75949, upload-time = "2025-10-01T02:14:41.687Z" } wheels = [ From 38dfe0c06c8c120a0aec0eae6852f6ca768c3d30 Mon Sep 17 00:00:00 2001 From: Sina Date: Tue, 25 Aug 2026 16:22:44 -0400 Subject: [PATCH 5/6] fix: lint --- tests/test_rules_customization.py | 1 + 1 file changed, 1 insertion(+) diff --git a/tests/test_rules_customization.py b/tests/test_rules_customization.py index b928dbd7..c6e5d755 100644 --- a/tests/test_rules_customization.py +++ b/tests/test_rules_customization.py @@ -446,6 +446,7 @@ def test_patch_matches_on_labels(self): # The alert in the other app is unaffected. self.assertEqual(_find_rule(result, "app-2", "group_c", "OtherAlert")["expr"], "x > 0") + class TestApplySemantics(unittest.TestCase): def test_input_is_not_mutated(self): config = """ From 02ef0b6c5e844f2163773386b2ba32e15765bf7c Mon Sep 17 00:00:00 2001 From: Sina P <55766091+sinapah@users.noreply.github.com> Date: Mon, 31 Aug 2026 11:11:41 -0400 Subject: [PATCH 6/6] feat: alert knob using pydantic (#207) * feat: alert knob using pydantic * fix: comments --- src/cosl/rules_customization.py | 223 ++++++++++++++---------------- tests/test_rules_customization.py | 18 ++- 2 files changed, 117 insertions(+), 124 deletions(-) diff --git a/src/cosl/rules_customization.py b/src/cosl/rules_customization.py index 80dd17a7..f6a8c2ee 100644 --- a/src/cosl/rules_customization.py +++ b/src/cosl/rules_customization.py @@ -34,129 +34,114 @@ from typing import Any, Dict, List, Mapping, Optional, cast import yaml +from pydantic import ( + BaseModel, + ConfigDict, + Field, + ValidationError, + field_validator, + model_validator, +) from .types import OfficialRuleFileFormat logger = logging.getLogger(__name__) -_VALID_TOP_LEVEL_KEYS = frozenset({"remove", "patch"}) -_VALID_WHERE_KEYS = frozenset({"alert", "group", "labels", "annotations"}) -_VALID_SET_KEYS = frozenset({"alert", "expr", "for", "labels", "annotations"}) +# --------------------------------------------------------------------------- +# Pydantic models for config validation +# --------------------------------------------------------------------------- -class AlertRulesCustomizationError(Exception): - """Raised when the alert rules customization configuration is invalid.""" +class _WhereBlock(BaseModel): + alert: Optional[str] = None + group: Optional[str] = None + labels: Optional[Dict[str, str]] = None + annotations: Optional[Dict[str, str]] = None -def _validate_where(where: Any, context: str) -> Dict[str, Any]: - """Validate a ``where`` selector block and return it as a plain dict. + model_config = ConfigDict(extra="forbid") - Raises: - AlertRulesCustomizationError: if the where block is missing, not a mapping, - empty, contains unknown keys, or has wrongly-typed values. - """ - if not isinstance(where, collections.abc.Mapping): - raise AlertRulesCustomizationError(f"{context}: 'where' must be a mapping") - validated_where: Dict[str, Any] = dict(cast(Mapping[Any, Any], where)) - if not validated_where: - raise AlertRulesCustomizationError(f"{context}: 'where' must not be empty") - unknown_keys = set(validated_where.keys()) - _VALID_WHERE_KEYS - if unknown_keys: - raise AlertRulesCustomizationError( - f"{context}: unknown 'where' keys {sorted(unknown_keys)}; " - f"expected a subset of {sorted(_VALID_WHERE_KEYS)}" - ) - for key in ("alert", "group"): - if key in validated_where and not isinstance(validated_where[key], str): - raise AlertRulesCustomizationError(f"{context}: 'where.{key}' must be a string") - for key in ("labels", "annotations"): - if key in validated_where and not isinstance( - validated_where[key], collections.abc.Mapping + @field_validator("labels", "annotations") + @classmethod + def _validate_mapping(cls, v: Any) -> Optional[Dict[str, str]]: # type: ignore[return] + if not isinstance(v, collections.abc.Mapping): + raise ValueError("must be a mapping of key-value pairs") + return v # type: ignore[return-value] + + @model_validator(mode="after") + def _not_empty(self): + if not any([self.alert, self.group, self.labels, self.annotations]): + valid_keys = sorted(_WhereBlock.model_fields.keys()) + raise ValueError(f"'where' must have at least one of: {valid_keys}") + return self + + +class _SetBlock(BaseModel): + alert: Optional[str] = None + expr: Optional[str] = None + for_: Optional[str] = Field(default=None, alias="for") + labels: Optional[Dict[str, str]] = None + annotations: Optional[Dict[str, str]] = None + + model_config = ConfigDict(extra="forbid", populate_by_name=True) + + @field_validator("labels", "annotations") + @classmethod + def _validate_mapping(cls, v: Any) -> Optional[Dict[str, str]]: # type: ignore[return] + if not isinstance(v, collections.abc.Mapping): + raise ValueError("must be a mapping of key-value pairs") + return v # type: ignore[return-value] + + @model_validator(mode="after") + def _not_empty(self): + if not any( + [ + self.alert, + self.expr, + self.for_, + self.labels, + self.annotations, + ] ): - raise AlertRulesCustomizationError( - f"{context}: 'where.{key}' must be a mapping of key-value pairs" + valid_keys = sorted( + f.alias if f.alias else k for k, f in _SetBlock.model_fields.items() ) - return validated_where + raise ValueError(f"'set' must have at least one of: {valid_keys}") + return self -def _validate_set(set_block: Any, context: str) -> Dict[str, Any]: - """Validate a patch ``set`` block and return it as a plain dict. +class _RemoveOperation(BaseModel): + where: _WhereBlock - Raises: - AlertRulesCustomizationError: if the set block is missing, not a mapping, - contains unknown keys, or has wrongly-typed values. - """ - if not isinstance(set_block, collections.abc.Mapping): - raise AlertRulesCustomizationError(f"{context}: 'set' must be a mapping") - validated_set: Dict[str, Any] = dict(cast(Mapping[Any, Any], set_block)) - unknown_keys = set(validated_set.keys()) - _VALID_SET_KEYS - if unknown_keys: - raise AlertRulesCustomizationError( - f"{context}: unknown 'set' keys {sorted(unknown_keys)}; " - f"expected a subset of {sorted(_VALID_SET_KEYS)}" - ) - for key in ("alert", "expr", "for"): - if key in validated_set and not isinstance(validated_set[key], str): - raise AlertRulesCustomizationError(f"{context}: 'set.{key}' must be a string") - for key in ("labels", "annotations"): - if key in validated_set and not isinstance(validated_set[key], collections.abc.Mapping): - raise AlertRulesCustomizationError( - f"{context}: 'set.{key}' must be a mapping of key-value pairs" - ) - return validated_set + model_config = ConfigDict(extra="forbid") -def _validate_remove(entries: Any) -> List[Dict[str, Any]]: - """Validate the ``remove`` operation list and return it. +class _PatchOperation(BaseModel): + where: _WhereBlock + set: _SetBlock + + model_config = ConfigDict(extra="forbid") + + +class _RulesCustomizationConfig(BaseModel): + remove: Optional[List[_RemoveOperation]] = None + patch: Optional[List[_PatchOperation]] = None + + model_config = ConfigDict(extra="forbid") - Raises: - AlertRulesCustomizationError: if the operation list is malformed. - """ - if entries is None: - return [] - if not isinstance(entries, list): - raise AlertRulesCustomizationError("'remove' must be a list of operations") - validated: List[Dict[str, Any]] = [] - for index, item in enumerate(cast(List[Any], entries)): - context = f"remove[{index}]" - if not isinstance(item, collections.abc.Mapping): - raise AlertRulesCustomizationError(f"{context} must be a mapping with a 'where' key") - entry: Dict[str, Any] = dict(cast(Mapping[Any, Any], item)) - if "where" not in entry: - raise AlertRulesCustomizationError(f"{context}: missing required key 'where'") - validated.append({"where": _validate_where(entry["where"], context)}) - return validated - - -def _validate_patch(entries: Any) -> List[Dict[str, Any]]: - """Validate the ``patch`` operation list and return it. - - Raises: - AlertRulesCustomizationError: if the operation list is malformed. - """ - if entries is None: - return [] - if not isinstance(entries, list): - raise AlertRulesCustomizationError("'patch' must be a list of operations") - validated: List[Dict[str, Any]] = [] - for index, item in enumerate(cast(List[Any], entries)): - context = f"patch[{index}]" - if not isinstance(item, collections.abc.Mapping): - raise AlertRulesCustomizationError( - f"{context} must be a mapping with 'where' and 'set' keys" - ) - entry: Dict[str, Any] = dict(cast(Mapping[Any, Any], item)) - if "where" not in entry: - raise AlertRulesCustomizationError(f"{context}: missing required key 'where'") - if "set" not in entry: - raise AlertRulesCustomizationError(f"{context}: missing required key 'set'") - validated.append( - { - "where": _validate_where(entry["where"], context), - "set": _validate_set(entry["set"], context), - } - ) - return validated + +class AlertRulesCustomizationError(Exception): + """Raised when the alert rules customization configuration is invalid.""" + + +def _format_pydantic_error(err: ValidationError) -> str: + """Format a pydantic ValidationError into a single human-readable message.""" + messages: List[str] = [] + for error in err.errors(): + loc = ".".join(str(p) for p in error["loc"]) if error["loc"] else "root" + msg: str = str(error["msg"]).rstrip(".") + messages.append(f"{loc}: {msg}") + return "; ".join(messages) class AlertRulesCustomization: @@ -197,7 +182,6 @@ def from_yaml(cls, config_string: str) -> "AlertRulesCustomization": empty ``where`` selectors. """ if not config_string or not config_string.strip(): - # Empty or whitespace-only config: no-op. return cls() try: @@ -206,26 +190,31 @@ def from_yaml(cls, config_string: str) -> "AlertRulesCustomization": raise AlertRulesCustomizationError(f"invalid YAML: {e}") from e if parsed is None: - # Config parsing to null: no-op. return cls() if not isinstance(parsed, collections.abc.Mapping): + valid_keys = sorted(_RulesCustomizationConfig.model_fields.keys()) raise AlertRulesCustomizationError( - f"configuration must be a mapping with keys {sorted(_VALID_TOP_LEVEL_KEYS)}; " + f"configuration must be a mapping with keys {valid_keys}; " f"got {type(parsed).__name__}" ) - config: Dict[str, Any] = dict(cast(Mapping[Any, Any], parsed)) - unknown_keys = set(config.keys()) - _VALID_TOP_LEVEL_KEYS - if unknown_keys: - raise AlertRulesCustomizationError( - f"unknown top-level keys {sorted(unknown_keys)}; " - f"expected a subset of {sorted(_VALID_TOP_LEVEL_KEYS)}" - ) + try: + config = _RulesCustomizationConfig.model_validate(parsed) + except ValidationError as e: + raise AlertRulesCustomizationError(_format_pydantic_error(e)) from e return cls( - remove=_validate_remove(config.get("remove")), - patch=_validate_patch(config.get("patch")), + remove=( + [op.model_dump(exclude_none=True, by_alias=True) for op in config.remove] + if config.remove + else None + ), + patch=( + [op.model_dump(exclude_none=True, by_alias=True) for op in config.patch] + if config.patch + else None + ), ) def apply( diff --git a/tests/test_rules_customization.py b/tests/test_rules_customization.py index c6e5d755..f271f5c9 100644 --- a/tests/test_rules_customization.py +++ b/tests/test_rules_customization.py @@ -86,7 +86,7 @@ def test_unknown_top_level_key_raises(self): - where: alert: Foo """ - with self.assertRaisesRegex(AlertRulesCustomizationError, "top-level"): + with self.assertRaises(AlertRulesCustomizationError): AlertRulesCustomization.from_yaml(config) def test_remove_missing_where_raises(self): @@ -94,11 +94,13 @@ def test_remove_missing_where_raises(self): AlertRulesCustomization.from_yaml("remove:\n - alert: Foo") def test_remove_empty_where_raises(self): - with self.assertRaisesRegex(AlertRulesCustomizationError, "'where' must not be empty"): + with self.assertRaisesRegex( + AlertRulesCustomizationError, "'where' must have at least one of" + ): AlertRulesCustomization.from_yaml("remove:\n - where: {}") def test_remove_unknown_where_key_raises(self): - with self.assertRaisesRegex(AlertRulesCustomizationError, "'where' keys"): + with self.assertRaises(AlertRulesCustomizationError): AlertRulesCustomization.from_yaml("remove:\n - where:\n expr: up < 1") def test_patch_missing_where_raises(self): @@ -110,24 +112,26 @@ def test_patch_missing_set_raises(self): AlertRulesCustomization.from_yaml("patch:\n - where:\n alert: Foo") def test_patch_empty_where_raises(self): - with self.assertRaisesRegex(AlertRulesCustomizationError, "'where' must not be empty"): + with self.assertRaisesRegex( + AlertRulesCustomizationError, "'where' must have at least one of" + ): AlertRulesCustomization.from_yaml("patch:\n - where: {}\n set:\n for: 5m") def test_patch_unknown_where_key_raises(self): - with self.assertRaisesRegex(AlertRulesCustomizationError, "'where' keys"): + with self.assertRaises(AlertRulesCustomizationError): AlertRulesCustomization.from_yaml( "patch:\n - where:\n record: some:record\n set:\n expr: up" ) def test_patch_unknown_set_key_raises(self): - with self.assertRaisesRegex(AlertRulesCustomizationError, "'set' keys"): + with self.assertRaises(AlertRulesCustomizationError): AlertRulesCustomization.from_yaml( "patch:\n - where:\n alert: Foo\n set:\n duration: 5m" ) def test_operations_not_a_list_raises(self): for key in ("remove", "patch"): - with self.assertRaisesRegex(AlertRulesCustomizationError, f"'{key}'"): + with self.assertRaises(AlertRulesCustomizationError): AlertRulesCustomization.from_yaml(f"{key}: not-a-list") def test_malformed_operation_entries_raise(self):