feat(11-06): add the tide fake Centrifugo recorder, broadcast goldens and parity:broadcasts
- 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
This commit is contained in:
Jakub Zych
370
modules/tide/centrifugo.go
Normal file
370
modules/tide/centrifugo.go
Normal file
@@ -0,0 +1,370 @@
|
||||
package tide
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/subtle"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"net"
|
||||
"net/http"
|
||||
"sort"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
// DefaultCentrifugoListen is the loopback address the fake Centrifugo
|
||||
// recorder binds when no other address is given.
|
||||
const DefaultCentrifugoListen = "127.0.0.1:8424"
|
||||
|
||||
// MaxPublicationBody caps one recorded Centrifugo API request body.
|
||||
const MaxPublicationBody = 1 << 20
|
||||
|
||||
// DefaultBroadcastSettle is how long RecordBroadcasts waits after a step
|
||||
// for publications that arrive after its HTTP response.
|
||||
const DefaultBroadcastSettle = 500 * time.Millisecond
|
||||
|
||||
// Publication is one publish or broadcast request received by the fake
|
||||
// Centrifugo recorder. The Authorization header value is never kept: only
|
||||
// whether it carried the expected API key.
|
||||
type Publication struct {
|
||||
Method string `yaml:"method"`
|
||||
Path string `yaml:"path"`
|
||||
Authorization bool `yaml:"authorization"`
|
||||
Body json.RawMessage `yaml:"body"`
|
||||
}
|
||||
|
||||
// CentrifugoRecorderOptions configures NewCentrifugoRecorder.
|
||||
type CentrifugoRecorderOptions struct {
|
||||
// APIKey is the key the backend under test sends as
|
||||
// "Authorization: apikey <key>". It is only compared, never stored.
|
||||
APIKey string
|
||||
}
|
||||
|
||||
// CentrifugoRecorder is a fake Centrifugo HTTP API. It records every
|
||||
// publish and broadcast request and answers the other server API calls a
|
||||
// backend makes with empty successful results.
|
||||
type CentrifugoRecorder struct {
|
||||
apiKey string
|
||||
mu sync.Mutex
|
||||
pubs []Publication
|
||||
}
|
||||
|
||||
// NewCentrifugoRecorder returns an empty recorder.
|
||||
func NewCentrifugoRecorder(opts CentrifugoRecorderOptions) *CentrifugoRecorder {
|
||||
return &CentrifugoRecorder{apiKey: opts.APIKey}
|
||||
}
|
||||
|
||||
// ServeHTTP records POST .../publish and .../broadcast and answers
|
||||
// {"result":{}}. .../presence answers an empty presence map, .../unsubscribe
|
||||
// and .../info answer {"result":{}}, anything else is 404.
|
||||
func (c *CentrifugoRecorder) ServeHTTP(w http.ResponseWriter, r *http.Request) {
|
||||
path := strings.TrimRight(r.URL.Path, "/")
|
||||
recorded := strings.HasSuffix(path, "/publish") || strings.HasSuffix(path, "/broadcast")
|
||||
answered := strings.HasSuffix(path, "/presence") || strings.HasSuffix(path, "/unsubscribe") || strings.HasSuffix(path, "/info")
|
||||
if !recorded && !answered {
|
||||
http.NotFound(w, r)
|
||||
return
|
||||
}
|
||||
if r.Method != http.MethodPost {
|
||||
w.Header().Set("Allow", http.MethodPost)
|
||||
http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
|
||||
return
|
||||
}
|
||||
raw, err := io.ReadAll(io.LimitReader(r.Body, MaxPublicationBody+1))
|
||||
if err != nil {
|
||||
http.Error(w, "read body", http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
if len(raw) > MaxPublicationBody {
|
||||
http.Error(w, "body too large", http.StatusRequestEntityTooLarge)
|
||||
return
|
||||
}
|
||||
if recorded {
|
||||
c.mu.Lock()
|
||||
c.pubs = append(c.pubs, Publication{
|
||||
Method: r.Method,
|
||||
Path: r.URL.Path,
|
||||
Authorization: c.authorized(r.Header.Get("Authorization")),
|
||||
Body: json.RawMessage(raw),
|
||||
})
|
||||
c.mu.Unlock()
|
||||
}
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
if strings.HasSuffix(path, "/presence") {
|
||||
_, _ = io.WriteString(w, `{"result":{"presence":{}}}`)
|
||||
return
|
||||
}
|
||||
_, _ = io.WriteString(w, `{"result":{}}`)
|
||||
}
|
||||
|
||||
func (c *CentrifugoRecorder) authorized(header string) bool {
|
||||
if c.apiKey == "" {
|
||||
return false
|
||||
}
|
||||
want := "apikey " + c.apiKey
|
||||
return subtle.ConstantTimeCompare([]byte(header), []byte(want)) == 1
|
||||
}
|
||||
|
||||
// Publications returns a copy of the recorded requests in arrival order.
|
||||
func (c *CentrifugoRecorder) Publications() []Publication {
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
out := make([]Publication, len(c.pubs))
|
||||
for i, p := range c.pubs {
|
||||
p.Body = append(json.RawMessage(nil), p.Body...)
|
||||
out[i] = p
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// Reset forgets every recorded request.
|
||||
func (c *CentrifugoRecorder) Reset() {
|
||||
c.mu.Lock()
|
||||
c.pubs = nil
|
||||
c.mu.Unlock()
|
||||
}
|
||||
|
||||
// ListenAndServe serves the recorder on addr until ctx is cancelled. addr
|
||||
// must be a loopback address, the same rule the recording proxy applies.
|
||||
func (c *CentrifugoRecorder) ListenAndServe(ctx context.Context, addr string) error {
|
||||
if err := requireLoopbackAddr(addr); err != nil {
|
||||
return err
|
||||
}
|
||||
ln, err := net.Listen("tcp", addr)
|
||||
if err != nil {
|
||||
return fmt.Errorf("tide: centrifugo recorder listen %s: %w", addr, err)
|
||||
}
|
||||
srv := &http.Server{Handler: c, ReadHeaderTimeout: 10 * time.Second}
|
||||
done := make(chan struct{})
|
||||
go func() {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
shutCtx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
|
||||
_ = srv.Shutdown(shutCtx)
|
||||
cancel()
|
||||
case <-done:
|
||||
}
|
||||
}()
|
||||
err = srv.Serve(ln)
|
||||
close(done)
|
||||
if errors.Is(err, http.ErrServerClosed) {
|
||||
return nil
|
||||
}
|
||||
return err
|
||||
}
|
||||
|
||||
// BroadcastConfig drives RecordBroadcasts.
|
||||
type BroadcastConfig struct {
|
||||
// Target is the loopback base URL of the reference backend.
|
||||
Target string
|
||||
// Listen is the loopback address of the recorder the backend publishes
|
||||
// to. Empty means DefaultCentrifugoListen.
|
||||
Listen string
|
||||
// APIKey is the Centrifugo API key the backend is configured with.
|
||||
APIKey string
|
||||
// Store holds and receives captured variables (the private vars file).
|
||||
Store *Store
|
||||
// Rules are optional capture rules merged into steps without captures.
|
||||
Rules Rules
|
||||
// Step names the flow step whose publications are recorded. Earlier
|
||||
// steps run first as setup and their publications are discarded; later
|
||||
// steps do not run. Empty records the whole flow.
|
||||
Step string
|
||||
// IDs names the id:* variables replaced by placeholders. Empty means
|
||||
// every id:* placeholder or capture the flow mentions.
|
||||
IDs []string
|
||||
// Name is the golden name. Empty uses the step, else the flow name.
|
||||
Name string
|
||||
// Settle is the wait for publications that arrive after a response.
|
||||
Settle time.Duration
|
||||
// Client and MaxBody are passed to RecordFlow.
|
||||
Client *http.Client
|
||||
MaxBody int64
|
||||
}
|
||||
|
||||
// RecordBroadcasts runs spec against the reference backend while a
|
||||
// loopback CentrifugoRecorder stands in for Centrifugo, then returns the
|
||||
// normalised publications as a golden. It fails when nothing is published.
|
||||
func RecordBroadcasts(ctx context.Context, spec Flow, cfg BroadcastConfig) (BroadcastGolden, error) {
|
||||
if err := validateFlow(spec); err != nil {
|
||||
return BroadcastGolden{}, err
|
||||
}
|
||||
if _, err := parseLoopbackUpstream(cfg.Target); err != nil {
|
||||
return BroadcastGolden{}, fmt.Errorf("tide: broadcast target: %w", err)
|
||||
}
|
||||
if strings.TrimSpace(cfg.APIKey) == "" {
|
||||
return BroadcastGolden{}, fmt.Errorf("tide: broadcast recording needs the backend's Centrifugo API key")
|
||||
}
|
||||
listen := cfg.Listen
|
||||
if listen == "" {
|
||||
listen = DefaultCentrifugoListen
|
||||
}
|
||||
if err := requireLoopbackAddr(listen); err != nil {
|
||||
return BroadcastGolden{}, err
|
||||
}
|
||||
idx := -1
|
||||
if cfg.Step != "" {
|
||||
for i, s := range spec.Steps {
|
||||
if s.ID == cfg.Step {
|
||||
idx = i
|
||||
break
|
||||
}
|
||||
}
|
||||
if idx < 0 {
|
||||
return BroadcastGolden{}, fmt.Errorf("tide: flow %q has no step %q", spec.Name, cfg.Step)
|
||||
}
|
||||
}
|
||||
store := cfg.Store
|
||||
if store == nil {
|
||||
store = mustMemoryStore()
|
||||
}
|
||||
settle := cfg.Settle
|
||||
if settle <= 0 {
|
||||
settle = DefaultBroadcastSettle
|
||||
}
|
||||
|
||||
rec := NewCentrifugoRecorder(CentrifugoRecorderOptions{APIKey: cfg.APIKey})
|
||||
srvCtx, cancel := context.WithCancel(ctx)
|
||||
defer cancel()
|
||||
errCh := make(chan error, 1)
|
||||
go func() { errCh <- rec.ListenAndServe(srvCtx, listen) }()
|
||||
if err := waitListening(ctx, listen, errCh); err != nil {
|
||||
return BroadcastGolden{}, err
|
||||
}
|
||||
|
||||
record := func(steps []Step) error {
|
||||
sub := spec
|
||||
sub.Steps = steps
|
||||
_, err := RecordFlow(ctx, sub, RecordConfig{Target: cfg.Target, Client: cfg.Client, MaxBody: cfg.MaxBody, Store: store, Rules: cfg.Rules})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return sleepCtx(ctx, settle)
|
||||
}
|
||||
steps := spec.Steps
|
||||
if idx >= 0 {
|
||||
if idx > 0 {
|
||||
if err := record(spec.Steps[:idx]); err != nil {
|
||||
return BroadcastGolden{}, err
|
||||
}
|
||||
rec.Reset()
|
||||
}
|
||||
steps = spec.Steps[idx : idx+1]
|
||||
}
|
||||
if err := record(steps); err != nil {
|
||||
return BroadcastGolden{}, err
|
||||
}
|
||||
pubs := rec.Publications()
|
||||
if len(pubs) == 0 {
|
||||
return BroadcastGolden{}, fmt.Errorf("tide: flow %q published nothing to %s; is the backend's Centrifugo API URL pointed at the recorder?", spec.Name, listen)
|
||||
}
|
||||
|
||||
names := cfg.IDs
|
||||
if len(names) == 0 {
|
||||
names = flowIDNames(spec)
|
||||
}
|
||||
idStore := mustMemoryStore()
|
||||
for _, n := range names {
|
||||
v, ok := store.Get(n)
|
||||
if !ok || v == "" {
|
||||
return BroadcastGolden{}, fmt.Errorf("tide: id variable %q is not in the vars store", n)
|
||||
}
|
||||
idStore.Set(n, v)
|
||||
}
|
||||
norm, err := NormalizePublications(pubs, idStore)
|
||||
if err != nil {
|
||||
return BroadcastGolden{}, err
|
||||
}
|
||||
for i, p := range norm {
|
||||
if strings.Contains(string(p.Body), cfg.APIKey) {
|
||||
return BroadcastGolden{}, fmt.Errorf("tide: publication %d body contains the Centrifugo API key", i)
|
||||
}
|
||||
}
|
||||
g := BroadcastGolden{
|
||||
Version: CurrentVersion,
|
||||
Name: cfg.Name,
|
||||
Flow: spec.Name,
|
||||
Publications: norm,
|
||||
}
|
||||
if cfg.Step != "" {
|
||||
g.Flow += "#" + cfg.Step
|
||||
}
|
||||
if g.Name == "" {
|
||||
g.Name = cfg.Step
|
||||
}
|
||||
if g.Name == "" {
|
||||
g.Name = spec.Name
|
||||
}
|
||||
return g, nil
|
||||
}
|
||||
|
||||
func waitListening(ctx context.Context, addr string, errCh <-chan error) error {
|
||||
deadline := time.Now().Add(5 * time.Second)
|
||||
for {
|
||||
select {
|
||||
case err := <-errCh:
|
||||
if err == nil {
|
||||
err = fmt.Errorf("stopped")
|
||||
}
|
||||
return fmt.Errorf("tide: centrifugo recorder: %w", err)
|
||||
default:
|
||||
}
|
||||
conn, err := net.DialTimeout("tcp", addr, 200*time.Millisecond)
|
||||
if err == nil {
|
||||
_ = conn.Close()
|
||||
return nil
|
||||
}
|
||||
if time.Now().After(deadline) {
|
||||
return fmt.Errorf("tide: centrifugo recorder did not start on %s: %w", addr, err)
|
||||
}
|
||||
if err := sleepCtx(ctx, 20*time.Millisecond); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func sleepCtx(ctx context.Context, d time.Duration) error {
|
||||
t := time.NewTimer(d)
|
||||
defer t.Stop()
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return ctx.Err()
|
||||
case <-t.C:
|
||||
return nil
|
||||
}
|
||||
}
|
||||
|
||||
// flowIDNames lists the id:* variables a flow mentions in request
|
||||
// placeholders or captures, sorted.
|
||||
func flowIDNames(spec Flow) []string {
|
||||
seen := map[string]struct{}{}
|
||||
add := func(s string) {
|
||||
for _, m := range placeholderRe.FindAllStringSubmatch(s, -1) {
|
||||
if strings.HasPrefix(m[1], "id:") {
|
||||
seen[m[1]] = struct{}{}
|
||||
}
|
||||
}
|
||||
}
|
||||
for _, step := range spec.Steps {
|
||||
add(step.Request.Path)
|
||||
add(step.Request.Query)
|
||||
add(string(step.Request.Body))
|
||||
for _, v := range step.Request.Headers {
|
||||
add(v)
|
||||
}
|
||||
for _, c := range step.Capture {
|
||||
if strings.HasPrefix(c.As, "id:") {
|
||||
seen[c.As] = struct{}{}
|
||||
}
|
||||
}
|
||||
}
|
||||
out := make([]string, 0, len(seen))
|
||||
for n := range seen {
|
||||
out = append(out, n)
|
||||
}
|
||||
sort.Strings(out)
|
||||
return out
|
||||
}
|
||||
Reference in New Issue
Block a user