Merge candidate 20260906T000001Z-core-runtime

Change-source: pulse-maintainer

internal/notifications/ntfy_transition_test.go

@@ -1,6 +1,7 @@
 package notifications
 
 import (
+	"encoding/json"
 	"io"
 	"net/http"
 	"reflect"
@@ -14,6 +15,16 @@ import (
 // Exercise the shared destination across transitions: generated firing headers
 // must not leak into the stored configuration and override recovery metadata.
 func TestNtfySeverityRecoveryTransition(t *testing.T) {
+	for _, queued := range []bool{false, true} {
+		name := "direct"
+		if queued {
+			name = "queued"
+		}
+		t.Run(name, func(t *testing.T) { testNtfySeverityRecoveryTransition(t, queued) })
+	}
+}
+
+func testNtfySeverityRecoveryTransition(t *testing.T, queued bool) {
 	type receipt struct {
 		header http.Header
 		body   string
@@ -28,7 +39,7 @@ func TestNtfySeverityRecoveryTransition(t *testing.T) {
 		w.WriteHeader(http.StatusAccepted)
 	}))
 	defer server.Close()
-	manager := NewNotificationManager("")
+	manager := NewNotificationManagerWithDataDir("", t.TempDir())
 	defer manager.Stop()
 	manager.webhookClient = server.Client()
 	if err := manager.UpdateAllowedPrivateCIDRs("127.0.0.1/32"); err != nil {
@@ -36,6 +47,7 @@ func TestNtfySeverityRecoveryTransition(t *testing.T) {
 	}
 	webhook := WebhookConfig{Name: "transition", URL: server.URL + "/topic", Enabled: true, Service: "ntfy",
 		Headers: map[string]string{"X-Static": "preserved"}}
+	manager.AddWebhook(webhook)
 	originalHeaders := map[string]string{"X-Static": "preserved"}
 	alert := &alerts.Alert{ID: "transition", Type: "cpu", ResourceID: "vm-1", ResourceName: "database", Node: "node-a",
 		Message: "CPU above threshold", Value: 99, Threshold: 90, StartTime: time.Now().Add(-time.Minute)}
@@ -54,7 +66,25 @@ func TestNtfySeverityRecoveryTransition(t *testing.T) {
 			alert.Level = step.level
 			before := *alert
 			var err error
-			if step.resolved {
+			if queued {
+				config, marshalErr := json.Marshal(webhook)
+				if marshalErr != nil {
+					t.Fatal(marshalErr)
+				}
+				kind := "webhook"
+				payload := alert.Clone()
+				if step.resolved {
+					kind += "_resolved"
+					annotateResolvedMetadata(payload, time.Now())
+				}
+				if manager.queue == nil {
+					t.Fatal("notification queue unavailable")
+				}
+				err = manager.queue.Enqueue(&QueuedNotification{
+					ID: step.name, Type: kind, Status: QueueStatusPending,
+					Config: config, Alerts: []*alerts.Alert{payload}, MaxAttempts: 1,
+				})
+			} else if step.resolved {
 				err = manager.sendResolvedWebhook(webhook, []*alerts.Alert{alert}, time.Now())
 			} else {
 				err = manager.sendGroupedWebhook(webhook, []*alerts.Alert{alert})
@@ -86,6 +116,24 @@ func TestNtfySeverityRecoveryTransition(t *testing.T) {
 			if !reflect.DeepEqual(*alert, before) {
 				t.Error("source alert mutated")
 			}
+			if queued {
+				// Receipt precedes the queue commit. Wait for completion before
+				// advancing the same alert identity to its next lifecycle state.
+				deadline := time.Now().Add(3 * time.Second)
+				for {
+					var status string
+					if err := manager.queue.db.QueryRow("SELECT status FROM notification_queue WHERE id = ?", step.name).Scan(&status); err != nil {
+						t.Fatal(err)
+					}
+					if status == string(QueueStatusSent) {
+						break
+					}
+					if time.Now().After(deadline) {
+						t.Fatalf("received HTTP request but queue status remained %s", status)
+					}
+					time.Sleep(time.Millisecond)
+				}
+			}
 		})
 	}
 }