nexarch/internal/notifyapi/qa07_e2e_test.go

package notifyapi

import (
	"context"
	"net/http"
	"net/http/httptest"
	"sync"
	"testing"
)

// QA-07 Pruefung 3: Zustellprotokoll (notification_jobs, Core CFG-02) wird
// gegen die aus den simulierten Workflow-Prozesspfaden (Ablehnung, Eskalation,
// Fristablauf) erwarteten Ereignisse abgeglichen -- jedes erwartete Ereignis
// erscheint genau einmal, kein Duplikat, keine Luecke. Treibt denselben
// HTTP-Pfad an, den workflow/internal/escalationnotify.DispatchDue und
// Aehnliches live gegen Core CFG-05 nutzen.
func TestE2E_DeliveryLogMatchesExpectedWorkflowEvents(t *testing.T) {
	prefs, dispatcher, pool := setupTest(t)
	ctx := context.Background()

	mux := http.NewServeMux()
	Mount(mux, prefs, dispatcher, testToken)
	server := httptest.NewServer(mux)
	defer server.Close()

	tenant := "acme"
	if err := prefs.Set(ctx, tenant, "cfg05_qa07_muted", "workflow.task.escalated", "email", false); err != nil {
		t.Fatalf("praeferenz deaktivieren: %v", err)
	}

	type expectedEvent struct {
		userID    string
		eventType string
		recipient string
	}
	events := []expectedEvent{
		{userID: "cfg05_qa07_alice", eventType: "workflow.approval.rejected", recipient: "cfg05_qa07_alice@acme.example"},
		{userID: "cfg05_qa07_bob", eventType: "workflow.approval.delegated", recipient: "cfg05_qa07_bob@acme.example"},
		{userID: "cfg05_qa07_carol", eventType: "workflow.task.escalated", recipient: "cfg05_qa07_carol@acme.example"},
		{userID: "cfg05_qa07_dave", eventType: "workflow.task.reminder_due", recipient: "cfg05_qa07_dave@acme.example"},
		{userID: "cfg05_qa07_muted", eventType: "workflow.task.escalated", recipient: "cfg05_qa07_muted@acme.example"},
	}

	var wg sync.WaitGroup
	for _, ev := range events {
		wg.Add(1)
		go func(ev expectedEvent) {
			defer wg.Done()
			resp := post(t, server, testToken, enqueueRequest{
				TenantSlug: tenant, UserID: ev.userID, EventType: ev.eventType,
				Channel: "email", Recipient: ev.recipient,
				Payload: map[string]any{"task_id": "qa07-" + ev.userID},
			})
			_ = resp.Body.Close()
			if resp.StatusCode != http.StatusOK {
				t.Errorf("enqueue fuer %s/%s: status = %d", ev.userID, ev.eventType, resp.StatusCode)
			}
		}(ev)
	}
	wg.Wait()

	rows, err := pool.Query(ctx, `
		SELECT recipient, payload->>'task_id'
		FROM notification_jobs
		WHERE recipient LIKE 'cfg05\_qa07\_%' ESCAPE '\'
	`)
	if err != nil {
		t.Fatalf("zustellprotokoll lesen: %v", err)
	}
	defer rows.Close()

	seen := map[string]int{}
	for rows.Next() {
		var recipient, taskID string
		if err := rows.Scan(&recipient, &taskID); err != nil {
			t.Fatalf("zeile lesen: %v", err)
		}
		seen[recipient]++
	}
	if err := rows.Err(); err != nil {
		t.Fatalf("iterieren: %v", err)
	}

	for _, ev := range events {
		if ev.userID == "cfg05_qa07_muted" {
			continue
		}
		if seen[ev.recipient] != 1 {
			t.Fatalf("empfaenger %s: %d eintraege im zustellprotokoll, want genau 1 (kein duplikat, keine luecke)", ev.recipient, seen[ev.recipient])
		}
	}
	if n := seen["cfg05_qa07_muted@acme.example"]; n != 0 {
		t.Fatalf("deaktivierter kanal hat trotzdem %d eintraege im zustellprotokoll erzeugt, want 0", n)
	}
}