diff --git a/internal/notifications/ntfy_transition_test.go b/internal/notifications/ntfy_transition_test.go index 162c7230f..398df6f3e 100644 --- a/internal/notifications/ntfy_transition_test.go +++ b/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) + } + } }) } }