Mikkel Georgsen
e07ad922eb
- Add PendingOp struct with UUID, type, payload, created_at, attempts - Add WAQ type backed by DragonFlyDB via go-redis v9 - Implement Enqueue (RPUSH), Dequeue (BLPOP), Len (LLEN) - Custom URL parser handles passwords with forward slashes - Unit tests pass; integration test passes against DragonFlyDB at 10.5.0.10:6379
85 lines
1.9 KiB
Go
85 lines
1.9 KiB
Go
package queue_test
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"os"
|
|
"testing"
|
|
"time"
|
|
|
|
"git.georgsen.dk/hwlab/internal/queue"
|
|
)
|
|
|
|
func TestPendingOpJSON(t *testing.T) {
|
|
payload := map[string]string{"device_id": "42", "hw_id": "HW-00001"}
|
|
op, err := queue.NewPendingOp("create_device", payload)
|
|
if err != nil {
|
|
t.Fatalf("NewPendingOp: %v", err)
|
|
}
|
|
if op.ID == "" {
|
|
t.Error("ID should be a UUID")
|
|
}
|
|
if op.Type != "create_device" {
|
|
t.Errorf("Type: want create_device, got %s", op.Type)
|
|
}
|
|
// Round-trip JSON
|
|
data, _ := json.Marshal(op)
|
|
var op2 queue.PendingOp
|
|
if err := json.Unmarshal(data, &op2); err != nil {
|
|
t.Fatalf("unmarshal: %v", err)
|
|
}
|
|
if op2.ID != op.ID {
|
|
t.Errorf("ID mismatch after round-trip: %s != %s", op2.ID, op.ID)
|
|
}
|
|
}
|
|
|
|
func TestNewWAQInvalidURL(t *testing.T) {
|
|
_, err := queue.NewWAQ("not-a-redis-url")
|
|
if err == nil {
|
|
t.Error("expected error for invalid URL")
|
|
}
|
|
}
|
|
|
|
// dragonflyURL returns the DragonFlyDB URL from env, or the default from .env.
|
|
func dragonflyURL(t *testing.T) string {
|
|
t.Helper()
|
|
url := os.Getenv("HWLAB_DRAGONFLY_URL")
|
|
if url == "" {
|
|
url = "redis://:nUq/IfoIQJf/kouckKHRQOk7vV0NwCuI@10.5.0.10:6379"
|
|
}
|
|
return url
|
|
}
|
|
|
|
func TestWAQEnqueueDequeue(t *testing.T) {
|
|
waq, err := queue.NewWAQ(dragonflyURL(t))
|
|
if err != nil {
|
|
t.Skipf("DragonFlyDB unavailable: %v", err)
|
|
}
|
|
defer waq.Close()
|
|
|
|
ctx := context.Background()
|
|
|
|
op, _ := queue.NewPendingOp("test_op", map[string]string{"test": "value"})
|
|
if err := waq.Enqueue(ctx, op); err != nil {
|
|
t.Fatalf("Enqueue: %v", err)
|
|
}
|
|
|
|
n, err := waq.Len(ctx)
|
|
if err != nil {
|
|
t.Fatalf("Len: %v", err)
|
|
}
|
|
if n < 1 {
|
|
t.Error("expected at least 1 item after enqueue")
|
|
}
|
|
|
|
got, err := waq.Dequeue(ctx, 2*time.Second)
|
|
if err != nil {
|
|
t.Fatalf("Dequeue: %v", err)
|
|
}
|
|
if got == nil {
|
|
t.Fatal("expected op from dequeue, got nil")
|
|
}
|
|
if got.ID != op.ID {
|
|
t.Errorf("dequeued op ID mismatch: want %s, got %s", op.ID, got.ID)
|
|
}
|
|
}
|