9ecbf74a22
- CentrifugoRecorder records publish/broadcast requests (method, path, whether the API key matched, JSON body) and binds loopback only - BroadcastGolden load/write, NormalizePublications (timestamps, actor, captured ids only) and DiffPublications (structural, key order ignored) - RecordBroadcasts runs a flow or one step against a loopback backend - summer parity:broadcasts wraps it; README documents format and rules
237 lines
8.5 KiB
Go
237 lines
8.5 KiB
Go
package tide
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"io"
|
|
"net"
|
|
"net/http"
|
|
"net/http/httptest"
|
|
"os"
|
|
"path/filepath"
|
|
"strings"
|
|
"testing"
|
|
"time"
|
|
)
|
|
|
|
func postJSON(t *testing.T, url, auth, body string) *http.Response {
|
|
t.Helper()
|
|
req, err := http.NewRequest(http.MethodPost, url, strings.NewReader(body))
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if auth != "" {
|
|
req.Header.Set("Authorization", auth)
|
|
}
|
|
resp, err := http.DefaultClient.Do(req)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
t.Cleanup(func() { _ = resp.Body.Close() })
|
|
return resp
|
|
}
|
|
|
|
func TestCentrifugoRecorderRecordsPublishAndBroadcast(t *testing.T) {
|
|
rec := NewCentrifugoRecorder(CentrifugoRecorderOptions{APIKey: "k1"})
|
|
srv := httptest.NewServer(rec)
|
|
t.Cleanup(srv.Close)
|
|
|
|
resp := postJSON(t, srv.URL+"/api/publish", "apikey k1", `{"channel":"room:1","data":{}}`)
|
|
body, _ := io.ReadAll(resp.Body)
|
|
if resp.StatusCode != 200 || string(body) != `{"result":{}}` {
|
|
t.Fatalf("publish answer %d %s", resp.StatusCode, body)
|
|
}
|
|
postJSON(t, srv.URL+"/api/broadcast", "apikey wrong", `{"channels":["a:1","b:2"]}`)
|
|
resp = postJSON(t, srv.URL+"/api/presence", "apikey k1", `{"channel":"room:1"}`)
|
|
body, _ = io.ReadAll(resp.Body)
|
|
if string(body) != `{"result":{"presence":{}}}` {
|
|
t.Fatalf("presence answer %s", body)
|
|
}
|
|
if resp := postJSON(t, srv.URL+"/api/other", "", `{}`); resp.StatusCode != 404 {
|
|
t.Fatalf("unknown path %d", resp.StatusCode)
|
|
}
|
|
|
|
pubs := rec.Publications()
|
|
if len(pubs) != 2 {
|
|
t.Fatalf("publications %d, want 2 (presence is not recorded)", len(pubs))
|
|
}
|
|
if pubs[0].Path != "/api/publish" || !pubs[0].Authorization || pubs[0].Method != "POST" {
|
|
t.Fatalf("first %+v", pubs[0])
|
|
}
|
|
if pubs[1].Path != "/api/broadcast" || pubs[1].Authorization {
|
|
t.Fatalf("second %+v", pubs[1])
|
|
}
|
|
rec.Reset()
|
|
if len(rec.Publications()) != 0 {
|
|
t.Fatal("reset kept publications")
|
|
}
|
|
}
|
|
|
|
func TestCentrifugoRecorderRefusesNonLoopback(t *testing.T) {
|
|
rec := NewCentrifugoRecorder(CentrifugoRecorderOptions{})
|
|
if err := rec.ListenAndServe(t.Context(), "0.0.0.0:0"); err == nil || !strings.Contains(err.Error(), "loopback") {
|
|
t.Fatalf("want loopback refusal, got %v", err)
|
|
}
|
|
}
|
|
|
|
func TestNormalizePublications(t *testing.T) {
|
|
store := mustMemoryStore()
|
|
store.Set("id:collection", "12")
|
|
store.Set("id:album", "40")
|
|
store.Set("jwt:alice", "not-an-id")
|
|
pubs := []Publication{{
|
|
Method: "POST",
|
|
Path: "/api/publish",
|
|
Body: json.RawMessage(`{"channel":"collection:12","data":{"event":"deleted.acme.album",` +
|
|
`"payload":{"id":40,"collection_id":12,"action":"deleted","actor":{"user_id":3,"name":null},` +
|
|
`"timestamp":"2026-09-30T10:00:00+00:00","count":12},"timestamp":"2026-09-30T10:00:01+00:00"}}`),
|
|
}}
|
|
got, err := NormalizePublications(pubs, store)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
want := `{"channel":"collection:{{id:collection}}","data":{"event":"deleted.acme.album",` +
|
|
`"payload":{"id":{{id:album}},"collection_id":{{id:collection}},"action":"deleted","actor":"{{actor}}",` +
|
|
`"timestamp":"{{timestamp}}","count":12},"timestamp":"{{timestamp}}"}}`
|
|
if string(got[0].Body) != want {
|
|
t.Fatalf("normalised\n got %s\nwant %s", got[0].Body, want)
|
|
}
|
|
|
|
// A Z timestamp and an actor with extra keys are left alone.
|
|
odd := []Publication{{Method: "POST", Path: "/p", Body: json.RawMessage(
|
|
`{"data":{"payload":{"actor":{"user_id":1,"name":"x","role":"a"}},"timestamp":"2026-09-30T10:00:00Z"}}`)}}
|
|
got, err = NormalizePublications(odd, store)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if !strings.Contains(string(got[0].Body), `"2026-09-30T10:00:00Z"`) || !strings.Contains(string(got[0].Body), `"role":"a"`) {
|
|
t.Fatalf("over-normalised %s", got[0].Body)
|
|
}
|
|
|
|
store.Set("id:other", "12")
|
|
if _, err := NormalizePublications(pubs, store); err == nil || !strings.Contains(err.Error(), "ambiguous") {
|
|
t.Fatalf("want ambiguity error, got %v", err)
|
|
}
|
|
}
|
|
|
|
func TestDiffPublications(t *testing.T) {
|
|
want := []Publication{{Method: "POST", Path: "/api/publish", Authorization: true,
|
|
Body: json.RawMessage(`{"channel":"c:{{id:c}}","data":{"payload":{"id":{{id:a}},"n":1}}}`)}}
|
|
same := []Publication{{Method: "POST", Path: "/api/publish", Authorization: true,
|
|
Body: json.RawMessage(`{"data":{"payload":{"n":1,"id":{{id:a}}}},"channel":"c:{{id:c}}"}`)}}
|
|
if d := DiffPublications(want, same); len(d) != 0 {
|
|
t.Fatalf("key order must not matter: %+v", d)
|
|
}
|
|
stringID := []Publication{{Method: "POST", Path: "/api/broadcast", Authorization: false,
|
|
Body: json.RawMessage(`{"channel":"c:{{id:c}}","data":{"payload":{"id":"{{id:a}}","n":1}}}`)}}
|
|
d := DiffPublications(want, stringID)
|
|
paths := map[string]bool{}
|
|
for _, x := range d {
|
|
paths[x.Path] = true
|
|
}
|
|
for _, p := range []string{"$[0].path", "$[0].authorization", "$[0].body.data.payload.id"} {
|
|
if !paths[p] {
|
|
t.Fatalf("missing diff at %s in %+v", p, d)
|
|
}
|
|
}
|
|
if d := DiffPublications(want, nil); len(d) != 1 || d[0].Path != "$" {
|
|
t.Fatalf("count diff %+v", d)
|
|
}
|
|
}
|
|
|
|
func TestBroadcastGoldenRoundTrip(t *testing.T) {
|
|
dir := t.TempDir()
|
|
path := filepath.Join(dir, "g.yaml")
|
|
g := BroadcastGolden{Version: CurrentVersion, Name: "deleted", Flow: "flows/x#delete", Pending: "later",
|
|
Publications: []Publication{{Method: "POST", Path: "/api/publish", Authorization: true,
|
|
Body: json.RawMessage(`{"channel":"c:{{id:c}}","data":{"payload":{"id":{{id:a}}}}}`)}}}
|
|
if err := WriteBroadcastGolden(path, g); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
back, err := LoadBroadcastGolden(path)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if back.Name != g.Name || back.Flow != g.Flow || back.Pending != g.Pending || len(back.Publications) != 1 {
|
|
t.Fatalf("round trip %+v", back)
|
|
}
|
|
if d := DiffPublications(g.Publications, back.Publications); len(d) != 0 {
|
|
t.Fatalf("round trip diff %+v", d)
|
|
}
|
|
|
|
leak := g
|
|
leak.Publications = []Publication{{Method: "POST", Path: "/p",
|
|
Body: json.RawMessage(`{"token":"eyJhbGciOiJIUzI1NiJ9.eyJzdWIiOiIxIn0.c2lnbmF0dXJl"}`)}}
|
|
if err := WriteBroadcastGolden(filepath.Join(dir, "leak.yaml"), leak); err == nil {
|
|
t.Fatal("a JWT-shaped body must be refused")
|
|
}
|
|
if err := os.WriteFile(filepath.Join(dir, "bad.yaml"), []byte("version: 1\nname: x\nextra: 1\npublications: []\n"), 0o644); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if _, err := LoadBroadcastGolden(filepath.Join(dir, "bad.yaml")); err == nil {
|
|
t.Fatal("unknown field must be rejected")
|
|
}
|
|
}
|
|
|
|
func freeLoopbackAddr(t *testing.T) string {
|
|
t.Helper()
|
|
ln, err := net.Listen("tcp", "127.0.0.1:0")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
addr := ln.Addr().String()
|
|
_ = ln.Close()
|
|
return addr
|
|
}
|
|
|
|
func TestRecordBroadcastsStep(t *testing.T) {
|
|
listen := freeLoopbackAddr(t)
|
|
publish := func(body string) {
|
|
req, _ := http.NewRequest(http.MethodPost, "http://"+listen+"/api/publish", strings.NewReader(body))
|
|
req.Header.Set("Authorization", "apikey test-key")
|
|
if resp, err := http.DefaultClient.Do(req); err == nil {
|
|
_ = resp.Body.Close()
|
|
}
|
|
}
|
|
backend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
|
w.Header().Set("Content-Type", "application/json")
|
|
switch r.Method {
|
|
case http.MethodPost:
|
|
publish(`{"channel":"room:5","data":{"event":"created","payload":{"id":77},"timestamp":"2026-09-30T10:00:00+00:00"}}`)
|
|
_, _ = io.WriteString(w, `{"data":{"id":77}}`)
|
|
case http.MethodDelete:
|
|
publish(`{"channel":"room:5","data":{"event":"deleted","payload":{"id":77},"timestamp":"2026-09-30T10:00:02+00:00"}}`)
|
|
_, _ = io.WriteString(w, `{"message":"deleted"}`)
|
|
}
|
|
}))
|
|
t.Cleanup(backend.Close)
|
|
|
|
store := mustMemoryStore()
|
|
store.Set("id:room", "5")
|
|
spec := Flow{Version: CurrentVersion, Name: "items", Steps: []Step{
|
|
{ID: "create", Request: Request{Method: "POST", Path: "/items"},
|
|
Capture: []CaptureRule{{From: "response.json", Path: "$.data.id", As: "id:item"}}},
|
|
{ID: "delete", Request: Request{Method: "DELETE", Path: "/items/{{id:item}}"}},
|
|
}}
|
|
ctx, cancel := context.WithTimeout(t.Context(), 10*time.Second)
|
|
defer cancel()
|
|
g, err := RecordBroadcasts(ctx, spec, BroadcastConfig{
|
|
Target: backend.URL, Listen: listen, APIKey: "test-key", Store: store,
|
|
Step: "delete", IDs: []string{"id:room", "id:item"}, Settle: 50 * time.Millisecond,
|
|
})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if g.Name != "delete" || g.Flow != "items#delete" || len(g.Publications) != 1 {
|
|
t.Fatalf("golden %+v", g)
|
|
}
|
|
want := `{"channel":"room:{{id:room}}","data":{"event":"deleted","payload":{"id":{{id:item}}},"timestamp":"{{timestamp}}"}}`
|
|
if string(g.Publications[0].Body) != want || !g.Publications[0].Authorization {
|
|
t.Fatalf("publication %+v body %s", g.Publications[0], g.Publications[0].Body)
|
|
}
|
|
|
|
if _, err := RecordBroadcasts(ctx, spec, BroadcastConfig{Target: "http://192.0.2.1:80", APIKey: "k"}); err == nil {
|
|
t.Fatal("a non-loopback target must be refused")
|
|
}
|
|
}
|