From 16d78135c7e98d09820cb4c9903b13de017de8dc Mon Sep 17 00:00:00 2001
From: Hao Zhou <zhuhz@amazon.com>
Date: Wed, 3 Jan 2024 19:25:11 +0000
Subject: [PATCH] add data race detection in test

---
 Makefile                  |  2 +-
 pkg/worker/worker_test.go | 21 ++++++++++++++++++---
 2 files changed, 19 insertions(+), 4 deletions(-)

diff --git a/Makefile b/Makefile
index 54f28989..281e23c0 100644
--- a/Makefile
+++ b/Makefile
@@ -38,7 +38,7 @@ verify:
 
 ## Run unit tests
 test: verify
-	go test ./pkg/... ./controllers/... ./webhooks/... -coverprofile cover.out
+	go test -race ./pkg/... ./controllers/... ./webhooks/... -coverprofile cover.out
 
 test-e2e:
 	KUBE_CONFIG_PATH=${KUBE_CONFIG_PATH} REGION=${AWS_REGION} CLUSTER_NAME=${CLUSTER_NAME} ./scripts/test/run-integration-tests.sh
diff --git a/pkg/worker/worker_test.go b/pkg/worker/worker_test.go
index 1d1c84d6..c72e303e 100644
--- a/pkg/worker/worker_test.go
+++ b/pkg/worker/worker_test.go
@@ -16,6 +16,7 @@ package worker
 import (
 	"context"
 	"fmt"
+	"sync"
 	"testing"
 	"time"
 
@@ -34,15 +35,18 @@ var (
 	maxRequeue                      = 3
 )
 
+var mu sync.RWMutex
+
 func GetMockWorkerPool(ctx context.Context) Worker {
 	log := zap.New(zap.UseDevMode(true)).WithValues("worker resource Id", resourceName)
 	return NewDefaultWorkerPool(resourceName, workerCount, maxRequeue, log, ctx)
 }
 
 func MockWorkerFunc(job interface{}) (result ctrl.Result, err error) {
+	mu.Lock()
+	defer mu.Unlock()
 	v := job.(*int)
 	*v++
-	time.Sleep(time.Millisecond * mockTimeToProcessWorkerFunc)
 
 	return ctrl.Result{}, nil
 }
@@ -75,8 +79,11 @@ func TestWorker_SubmitJob(t *testing.T) {
 	time.Sleep(time.Millisecond * (mockTimeToProcessWorkerFunc + bufferTimeBwWorkerFuncExecution) * time.Duration(jobCount))
 
 	// Verify job completed.
-	assert.Equal(t, job1, 1)
-	assert.Equal(t, job2, 1)
+	mu.RLock()
+	defer mu.RUnlock()
+	for _, j := range []int{job1, job2} {
+		assert.Equal(t, j, 1)
+	}
 }
 
 func TestWorker_SubmitJob_RequeueOnError(t *testing.T) {
@@ -84,6 +91,8 @@ func TestWorker_SubmitJob_RequeueOnError(t *testing.T) {
 	defer cancel()
 
 	workerFunc := func(job interface{}) (result ctrl.Result, err error) {
+		mu.Lock()
+		defer mu.Unlock()
 		invoked := job.(*int)
 		*invoked++
 
@@ -100,7 +109,9 @@ func TestWorker_SubmitJob_RequeueOnError(t *testing.T) {
 	time.Sleep((mockTimeToProcessWorkerFunc + bufferTimeBwWorkerFuncExecution) * time.Millisecond * time.Duration(maxRequeue))
 
 	// expected invocation = max requeue + the first invocation
+	mu.RLock()
 	assert.Equal(t, maxRequeue+1, invoked)
+	mu.RUnlock()
 }
 
 func TestWorker_SubmitJob_NotRequeueOnError(t *testing.T) {
@@ -108,6 +119,8 @@ func TestWorker_SubmitJob_NotRequeueOnError(t *testing.T) {
 	defer cancel()
 
 	workerFunc := func(job interface{}) (result ctrl.Result, err error) {
+		mu.Lock()
+		defer mu.Unlock()
 		invoked := job.(*int)
 		*invoked++
 
@@ -127,5 +140,7 @@ func TestWorker_SubmitJob_NotRequeueOnError(t *testing.T) {
 	actualInqueue := 1
 	// invoked should be only incremented once
 	assert.NotEqual(t, maxRequeue, actualInqueue)
+	mu.RLock()
 	assert.Equal(t, actualInqueue, invoked)
+	mu.RUnlock()
 }