48 lines
1.5 KiB
Go
48 lines
1.5 KiB
Go
|
package pipeline
|
||
|
|
|
||
|
|
import (
|
||
|
|
"context"
|
||
|
|
"testing"
|
||
|
|
)
|
||
|
|
|
||
|
|
func TestMemoryIdempotencyStoreLeaseLifecycle(t *testing.T) {
|
||
|
|
store := NewMemoryIdempotencyStore()
|
||
|
|
ctx := context.Background()
|
||
|
|
|
||
|
|
first, err := store.Acquire(ctx, "key-1")
|
||
|
|
if err != nil || first.State != IdempotencyAcquired || first.Token == "" {
|
||
|
|
t.Fatalf("first lease=%#v err=%v", first, err)
|
||
|
|
}
|
||
|
|
second, err := store.Acquire(ctx, "key-1")
|
||
|
|
if err != nil || second.State != IdempotencyInProgress {
|
||
|
|
t.Fatalf("second lease=%#v err=%v", second, err)
|
||
|
|
}
|
||
|
|
if err := store.Release(ctx, "key-1", "stale-token"); err == nil {
|
||
|
|
t.Fatal("stale release succeeded")
|
||
|
|
}
|
||
|
|
if err := store.Release(ctx, "key-1", first.Token); err != nil {
|
||
|
|
t.Fatalf("release: %v", err)
|
||
|
|
}
|
||
|
|
|
||
|
|
reacquired, err := store.Acquire(ctx, "key-1")
|
||
|
|
if err != nil || reacquired.State != IdempotencyAcquired || reacquired.Token == first.Token {
|
||
|
|
t.Fatalf("reacquired lease=%#v err=%v", reacquired, err)
|
||
|
|
}
|
||
|
|
if err := store.Complete(ctx, "key-1", reacquired.Token); err != nil {
|
||
|
|
t.Fatalf("complete: %v", err)
|
||
|
|
}
|
||
|
|
afterComplete, err := store.Acquire(ctx, "key-1")
|
||
|
|
if err != nil || afterComplete.State != IdempotencyAcquired || afterComplete.Token == reacquired.Token {
|
||
|
|
t.Fatalf("lease after complete=%#v err=%v", afterComplete, err)
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
func TestMemoryIdempotencyStoreHonorsCanceledContext(t *testing.T) {
|
||
|
|
store := NewMemoryIdempotencyStore()
|
||
|
|
ctx, cancel := context.WithCancel(context.Background())
|
||
|
|
cancel()
|
||
|
|
if _, err := store.Acquire(ctx, "key-1"); err == nil {
|
||
|
|
t.Fatal("Acquire accepted canceled context")
|
||
|
|
}
|
||
|
|
}
|