Something went wrong. Try again.
Monorepo for Tangled
Something went wrong. Try again.
Go
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146package state
import ( "context" "io" "log/slog" "net/http" "net/http/httptest" "path/filepath" "strings" "testing" "time"
"tangled.org/core/appview/db" "tangled.org/core/appview/pipelines" ec "tangled.org/core/eventconsumer" "tangled.org/core/eventconsumer/cursor" "tangled.org/core/eventstream" "tangled.org/core/notifier" spindledb "tangled.org/core/spindle/db" spindlemodels "tangled.org/core/spindle/models")
func TestColdStart_SpindleEventsRebuildPipelineStatuses(t *testing.T) { ctx := t.Context()
spindleDB, err := spindledb.Make(ctx, filepath.Join(t.TempDir(), "spindle.db")) if err != nil { t.Fatalf("spindle Make: %v", err) } t.Cleanup(func() { spindleDB.Close() })
n := notifier.New() workflowId := spindlemodels.WorkflowId{ PipelineId: spindlemodels.PipelineId{Knot: "knot.boltless.example", Rkey: "pipeline-rk1"}, Name: "build", } for _, step := range []func() error{ func() error { return spindleDB.StatusPending(workflowId, &n) }, func() error { return spindleDB.StatusRunning(workflowId, &n) }, func() error { return spindleDB.StatusSuccess(workflowId, &n) }, } { if err := step(); err != nil { t.Fatalf("seed spindle event: %v", err) } }
mux := http.NewServeMux() mux.HandleFunc("/events", func(w http.ResponseWriter, r *http.Request) { _ = eventstream.Stream(w, r, eventstream.StreamConfig{ Backend: spindleDB, Notifier: &n, Logger: slog.New(slog.NewTextHandler(io.Discard, nil)), }) }) srv := httptest.NewServer(mux) t.Cleanup(srv.Close) source := ec.Source{Kind: "test", Host: strings.TrimPrefix(srv.URL, "http://")}
appviewDB, err := db.Make(ctx, filepath.Join(t.TempDir(), "appview.db")) if err != nil { t.Fatalf("appview Make: %v", err) } t.Cleanup(func() { appviewDB.Close() })
logger := slog.New(slog.NewTextHandler(io.Discard, nil)) processFunc := spindleIngester(appviewDB, pipelines.NewStatusNotifier())
cfg := ec.ConsumerConfig{ ProcessFunc: processFunc, WorkerCount: 1, QueueSize: 16, ConnectionTimeout: 2 * time.Second, CursorStore: &cursor.MemoryStore{}, URLFunc: ec.DefaultURL(true), Logger: logger, } c := ec.NewConsumer(cfg)
consumerCtx, cancel := context.WithCancel(ctx) defer cancel() c.Start(consumerCtx) c.AddSource(consumerCtx, source)
deadline := time.Now().Add(3 * time.Second) for time.Now().Before(deadline) { var n int if err := appviewDB.QueryRow(`select count(*) from pipeline_statuses`).Scan(&n); err != nil { t.Fatalf("count: %v", err) } if n >= 3 { break } time.Sleep(20 * time.Millisecond) }
rows, err := appviewDB.Query(` select spindle, pipeline_knot, pipeline_rkey, workflow, status from pipeline_statuses order by created asc `) if err != nil { t.Fatalf("query: %v", err) } defer rows.Close()
type rec struct { spindle, knot, rkey, workflow, status string } var got []rec for rows.Next() { var r rec if err := rows.Scan(&r.spindle, &r.knot, &r.rkey, &r.workflow, &r.status); err != nil { t.Fatalf("scan: %v", err) } got = append(got, r) }
if len(got) != 3 { t.Fatalf("pipeline_statuses rows = %d, want 3: %+v", len(got), got) }
wantStatuses := []string{"pending", "running", "success"} gotStatuses := map[string]bool{} for _, r := range got { gotStatuses[r.status] = true if r.spindle != source.Host { t.Errorf("spindle = %q, want %q", r.spindle, source.Host) } if r.knot != workflowId.Knot { t.Errorf("pipeline_knot = %q, want %q", r.knot, workflowId.Knot) } if r.rkey != workflowId.Rkey { t.Errorf("pipeline_rkey = %q, want %q", r.rkey, workflowId.Rkey) } if r.workflow != workflowId.Name { t.Errorf("workflow = %q, want %q", r.workflow, workflowId.Name) } } for _, want := range wantStatuses { if !gotStatuses[want] { t.Errorf("missing status %q in projection", want) } }}