Vihren Atlas

Atlas › Test

TestDoesntGrowPastMax

go.temporal.io/server/common/goro/TestAdaptivePoolSuite/TestDoesntGrowPastMax

Package
go.temporal.io/server/common/goro
Suite / test hierarchy
TestAdaptivePoolSuite/TestDoesntGrowPastMax
Test
TestDoesntGrowPastMax
Introduced at
TestDoesntGrowPastMax Test frontier
Covered ranges
43
Covered lines
144
Covered files
6

Covered source

Expand a file to inspect source; the > gutter marks covered lines.

go.temporal.io/server/common/testing/parallelsuite/suite.go 84 covered LOC · 22 ranges

Open complete file

74 //
75 //nolint:revive // ctx is last so callers can pass nil to mean "no override"; SA1012 forbids passing nil as the first ctx arg.
76 > func (s *Suite[T]) copySuite(t *testing.T, parallel bool, assertT require.TestingT, ctx context.Context) testingSuite { suite.go
77 > cp := reflect.New(reflect.TypeFor[T]().Elem()).Interface().(T)
78 > cp.initSuite(t, parallel, assertT, ctx)
79 > return cp
80 > }
81
82 //nolint:revive // see copySuite above.
83 > func (s *Suite[T]) initSuite(t *testing.T, parallel bool, assertT require.TestingT, ctx context.Context) { suite.go
84 > g := &s.guardT
85 > g.name = t.Name()
86 > g.T = t
87 > g.hasSubtests.Store(false)
88 > s.runParallel = parallel
89 > s.ctx = ctx
90 > s.ctxOnce = sync.Once{}
91 > if s.runParallel {
92 > t.Parallel() //nolint:testifylint // parallelsuite intentionally supports parallel tests
93 > }
94 > if assertT == nil {
95 > assertT = g
96 > }
97 > s.assertT = assertT
98 > s.Assertions = require.New(assertT)
99 > s.ProtoAssertions = protorequire.New(assertT)
100 > s.HistoryRequire = historyrequire.New(assertT)
101 }
102
177 //
178 // The suite must embed [Suite] and have no other fields.
179 > func Run[T testingSuite](t *testing.T, s T, args ...any) { suite.go
180 > run(t, s, true, args...)
181 > }
182
183 // RunLegacySequential behaves like [Run] but does not mark test methods as parallel.
189 }
190
191 > func run[T testingSuite](t *testing.T, s T, methodsParallel bool, args ...any) { suite.go
192 > t.Helper()
193 >
194 > typ := reflect.TypeFor[T]()
195 > if typ.Kind() != reflect.Pointer || typ.Elem().Kind() != reflect.Struct {
196 panic(fmt.Sprintf("parallelsuite.Run: suite must be a pointer to a struct, got %v", typ))
197 }
198 > structType := typ.Elem() suite.go
199 >
200 > validateSuiteStruct(structType)
201 >
202 > methods := discoverTestMethods(typ, structType, args)
203 > if len(methods) == 0 {
204 panic(fmt.Sprintf("parallelsuite.Run: suite %s has no Test* methods", structType.Name()))
205 }
206
207 > methods = applyTestifyMFilter(methods) suite.go
208 > if len(methods) == 0 {
209 return // all methods filtered by -testify.m; nothing to run
210 }
211
212 > argVals := make([]reflect.Value, len(args)) suite.go
213 > for i, a := range args {
214 argVals[i] = reflect.ValueOf(a)
215 }
216
217 > s.initSuite(t, true, nil, nil) suite.go
218 >
219 > for _, method := range methods {
220 > t.Run(method.Name, func(t *testing.T) {
221 > cpS := s.copySuite(t, methodsParallel, nil, nil)
222 > callArgs := append([]reflect.Value{reflect.ValueOf(cpS)}, argVals...)
223 > method.Func.Call(callArgs)
224 > })
225 }
226 }
228 var inheritedMethods map[string]bool
229
230 > func init() { suite.go
231 > type ds struct{ Suite[*ds] }
232 > ptrType := reflect.TypeFor[*ds]()
233 > inheritedMethods = make(map[string]bool, ptrType.NumMethod())
234 > for method := range ptrType.Methods() {
235 > inheritedMethods[method.Name] = true
236 > }
237 }
238
239 > func validateSuiteStruct(structType reflect.Type) { suite.go
240 > if !strings.HasSuffix(structType.Name(), "Suite") {
241 panic(fmt.Sprintf("parallelsuite.Run: struct name %q must end with \"Suite\"", structType.Name()))
242 }
243
244 > if structType.NumField() != 1 { suite.go
245 panic(fmt.Sprintf(
246 "parallelsuite.Run: suite %s must have no fields besides the embedded parallelsuite.Suite; "+
249 ))
250 }
251 > f := structType.Field(0) suite.go
252 > if !f.Anonymous {
253 panic(fmt.Sprintf(
254 "parallelsuite.Run: suite %s must embed parallelsuite.Suite, found named field %q",
263 // The flag is registered by testify's suite package (imported above); we share
264 // that registration via flag.Lookup rather than registering it a second time.
265 > func applyTestifyMFilter(methods []reflect.Method) []reflect.Method { suite.go
266 > f := flag.Lookup("testify.m")
267 > if f == nil {
268 return methods
269 }
270 > pattern := f.Value.String() suite.go
271 > if pattern == "" {
272 > return methods suite.go
273 > }
274 re, err := regexp.Compile(pattern)
275 if err != nil {
285 }
286
287 > func discoverTestMethods(ptrType, structType reflect.Type, args []any) []reflect.Method { suite.go
288 > expectedNumIn := 1 + len(args)
289 >
290 > for method := range ptrType.Methods() {
291 > name := method.Name
292 > if !strings.HasPrefix(name, "Test") && !inheritedMethods[name] {
293 panic(fmt.Sprintf(
294 "parallelsuite.Run: suite %s has exported method %s that does not start with Test; "+
299 }
300
301 > var methods []reflect.Method suite.go
302 > for method := range ptrType.Methods() {
303 > if !strings.HasPrefix(method.Name, "Test") {
304 > continue
305 }
306
307 > mt := method.Type suite.go
308 > if mt.NumOut() != 0 {
309 panic(fmt.Sprintf(
310 "parallelsuite.Run: method %s.%s must not have return values, got %v",
312 ))
313 }
314 > if mt.NumIn() != expectedNumIn { suite.go
315 panic(fmt.Sprintf(
316 "parallelsuite.Run: method %s.%s has wrong number of parameters: expected %d, got %d (%v)",
319 }
320
321 > for j, a := range args { suite.go
322 paramType := mt.In(1 + j)
323 argType := reflect.TypeOf(a)
330 }
331
332 > methods = append(methods, method) suite.go
333 }
334 > return methods suite.go
335 }
go.temporal.io/server/common/goro/adaptive_pool.go 44 covered LOC · 16 ranges

Open complete file

32 targetDelay time.Duration,
33 shrinkFactor float64,
34 > ) *AdaptivePool { adaptive_pool.go
35 > p := &AdaptivePool{
36 > ts: ts,
37 > minWorkers: minWorkers,
38 > maxWorkers: maxWorkers,
39 > targetDelay: targetDelay,
40 > shrinkFactor: shrinkFactor,
41 > ch: make(chan func()),
42 > stopCh: make(chan struct{}),
43 > }
44 > for range minWorkers {
45 > go p.work() adaptive_pool.go
46 > }
47 > p.workers.Store(int64(minWorkers)) adaptive_pool.go
48 > return p
49 }
50
52 // When Stop is called, concurrent calls to Do may or may not call their function, and future
53 // calls definitely won't.
54 > func (p *AdaptivePool) Stop() { adaptive_pool.go
55 > close(p.stopCh)
56 > }
57
58 // Do calls f() on a worker goroutine. If the call can't be started within targetDelay, it adds
59 // another worker. If Stop is called concurrently, Do may or may not call f. If Stop has been
60 // called already, Do does nothing.
61 > func (p *AdaptivePool) Do(f func()) { adaptive_pool.go
62 > // try send first
63 > select {
64 > case p.ch <- f: adaptive_pool.go
65 > return
66 > default: adaptive_pool.go
67 }
68
69 // we might want to add a worker, send with timeout
70 > have := p.workers.Load() adaptive_pool.go
71 > if have < int64(p.maxWorkers) {
72 timech, timer := p.ts.NewTimer(p.targetDelay)
73 select {
87
88 // blocking send
89 > select { adaptive_pool.go
90 > case p.ch <- f:
91 case <-p.stopCh:
92 }
93 }
94
95 > func (p *AdaptivePool) work() { adaptive_pool.go
96 > for {
97 > // try receive first
98 > select {
99 > case f := <-p.ch: adaptive_pool.go
100 > f()
101 > continue
102 > default: adaptive_pool.go
103 }
104
105 > have := p.workers.Load() adaptive_pool.go
106 > if have > int64(p.minWorkers) {
107 // we might want to exit, receive with timeout
108 // jitter this so we shrink slower than we grow
134
135 // NumWorkers returns the current number of workers. Probably only useful for testing.
136 > func (p *AdaptivePool) NumWorkers() int { adaptive_pool.go
137 > return int(p.workers.Load())
138 > }
go.temporal.io/server/common/clock/event_time_source.go 5 covered LOC · 1 range

Open complete file

39
40 // NewEventTimeSource returns a EventTimeSource with the current time set to Unix zero: 1970-01-01 00:00:00 +0000 UTC.
41 > func NewEventTimeSource() *EventTimeSource { event_time_source.go
42 > return &EventTimeSource{
43 > now: time.Unix(0, 0),
44 > }
45 > }
46
47 // Some clients depend on the fact that the runtime's timers do _not_ run synchronously.
go.temporal.io/server/common/testing/historyrequire/history_require.go 5 covered LOC · 1 range

Open complete file

36 )
37
38 > func New(t require.TestingT) HistoryRequire { history_require.go
39 > return HistoryRequire{
40 > t: t,
41 > }
42 > }
43
44 // TODO (maybe):
go.temporal.io/server/common/testing/parallelsuite/guard.go 3 covered LOC · 2 ranges

Open complete file

16 }
17
18 > func (g *guardT) Helper() { guard.go
19 > if g.hasSubtests.Load() {
20 panic(fmt.Sprintf(
21 "parallelsuite: assertion called on %q after Run() was called; "+
24 ))
25 }
26 > g.T.Helper() guard.go
27 }
28
go.temporal.io/server/common/testing/protorequire/require.go 3 covered LOC · 1 range

Open complete file

38 }
39
40 > func New(t require.TestingT) ProtoAssertions { require.go
41 > return ProtoAssertions{t}
42 > }
43
44 // ProtoEqual compares two proto messages for equality using proto semantics. Options can be passed to customize