package ntfy import ( "context" "io" "net/http" "net/http/httptest" "sync/atomic" "testing" "time" "lazyntfy/internal/config" ) func TestPrimeAvoidsReplayingRestoredHistory(t *testing.T) { gotSince := make(chan string, 1) srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { gotSince <- r.URL.Query().Get("since") w.WriteHeader(http.StatusOK) if f, ok := w.(http.Flusher); ok { f.Flush() } <-r.Context().Done() })) defer srv.Close() mgr := NewManager(NewClient()) mgr.Prime(map[string]int64{srv.URL + "/alerts": 1234}) ctx, cancel := context.WithCancel(context.Background()) defer cancel() mgr.Start(ctx, []config.Sub{{ Origin: srv.URL, Name: "alerts", Auth: config.Auth{Token: "tk_test"}, }}, "all") select { case since := <-gotSince: if since != "1234" { t.Fatalf("first since %q", since) } case <-time.After(3 * time.Second): t.Fatal("timed out waiting for the first request") } } func TestManagerReconnectsWithCursor(t *testing.T) { var hits atomic.Int32 secondSince := make(chan string, 1) srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { n := hits.Add(1) if r.Header.Get("Authorization") != "Bearer tk_test" { t.Errorf("missing auth") } if n == 1 { if r.URL.Query().Get("since") != "all" { t.Errorf("first since %q", r.URL.Query().Get("since")) } _, _ = io.WriteString(w, "{\"id\":\"m1\",\"time\":100,\"event\":\"message\",\"topic\":\"alerts\",\"message\":\"hi\"}\n") return } secondSince <- r.URL.Query().Get("since") w.WriteHeader(http.StatusOK) if f, ok := w.(http.Flusher); ok { f.Flush() } <-r.Context().Done() })) defer srv.Close() mgr := NewManager(NewClient()) mgr.SetBackoff(5*time.Millisecond, 20*time.Millisecond) ctx, cancel := context.WithCancel(context.Background()) defer cancel() mgr.Start(ctx, []config.Sub{{ Origin: srv.URL, Name: "alerts", Auth: config.Auth{Token: "tk_test"}, }}, "all") deadline := time.After(3 * time.Second) sawMessage := false for !sawMessage { select { case n := <-mgr.Notices(): if n.Event != nil && n.Event.ID == "m1" { sawMessage = true } case <-deadline: t.Fatal("timed out waiting for the first message") } } select { case since := <-secondSince: if since != "100" { t.Fatalf("reconnect since %q", since) } case <-time.After(3 * time.Second): t.Fatalf("timed out waiting for reconnect, hits=%d", hits.Load()) } }