package notify import ( "context" "net/http" "net/http/httptest" "net/url" "strings" "sync" "sync/atomic" "testing" "time" ) const tmpl = "https://admin.example/api/limits?level=$LEVEL&id=$ID&flow=$FLOW&period=$PERIOD&by=$ADDRESS&count=$COUNT&limit=$LIMIT" func TestBuildURL(t *testing.T) { ev := Event{Level: "A", ID: "me@example.com", Flow: "o", Address: "me@example.com", Count: 100, Limit: 100} got := BuildURL(tmpl, ev) want := "https://admin.example/api/limits?level=A&id=me%40example.com&flow=o&period=d&by=me%40example.com&count=100&limit=100" if got != want { t.Errorf("BuildURL =\n %s\nwant\n %s", got, want) } // Dangerous characters in values are escaped, templates without placeholders pass through. ev.ID = "a b&c=d@x.y" u, err := url.Parse(BuildURL(tmpl, ev)) if err != nil { t.Fatal(err) } if u.Query().Get("id") != "a b&c=d@x.y" { t.Errorf("id round-trip = %q", u.Query().Get("id")) } if got := BuildURL("https://x/static", ev); got != "https://x/static" { t.Errorf("static template changed: %s", got) } } func TestURLFor(t *testing.T) { cfg := Config{OrgURL: "o", AddressURL: "a", DomainURL: "d"} for level, want := range map[string]string{"O": "o", "A": "a", "D": "d", "X": ""} { if got := cfg.URLFor(level); got != want { t.Errorf("URLFor(%s) = %q, want %q", level, got, want) } } } // capture is an httptest handler recording requests. type capture struct { mu sync.Mutex requests []*http.Request status []int // status to answer per call; last entry repeats calls atomic.Int32 delay time.Duration inflight atomic.Int32 maxSeen atomic.Int32 } func (c *capture) ServeHTTP(w http.ResponseWriter, r *http.Request) { n := c.inflight.Add(1) for { m := c.maxSeen.Load() if n <= m || c.maxSeen.CompareAndSwap(m, n) { break } } defer c.inflight.Add(-1) if c.delay > 0 { time.Sleep(c.delay) } i := int(c.calls.Add(1)) - 1 c.mu.Lock() c.requests = append(c.requests, r.Clone(context.Background())) c.mu.Unlock() status := http.StatusOK if len(c.status) > 0 { if i < len(c.status) { status = c.status[i] } else { status = c.status[len(c.status)-1] } } w.WriteHeader(status) } type hooks struct { mu sync.Mutex sent []sentCall failed []failedCall done chan struct{} } type sentCall struct { ev Event url string attempt int } type failedCall struct { ev Event url string attempts int err error } func newHooks() *hooks { return &hooks{done: make(chan struct{}, 64)} } func (h *hooks) Hooks() Hooks { return Hooks{ Sent: func(ev Event, url string, attempt int) { h.mu.Lock() h.sent = append(h.sent, sentCall{ev, url, attempt}) h.mu.Unlock() h.done <- struct{}{} }, Failed: func(ev Event, url string, attempts int, err error) { h.mu.Lock() h.failed = append(h.failed, failedCall{ev, url, attempts, err}) h.mu.Unlock() h.done <- struct{}{} }, } } func (h *hooks) wait(t *testing.T, n int) { t.Helper() for i := 0; i < n; i++ { select { case <-h.done: case <-time.After(5 * time.Second): t.Fatalf("timeout waiting for notification outcome %d/%d", i+1, n) } } } func fastConfig(orgURL, addrURL, domURL string) Config { return Config{ OrgURL: orgURL, AddressURL: addrURL, DomainURL: domURL, User: "api", Password: "secret", Timeout: 2 * time.Second, Attempts: 5, Backoff: time.Millisecond, MaxConcurrent: 8, } } func TestNotify_SendsToLevelURLWithBasicAuth(t *testing.T) { org, addr, dom := &capture{}, &capture{}, &capture{} so, sa, sd := httptest.NewServer(org), httptest.NewServer(addr), httptest.NewServer(dom) defer so.Close() defer sa.Close() defer sd.Close() q := "/api/limits?level=$LEVEL&id=$ID&flow=$FLOW&period=$PERIOD&by=$ADDRESS&count=$COUNT" h := newHooks() n := New(fastConfig(so.URL+q, sa.URL+q, sd.URL+q), h.Hooks()) defer n.Shutdown(context.Background()) ev := Event{TraceID: "tr", Level: "A", ID: "me@example.com", Flow: "o", Address: "me@example.com", Count: 100, Limit: 100} if !n.Notify(ev) { t.Fatal("Notify returned false for a configured level") } h.wait(t, 1) if org.calls.Load() != 0 || dom.calls.Load() != 0 || addr.calls.Load() != 1 { t.Fatalf("calls org=%d addr=%d dom=%d, want only addr", org.calls.Load(), addr.calls.Load(), dom.calls.Load()) } r := addr.requests[0] if r.Method != http.MethodGet { t.Errorf("method = %s", r.Method) } if u, p, ok := r.BasicAuth(); !ok || u != "api" || p != "secret" { t.Errorf("basic auth = %q %q %v", u, p, ok) } qv := r.URL.Query() for k, want := range map[string]string{"level": "A", "id": "me@example.com", "flow": "o", "period": "d", "by": "me@example.com", "count": "100"} { if qv.Get(k) != want { t.Errorf("query %s = %q, want %q", k, qv.Get(k), want) } } if len(h.sent) != 1 || h.sent[0].attempt != 1 || h.sent[0].ev != ev || !strings.HasPrefix(h.sent[0].url, sa.URL) { t.Errorf("sent hook = %+v", h.sent) } // Org and domain levels route to their own URLs. n.Notify(Event{Level: "O", ID: "7"}) n.Notify(Event{Level: "D", ID: "example.com"}) h.wait(t, 2) if org.calls.Load() != 1 || dom.calls.Load() != 1 { t.Errorf("calls org=%d dom=%d, want 1 each", org.calls.Load(), dom.calls.Load()) } } func TestNotify_EmptyURLSkips(t *testing.T) { h := newHooks() n := New(fastConfig("", "", ""), h.Hooks()) if n.Notify(Event{Level: "A"}) || n.Notify(Event{Level: "Z"}) { t.Error("Notify should return false when no URL is configured for the level") } if err := n.Shutdown(context.Background()); err != nil { t.Fatal(err) } if len(h.sent)+len(h.failed) != 0 { t.Errorf("no hooks expected, got sent=%v failed=%v", h.sent, h.failed) } } func TestNotify_NoBasicAuthWithoutUser(t *testing.T) { c := &capture{} s := httptest.NewServer(c) defer s.Close() h := newHooks() cfg := fastConfig(s.URL+"/l?level=$LEVEL", "", "") cfg.User, cfg.Password = "", "" n := New(cfg, h.Hooks()) n.Notify(Event{Level: "O"}) h.wait(t, 1) n.Shutdown(context.Background()) if _, _, ok := c.requests[0].BasicAuth(); ok { t.Error("Authorization header sent although no user is configured") } } func TestNotify_RetriesThenSucceeds(t *testing.T) { c := &capture{status: []int{500, 503, 200}} s := httptest.NewServer(c) defer s.Close() h := newHooks() n := New(fastConfig(s.URL+"/l?level=$LEVEL", "", ""), h.Hooks()) n.Notify(Event{Level: "O", ID: "7"}) h.wait(t, 1) n.Shutdown(context.Background()) if c.calls.Load() != 3 || len(h.sent) != 1 || h.sent[0].attempt != 3 || len(h.failed) != 0 { t.Errorf("calls=%d sent=%+v failed=%+v; want success on attempt 3", c.calls.Load(), h.sent, h.failed) } } func TestNotify_GivesUpAfterAttempts(t *testing.T) { c := &capture{status: []int{500}} s := httptest.NewServer(c) defer s.Close() h := newHooks() n := New(fastConfig(s.URL+"/l?level=$LEVEL", "", ""), h.Hooks()) n.Notify(Event{Level: "O", ID: "7"}) h.wait(t, 1) n.Shutdown(context.Background()) if c.calls.Load() != 5 || len(h.failed) != 1 || h.failed[0].attempts != 5 || len(h.sent) != 0 { t.Errorf("calls=%d failed=%+v sent=%+v; want 5 attempts then failure", c.calls.Load(), h.failed, h.sent) } if !strings.Contains(h.failed[0].err.Error(), "500") { t.Errorf("failure error should carry the status: %v", h.failed[0].err) } } // 4xx means a configuration problem (bad credentials, wrong URL): retrying is pointless. func TestNotify_NoRetryOn4xx(t *testing.T) { c := &capture{status: []int{401}} s := httptest.NewServer(c) defer s.Close() h := newHooks() n := New(fastConfig(s.URL+"/l?level=$LEVEL", "", ""), h.Hooks()) n.Notify(Event{Level: "O", ID: "7"}) h.wait(t, 1) n.Shutdown(context.Background()) if c.calls.Load() != 1 || len(h.failed) != 1 || h.failed[0].attempts != 1 { t.Errorf("calls=%d failed=%+v; want a single attempt", c.calls.Load(), h.failed) } } func TestNotify_TimeoutIsRetried(t *testing.T) { c := &capture{delay: 300 * time.Millisecond} s := httptest.NewServer(c) defer s.Close() h := newHooks() cfg := fastConfig(s.URL+"/l?level=$LEVEL", "", "") cfg.Timeout = 50 * time.Millisecond cfg.Attempts = 2 n := New(cfg, h.Hooks()) n.Notify(Event{Level: "O", ID: "7"}) h.wait(t, 1) n.Shutdown(context.Background()) if len(h.failed) != 1 || h.failed[0].attempts != 2 { t.Errorf("failed=%+v; want failure after 2 timed-out attempts", h.failed) } } func TestNotify_BoundedConcurrency(t *testing.T) { c := &capture{delay: 50 * time.Millisecond} s := httptest.NewServer(c) defer s.Close() h := newHooks() cfg := fastConfig(s.URL+"/l?level=$LEVEL", "", "") cfg.MaxConcurrent = 2 n := New(cfg, h.Hooks()) for i := 0; i < 10; i++ { n.Notify(Event{Level: "O", ID: "7"}) } h.wait(t, 10) n.Shutdown(context.Background()) if c.maxSeen.Load() > 2 { t.Errorf("max concurrent requests = %d, want <= 2", c.maxSeen.Load()) } if c.calls.Load() != 10 { t.Errorf("calls = %d, want 10", c.calls.Load()) } } // Shutdown waits for in-flight notifications, bounded by the context. func TestNotify_ShutdownDrainsAndRespectsContext(t *testing.T) { c := &capture{delay: 200 * time.Millisecond} s := httptest.NewServer(c) defer s.Close() h := newHooks() n := New(fastConfig(s.URL+"/l?level=$LEVEL", "", ""), h.Hooks()) n.Notify(Event{Level: "O", ID: "7"}) ctx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond) defer cancel() if err := n.Shutdown(ctx); err == nil { t.Error("Shutdown with an expired context should return its error") } if err := n.Shutdown(context.Background()); err != nil { t.Errorf("second Shutdown: %v", err) } if c.calls.Load() != 1 || len(h.sent) != 1 { t.Errorf("in-flight notification was not completed: calls=%d sent=%d", c.calls.Load(), len(h.sent)) } if n.Notify(Event{Level: "O"}) { t.Error("Notify after Shutdown should be refused") } } func TestNew_Defaults(t *testing.T) { n := New(Config{OrgURL: "https://x/?level=$LEVEL"}, Hooks{}) defer n.Shutdown(context.Background()) if n.cfg.Attempts != 5 || n.cfg.Backoff != time.Second || n.cfg.MaxConcurrent != 8 || n.cfg.Timeout != 5*time.Second { t.Errorf("defaults not applied: %+v", n.cfg) } }