test(11-07): cover lighthouse realtime and the Centrifugo driver
- lighthouse: TestSuppression (Widget silenced, Gadget not, nesting, stale outer ctx), TestBulkEmitsOnce, TestBroadcastEdges (zero-key batch, update actor, id-only delete, method contract, multi-channel, savepoint), TestBroadcastPublishFailure, TestFromSelectsDriver, TestMountSurfaces, TestRegistry under -race, drivers, args JSON, Bind (coverage 91.7%) - centrifugo: TestTokenClaims, TestTokenHandler, TestClientRequests, TestClientLoadConfig and a TestProxy table porting the WinterCMS WS-005, WS-007 and WS-013 cases (coverage 92.4%)
This commit is contained in:
Jakub Zych
@@ -7,12 +7,14 @@ import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"reflect"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.golem15.com/golem15/summercms/modules/backpack"
|
||||
"git.golem15.com/golem15/summercms/modules/bouncer"
|
||||
"git.golem15.com/golem15/summercms/modules/compass"
|
||||
"git.golem15.com/golem15/summercms/modules/conga"
|
||||
"git.golem15.com/golem15/summercms/modules/lagoon"
|
||||
@@ -360,3 +362,425 @@ func TestBroadcastTx(t *testing.T) {
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
// Sprocket broadcasts through every optional method contract.
|
||||
type Sprocket struct {
|
||||
ID uint `gorm:"column:id;primaryKey"`
|
||||
Name string `gorm:"column:name"`
|
||||
Channels string `gorm:"column:channels"`
|
||||
}
|
||||
|
||||
func (Sprocket) TableName() string { return "acme_sprockets" }
|
||||
|
||||
func (s *Sprocket) BroadcastChannels(_ context.Context, tx *gorm.DB) ([]string, error) {
|
||||
if s.Channels == "" {
|
||||
return nil, nil
|
||||
}
|
||||
if s.Channels == "fail" {
|
||||
return nil, errors.New("channels failed")
|
||||
}
|
||||
if s.Channels == "abort-tx" {
|
||||
// A failed statement aborts a Postgres transaction unless it runs
|
||||
// inside a savepoint.
|
||||
return nil, tx.Exec(`SELECT * FROM acme_missing_table`).Error
|
||||
}
|
||||
return strings.Split(s.Channels, ","), nil
|
||||
}
|
||||
|
||||
func (Sprocket) BroadcastAlias() string { return "acme.sprocket" }
|
||||
func (Sprocket) BroadcastTTL() int { return 15 }
|
||||
|
||||
func (s *Sprocket) BroadcastPayload(_ context.Context, _ *gorm.DB, ev Event) (any, error) {
|
||||
if s.Name == "bad-payload" {
|
||||
return nil, errors.New("payload failed")
|
||||
}
|
||||
return struct {
|
||||
ID uint `json:"id"`
|
||||
Action Action `json:"action"`
|
||||
TTL int `json:"ttl"`
|
||||
Actor Actor `json:"actor"`
|
||||
}{s.ID, ev.Action, ev.TTL, ev.Actor}, nil
|
||||
}
|
||||
|
||||
func (s *Sprocket) ShouldBroadcast(a Action) bool { return a != ActionUpdated }
|
||||
|
||||
// TestSuppression covers D-08: WithoutBroadcasting[T] silences one model
|
||||
// type for the ctx it hands out and nothing else. Another type still
|
||||
// broadcasts, nesting adds types, a pointer type argument names the
|
||||
// struct, and a write through a stale outer ctx is not suppressed.
|
||||
func TestSuppression(t *testing.T) {
|
||||
env := newLHEnv(t, nil, func(t *testing.T, svc *Service) {
|
||||
if err := Bind[Gadget](svc, Binding[Gadget]{Channels: func(_ context.Context, _ *gorm.DB, m *Gadget) ([]string, error) {
|
||||
return []string{fmt.Sprintf("gadgets:%d", m.OwnerID)}, nil
|
||||
}}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
})
|
||||
mem := env.memory(t)
|
||||
ctx := t.Context()
|
||||
events := func(pubs []Publication) []string {
|
||||
var out []string
|
||||
for _, p := range pubs {
|
||||
out = append(out, p.Event+"@"+strings.Join(p.Channels, ","))
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
t.Run("silences_widget_not_gadget", func(t *testing.T) {
|
||||
from := len(mem.Publications())
|
||||
err := WithoutBroadcasting[Widget](ctx, func(ctx context.Context) error {
|
||||
return lagoon.Transaction(ctx, env.gdb, func(ctx context.Context, tx *gorm.DB) error {
|
||||
if err := tx.WithContext(ctx).Create(&Widget{Name: "quiet", OwnerID: 1}).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
return tx.WithContext(ctx).Create(&Gadget{Name: "loud", OwnerID: 1}).Error
|
||||
})
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
got := events(waitPublications(t, mem, from, 1))
|
||||
if len(got) != 1 || got[0] != "created.lighthouse.gadget@gadgets:1" {
|
||||
t.Fatalf("publications = %v, want only the gadget", got)
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("nesting_and_pointer_type_argument", func(t *testing.T) {
|
||||
before := env.broadcastJobs(t)
|
||||
err := WithoutBroadcasting[*Widget](ctx, func(ctx context.Context) error {
|
||||
return WithoutBroadcasting[Gadget](ctx, func(ctx context.Context) error {
|
||||
if err := env.gdb.WithContext(ctx).Create(&Widget{Name: "nested", OwnerID: 2}).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
return env.gdb.WithContext(ctx).Create(&Gadget{Name: "nested", OwnerID: 2}).Error
|
||||
})
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if after := env.broadcastJobs(t); after != before {
|
||||
t.Fatalf("broadcast jobs %d -> %d with both types suppressed", before, after)
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("stale_outer_ctx_is_not_suppressed", func(t *testing.T) {
|
||||
from := len(mem.Publications())
|
||||
outer := ctx
|
||||
err := WithoutBroadcasting[Widget](ctx, func(context.Context) error {
|
||||
return env.gdb.WithContext(outer).Create(&Widget{Name: "outer", OwnerID: 3}).Error
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
got := events(waitPublications(t, mem, from, 1))
|
||||
if got[0] != "created.lighthouse.widget@widgets:3" {
|
||||
t.Fatalf("publications = %v", got)
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("fn_error_is_returned", func(t *testing.T) {
|
||||
boom := errors.New("boom")
|
||||
if err := WithoutBroadcasting[Widget](nil, func(ctx context.Context) error {
|
||||
if ctx == nil {
|
||||
t.Error("nil ctx was not replaced")
|
||||
}
|
||||
return boom
|
||||
}); !errors.Is(err, boom) {
|
||||
t.Fatalf("err = %v", err)
|
||||
}
|
||||
if suppressed(nil, reflect.TypeFor[Widget]()) {
|
||||
t.Fatal("a nil ctx suppresses")
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
type bulkPayload struct {
|
||||
Reason string `json:"reason"`
|
||||
Count int `json:"count"`
|
||||
}
|
||||
|
||||
// TestBulkEmitsOnce covers D-08 and RT-03: N suppressed creates plus one
|
||||
// Emit in the same transaction publish exactly one summary event, with the
|
||||
// payload bytes in declaration order, and nothing when the transaction
|
||||
// rolls back.
|
||||
func TestBulkEmitsOnce(t *testing.T) {
|
||||
env := newLHEnv(t, map[string]any{"realtime.broadcast_namespace": "Acme"}, nil)
|
||||
mem := env.memory(t)
|
||||
ctx := t.Context()
|
||||
|
||||
from := len(mem.Publications())
|
||||
err := WithoutBroadcasting[Widget](ctx, func(ctx context.Context) error {
|
||||
return lagoon.Transaction(ctx, env.gdb, func(ctx context.Context, tx *gorm.DB) error {
|
||||
for i := range 3 {
|
||||
if err := tx.WithContext(ctx).Create(&Widget{Name: fmt.Sprintf("bulk-%d", i), OwnerID: 9}).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return env.svc.Emit(ctx, tx, Broadcast{Channels: []string{"Widgets:9"}, Event: "collection.bulk_updated", Payload: bulkPayload{Reason: "bulk_create", Count: 3}})
|
||||
})
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
p := waitPublications(t, mem, from, 1)[0]
|
||||
if p.Event != "collection.bulk_updated" || p.Channels[0] != "acme:widgets:9" || p.Method != "publish" {
|
||||
t.Fatalf("publication = %+v", p)
|
||||
}
|
||||
if string(p.Payload) != `{"reason":"bulk_create","count":3}` {
|
||||
t.Fatalf("payload = %s, want the ordered bytes", p.Payload)
|
||||
}
|
||||
|
||||
before := env.broadcastJobs(t)
|
||||
rollback := errors.New("rollback")
|
||||
err = lagoon.Transaction(ctx, env.gdb, func(ctx context.Context, tx *gorm.DB) error {
|
||||
if err := env.svc.Emit(ctx, tx, Broadcast{Channels: []string{"widgets:9"}, Event: "collection.bulk_updated", Payload: bulkPayload{Reason: "bulk_create", Count: 1}}); err != nil {
|
||||
return err
|
||||
}
|
||||
return rollback
|
||||
})
|
||||
if !errors.Is(err, rollback) {
|
||||
t.Fatalf("err = %v", err)
|
||||
}
|
||||
if after := env.broadcastJobs(t); after != before {
|
||||
t.Fatalf("an Emit in a rolled-back transaction left %d job(s)", after-before)
|
||||
}
|
||||
|
||||
// Emit without a transaction enqueues on its own; several channels are
|
||||
// one broadcast.
|
||||
from = len(mem.Publications())
|
||||
if err := env.svc.Emit(nil, nil, Broadcast{Channels: []string{"a:1", "b:2"}, Event: "acme.pinged", Payload: nil}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
p = waitPublications(t, mem, from, 1)[0]
|
||||
if p.Method != "broadcast" || strings.Join(p.Channels, ",") != "acme:a:1,acme:b:2" || string(p.Payload) != "null" {
|
||||
t.Fatalf("publication = %+v payload %s", p, p.Payload)
|
||||
}
|
||||
|
||||
if err := env.svc.Emit(ctx, env.gdb, Broadcast{Event: "x"}); err != nil {
|
||||
t.Fatalf("Emit without channels = %v, want nil", err)
|
||||
}
|
||||
if err := env.svc.Emit(ctx, env.gdb, Broadcast{Channels: []string{"a:1"}}); err == nil {
|
||||
t.Fatal("Emit without an event name succeeded")
|
||||
}
|
||||
if err := env.svc.Emit(ctx, env.gdb, Broadcast{Channels: []string{"a:1"}, Event: "x", Payload: make(chan int)}); err == nil {
|
||||
t.Fatal("Emit with an unencodable payload succeeded")
|
||||
}
|
||||
var nilSvc *Service
|
||||
if err := nilSvc.Emit(ctx, env.gdb, Broadcast{Channels: []string{"a:1"}, Event: "x"}); err == nil {
|
||||
t.Fatal("Emit on a nil service succeeded")
|
||||
}
|
||||
}
|
||||
|
||||
// failingDriver fails every publication.
|
||||
type failingDriver struct{ calls *int32 }
|
||||
|
||||
func (failingDriver) Name() string { return "acme-failing" }
|
||||
func (failingDriver) Routes() []Route { return nil }
|
||||
func (d failingDriver) Publish(context.Context, string, string, json.RawMessage) error {
|
||||
return errors.New("realtime server is down")
|
||||
}
|
||||
func (d failingDriver) Broadcast(context.Context, []string, string, json.RawMessage) error {
|
||||
return errors.New("realtime server is down")
|
||||
}
|
||||
|
||||
func init() {
|
||||
RegisterDriver("acme-failing", func(*backpack.App, *Service) (Driver, error) { return failingDriver{}, nil })
|
||||
}
|
||||
|
||||
// TestBroadcastEdges covers the remaining broadcast rules: a zero-key batch
|
||||
// update is skipped, updates and deletes broadcast (a delete through an
|
||||
// id-only model carries the reloaded row), the method-based contract
|
||||
// (alias, ttl, payload, filter), multi-channel broadcasts, the actor of a
|
||||
// frontend and a backend principal, failures inside the savepoint that
|
||||
// never abort the write, and a failed publish that is logged and not
|
||||
// retried.
|
||||
func TestBroadcastEdges(t *testing.T) {
|
||||
env := newLHEnv(t, nil, nil)
|
||||
name := "Ann"
|
||||
env.svc.SetUserLookup(func(_ context.Context, id uint) (User, bool, error) {
|
||||
switch id {
|
||||
case 7:
|
||||
return User{ID: 7, Name: &name}, true, nil
|
||||
case 8:
|
||||
return User{}, false, errors.New("lookup failed")
|
||||
}
|
||||
return User{}, false, nil
|
||||
})
|
||||
mem := env.memory(t)
|
||||
ctx := t.Context()
|
||||
|
||||
w := Widget{Name: "edge", OwnerID: 4}
|
||||
if err := env.gdb.WithContext(ctx).Create(&w).Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
waitPublications(t, mem, 0, 1)
|
||||
|
||||
t.Run("zero_key_batch_update_is_skipped", func(t *testing.T) {
|
||||
before := env.broadcastJobs(t)
|
||||
if err := env.gdb.WithContext(ctx).Model(&Widget{}).Where("owner_id = ?", 4).Update("name", "batch").Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if after := env.broadcastJobs(t); after != before {
|
||||
t.Fatalf("a zero-key batch update enqueued %d job(s)", after-before)
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("update_carries_the_frontend_actor", func(t *testing.T) {
|
||||
from := len(mem.Publications())
|
||||
uctx := bouncer.WithUser(ctx, &bouncer.Principal{ID: 7})
|
||||
w.Name = "renamed"
|
||||
if err := env.gdb.WithContext(uctx).Save(&w).Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
p := waitPublications(t, mem, from, 1)[0]
|
||||
var body struct {
|
||||
Actor Actor `json:"actor"`
|
||||
}
|
||||
if err := json.Unmarshal(p.Payload, &body); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if p.Event != "updated.lighthouse.widget" || body.Actor.UserID == nil || *body.Actor.UserID != 7 || body.Actor.Name == nil || *body.Actor.Name != "Ann" {
|
||||
t.Fatalf("publication %s payload %s", p.Event, p.Payload)
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("actor_rules", func(t *testing.T) {
|
||||
if a := env.svc.Actor(bouncer.WithUser(ctx, &bouncer.Principal{ID: 3, Backend: true})); a.UserID != nil || *a.Name != "System" {
|
||||
t.Fatalf("backend actor = %+v, want System", a)
|
||||
}
|
||||
if a := env.svc.Actor(bouncer.WithUser(ctx, &bouncer.Principal{ID: 8})); a.UserID == nil || *a.UserID != 8 || a.Name != nil {
|
||||
t.Fatalf("lookup-error actor = %+v, want the id and a null name", a)
|
||||
}
|
||||
if a := env.svc.Actor(bouncer.WithUser(ctx, &bouncer.Principal{ID: 99})); a.UserID == nil || a.Name != nil {
|
||||
t.Fatalf("unknown-user actor = %+v", a)
|
||||
}
|
||||
raw, _ := json.Marshal(SystemActor())
|
||||
if string(raw) != `{"user_id":null,"name":"System"}` {
|
||||
t.Fatalf("SystemActor = %s", raw)
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("delete_of_an_id_only_model_carries_the_row", func(t *testing.T) {
|
||||
from := len(mem.Publications())
|
||||
if err := env.gdb.WithContext(ctx).Delete(&Widget{ID: w.ID}).Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
p := waitPublications(t, mem, from, 1)[0]
|
||||
var body struct {
|
||||
Model Widget `json:"model"`
|
||||
}
|
||||
if err := json.Unmarshal(p.Payload, &body); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if p.Event != "deleted.lighthouse.widget" || p.Channels[0] != "widgets:4" || body.Model.Name != "renamed" {
|
||||
t.Fatalf("publication %s %v payload %s", p.Event, p.Channels, p.Payload)
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("method_contract_and_multi_channel", func(t *testing.T) {
|
||||
from := len(mem.Publications())
|
||||
s := Sprocket{Name: "multi", Channels: "Room:1,room:2"}
|
||||
if err := env.gdb.WithContext(ctx).Create(&s).Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
p := waitPublications(t, mem, from, 1)[0]
|
||||
if p.Method != "broadcast" || strings.Join(p.Channels, ",") != "room:1,room:2" || p.Event != "created.acme.sprocket" {
|
||||
t.Fatalf("publication = %+v", p)
|
||||
}
|
||||
if !strings.HasPrefix(string(p.Payload), fmt.Sprintf(`{"id":%d,"action":"created","ttl":15,`, s.ID)) {
|
||||
t.Fatalf("payload = %s", p.Payload)
|
||||
}
|
||||
before := env.broadcastJobs(t)
|
||||
s.Name = "filtered"
|
||||
if err := env.gdb.WithContext(ctx).Save(&s).Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if after := env.broadcastJobs(t); after != before {
|
||||
t.Fatal("ShouldBroadcast(updated) = false did not veto the update")
|
||||
}
|
||||
if err := env.gdb.WithContext(ctx).Create(&Sprocket{Name: "silent"}).Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if after := env.broadcastJobs(t); after != before {
|
||||
t.Fatal("a model without channels enqueued a job")
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("failures_are_logged_and_never_abort_the_write", func(t *testing.T) {
|
||||
before := env.broadcastJobs(t)
|
||||
err := lagoon.Transaction(ctx, env.gdb, func(ctx context.Context, tx *gorm.DB) error {
|
||||
if err := tx.WithContext(ctx).Create(&Sprocket{Name: "chan-fail", Channels: "fail"}).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
if err := tx.WithContext(ctx).Create(&Sprocket{Name: "bad-payload", Channels: "room:3"}).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
// The transaction is still usable after both failures.
|
||||
return tx.WithContext(ctx).Create(&Widget{Name: "after-failures", OwnerID: 0}).Error
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("a broadcast failure aborted the write: %v", err)
|
||||
}
|
||||
var n int64
|
||||
if err := env.gdb.Model(&Sprocket{}).Where("name IN ?", []string{"chan-fail", "bad-payload"}).Count(&n).Error; err != nil || n != 2 {
|
||||
t.Fatalf("committed sprockets = %d (err %v), want 2", n, err)
|
||||
}
|
||||
if after := env.broadcastJobs(t); after != before {
|
||||
t.Fatalf("failed broadcasts enqueued %d job(s)", after-before)
|
||||
}
|
||||
if env.logs.count("realtime: broadcast skipped") < 2 {
|
||||
t.Fatalf("no skip warnings:\n%s", env.logs.all())
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("a_failed_query_inside_the_savepoint_keeps_the_transaction", func(t *testing.T) {
|
||||
err := lagoon.Transaction(ctx, env.gdb, func(ctx context.Context, tx *gorm.DB) error {
|
||||
if err := tx.WithContext(ctx).Create(&Sprocket{Name: "abort", Channels: "abort-tx"}).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
return tx.WithContext(ctx).Create(&Widget{Name: "after-abort", OwnerID: 0}).Error
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("a failed query in the broadcast savepoint aborted the write: %v", err)
|
||||
}
|
||||
var n int64
|
||||
if err := env.gdb.Model(&Widget{}).Where("name = ?", "after-abort").Count(&n).Error; err != nil || n != 1 {
|
||||
t.Fatalf("write after the failed broadcast query: %d rows (err %v)", n, err)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
// TestBroadcastPublishFailure covers D-09: a failed publish is logged at
|
||||
// Warn without the payload and the one-attempt job is not retried.
|
||||
func TestBroadcastPublishFailure(t *testing.T) {
|
||||
env := newLHEnv(t, map[string]any{"realtime.driver": "acme-failing"}, nil)
|
||||
ctx := t.Context()
|
||||
if err := env.gdb.WithContext(ctx).Create(&Widget{Name: "secret-payload-value", OwnerID: 6}).Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
deadline := time.Now().Add(10 * time.Second)
|
||||
for env.logs.count("realtime: broadcast failed") == 0 {
|
||||
if time.Now().After(deadline) {
|
||||
t.Fatalf("no failure log:\n%s", env.logs.all())
|
||||
}
|
||||
time.Sleep(20 * time.Millisecond)
|
||||
}
|
||||
rec, _ := env.logs.find("realtime: broadcast failed")
|
||||
if rec.level != slog.LevelWarn || !strings.Contains(rec.attrs["channels"], "widgets:6") || strings.Contains(env.logs.all(), "secret-payload-value") {
|
||||
t.Fatalf("failure log = %+v", rec)
|
||||
}
|
||||
var state string
|
||||
var attempt int
|
||||
for time.Now().Before(deadline) {
|
||||
if err := env.db.QueryRowContext(ctx, `SELECT state::text, attempt FROM river_job WHERE kind = 'summer.broadcast' ORDER BY id DESC LIMIT 1`).Scan(&state, &attempt); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if state == "completed" {
|
||||
break
|
||||
}
|
||||
time.Sleep(20 * time.Millisecond)
|
||||
}
|
||||
if state != "completed" || attempt != 1 {
|
||||
t.Fatalf("job state %q attempt %d, want completed after one attempt (never retried)", state, attempt)
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user