20209820564

Committed 14 Dec 2025 03:03PM UTC coverage: 86.235% (-0.04%) from 86.275%

Build # 20209820564

Build Type

Pull #1139

github

Committed by

web-flow

Commit Message

Merge 175ab0b96 into 2f0371f90

Pull Request Pull Request #1139: chore(deps): bump SwissDataScienceCenter/renku-actions from 1.19.1 to 1.20.0

Run Details

23901 of 27716 relevant lines covered (86.24%)

1.51 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

30.95

/components/renku_data_services/notifications/core.py

"""Business logic for notifications."""

import logging

from renku_data_services.notifications import apispec, models

logger = logging.getLogger(__name__)


def validate_unsaved_alert(alert: apispec.AlertPost) -> models.UnsavedAlert:
    """Validate the creation of a new alert."""
    return models.UnsavedAlert(
        title=alert.title,
        message=alert.message,
        event_type=alert.event_type,
        user_id=alert.user_id,
        session_name=alert.session_name,
    )


def validate_alert_patch(patch: apispec.AlertPatch) -> models.AlertPatch:
    """Validate the patch for an update."""
    return models.AlertPatch(
        resolved=patch.resolved,
    )


def transform_alert_to_unsaved_alert(alert: apispec.AlertmanagerAlert) -> models.UnsavedAlert | None:
    """Transform a single Alertmanager alert to an UnsavedAlert."""
    labels = alert.labels
    annotations = alert.annotations

    user_id = labels.get("safe_username")
    if not user_id:
        logger.warning("Alert is missing 'safe_username' label, skipping: %s", alert)
        return None

    session_name = labels.get("statefulset")
    if not session_name:
        # We discard alerts without a session for now
        return None

    title = annotations.get("title")
    if not title:
        logger.warning("Alert is missing 'title' annotation, skipping: %s", alert)
        return None

    message = annotations.get("description")
    if not message:
        logger.warning("Alert is missing 'description' annotation, skipping: %s", alert)
        return None

    event_type = labels.get("alertname")
    if not event_type:
        logger.warning("Alert is missing 'alertname' label, skipping: %s", alert)
        return None

    return models.UnsavedAlert(
        title=title,
        message=message,
        event_type=event_type,
        user_id=user_id,
        session_name=session_name,
    )


def alertmanager_webhook_to_unsaved_alerts(
    webhook: apispec.AlertmanagerWebhook,
) -> tuple[list[models.UnsavedAlert], list[models.UnsavedAlert]]:
    """Transform Alertmanager webhook payload to a tuple of firing and resolved UnsavedAlerts."""

    firing_alerts: list[models.UnsavedAlert] = []
    resolved_alerts: list[models.UnsavedAlert] = []

    for alert in webhook.alerts:
        unsaved_alert = transform_alert_to_unsaved_alert(alert)

        if unsaved_alert is None:
            continue

        if alert.status == apispec.Status.firing:
            firing_alerts.append(unsaved_alert)
        elif alert.status == apispec.Status.resolved:
            resolved_alerts.append(unsaved_alert)

    return firing_alerts, resolved_alerts

1	"""Business logic for notifications."""
2
3	import logging	2✔
4
5	from renku_data_services.notifications import apispec, models	2✔
6
7	logger = logging.getLogger(__name__)	2✔
8
9
10	def validate_unsaved_alert(alert: apispec.AlertPost) -> models.UnsavedAlert:	2✔
11	"""Validate the creation of a new alert."""
12	return models.UnsavedAlert(	1✔
13	title=alert.title,
14	message=alert.message,
15	event_type=alert.event_type,
16	user_id=alert.user_id,
17	session_name=alert.session_name,
18	)
19
20
21	def validate_alert_patch(patch: apispec.AlertPatch) -> models.AlertPatch:	2✔
22	"""Validate the patch for an update."""
23	return models.AlertPatch(	1✔
24	resolved=patch.resolved,
25	)
26
27
28	def transform_alert_to_unsaved_alert(alert: apispec.AlertmanagerAlert) -> models.UnsavedAlert \| None:	2✔
29	"""Transform a single Alertmanager alert to an UnsavedAlert."""
30	labels = alert.labels	×
31	annotations = alert.annotations	×
32
33	user_id = labels.get("safe_username")	×
34	if not user_id:	×
35	logger.warning("Alert is missing 'safe_username' label, skipping: %s", alert)	×
36	return None	×
37
38	session_name = labels.get("statefulset")	×
39	if not session_name:	×
40	# We discard alerts without a session for now
41	return None	×
42
43	title = annotations.get("title")	×
44	if not title:	×
45	logger.warning("Alert is missing 'title' annotation, skipping: %s", alert)	×
46	return None	×
47
48	message = annotations.get("description")	×
49	if not message:	×
50	logger.warning("Alert is missing 'description' annotation, skipping: %s", alert)	×
51	return None	×
52
53	event_type = labels.get("alertname")	×
54	if not event_type:	×
55	logger.warning("Alert is missing 'alertname' label, skipping: %s", alert)	×
56	return None	×
57
58	return models.UnsavedAlert(	×
59	title=title,
60	message=message,
61	event_type=event_type,
62	user_id=user_id,
63	session_name=session_name,
64	)
65
66
67	def alertmanager_webhook_to_unsaved_alerts(	2✔
68	webhook: apispec.AlertmanagerWebhook,
69	) -> tuple[list[models.UnsavedAlert], list[models.UnsavedAlert]]:
70	"""Transform Alertmanager webhook payload to a tuple of firing and resolved UnsavedAlerts."""
71
72	firing_alerts: list[models.UnsavedAlert] = []	1✔
73	resolved_alerts: list[models.UnsavedAlert] = []	1✔
74
75	for alert in webhook.alerts:	1✔
76	unsaved_alert = transform_alert_to_unsaved_alert(alert)	×
77
78	if unsaved_alert is None:	×
79	continue	×
80
81	if alert.status == apispec.Status.firing:	×
82	firing_alerts.append(unsaved_alert)	×
83	elif alert.status == apispec.Status.resolved:	×
84	resolved_alerts.append(unsaved_alert)	×
85
86	return firing_alerts, resolved_alerts	1✔

SwissDataScienceCenter / renku-data-services / 20209820564

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous