fix(alerting): validate configuration without untyped keyword expansion

This commit is contained in:
XAVIER ALMENDROS 2026-10-03 12:29:25 +02:00
parent d1304a7b36
commit 79ffcb94f1
3 changed files with 30 additions and 3 deletions

View file

@ -115,7 +115,7 @@ class SlackAlerting(CustomBatchLogger):
)
self.alert_to_webhook_url = process_slack_alerting_variables(alert_to_webhook_url=alert_to_webhook_url)
self.is_running = False
self.alerting_args = SlackAlertingArgs(**alerting_args)
self.alerting_args = SlackAlertingArgs.model_validate(alerting_args)
self.default_webhook_url = default_webhook_url
self.flush_lock = asyncio.Lock()
self.periodic_started = False
@ -155,7 +155,7 @@ class SlackAlerting(CustomBatchLogger):
if alert_types is not None:
self.alert_types = alert_types
if alerting_args is not None:
self.alerting_args = SlackAlertingArgs(**alerting_args)
self.alerting_args = SlackAlertingArgs.model_validate(alerting_args)
self._ensure_periodic_flush_task()
if alert_type_config is not None:
for key, val in alert_type_config.items():

View file

@ -3,7 +3,7 @@ import time
from collections.abc import Mapping
from datetime import datetime as dt
from enum import Enum
from typing import Annotated, Any, Final, Literal, Optional, Union
from typing import Annotated, Final
from pydantic import BaseModel, Field
from typing_extensions import NotRequired, ReadOnly, TypedDict

View file

@ -822,6 +822,33 @@ def test_slack_budget_key_alias_patterns_reject_invalid_config(patterns: object)
SlackAlertingArgs.model_validate({"slack_budget_alert_key_aliases": patterns})
@pytest.mark.asyncio
@pytest.mark.parametrize("patterns, delivered", (([], False), (["github-example-*"], True)))
async def test_slack_budget_key_alias_filter_reload_validates_and_changes_delivery(
patterns: list[str], delivered: bool
) -> None:
http_handler: Final = _webhook_accepting_posts()
slack_alerting: Final = SlackAlerting(
alerting=["slack"],
default_webhook_url=SLACK_WEBHOOK_URL,
async_http_handler=http_handler,
)
slack_alerting.update_values(alerting_args={"slack_budget_alert_key_aliases": patterns})
with pytest.raises(ValidationError, match="slack_budget_alert_key_aliases"):
slack_alerting.update_values(alerting_args={"slack_budget_alert_key_aliases": "github-example-*"})
await slack_alerting.send_alert(
message=THRESHOLD_ALERT,
level="High",
alert_type=AlertType.budget_alerts,
alerting_metadata={},
user_info=_budget_webhook_event("github-example-api"),
)
await slack_alerting.flush_queue()
assert http_handler.post.await_count == int(delivered)
if delivered:
assert THRESHOLD_ALERT in _posted_slack_bodies(http_handler)[0]["text"]
def _periodic_flush_tasks() -> list[asyncio.Task[object]]:
return [
t