fix queuer job tests

Author: siherrmann

core/listener.go

@@ -11,13 +11,13 @@ type Listener[T any] struct {
 
 func NewListener[T any]() *Listener[T] {
 	return &Listener[T]{
-		Channel: make(chan T, 1),
+		Channel: make(chan T),
 	}
 }
 
 func (l *Listener[T]) Listen(ctx context.Context, notifyFunction func(data T)) {
+	log.Printf("Context: %v, Channel: %v", ctx, l.Channel)
 	for {
-		log.Printf("Context: %v, Channel: %v", ctx, l.Channel)
 		select {
 		case <-ctx.Done():
 			return
@@ -30,6 +30,10 @@ func (l *Listener[T]) Listen(ctx context.Context, notifyFunction func(data T)) {
 }
 
 func (l *Listener[T]) Notify(data T) {
-	log.Printf("Listener notified with data: %v", data)
-	l.Channel <- data
+	select {
+	case l.Channel <- data:
+		log.Printf("Listener notified with data: %v", data)
+	default:
+		log.Printf("No listener for data: %v", data)
+	}
 }

queuerJob.go

@@ -86,6 +86,7 @@ func (q *Queuer) CancelAllJobsByWorker(workerRid uuid.UUID, entries int) error {
 		return fmt.Errorf("error selecting jobs by worker RID %v: %v", workerRid, err)
 	}
 
+	log.Printf("Cancelling %d jobs for worker with RID %v", len(jobs), workerRid)
 	for _, job := range jobs {
 		err := q.cancelJob(job)
 		if err != nil {
@@ -288,6 +289,7 @@ func (q *Queuer) scheduleJob(job *model.Job) {
 }
 
 func (q *Queuer) cancelJob(job *model.Job) error {
+	log.Printf("Cancelling job with RID %v", job.RID)
 	switch job.Status {
 	case model.JobStatusRunning:
 		jobRunner, found := q.activeRunners.Load(job.RID)

queuerJob_test.go

@@ -65,6 +65,10 @@ func TestAddJob(t *testing.T) {
 
 func TestAddJobRunning(t *testing.T) {
 	testQueuer := newQueuerMock("TestQueuer", 100)
+	testQueuer.AddTask(TaskMock)
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	testQueuer.Start(ctx, cancel)
 
 	t.Run("Successfully runs a job without options", func(t *testing.T) {
 		job, err := testQueuer.AddJob(TaskMock, 1, "2")
@@ -74,17 +78,11 @@ func TestAddJobRunning(t *testing.T) {
 		assert.NoError(t, err, "GetJob should not return an error")
 		assert.NotNil(t, queuedJob, "GetJob should return the job that is currently running")
 
-		// Initialize the queuer and start processing jobs
-		testQueuer.AddTask(TaskMock)
-		ctx, cancel := context.WithCancel(context.Background())
-		defer cancel()
-		testQueuer.Start(ctx, cancel)
-
 		time.Sleep(2 * time.Second)
 
 		jobNotExisting, err := testQueuer.GetJob(job.RID)
-		assert.Error(t, err, "GetJob should return an error for cancelled job")
-		assert.Nil(t, jobNotExisting, "GetJob should return nil for cancelled job")
+		assert.Error(t, err, "GetJob should return an error for ended job")
+		assert.Nil(t, jobNotExisting, "GetJob should return nil for ended job")
 
 		jobArchived, err := testQueuer.dbJob.SelectJobFromArchive(job.RID)
 		assert.NoError(t, err, "SelectJobFromArchive should not return an error for archived job")
@@ -113,24 +111,19 @@ func TestAddJobRunning(t *testing.T) {
 		require.NoError(t, err, "GetJob should not return an error")
 		require.NotNil(t, queuedJob, "GetJob should return the job that is currently running")
 
-		// Initialize the queuer and start processing jobs
-		testQueuer.AddTask(TaskMock)
-		ctx, cancel := context.WithCancel(context.Background())
-		defer cancel()
-		testQueuer.Start(ctx, cancel)
+		time.Sleep(500 * time.Millisecond)
 
-		// Wait for the job to be running (not scheduled because of less than 1 minute delay)
-		time.Sleep(1 * time.Second)
 		jobScheduled, err := testQueuer.GetJob(job.RID)
 		assert.NoError(t, err, "GetJob should not return an error for running job")
-		assert.NotNil(t, jobScheduled, "GetJob should return the job that is currently running")
+		require.NotNil(t, jobScheduled, "GetJob should return the job that is currently running")
 		assert.Equal(t, model.JobStatusRunning, jobScheduled.Status, "Job should be in Scheduled status")
 
 		// Wait for schedule time and job execution
 		time.Sleep(15*time.Second + maxDeviation)
+
 		jobNotExisting, err := testQueuer.GetJob(job.RID)
-		assert.Error(t, err, "GetJob should return an error for cancelled job")
-		assert.Nil(t, jobNotExisting, "GetJob should return nil for cancelled job")
+		assert.Error(t, err, "GetJob should return an error for ended job")
+		assert.Nil(t, jobNotExisting, "GetJob should return nil for ended job")
 
 		jobArchived, err := testQueuer.dbJob.SelectJobFromArchive(job.RID)
 		assert.NoError(t, err, "SelectJobFromArchive should not return an error for archived job")
@@ -274,7 +267,13 @@ func TestCancelJob(t *testing.T) {
 }
 
 func TestCancelJobRunning(t *testing.T) {
+	// Only works with a running queuer because the worker needs to process jobs
+	// to be able to cancel them.
 	testQueuer := newQueuerMock("TestQueuer", 100)
+	testQueuer.AddTask(TaskMock)
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	testQueuer.Start(ctx, cancel)
 
 	t.Run("Successfully cancels a running job", func(t *testing.T) {
 		job, err := testQueuer.AddJob(TaskMock, 3, "2")
@@ -284,12 +283,6 @@ func TestCancelJobRunning(t *testing.T) {
 		assert.NoError(t, err, "GetJob should not return an error")
 		assert.NotNil(t, queuedJob, "GetJob should return the job that is currently running")
 
-		// Initialize the queuer and start processing jobs
-		testQueuer.AddTask(TaskMock)
-		ctx, cancel := context.WithCancel(context.Background())
-		defer cancel()
-		testQueuer.Start(ctx, cancel)
-
 		time.Sleep(1 * time.Second)
 
 		cancelledJob, err := testQueuer.CancelJob(job.RID)
@@ -311,32 +304,27 @@ func TestCancelAllJobsByWorkerRunning(t *testing.T) {
 	// Only works with a running queuer because the worker needs to process jobs
 	// to be able to cancel them.
 	testQueuer := newQueuerMock("TestQueuer", 100)
+	testQueuer.AddTask(TaskMock)
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	testQueuer.Start(ctx, cancel)
 
 	t.Run("Successfully cancels all jobs by worker RID", func(t *testing.T) {
-		worker, err := testQueuer.GetWorker(testQueuer.worker.RID)
-		require.NoError(t, err, "GetWorker should not return an error on success")
-
 		job1, err := testQueuer.AddJob(TaskMock, 10, "2")
 		require.NoError(t, err, "AddJob should not return an error on success")
-		job1.WorkerRID = worker.RID
 
 		job2, err := testQueuer.AddJob(TaskMock, 10, "4")
 		require.NoError(t, err, "AddJob should not return an error on success")
-		job2.WorkerRID = worker.RID
-
-		// Initialize the queuer and start processing jobs
-		testQueuer.AddTask(TaskMock)
-		ctx, cancel := context.WithCancel(context.Background())
-		defer cancel()
-		testQueuer.Start(ctx, cancel)
 
 		time.Sleep(1 * time.Second)
 
-		jobs, err := testQueuer.dbJob.SelectAllJobsByWorkerRID(worker.RID, 0, 10)
+		jobs, err := testQueuer.dbJob.SelectAllJobsByWorkerRID(testQueuer.worker.RID, 0, 10)
 		assert.NoError(t, err, "SelectAllJobsByWorkerRID should not return an error")
-		assert.Len(t, jobs, 2, "There should be two jobs for the worker")
+		require.Len(t, jobs, 2, "There should be two jobs for the worker")
+		assert.Equal(t, model.JobStatusRunning, jobs[0].Status, "Job1 should be in Running status")
+		assert.Equal(t, model.JobStatusRunning, jobs[1].Status, "Job2 should be in Running status")
 
-		err = testQueuer.CancelAllJobsByWorker(worker.RID, 10)
+		err = testQueuer.CancelAllJobsByWorker(testQueuer.worker.RID, 10)
 		assert.NoError(t, err, "CancelAllJobsByWorker should not return an error on success")
 
 		jobs, err = testQueuer.GetJobs(0, 10)

queuerListener_test.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"queuer/model"
 	"testing"
+	"time"
 
 	"github.com/google/uuid"
 	"github.com/stretchr/testify/assert"
@@ -12,52 +13,52 @@ import (
 
 func TestListenForJobUpdate(t *testing.T) {
 	q := newQueuerMock("testQueuer", 1)
-	require.NotNil(t, q, "expected queuer to be initialized")
-	require.NotNil(t, q.jobUpdateListener, "expected jobUpdateListener to be initialized")
-
-	// Start the queuer
 	ctx, cancel := context.WithCancel(context.Background())
 	q.Start(ctx, cancel)
 
-	// Use a channel to capture the notification
 	data := &model.Job{RID: uuid.New()}
 	notifyChannel := make(chan *model.Job)
 	err := q.ListenForJobUpdate(func(d *model.Job) {
 		notifyChannel <- d
 	})
 	require.NoError(t, err, "expected to successfully listen for job updates")
 
+	time.Sleep(1 * time.Second)
+
 	// Notify the listener manually
 	q.jobUpdateListener.Notify(data)
 
-	// Check if the data was received
-	receivedData := <-notifyChannel
-	assert.NotNil(t, receivedData, "expected to receive job data")
-	assert.Equal(t, data.RID, receivedData.RID, "expected to receive the same job RID")
+	select {
+	case receivedData := <-notifyChannel:
+		assert.NotNil(t, receivedData, "expected to receive job data")
+		assert.Equal(t, data.RID, receivedData.RID, "expected to receive the same job RID")
+	default:
+		t.Fatal("expected to receive job data, but got nothing")
+	}
 }
 
 func TestListenForJobDelete(t *testing.T) {
 	q := newQueuerMock("testQueuer", 1)
-	require.NotNil(t, q, "expected queuer to be initialized")
-	require.NotNil(t, q.jobUpdateListener, "expected jobUpdateListener to be initialized")
-
-	// Start the queuer
 	ctx, cancel := context.WithCancel(context.Background())
 	q.Start(ctx, cancel)
 
-	// Use a channel to capture the notification
 	data := &model.Job{RID: uuid.New()}
 	notifyChannel := make(chan *model.Job)
 	err := q.ListenForJobDelete(func(d *model.Job) {
 		notifyChannel <- d
 	})
 	require.NoError(t, err, "expected to successfully listen for job deletions")
 
+	time.Sleep(1 * time.Second)
+
 	// Notify the listener manually
 	q.jobDeleteListener.Notify(data)
 
-	// Check if the data was received
-	receivedData := <-notifyChannel
-	assert.NotNil(t, receivedData, "expected to receive job data")
-	assert.Equal(t, data.RID, receivedData.RID, "expected to receive the same job RID")
+	select {
+	case receivedData := <-notifyChannel:
+		assert.NotNil(t, receivedData, "expected to receive job data")
+		assert.Equal(t, data.RID, receivedData.RID, "expected to receive the same job RID")
+	default:
+		t.Fatal("expected to receive job data, but got nothing")
+	}
 }