@@ -169,3 +169,128 @@ func TestStreamModule_StopWithoutStart(t *testing.T) {
169169 t .Errorf ("Stop() without Start: expected DeadlineExceeded, got %v" , err )
170170 }
171171}
172+
173+ func TestStreamModule_StatusTransitions (t * testing.T ) {
174+ t .Run ("initial status is stopped" , func (t * testing.T ) {
175+ m , err := newStreamModule ("test-stream" , map [string ]any {
176+ "input" : map [string ]any {"generate" : map [string ]any {}},
177+ "output" : map [string ]any {"drop" : map [string ]any {}},
178+ })
179+ if err != nil {
180+ t .Fatalf ("newStreamModule() error = %v" , err )
181+ }
182+ if got := m .Status (); got != streamStopped {
183+ t .Errorf ("initial status = %q, want %q" , got , streamStopped )
184+ }
185+ })
186+
187+ t .Run ("status is running after successful start" , func (t * testing.T ) {
188+ m , err := newStreamModule ("test-stream" , map [string ]any {
189+ "input" : map [string ]any {
190+ "generate" : map [string ]any {
191+ "mapping" : `root = {"test": "data"}` ,
192+ "count" : 0 ,
193+ "interval" : "1s" ,
194+ },
195+ },
196+ "output" : map [string ]any {
197+ "drop" : map [string ]any {},
198+ },
199+ })
200+ if err != nil {
201+ t .Fatalf ("newStreamModule() error = %v" , err )
202+ }
203+
204+ if err := m .Init (); err != nil {
205+ t .Fatalf ("Init() error = %v" , err )
206+ }
207+
208+ ctx := context .Background ()
209+ if err := m .Start (ctx ); err != nil {
210+ t .Fatalf ("Start() error = %v" , err )
211+ }
212+
213+ if got := m .Status (); got != streamRunning {
214+ t .Errorf ("status after Start = %q, want %q" , got , streamRunning )
215+ }
216+
217+ // Allow goroutine to begin running before stopping.
218+ time .Sleep (50 * time .Millisecond )
219+
220+ stopCtx , cancel := context .WithTimeout (context .Background (), 2 * time .Second )
221+ defer cancel ()
222+ _ = m .Stop (stopCtx )
223+ })
224+
225+ t .Run ("status is stopped after successful stop" , func (t * testing.T ) {
226+ m , err := newStreamModule ("test-stream" , map [string ]any {
227+ "input" : map [string ]any {
228+ "generate" : map [string ]any {
229+ "mapping" : `root = {"test": "data"}` ,
230+ "count" : 0 ,
231+ "interval" : "1s" ,
232+ },
233+ },
234+ "output" : map [string ]any {
235+ "drop" : map [string ]any {},
236+ },
237+ })
238+ if err != nil {
239+ t .Fatalf ("newStreamModule() error = %v" , err )
240+ }
241+
242+ if err := m .Init (); err != nil {
243+ t .Fatalf ("Init() error = %v" , err )
244+ }
245+
246+ ctx := context .Background ()
247+ if err := m .Start (ctx ); err != nil {
248+ t .Fatalf ("Start() error = %v" , err )
249+ }
250+
251+ // Allow goroutine to begin running before stopping.
252+ time .Sleep (50 * time .Millisecond )
253+
254+ stopCtx , cancel := context .WithTimeout (context .Background (), 2 * time .Second )
255+ defer cancel ()
256+
257+ if err := m .Stop (stopCtx ); err != nil {
258+ t .Fatalf ("Stop() error = %v" , err )
259+ }
260+
261+ if got := m .Status (); got != streamStopped {
262+ t .Errorf ("status after Stop = %q, want %q" , got , streamStopped )
263+ }
264+ })
265+
266+ t .Run ("status is errored after failed start" , func (t * testing.T ) {
267+ m , err := newStreamModule ("bad-stream" , map [string ]any {
268+ "input" : map [string ]any {
269+ "unknown_input_type" : map [string ]any {
270+ "invalid" : "config" ,
271+ },
272+ },
273+ "output" : map [string ]any {
274+ "drop" : map [string ]any {},
275+ },
276+ })
277+ if err != nil {
278+ t .Fatalf ("newStreamModule() error = %v" , err )
279+ }
280+
281+ if err := m .Init (); err != nil {
282+ t .Fatalf ("Init() error = %v" , err )
283+ }
284+
285+ ctx := context .Background ()
286+ if err := m .Start (ctx ); err == nil {
287+ t .Error ("Start() expected error for invalid config, got nil" )
288+ _ = m .Stop (context .Background ())
289+ return
290+ }
291+
292+ if got := m .Status (); got != streamErrored {
293+ t .Errorf ("status after failed Start = %q, want %q" , got , streamErrored )
294+ }
295+ })
296+ }
0 commit comments