Something went wrong. Try again.
Monorepo for Tangled tangled.org
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192package db
import ( "context" "database/sql" "path/filepath" "slices" "testing" "time"
"tangled.org/core/notifier" "tangled.org/core/spindle/models" "tangled.org/core/workflow")
func enqueueBare(ctx context.Context, d *DB, repoDid, pipeline string) error { record := &models.PipelineRecord{ID: models.PipelineId(pipeline), RepoDID: repoDid} return d.EnqueuePipeline(ctx, repoDid, record, nil, workflow.PipelineSpec{}, false, "", "", nil)}
func TestJobTraceContextSurvivesEnqueue(t *testing.T) { ctx := context.Background() database, err := Make(ctx, filepath.Join(t.TempDir(), "spindle.db")) if err != nil { t.Fatal(err) } t.Cleanup(func() { database.Close() })
pipelineID := models.PipelineId("pipeline") const traceparent = "00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01" const tracestate = "vendor=value" record := &models.PipelineRecord{ID: pipelineID, RepoDID: "did:web:repo.example.com"} if err := database.EnqueuePipeline( ctx, record.RepoDID, record, &workflow.TriggerRepo{}, workflow.PipelineSpec{}, false, traceparent, tracestate, nil, ); err != nil { t.Fatal(err) }
job, err := database.DequeueJob(ctx, nil) if err != nil { t.Fatal(err) } if job == nil { t.Fatal("dequeued job is nil") } if job.Traceparent != traceparent || job.Tracestate != tracestate { t.Fatalf("trace context = (%q, %q), want (%q, %q)", job.Traceparent, job.Tracestate, traceparent, tracestate) } if created := time.Unix(0, job.CreatedAtNs); created.Before(time.Now().Add(-5*time.Second)) || created.After(time.Now().Add(time.Second)) { t.Fatalf("created_at_ns = %v, want current enqueue time", created) }}
func TestEnqueuePipelineCommitsJobAndStatusesTogether(t *testing.T) { ctx := context.Background() database, err := Make(ctx, filepath.Join(t.TempDir(), "spindle.db")) if err != nil { t.Fatal(err) } t.Cleanup(func() { database.Close() })
n := notifier.New() pipelineID := models.PipelineId("pipeline") tpl := workflow.PipelineSpec{Workflows: []*workflow.WorkflowSpec{{Name: "build"}}} record := &models.PipelineRecord{ID: pipelineID, RepoDID: "did:web:repo.example.com"} if err := database.EnqueuePipeline( ctx, record.RepoDID, record, nil, tpl, false, "", "", &n, ); err != nil { t.Fatal(err) }
status, err := database.GetStatus(models.WorkflowId{PipelineId: pipelineID, Name: "build"}) if err != nil { t.Fatal(err) } if status != models.StatusKindPending { t.Fatalf("status = %q, want pending", status) } job, err := database.DequeueJob(ctx, nil) if err != nil { t.Fatal(err) } if job == nil || job.CreatedAtNs == 0 { t.Fatalf("dequeued job = %+v, want nanosecond admission timestamp", job) }}
func TestMakeMigratesLegacyJobTimestampToNanoseconds(t *testing.T) { ctx := context.Background() path := filepath.Join(t.TempDir(), "spindle.db") raw, err := sql.Open("sqlite3", path) if err != nil { t.Fatal(err) } _, err = raw.Exec(` create table jobs ( id integer primary key autoincrement, repo_did text not null, pipeline_id_knot text not null, pipeline_id_rkey text not null, source_repo text, tpl text not null, traceparent text not null default '', tracestate text not null default '', created_at integer not null ); create table migrations ( id integer primary key autoincrement, name text unique ); insert into jobs ( repo_did, pipeline_id_knot, pipeline_id_rkey, tpl, created_at ) values ('did:web:repo.example.com', 'knot.example.com', 'pipeline', '{}', 1700000000); `) if err != nil { t.Fatal(err) } if err := raw.Close(); err != nil { t.Fatal(err) }
database, err := Make(ctx, path) if err != nil { t.Fatal(err) } t.Cleanup(func() { database.Close() }) job, err := database.DequeueJob(ctx, nil) if err != nil { t.Fatal(err) } if job == nil || job.CreatedAtNs != 1700000000*int64(time.Second) { t.Fatalf("created_at_ns = %v, want converted legacy seconds", job) }}
func TestDequeueJobPrefersTheOwnerWithFewerRunning(t *testing.T) { ctx := context.Background() database, err := Make(ctx, filepath.Join(t.TempDir(), "spindle.db")) if err != nil { t.Fatal(err) } t.Cleanup(func() { database.Close() })
if err := database.AddRepo(Repo{Knot: "knot.example.com", Owner: "did:plc:busy", Rkey: "busy", RepoDid: "did:plc:busyrepo"}); err != nil { t.Fatal(err) } enqueue := func(repoDid, pipeline string) { t.Helper() if err := enqueueBare(ctx, database, repoDid, pipeline); err != nil { t.Fatal(err) } } enqueue("did:plc:busyrepo", "busy-1") enqueue("did:plc:busyrepo", "busy-2") // no repo row yet, so the repo stands in for its owner enqueue("did:plc:quietrepo", "quiet-1")
running := map[string]int{"did:plc:busy": 1} var order []string for { job, err := database.DequeueJob(ctx, running) if err != nil { t.Fatal(err) } if job == nil { break } order = append(order, string(job.PipelineId)) running[job.Owner]++ } want := []string{"quiet-1", "busy-1", "busy-2"} if !slices.Equal(order, want) { t.Fatalf("dequeue order = %v, want %v", order, want) }}