72 lines
2.2 KiB
Go
72 lines
2.2 KiB
Go
|
package repository
|
||
|
|
|
||
|
|
import (
|
||
|
|
"context"
|
||
|
|
"testing"
|
||
|
|
|
||
|
|
goredis "github.com/redis/go-redis/v9"
|
||
|
|
)
|
||
|
|
|
||
|
|
func TestRedisQueue_DedupeAndSchedulerLock(t *testing.T) {
|
||
|
|
ctx := context.Background()
|
||
|
|
client := goredis.NewClient(&goredis.Options{Addr: "127.0.0.1:6379"})
|
||
|
|
if err := client.Ping(ctx).Err(); err != nil {
|
||
|
|
t.Skip("redis not available:", err)
|
||
|
|
}
|
||
|
|
defer client.Close()
|
||
|
|
|
||
|
|
repo := NewRedisQueueRepository(client)
|
||
|
|
ok, err := repo.TryAcquireDedupe(ctx, "demo", "hash1", "job-1", 60)
|
||
|
|
if err != nil || !ok {
|
||
|
|
t.Fatalf("TryAcquireDedupe() = %v, %v", ok, err)
|
||
|
|
}
|
||
|
|
ok, err = repo.TryAcquireDedupe(ctx, "demo", "hash1", "job-2", 60)
|
||
|
|
if err != nil || ok {
|
||
|
|
t.Fatalf("duplicate TryAcquireDedupe() = %v, %v", ok, err)
|
||
|
|
}
|
||
|
|
if err := repo.ReleaseDedupe(ctx, "demo", "hash1"); err != nil {
|
||
|
|
t.Fatalf("ReleaseDedupe() error = %v", err)
|
||
|
|
}
|
||
|
|
|
||
|
|
ok, err = repo.TrySchedulerLock(ctx, "holder-a", 30)
|
||
|
|
if err != nil || !ok {
|
||
|
|
t.Fatalf("TrySchedulerLock() = %v, %v", ok, err)
|
||
|
|
}
|
||
|
|
ok, err = repo.TrySchedulerLock(ctx, "holder-b", 30)
|
||
|
|
if err != nil || ok {
|
||
|
|
t.Fatalf("duplicate TrySchedulerLock() = %v, %v", ok, err)
|
||
|
|
}
|
||
|
|
_ = repo.ReleaseSchedulerLock(ctx, "holder-a")
|
||
|
|
}
|
||
|
|
|
||
|
|
func TestRedisQueue_LockOwnerGuards(t *testing.T) {
|
||
|
|
ctx := context.Background()
|
||
|
|
client := goredis.NewClient(&goredis.Options{Addr: "127.0.0.1:6379"})
|
||
|
|
if err := client.Ping(ctx).Err(); err != nil {
|
||
|
|
t.Skip("redis not available:", err)
|
||
|
|
}
|
||
|
|
defer client.Close()
|
||
|
|
|
||
|
|
repo := NewRedisQueueRepository(client)
|
||
|
|
jobID := "owner-guard-job"
|
||
|
|
_ = client.Del(ctx, lockKey(jobID))
|
||
|
|
|
||
|
|
ok, err := repo.TryLock(ctx, jobID, "worker-a", 30)
|
||
|
|
if err != nil || !ok {
|
||
|
|
t.Fatalf("TryLock() = %v, %v", ok, err)
|
||
|
|
}
|
||
|
|
if err := repo.ReleaseLock(ctx, jobID, "worker-b"); err != nil {
|
||
|
|
t.Fatalf("ReleaseLock(worker-b) error = %v", err)
|
||
|
|
}
|
||
|
|
if value, err := client.Get(ctx, lockKey(jobID)).Result(); err != nil || value != "worker-a" {
|
||
|
|
t.Fatalf("lock value after wrong release = %q, %v", value, err)
|
||
|
|
}
|
||
|
|
if err := repo.RefreshLock(ctx, jobID, "worker-b", 30); err == nil {
|
||
|
|
t.Fatal("RefreshLock(worker-b) error = nil, want error")
|
||
|
|
}
|
||
|
|
if err := repo.ReleaseLock(ctx, jobID, "worker-a"); err != nil {
|
||
|
|
t.Fatalf("ReleaseLock(worker-a) error = %v", err)
|
||
|
|
}
|
||
|
|
_ = client.Del(ctx, lockKey(jobID))
|
||
|
|
}
|