Function assertEvents

pkg/acquisition/modules/http/http_test.go:409–445

(out chan pipeline.Event, expected []string, errChan chan error)

}

func assertEvents(out chan pipeline.Event, expected []string, errChan chan error) {
	readLines := []pipeline.Event{}

	for range expected {
		select {
		case event := <-out:
			readLines = append(readLines, event)
		case <-time.After(2 * time.Second):
			errChan <- errors.New("timeout waiting for event")
			return
		}
	}

	if len(readLines) != len(expected) {
		errChan <- fmt.Errorf("expected %d lines, got %d", len(expected), len(readLines))
		return
	}

	for i, evt := range readLines {
		if evt.Line.Raw != expected[i] {
			errChan <- fmt.Errorf(`expected %s, got '%+v'`, expected, evt.Line.Raw)
			return
		}

		if evt.Line.Src != "127.0.0.1" {
			errChan <- fmt.Errorf("expected '127.0.0.1', got '%s'", evt.Line.Src)
			return
		}

		if evt.Line.Module != "http" {
			errChan <- fmt.Errorf("expected 'http', got '%s'", evt.Line.Module)
			return
		}
	}

	errChan <- nil
}

func TestStreamingAcquisitionTimeout(t *testing.T) {
	ctx := t.Context()