53 lines
1.4 KiB
Go
53 lines
1.4 KiB
Go
package inventory
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"sync/atomic"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/docker/docker/api/types/events"
|
|
"github.com/stretchr/testify/assert"
|
|
"github.com/stretchr/testify/require"
|
|
)
|
|
|
|
type fakeEvents struct {
|
|
msgs chan events.Message
|
|
errs chan error
|
|
}
|
|
|
|
func newFakeEvents() *fakeEvents {
|
|
return &fakeEvents{msgs: make(chan events.Message, 10), errs: make(chan error, 1)}
|
|
}
|
|
|
|
func (f *fakeEvents) Events(context.Context, events.ListOptions) (<-chan events.Message, <-chan error) {
|
|
return f.msgs, f.errs
|
|
}
|
|
|
|
func TestWatch_DebouncesBursts(t *testing.T) {
|
|
src := newFakeEvents()
|
|
var calls atomic.Int32
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
done := make(chan error, 1)
|
|
go func() { done <- Watch(ctx, src, 30*time.Millisecond, func() { calls.Add(1) }) }()
|
|
|
|
for range 3 {
|
|
src.msgs <- events.Message{Type: events.ContainerEventType, Action: "start"}
|
|
}
|
|
require.Eventually(t, func() bool { return calls.Load() == 1 }, time.Second, 5*time.Millisecond)
|
|
|
|
src.msgs <- events.Message{Type: events.ServiceEventType, Action: "update"}
|
|
require.Eventually(t, func() bool { return calls.Load() == 2 }, time.Second, 5*time.Millisecond)
|
|
|
|
cancel()
|
|
assert.NoError(t, <-done)
|
|
}
|
|
|
|
func TestWatch_ReturnsStreamError(t *testing.T) {
|
|
src := newFakeEvents()
|
|
src.errs <- errors.New("daemon restarted")
|
|
err := Watch(context.Background(), src, time.Millisecond, func() {})
|
|
assert.ErrorContains(t, err, "daemon restarted")
|
|
}
|