/
engine_test.go
169 lines (140 loc) · 3.96 KB
/
engine_test.go
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
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
package promql
import (
"sync"
"testing"
"time"
"golang.org/x/net/context"
)
var noop = testStmt(func(context.Context) error {
return nil
})
func TestQueryConcurreny(t *testing.T) {
engine := NewEngine(nil)
defer engine.Stop()
block := make(chan struct{})
processing := make(chan struct{})
f1 := testStmt(func(context.Context) error {
processing <- struct{}{}
<-block
return nil
})
for i := 0; i < *maxConcurrentQueries; i++ {
q := engine.newTestQuery(f1)
go q.Exec()
select {
case <-processing:
// Expected.
case <-time.After(5 * time.Millisecond):
t.Fatalf("Query within concurrency threshold not being executed")
}
}
q := engine.newTestQuery(f1)
go q.Exec()
select {
case <-processing:
t.Fatalf("Query above concurrency threhosld being executed")
case <-time.After(5 * time.Millisecond):
// Expected.
}
// Terminate a running query.
block <- struct{}{}
select {
case <-processing:
// Expected.
case <-time.After(5 * time.Millisecond):
t.Fatalf("Query within concurrency threshold not being executed")
}
// Terminate remaining queries.
for i := 0; i < *maxConcurrentQueries; i++ {
block <- struct{}{}
}
}
func TestQueryTimeout(t *testing.T) {
*defaultQueryTimeout = 5 * time.Millisecond
defer func() {
// Restore default query timeout
*defaultQueryTimeout = 2 * time.Minute
}()
engine := NewEngine(nil)
defer engine.Stop()
f1 := testStmt(func(context.Context) error {
time.Sleep(10 * time.Millisecond)
return nil
})
// Timeouts are not exact but checked in designated places. For example between
// invoking test statements.
query := engine.newTestQuery(f1, f1)
res := query.Exec()
if res.Err == nil {
t.Fatalf("expected timeout error but got none")
}
if _, ok := res.Err.(ErrQueryTimeout); res.Err != nil && !ok {
t.Fatalf("expected timeout error but got: %s", res.Err)
}
}
func TestQueryCancel(t *testing.T) {
engine := NewEngine(nil)
defer engine.Stop()
// As for timeouts, cancellation is only checked at designated points. We ensure
// that we reach one of those points using the same method.
f1 := testStmt(func(context.Context) error {
time.Sleep(2 * time.Millisecond)
return nil
})
query1 := engine.newTestQuery(f1, f1)
query2 := engine.newTestQuery(f1, f1)
// Cancel query after starting it.
var wg sync.WaitGroup
var res *Result
wg.Add(1)
go func() {
res = query1.Exec()
wg.Done()
}()
time.Sleep(1 * time.Millisecond)
query1.Cancel()
wg.Wait()
if res.Err == nil {
t.Fatalf("expected cancellation error for query1 but got none")
}
if _, ok := res.Err.(ErrQueryCanceled); res.Err != nil && !ok {
t.Fatalf("expected cancellation error for query1 but got: %s", res.Err)
}
// Canceling query before starting it must have no effect.
query2.Cancel()
res = query2.Exec()
if res.Err != nil {
t.Fatalf("unexpeceted error on executing query2: %s", res.Err)
}
}
func TestEngineShutdown(t *testing.T) {
engine := NewEngine(nil)
handlerExecutions := 0
// Shutdown engine on first handler execution. Should handler execution ever become
// concurrent this test has to be adjusted accordingly.
f1 := testStmt(func(context.Context) error {
handlerExecutions++
engine.Stop()
time.Sleep(10 * time.Millisecond)
return nil
})
query1 := engine.newTestQuery(f1, f1)
query2 := engine.newTestQuery(f1, f1)
// Stopping the engine must cancel the base context. While executing queries is
// still possible, their context is canceled from the beginning and execution should
// terminate immediately.
res := query1.Exec()
if res.Err == nil {
t.Fatalf("expected error on shutdown during query but got none")
}
if handlerExecutions != 1 {
t.Fatalf("expected only one handler to be executed before query cancellation but got %d executions", handlerExecutions)
}
res2 := query2.Exec()
if res2.Err == nil {
t.Fatalf("expected error on querying shutdown engine but got none")
}
if handlerExecutions != 1 {
t.Fatalf("expected no handler execution for query after engine shutdown")
}
}