diff --git a/modules/tide/centrifugo_test.go b/modules/tide/centrifugo_test.go index b517c10..ff199f6 100644 --- a/modules/tide/centrifugo_test.go +++ b/modules/tide/centrifugo_test.go @@ -234,3 +234,106 @@ func TestRecordBroadcastsStep(t *testing.T) { t.Fatal("a non-loopback target must be refused") } } + +// TestCentrifugoRecorder covers the fake Centrifugo's edges (T-11-29): +// only publish and broadcast are recorded, the authorization flag is a +// comparison and the header value is never stored, other methods get 405, +// bodies over the cap get 413 and are not recorded, an empty configured +// key never counts as authorized, and ListenAndServe serves on loopback +// until its ctx ends. +func TestCentrifugoRecorder(t *testing.T) { + rec := NewCentrifugoRecorder(CentrifugoRecorderOptions{APIKey: "rec-key"}) + srv := httptest.NewServer(rec) + t.Cleanup(srv.Close) + + for _, path := range []string{"/api/unsubscribe", "/api/info", "/api/info/"} { + resp := postJSON(t, srv.URL+path, "apikey rec-key", `{}`) + body, _ := io.ReadAll(resp.Body) + if resp.StatusCode != 200 || string(body) != `{"result":{}}` || resp.Header.Get("Content-Type") != "application/json" { + t.Fatalf("%s answer %d %s", path, resp.StatusCode, body) + } + } + get, err := http.Get(srv.URL + "/api/publish") + if err != nil { + t.Fatal(err) + } + _ = get.Body.Close() + if get.StatusCode != http.StatusMethodNotAllowed || get.Header.Get("Allow") != http.MethodPost { + t.Fatalf("GET publish = %d allow %q", get.StatusCode, get.Header.Get("Allow")) + } + big := `{"pad":"` + strings.Repeat("x", MaxPublicationBody) + `"}` + if resp := postJSON(t, srv.URL+"/api/publish", "apikey rec-key", big); resp.StatusCode != http.StatusRequestEntityTooLarge { + t.Fatalf("oversized body = %d", resp.StatusCode) + } + postJSON(t, srv.URL+"/api/publish/", "apikey rec-key", `{"channel":"a:1"}`) + postJSON(t, srv.URL+"/api/publish", "Bearer rec-key", `{"channel":"a:2"}`) + postJSON(t, srv.URL+"/api/publish", "", `{"channel":"a:3"}`) + pubs := rec.Publications() + if len(pubs) != 3 { + t.Fatalf("publications = %d, want 3 (info, unsubscribe and the oversized body are not recorded)", len(pubs)) + } + if !pubs[0].Authorization || pubs[1].Authorization || pubs[2].Authorization { + t.Fatalf("authorization flags = %v %v %v", pubs[0].Authorization, pubs[1].Authorization, pubs[2].Authorization) + } + raw, _ := json.Marshal(pubs) + if strings.Contains(string(raw), "rec-key") { + t.Fatal("a recorded publication holds the Authorization value") + } + pubs[0].Body[0] = 'X' + if rec.Publications()[0].Body[0] != '{' { + t.Fatal("Publications returned the recorder's own body slice") + } + if NewCentrifugoRecorder(CentrifugoRecorderOptions{}).authorized("apikey ") { + t.Fatal("an empty configured key authorized a request") + } + + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + addr := ln.Addr().String() + _ = ln.Close() + ctx, cancel := context.WithCancel(t.Context()) + done := make(chan error, 1) + go func() { done <- rec.ListenAndServe(ctx, addr) }() + if err := waitListening(t.Context(), addr, done); err != nil { + t.Fatal(err) + } + postJSON(t, "http://"+addr+"/api/broadcast", "apikey rec-key", `{"channels":["a:1"]}`) + cancel() + select { + case err := <-done: + if err != nil { + t.Fatalf("ListenAndServe after cancel = %v", err) + } + case <-time.After(5 * time.Second): + t.Fatal("recorder did not stop") + } + if err := rec.ListenAndServe(t.Context(), "127.0.0.1:bad"); err == nil { + t.Fatal("an unusable address was accepted") + } + stopped := make(chan error, 1) + stopped <- nil + if err := waitListening(t.Context(), "127.0.0.1:1", stopped); err == nil || !strings.Contains(err.Error(), "stopped") { + t.Fatalf("waitListening on a stopped recorder = %v", err) + } + cctx, ccancel := context.WithCancel(t.Context()) + ccancel() + if err := sleepCtx(cctx, time.Second); err == nil { + t.Fatal("sleepCtx ignored a cancelled ctx") + } +} + +// TestFlowIDNames covers the id variables a broadcast recording masks by +// default: placeholders in the path, query, body and headers, and id +// captures, sorted and without duplicates. +func TestFlowIDNames(t *testing.T) { + spec := Flow{Steps: []Step{ + {Request: Request{Path: "/api/collections/{{id:collection}}/albums", Query: "artist={{id:artist}}", Headers: map[string]string{"X-Acme": "{{id:header}}"}}, + Capture: []CaptureRule{{As: "id:album"}, {As: "jwt:alice"}}}, + {Request: Request{Path: "/api/albums/{{id:album}}", Body: Body(`{"collection_id":{{id:collection}}}`)}}, + }} + if got := strings.Join(flowIDNames(spec), ","); got != "id:album,id:artist,id:collection,id:header" { + t.Fatalf("flowIDNames = %s", got) + } +}