queuer/queuerListener_test.go at main · siherrmann/queuer · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
package queuer

import (
	"context"
	"testing"
	"time"

	"github.com/google/uuid"
	"github.com/siherrmann/queuer/helper"
	"github.com/siherrmann/queuer/model"
	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"
)

func TestListenForJobUpdate(t *testing.T) {
	helper.SetTestDatabaseConfigEnvs(t, dbPort)
	q := NewQueuer("testQueuer", 1)
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	q.Start(ctx, cancel)

	data := &model.Job{RID: uuid.New()}
	notifyChannel := make(chan *model.Job, 1)
	err := q.ListenForJobUpdate(func(d *model.Job) {
		notifyChannel <- d
	})
	require.NoError(t, err, "expected to successfully listen for job updates")

	for i := 0; i < 100; i++ {
		// Notify the listener manually
		q.jobUpdateListener.Notify(data)

		q.jobUpdateListener.WaitForNotificationsProcessed()

		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")
		case <-time.After(1 * time.Second):
			t.Error("timed out after 1s waiting for job data")
		}
	}
}

func TestListenForJobDelete(t *testing.T) {
	helper.SetTestDatabaseConfigEnvs(t, dbPort)
	q := NewQueuer("testQueuer", 1)
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	q.Start(ctx, cancel)

	data := &model.Job{RID: uuid.New()}
	notifyChannel := make(chan *model.Job, 1)
	err := q.ListenForJobDelete(func(d *model.Job) {
		notifyChannel <- d
	})
	require.NoError(t, err, "expected to successfully listen for job deletions")

	for i := 0; i < 100; i++ {
		// Notify the listener manually
		q.jobDeleteListener.Notify(data)

		q.jobDeleteListener.WaitForNotificationsProcessed()

		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")
		case <-time.After(1 * time.Second):
			t.Error("timed out after 5s waiting for job data")
		}
	}
}