1package events
2
3import (
4 "sync"
5 "testing"
6 "time"
7
8 "github.com/stretchr/testify/assert"
9 "github.com/stretchr/testify/mock"
10 "github.com/stretchr/testify/require"
11)
12
13// MockEventHandler is a mock implementation of EventHandler
14type MockEventHandler struct {
15 mock.Mock
16 mu sync.Mutex
17 events []Event
18 canHandle bool
19}
20
21func (m *MockEventHandler) Handle(event Event) error {
22 m.mu.Lock()
23 defer m.mu.Unlock()
24 m.events = append(m.events, event)
25 args := m.Called(event)
26 return args.Error(0)
27}
28
29func (m *MockEventHandler) CanHandle(eventType EventType) bool {
30 args := m.Called(eventType)
31 return args.Bool(0)
32}
33
34func (m *MockEventHandler) GetReceivedEvents() []Event {
35 m.mu.Lock()
36 defer m.mu.Unlock()
37 return append([]Event{}, m.events...)
38}
39
40func TestInMemoryEventBus_NewInMemoryEventBus(t *testing.T) {
41 bus := NewInMemoryEventBus(50)
42
43 assert.NotNil(t, bus)
44 assert.Equal(t, 50, cap(bus.eventChan))
45 assert.Empty(t, bus.handlers)
46 assert.Equal(t, 50, bus.bufferSize)
47}
48
49func TestInMemoryEventBus_Subscribe(t *testing.T) {
50 bus := NewInMemoryEventBus(10)
51 handler := &MockEventHandler{}
52
53 bus.Subscribe(handler)
54
55 assert.Len(t, bus.handlers, 1)
56 assert.Contains(t, bus.handlers, handler)
57}
58
59func TestInMemoryEventBus_Subscribe_DuplicateHandler(t *testing.T) {
60 bus := NewInMemoryEventBus(10)
61 handler := &MockEventHandler{}
62
63 bus.Subscribe(handler)
64 bus.Subscribe(handler) // Subscribe same handler twice
65
66 // Implementation allows duplicates (handlers are appended to slice)
67 assert.Len(t, bus.handlers, 2)
68}
69
70func TestInMemoryEventBus_Unsubscribe(t *testing.T) {
71 bus := NewInMemoryEventBus(10)
72 handler1 := &MockEventHandler{}
73 handler2 := &MockEventHandler{}
74
75 bus.Subscribe(handler1)
76 bus.Subscribe(handler2)
77 assert.Len(t, bus.handlers, 2)
78
79 err := bus.Unsubscribe(handler1)
80
81 assert.NoError(t, err)
82 assert.Len(t, bus.handlers, 1)
83 // Check that handler2 is still in the list (compare by position)
84 assert.Equal(t, handler2, bus.handlers[0])
85}
86
87func TestInMemoryEventBus_Unsubscribe_NonExistentHandler(t *testing.T) {
88 bus := NewInMemoryEventBus(10)
89 handler1 := &MockEventHandler{}
90 handler2 := &MockEventHandler{}
91
92 bus.Subscribe(handler1)
93 err := bus.Unsubscribe(handler2) // Try to unsubscribe handler that wasn't subscribed
94
95 assert.Error(t, err)
96 assert.Contains(t, err.Error(), "handler not found")
97 assert.Len(t, bus.handlers, 1)
98 assert.Equal(t, handler1, bus.handlers[0])
99}
100
101func TestInMemoryEventBus_StartStop(t *testing.T) {
102 bus := NewInMemoryEventBus(10)
103
104 // Start the bus
105 err := bus.Start()
106 assert.NoError(t, err)
107
108 // Stop the bus
109 err = bus.Stop()
110 assert.NoError(t, err)
111}
112
113func TestInMemoryEventBus_Publish_Running(t *testing.T) {
114 bus := NewInMemoryEventBus(10)
115
116 handler := &MockEventHandler{}
117 handler.On("CanHandle", ImagePushed).Return(true)
118 handler.On("Handle", mock.AnythingOfType("Event")).Return(nil)
119
120 bus.Subscribe(handler)
121 bus.Start()
122 defer bus.Stop()
123
124 payload := ImagePushedPayload{
125 Name: "nginx:latest",
126 Reference: "latest",
127 }
128
129 bus.Publish(ImagePushed, payload)
130
131 // Give some time for async processing
132 time.Sleep(50 * time.Millisecond)
133
134 handler.AssertExpectations(t)
135 events := handler.GetReceivedEvents()
136 require.Len(t, events, 1)
137 assert.Equal(t, ImagePushed, events[0].Type)
138 assert.Equal(t, "nginx:latest", events[0].ImageName)
139 assert.Equal(t, "latest", events[0].Tag)
140}
141
142func TestInMemoryEventBus_Publish_NotRunning(t *testing.T) {
143 bus := NewInMemoryEventBus(10)
144
145 handler := &MockEventHandler{}
146 bus.Subscribe(handler)
147
148 payload := ImagePushedPayload{
149 Name: "nginx:latest",
150 Reference: "latest",
151 }
152
153 // Publish when bus is not running
154 bus.Publish(ImagePushed, payload)
155
156 // Give some time in case it processes (it shouldn't)
157 time.Sleep(10 * time.Millisecond)
158
159 // Handler should not have received any events
160 events := handler.GetReceivedEvents()
161 assert.Empty(t, events)
162}
163
164func TestInMemoryEventBus_HandlerFiltering(t *testing.T) {
165 bus := NewInMemoryEventBus(10)
166
167 // Handler that only handles ImagePushed events
168 imagePushedHandler := &MockEventHandler{}
169 imagePushedHandler.On("CanHandle", ImagePushed).Return(true)
170 imagePushedHandler.On("CanHandle", ConfigReload).Return(false)
171 imagePushedHandler.On("Handle", mock.AnythingOfType("Event")).Return(nil)
172
173 // Handler that only handles ConfigReload events
174 configReloadHandler := &MockEventHandler{}
175 configReloadHandler.On("CanHandle", ImagePushed).Return(false)
176 configReloadHandler.On("CanHandle", ConfigReload).Return(true)
177 configReloadHandler.On("Handle", mock.AnythingOfType("Event")).Return(nil)
178
179 bus.Subscribe(imagePushedHandler)
180 bus.Subscribe(configReloadHandler)
181 bus.Start()
182 defer bus.Stop()
183
184 // Publish ImagePushed event
185 bus.Publish(ImagePushed, ImagePushedPayload{
186 Name: "nginx:latest",
187 Reference: "latest",
188 })
189
190 // Publish ConfigReload event
191 bus.Publish(ConfigReload, nil)
192
193 // Give time for processing
194 time.Sleep(20 * time.Millisecond)
195
196 // Check that each handler only received appropriate events
197 imagePushedEvents := imagePushedHandler.GetReceivedEvents()
198 configReloadEvents := configReloadHandler.GetReceivedEvents()
199
200 require.Len(t, imagePushedEvents, 1)
201 assert.Equal(t, ImagePushed, imagePushedEvents[0].Type)
202
203 require.Len(t, configReloadEvents, 1)
204 assert.Equal(t, ConfigReload, configReloadEvents[0].Type)
205
206 imagePushedHandler.AssertExpectations(t)
207 configReloadHandler.AssertExpectations(t)
208}
209
210func TestInMemoryEventBus_HandlerError(t *testing.T) {
211 bus := NewInMemoryEventBus(10)
212
213 // Handler that returns an error
214 errorHandler := &MockEventHandler{}
215 errorHandler.On("CanHandle", ImagePushed).Return(true)
216 errorHandler.On("Handle", mock.AnythingOfType("Event")).Return(assert.AnError)
217
218 // Handler that succeeds
219 successHandler := &MockEventHandler{}
220 successHandler.On("CanHandle", ImagePushed).Return(true)
221 successHandler.On("Handle", mock.AnythingOfType("Event")).Return(nil)
222
223 bus.Subscribe(errorHandler)
224 bus.Subscribe(successHandler)
225 bus.Start()
226 defer bus.Stop()
227
228 payload := ImagePushedPayload{
229 Name: "nginx:latest",
230 Reference: "latest",
231 }
232
233 bus.Publish(ImagePushed, payload)
234
235 // Give time for processing
236 time.Sleep(20 * time.Millisecond)
237
238 // Both handlers should have been called despite error
239 errorHandler.AssertExpectations(t)
240 successHandler.AssertExpectations(t)
241
242 // Both should have received the event
243 errorEvents := errorHandler.GetReceivedEvents()
244 successEvents := successHandler.GetReceivedEvents()
245
246 assert.Len(t, errorEvents, 1)
247 assert.Len(t, successEvents, 1)
248}
249
250func TestInMemoryEventBus_ContextCancellation(t *testing.T) {
251 bus := NewInMemoryEventBus(10)
252
253 handler := &MockEventHandler{}
254 handler.On("CanHandle", ImagePushed).Return(true)
255 handler.On("Handle", mock.AnythingOfType("Event")).Return(nil)
256
257 bus.Subscribe(handler)
258 bus.Start()
259
260 // Stop the bus to trigger shutdown
261 err := bus.Stop()
262 assert.NoError(t, err)
263}
264
265func TestInMemoryEventBus_FullChannel(t *testing.T) {
266 // Create bus with very small buffer
267 bus := NewInMemoryEventBus(1)
268
269 handler := &MockEventHandler{}
270 handler.On("CanHandle", ImagePushed).Return(true)
271 // Make handler slow to process events
272 handler.On("Handle", mock.AnythingOfType("Event")).Return(nil).Run(func(args mock.Arguments) {
273 time.Sleep(50 * time.Millisecond)
274 })
275
276 bus.Subscribe(handler)
277 bus.Start()
278 defer bus.Stop()
279
280 payload := ImagePushedPayload{
281 Name: "nginx:latest",
282 Reference: "latest",
283 }
284
285 // Publish multiple events quickly to potentially fill the channel
286 for i := 0; i < 5; i++ {
287 bus.Publish(ImagePushed, payload)
288 }
289
290 // Give time for processing
291 time.Sleep(200 * time.Millisecond)
292
293 // Should still work, events might be dropped but no crash
294 events := handler.GetReceivedEvents()
295 assert.True(t, len(events) >= 1) // At least some events should be processed
296}
297
298func TestInMemoryEventBus_ConcurrentPublish(t *testing.T) {
299 bus := NewInMemoryEventBus(100)
300
301 handler := &MockEventHandler{}
302 handler.On("CanHandle", ImagePushed).Return(true)
303 handler.On("Handle", mock.AnythingOfType("Event")).Return(nil)
304
305 bus.Subscribe(handler)
306 bus.Start()
307 defer bus.Stop()
308
309 payload := ImagePushedPayload{
310 Name: "nginx:latest",
311 Reference: "latest",
312 }
313
314 // Publish events concurrently from multiple goroutines
315 numGoroutines := 10
316 eventsPerGoroutine := 5
317 var wg sync.WaitGroup
318
319 for i := 0; i < numGoroutines; i++ {
320 wg.Add(1)
321 go func() {
322 defer wg.Done()
323 for j := 0; j < eventsPerGoroutine; j++ {
324 bus.Publish(ImagePushed, payload)
325 }
326 }()
327 }
328
329 wg.Wait()
330
331 // Give time for processing
332 time.Sleep(100 * time.Millisecond)
333
334 events := handler.GetReceivedEvents()
335 expectedEvents := numGoroutines * eventsPerGoroutine
336 assert.Equal(t, expectedEvents, len(events))
337}