event/worker_pool_test.go at master · wppurking/event · 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
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
package event

import (
	"bytes"
	"fmt"
	"reflect"
	"testing"
	"time"

	"github.com/assembla/cony"
	"github.com/json-iterator/go"
	"github.com/stretchr/testify/assert"
)

type tstCtx struct {
	a int
	bytes.Buffer
}

func (c *tstCtx) record(s string) {
	_, _ = c.WriteString(s)
}

var tstCtxType = reflect.TypeOf(tstCtx{})

func TestWorkerPoolHandlerValidations(t *testing.T) {
	var cases = []struct {
		fn   interface{}
		good bool
	}{
		{func(j *Message) error { return nil }, true},
		{func(c *tstCtx, j *Message) error { return nil }, true},
		{func(c *tstCtx, j *Message) {}, false},
		{func(c *tstCtx, j *Message) string { return "" }, false},
		{func(c *tstCtx, j *Message) (error, string) { return nil, "" }, false},
		{func(c *tstCtx) error { return nil }, false},
		{func(c tstCtx, j *Message) error { return nil }, false},
		{func() error { return nil }, false},
		{func(c *tstCtx, j *Message, wat string) error { return nil }, false},
	}

	for i, testCase := range cases {
		r := isValidHandlerType(tstCtxType, reflect.ValueOf(testCase.fn))
		if testCase.good != r {
			t.Errorf("idx %d: should return %v but returned %v", i, testCase.good, r)
		}
	}
}

func TestWorkerPoolMiddlewareValidations(t *testing.T) {
	var cases = []struct {
		fn   interface{}
		good bool
	}{
		{func(j *Message, n NextMiddlewareFunc) error { return nil }, true},
		{func(c *tstCtx, j *Message, n NextMiddlewareFunc) error { return nil }, true},
		{func(c *tstCtx, j *Message) error { return nil }, false},
		{func(c *tstCtx, j *Message, n NextMiddlewareFunc) {}, false},
		{func(c *tstCtx, j *Message, n NextMiddlewareFunc) string { return "" }, false},
		{func(c *tstCtx, j *Message, n NextMiddlewareFunc) (error, string) { return nil, "" }, false},
		{func(c *tstCtx, n NextMiddlewareFunc) error { return nil }, false},
		{func(c tstCtx, j *Message, n NextMiddlewareFunc) error { return nil }, false},
		{func() error { return nil }, false},
		{func(c *tstCtx, j *Message, wat string) error { return nil }, false},
		{func(c *tstCtx, j *Message, n NextMiddlewareFunc, wat string) error { return nil }, false},
	}

	for i, testCase := range cases {
		r := isValidMiddlewareType(tstCtxType, reflect.ValueOf(testCase.fn))
		if testCase.good != r {
			t.Errorf("idx %d: should return %v but returned %v", i, testCase.good, r)
		}
	}
}

func TestWorkerPoolStartStop(t *testing.T) {
	ns := "work"
	enq := NewPublisher(ns, cony.URL(""))
	wp := NewWorkerPool(TestContext{}, 10, ns, enq, cony.URL(""))
	wp.Start()
	wp.Start()
	wp.Stop()
	wp.Stop()
	wp.Start()
	wp.Stop()
}

func TestWorkerPoolValidations(t *testing.T) {
	ns := "work"
	enq := NewPublisher(ns, cony.URL(""))
	wp := NewWorkerPool(TestContext{}, 10, ns, enq, cony.URL(""))

	func() {
		defer func() {
			if panicErr := recover(); panicErr != nil {
				assert.Regexp(t, "Your middleware function can have one of these signatures", fmt.Sprintf("%v", panicErr))
			} else {
				t.Errorf("expected a panic when using bad middleware")
			}
		}()

		wp.Middleware(TestWorkerPoolValidations)
	}()

	func() {
		defer func() {
			if panicErr := recover(); panicErr != nil {
				assert.Regexp(t, "Your handler function can have one of these signatures", fmt.Sprintf("%v", panicErr))
			} else {
				t.Errorf("expected a panic when using a bad handler")
			}
		}()

		wp.Consumer("wat", TestWorkerPoolValidations)
	}()
}

func TestWorkersPoolRunSingleThreaded(t *testing.T) {
}

func TestWorkerPoolPauseSingleThreadedJobs(t *testing.T) {
}

// Test Helpers
func (t *TestContext) SleepyJob(job *Message) error {
	msg := Q{}
	jsoniter.Unmarshal(job.Body, &msg)
	sleepTime := time.Duration(msg["sleep"].(int))
	time.Sleep(sleepTime * time.Millisecond)
	return nil
}

/*
func setupTestWorkerPool(pool *redis.Pool, namespace, jobName string, concurrency int, jobOpts ConsumerOptions) *WorkerPool {
	deleteQueue(pool, namespace, jobName)
	deleteRetryAndDead(pool, namespace)
	deletePausedAndLockedKeys(namespace, jobName, pool)

	wp := NewWorkerPool(TestContext{}, uint(concurrency), namespace, pool)
	wp.ConsumerWithOptions(jobName, jobOpts, (*TestContext).SleepyJob)
	// reset the backoff times to help with testing
	sleepBackoffsInMilliseconds = []int64{10, 10, 10, 10, 10}
	return wp
}
*/