Something went wrong. Try again.
Monorepo for Tangled
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150package engine
import ( "context" "errors" "testing" "time"
"tangled.org/core/spindle/models")
func TestSemaphoreSlotterDisabledDoesNotBlock(t *testing.T) { t.Parallel()
slotter := NewSemaphoreSlotter(0)
for range 10 { slot, err := slotter.AcquireWorkflowSlot(context.Background(), zeroWorkflowID(), nil, Wait) if err != nil { t.Fatalf("AcquireWorkflowSlot() error = %v", err) } slot.Release() }}
func TestSemaphoreSlotterBlocksUntilRelease(t *testing.T) { t.Parallel()
slotter := NewSemaphoreSlotter(1)
first, err := slotter.AcquireWorkflowSlot(context.Background(), zeroWorkflowID(), nil, Wait) if err != nil { t.Fatalf("first AcquireWorkflowSlot() error = %v", err) } releasedFirst := false defer func() { if !releasedFirst { first.Release() } }()
acquired := make(chan WorkflowSlot, 1) errs := make(chan error, 1) go func() { slot, err := slotter.AcquireWorkflowSlot(context.Background(), zeroWorkflowID(), nil, Wait) if err != nil { errs <- err return } acquired <- slot }()
assertNotAcquired(t, acquired, errs)
first.Release() releasedFirst = true
second := waitForSlot(t, acquired, errs) second.Release()}
func TestSemaphoreSlotterHonorsContextCancellation(t *testing.T) { t.Parallel()
slotter := NewSemaphoreSlotter(1)
first, err := slotter.AcquireWorkflowSlot(context.Background(), zeroWorkflowID(), nil, Wait) if err != nil { t.Fatalf("first AcquireWorkflowSlot() error = %v", err) } defer first.Release()
ctx, cancel := context.WithCancel(context.Background()) cancel()
_, err = slotter.AcquireWorkflowSlot(ctx, zeroWorkflowID(), nil, Wait) if !errors.Is(err, context.Canceled) { t.Fatalf("AcquireWorkflowSlot() error = %v, want context.Canceled", err) }}
func TestSemaphoreSlotterTryDisabledDoesNotReject(t *testing.T) { t.Parallel()
slotter := NewSemaphoreSlotter(0)
for range 10 { slot, err := slotter.AcquireWorkflowSlot(context.Background(), zeroWorkflowID(), nil, NoWait) if err != nil { t.Fatalf("AcquireWorkflowSlot(NoWait) error = %v", err) } slot.Release() }}
func TestSemaphoreSlotterTryRejectsWhenFull(t *testing.T) { t.Parallel()
slotter := NewSemaphoreSlotter(1)
first, err := slotter.AcquireWorkflowSlot(context.Background(), zeroWorkflowID(), nil, NoWait) if err != nil { t.Fatalf("first AcquireWorkflowSlot(NoWait) error = %v", err) }
if _, err := slotter.AcquireWorkflowSlot(context.Background(), zeroWorkflowID(), nil, NoWait); !errors.Is(err, ErrNoWorkflowSlots) { t.Fatalf("AcquireWorkflowSlot(NoWait) error = %v, want ErrNoWorkflowSlots", err) }
first.Release()
second, err := slotter.AcquireWorkflowSlot(context.Background(), zeroWorkflowID(), nil, NoWait) if err != nil { t.Fatalf("AcquireWorkflowSlot(NoWait) after release error = %v", err) } second.Release()}
func assertNotAcquired(t *testing.T, acquired <-chan WorkflowSlot, errs <-chan error) { t.Helper()
select { case slot := <-acquired: slot.Release() t.Fatal("AcquireWorkflowSlot() acquired a slot before one was released") case err := <-errs: t.Fatalf("AcquireWorkflowSlot() returned unexpected error: %v", err) case <-time.After(25 * time.Millisecond): }}
func waitForSlot(t *testing.T, acquired <-chan WorkflowSlot, errs <-chan error) WorkflowSlot { t.Helper()
select { case slot := <-acquired: return slot case err := <-errs: t.Fatalf("AcquireWorkflowSlot() returned error: %v", err) case <-time.After(time.Second): t.Fatal("timed out waiting for slot acquisition") }
return nil}
func zeroWorkflowID() models.WorkflowId { return models.WorkflowId{}}