package main import ( "context" "io" "net/http" "net/http/httptest" "strings" "sync/atomic" "testing" "time" ) func TestOpenOneAndBatchCarryAuthenticationAndCursor(t *testing.T) { var requests atomic.Int64 server := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) { requests.Add(1) if request.Header.Get("Authorization") != "Bearer token-1" || request.Header.Get("Accept") != "text/event-stream" { t.Errorf("headers=%v", request.Header) } if request.URL.Query().Get("resume") == "yes" && request.Header.Get("Last-Event-ID") != "41" { t.Errorf("Last-Event-ID=%q", request.Header.Get("Last-Event-ID")) } writer.Header().Set("Content-Type", "text/event-stream; charset=utf-8") _, _ = io.WriteString(writer, "id: 42\nevent: heartbeat\ndata: {}\n\n") })) defer server.Close() item, err := openOne(context.Background(), server.Client(), server.URL+"?resume=yes", "token-1", 41) if err != nil || item.cursor != 42 || item.body == nil { t.Fatalf("item=%+v err=%v", item, err) } closeAll([]connection{item}) items, err := openBatch(context.Background(), server.Client(), server.URL, "token-1", 2, 1000, nil) if err != nil || len(items) != 2 || requests.Load() != 3 { t.Fatalf("items=%d requests=%d err=%v", len(items), requests.Load(), err) } if err := waitHealthy(items, time.Millisecond); err != nil { t.Fatal(err) } closeAll(items) } func TestSSEValidationFailures(t *testing.T) { server := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, _ *http.Request) { writer.WriteHeader(http.StatusUnauthorized) _, _ = io.WriteString(writer, "denied") })) defer server.Close() if _, err := openOne(context.Background(), server.Client(), server.URL, "bad", 0); err == nil || !strings.Contains(err.Error(), "unexpected response 401") { t.Fatalf("unexpected response error=%v", err) } if err := waitHealthy(nil, 0); err == nil { t.Fatal("empty connection set accepted") } if err := waitHealthy([]connection{{}}, 0); err == nil || !strings.Contains(err.Error(), "not healthy") { t.Fatalf("unhealthy error=%v", err) } ctx, cancel := context.WithCancel(context.Background()) cancel() if _, err := openBatch(ctx, server.Client(), server.URL, "token", 2, 1, nil); err == nil { t.Fatal("cancelled batch unexpectedly succeeded") } }