From 02123df8cc15896f97c1580e49d8e2a991e37ddd Mon Sep 17 00:00:00 2001 From: sangtran Date: Thu, 3 Sep 2026 23:10:33 +0700 Subject: Moving all legacy quacker source code from github to cgit --- workerpool/workerpool_test.go | 608 ++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 608 insertions(+) create mode 100644 workerpool/workerpool_test.go (limited to 'workerpool/workerpool_test.go') diff --git a/workerpool/workerpool_test.go b/workerpool/workerpool_test.go new file mode 100644 index 0000000..851874b --- /dev/null +++ b/workerpool/workerpool_test.go @@ -0,0 +1,608 @@ +package workerpool + +import ( + "context" + "errors" + "fmt" + "sync" + "testing" + "time" +) + +type mockTask struct { + id string + fn func(ctx context.Context) error +} + +func (t *mockTask) Execute(ctx context.Context) error { + if t.fn != nil { + return t.fn(ctx) + } + return nil +} + +func (t *mockTask) GetID() string { + return t.id +} + +type taskResult struct { + taskID string + err error +} + +type mockObserver struct { + mu sync.Mutex + workerStarts []int + workerStops []int + tasksDone []taskResult + t *testing.T +} + +func (o *mockObserver) OnWorkerStart(workerID int) { + o.mu.Lock() + defer o.mu.Unlock() + o.workerStarts = append(o.workerStarts, workerID) + if o.t != nil { + o.t.Logf("observer: worker %d started", workerID) + } +} + +func (o *mockObserver) OnWorkerStop(workerID int) { + o.mu.Lock() + defer o.mu.Unlock() + o.workerStops = append(o.workerStops, workerID) + if o.t != nil { + o.t.Logf("observer: worker %d stopped", workerID) + } +} + +func (o *mockObserver) OnTaskDone(workerID int, taskID string, err error) { + o.mu.Lock() + defer o.mu.Unlock() + o.tasksDone = append(o.tasksDone, taskResult{taskID: taskID, err: err}) + if o.t != nil { + if err != nil { + o.t.Logf("observer: worker %d task %s failed: %v", workerID, taskID, err) + } else { + o.t.Logf("observer: worker %d task %s succeeded", workerID, taskID) + } + } +} + +func TestNewWorkerPool(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + opts []Option + shouldErr bool + }{ + { + name: "valid config with defaults", + opts: []Option{}, + shouldErr: false, + }, + { + name: "valid config with all options", + opts: []Option{ + WithName("test-pool"), + WithNumWorkers(4), + WithTaskQueueSize(100), + }, + shouldErr: false, + }, + { + name: "invalid number of workers (zero)", + opts: []Option{WithNumWorkers(0)}, + shouldErr: true, + }, + { + name: "invalid number of workers (negative)", + opts: []Option{WithNumWorkers(-1)}, + shouldErr: true, + }, + { + name: "invalid task queue size (zero)", + opts: []Option{WithTaskQueueSize(0)}, + shouldErr: true, + }, + { + name: "invalid task queue size (negative)", + opts: []Option{WithTaskQueueSize(-1)}, + shouldErr: true, + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + t.Parallel() + _, err := NewWorkerPool[*mockTask](t.Context(), test.opts...) + if (err != nil) != test.shouldErr { + t.Errorf("NewWorkerPool() error = %v, shouldErr %v", err, test.shouldErr) + } + }) + } +} + +func TestWorkerPool_Observer(t *testing.T) { + t.Parallel() + ctx := t.Context() + + obs := &mockObserver{t: t} + numWorkers := 2 + taskCount := 5 + + pool, err := NewWorkerPool[*mockTask](ctx, + WithName("observer-test"), + WithNumWorkers(numWorkers), + WithTaskQueueSize(10), + WithObserver(obs), + ) + if err != nil { + t.Fatalf("failed to create pool: %v", err) + } + + pool.Start() + + for i := range taskCount { + task := &mockTask{id: fmt.Sprintf("task-%d", i)} + if err := pool.Push(task); err != nil { + t.Errorf("failed to push task: %v", err) + } + } + + pool.StopAndWait() + + obs.mu.Lock() + defer obs.mu.Unlock() + + if len(obs.workerStarts) != numWorkers { + t.Errorf("expected %d worker starts, got %d", numWorkers, len(obs.workerStarts)) + } + if len(obs.workerStops) != numWorkers { + t.Errorf("expected %d worker stops, got %d", numWorkers, len(obs.workerStops)) + } + + if len(obs.tasksDone) != taskCount { + t.Errorf("expected %d tasks done, got %d", taskCount, len(obs.tasksDone)) + } + for _, r := range obs.tasksDone { + if r.err != nil { + t.Errorf("task %s should have succeeded, got error: %v", r.taskID, r.err) + } + } +} + +func TestWorkerPool_ObserverTaskFailed(t *testing.T) { + t.Parallel() + ctx := t.Context() + + obs := &mockObserver{t: t} + + pool, err := NewWorkerPool[*mockTask](ctx, + WithName("observer-fail-test"), + WithNumWorkers(1), + WithTaskQueueSize(10), + WithObserver(obs), + ) + if err != nil { + t.Fatalf("failed to create pool: %v", err) + } + + pool.Start() + + task := &mockTask{ + id: "fail-task", + fn: func(ctx context.Context) error { + return fmt.Errorf("intentional error") + }, + } + pool.Push(task) + + pool.StopAndWait() + + obs.mu.Lock() + defer obs.mu.Unlock() + + failures := 0 + for _, r := range obs.tasksDone { + if r.err != nil { + failures++ + } + } + + if failures != 1 { + t.Errorf("expected 1 failed task, got %d", failures) + } + if len(obs.tasksDone) > 0 && obs.tasksDone[0].taskID != "fail-task" { + t.Errorf("expected failed task id 'fail-task', got %s", obs.tasksDone[0].taskID) + } +} + +func TestWorkerPool_ObserverTaskPanic(t *testing.T) { + t.Parallel() + ctx := t.Context() + + panicCaught := make(chan string, 1) + + pool, err := NewWorkerPool[*mockTask](ctx, + WithName("observer-panic-test"), + WithNumWorkers(1), + WithTaskQueueSize(10), + WithPanicHandler(func(workerID int, taskID string, panicValue any) { + panicCaught <- taskID + }), + ) + if err != nil { + t.Fatalf("failed to create pool: %v", err) + } + + pool.Start() + + task := &mockTask{ + id: "panic-task", + fn: func(ctx context.Context) error { + panic("intentional panic") + }, + } + pool.Push(task) + + pool.StopAndWait() + + select { + case taskID := <-panicCaught: + if taskID != "panic-task" { + t.Errorf("expected panic from 'panic-task', got %s", taskID) + } + default: + t.Error("expected PanicHandler to be called") + } +} + +func TestWorkerPool_ConcurrentPush(t *testing.T) { + t.Parallel() + ctx := t.Context() + var wg sync.WaitGroup + taskCount := 20 + + pool, err := NewWorkerPool[*mockTask](ctx, + WithName("concurrent-push"), + WithNumWorkers(3), + WithTaskQueueSize(taskCount), + ) + if err != nil { + t.Fatalf("failed to create pool: %v", err) + } + + pool.Start() + defer pool.StopAndWait() + + for i := range taskCount { + wg.Add(1) + go func(id int) { + defer wg.Done() + task := &mockTask{id: fmt.Sprintf("task-%d", id)} + if err := pool.Push(task); err != nil { + t.Errorf("push failed: %v", err) + } + }(i) + } + + done := make(chan struct{}) + go func() { + wg.Wait() + close(done) + }() + + select { + case <-done: + case <-time.After(5 * time.Second): + t.Fatal("timeout waiting for concurrent push") + } +} + +func TestWorkerPool_TaskQueueFull(t *testing.T) { + t.Parallel() + ctx := t.Context() + + pool, err := NewWorkerPool[*mockTask](ctx, + WithName("queue-full"), + WithNumWorkers(1), + WithTaskQueueSize(2), + ) + if err != nil { + t.Fatalf("failed to create pool: %v", err) + } + + for i := range 2 { + task := &mockTask{id: fmt.Sprintf("task-%d", i)} + if err := pool.Push(task); err != nil { + t.Errorf("unexpected error on push %d: %v", i, err) + } + } + + task := &mockTask{id: "overflow-task"} + err = pool.Push(task) + if err == nil { + t.Fatal("expected error when queue is full, got nil") + } + if !errors.Is(err, ErrQueueFull) { + t.Fatalf("expected ErrQueueFull, got: %v", err) + } +} + +func TestWorkerPool_MultipleStop(t *testing.T) { + t.Parallel() + ctx := t.Context() + + pool, err := NewWorkerPool[*mockTask](ctx, + WithName("multiple-stop"), + WithNumWorkers(2), + WithTaskQueueSize(10), + ) + if err != nil { + t.Fatalf("failed to create pool: %v", err) + } + + pool.Start() + pool.Stop() + + func() { + defer func() { + if r := recover(); r != nil { + t.Errorf("Stop() panicked on second call: %v", r) + } + }() + pool.Stop() + }() + + pool.Wait() +} + +func TestWorkerPool_MetricsThreadSafe(t *testing.T) { + t.Parallel() + ctx := t.Context() + var wg sync.WaitGroup + taskCount := 20 + + pool, err := NewWorkerPool[*mockTask](ctx, + WithName("metrics-safe"), + WithNumWorkers(10), + WithTaskQueueSize(taskCount), + ) + if err != nil { + t.Fatalf("failed to create pool: %v", err) + } + + pool.Start() + defer pool.StopAndWait() + + for i := range taskCount { + wg.Add(1) + go func(id int) { + defer wg.Done() + task := &mockTask{ + id: fmt.Sprintf("task-%d", id), + fn: func(ctx context.Context) error { + time.Sleep(time.Millisecond) + return nil + }, + } + pool.Push(task) + }(i) + } + + concurrentReads := 100 + for range concurrentReads { + wg.Add(1) + go func() { + defer wg.Done() + for range 10 { + pool.GetMetrics() + } + }() + } + + wg.Wait() +} + +func TestWorkerPool_ContextCancellation(t *testing.T) { + t.Parallel() + ctx, cancel := context.WithCancel(t.Context()) + + pool, err := NewWorkerPool[*mockTask](ctx, + WithName("context-cancel"), + WithNumWorkers(2), + WithTaskQueueSize(10), + ) + if err != nil { + t.Fatalf("failed to create pool: %v", err) + } + + pool.Start() + + for i := range 5 { + task := &mockTask{id: fmt.Sprintf("task-%d", i)} + pool.Push(task) + } + + cancel() + + err = pool.Wait() + if err != nil && !errors.Is(err, context.Canceled) { + t.Errorf("expected context.Canceled, got: %v", err) + } +} + +func TestWorkerPool_ErrorHandler(t *testing.T) { + t.Parallel() + ctx := t.Context() + + errorHandlerCalled := make(chan struct{}) + + pool, err := NewWorkerPool[*mockTask](ctx, + WithName("error-handler-test"), + WithNumWorkers(1), + WithTaskQueueSize(10), + WithErrorHandler(func(workerID int, taskID string, err error) { + t.Logf("ErrorHandler called: worker=%d, task=%s, err=%v", workerID, taskID, err) + close(errorHandlerCalled) + }), + ) + if err != nil { + t.Fatalf("failed to create pool: %v", err) + } + + pool.Start() + + task := &mockTask{ + id: "error-task", + fn: func(ctx context.Context) error { + return fmt.Errorf("task error") + }, + } + pool.Push(task) + + select { + case <-errorHandlerCalled: + case <-time.After(time.Second): + t.Fatal("ErrorHandler was not called within timeout") + } + + pool.StopAndWait() +} + +func TestWorkerPool_PanicHandler(t *testing.T) { + t.Parallel() + ctx := t.Context() + + panicHandlerCalled := make(chan struct{}) + + pool, err := NewWorkerPool[*mockTask](ctx, + WithName("panic-handler-test"), + WithNumWorkers(1), + WithTaskQueueSize(10), + WithPanicHandler(func(workerID int, taskID string, panicValue any) { + t.Logf("PanicHandler called: worker=%d, task=%s, panic=%v", workerID, taskID, panicValue) + close(panicHandlerCalled) + }), + ) + if err != nil { + t.Fatalf("failed to create pool: %v", err) + } + + pool.Start() + + task := &mockTask{ + id: "panic-task", + fn: func(ctx context.Context) error { + panic("task panic") + }, + } + pool.Push(task) + + select { + case <-panicHandlerCalled: + case <-time.After(time.Second): + t.Fatal("PanicHandler was not called within timeout") + } + + pool.StopAndWait() +} + +func TestWorkerPool_PushContextCancelled(t *testing.T) { + t.Parallel() + ctx, cancel := context.WithCancel(t.Context()) + + pool, err := NewWorkerPool[*mockTask](ctx, + WithName("push-ctx-cancelled"), + WithNumWorkers(1), + WithTaskQueueSize(1), + ) + if err != nil { + t.Fatalf("failed to create pool: %v", err) + } + + task1 := &mockTask{id: "task-1"} + if err := pool.Push(task1); err != nil { + t.Fatalf("failed to fill queue: %v", err) + } + + cancel() + + task2 := &mockTask{id: "task-2"} + err = pool.Push(task2) + + if err == nil { + t.Fatal("expected error when pushing to full queue with cancelled context") + } + + if !errors.Is(err, ErrQueueFull) && !errors.Is(err, context.Canceled) { + t.Fatalf("expected ErrQueueFull or context.Canceled, got: %v", err) + } +} + +func TestWorkerPool_NoObserver(t *testing.T) { + t.Parallel() + ctx := t.Context() + + pool, err := NewWorkerPool[*mockTask](ctx, + WithName("no-observer"), + WithNumWorkers(2), + WithTaskQueueSize(10), + ) + if err != nil { + t.Fatalf("failed to create pool: %v", err) + } + + pool.Start() + + task1 := &mockTask{id: "success-task"} + pool.Push(task1) + + task2 := &mockTask{ + id: "fail-task", + fn: func(ctx context.Context) error { + return fmt.Errorf("error") + }, + } + pool.Push(task2) + + task3 := &mockTask{ + id: "panic-task", + fn: func(ctx context.Context) error { + panic("panic") + }, + } + pool.Push(task3) + + pool.StopAndWait() + + metrics := pool.GetMetrics() + if metrics.TasksFailed != 2 { + t.Errorf("expected 2 failed tasks, got %d", metrics.TasksFailed) + } +} + +func TestWorkerPool_StartMultipleTimes(t *testing.T) { + t.Parallel() + + pool, err := NewWorkerPool[*mockTask](t.Context(), + WithNumWorkers(1), + WithTaskQueueSize(1), + ) + if err != nil { + t.Fatalf("failed to create pool: %v", err) + } + + pool.Start() + defer pool.StopAndWait() + + defer func() { + if r := recover(); r == nil { + t.Error("expected panic when calling Start() multiple times") + } + }() + + pool.Start() +} -- cgit v1.3.1