<commit_before>package nakadi

import (
	"context"
	"testing"
	"time"

	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/mock"
	"github.com/stretchr/testify/require"
)

func TestStreamAPI_startStream(t *testing.T) {
	retryCh := make(chan error, 1)
	okCh := make(chan struct{}, 1)
	blockCh := make(chan time.Time, 1)
	stream := &mockStreamer{}

	_, opener, _ := setupMockStream(retryCh, okCh)

	opener.On("openStream").Once().Return(nil, assert.AnError)
	opener.On("openStream").Once().Return(stream, nil).WaitUntil(blockCh)
	stream.On("nextEvents").Return(Cursor{}, nil, assert.AnError).WaitUntil(blockCh)

	select {
	case err := <-retryCh:
		assert.EqualError(t, assert.AnError, err.Error())
	case <-time.After(50 * time.Millisecond):
		assert.Fail(t, "no retry error detected")
	}

	blockCh <- time.Now()
	select {
	case <-okCh:
		// nothing<commit_msg>Add test to reveal infinite looping in startStream<commit_after>package nakadi

import (
	"context"
	"testing"
	"time"

	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/mock"
	"github.com/stretchr/testify/require"
)

func TestStreamAPI_startStreamLoop(t *testing.T) {
	errorCh := make(chan error, 1)
	okCh := make(chan struct{})
	blockCh := make(chan time.Time, 1)
	stream := &mockStreamer{}
	streamAPI, opener, _ := setupMockStream(errorCh, okCh)

	opener.On("openStream").Once().Return(stream, nil)
	opener.On("openStream").Once().Return(nil, assert.AnError)
	stream.On("nextEvents").Return(Cursor{}, nil, assert.AnError).WaitUntil(blockCh)
	stream.On("closeStream").Return(nil)

	blockCh <- time.Now()
	<-okCh
	streamAPI.cancel()
	
	<-time.After(time.Second)
	//Wait for startStream to loop
	// If startStream tries to open the stream again after streamAPI is closed,
	// test will panic if openStream is called more than once.
}

func TestStreamAPI_startStream(t *testing.T) {
	retryCh := make(chan error, 1)
	okCh := make(chan struct{}, 1)
	blockCh := make(chan time.Time, 1)
	stream := &mockStreamer{}

	_, opener, _ := setupMockStream(retryCh, okCh)

	opener.On("openStream").Once().Return(nil, assert.AnError)
	opener.On("openStream").Once().Return(stream, nil).WaitUntil(blockCh)
	stream.On("nextEvents").Return(Cursor{}, nil, assert.AnError).WaitUntil(blockCh)

	select {
	case err := <-retryCh:
		assert.EqualError(t, assert.AnError, err.Error())
	case <-time.After(50 * time.Millisecond):
		assert.Fail(t, "no retry error detected")
	}

	blockCh <- time.Now()
	select {
	case <-okCh:
		// nothing
{'fraction_non_alphanumeric': 0.1624, 'fraction_numerical': 0.0048, 'mean_word_length': 4.940617577197149, 'pattern_counts': {'":': 0, '<': 14, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2706560', 'n_tokens_mistral': 924, 'n_tokens_neox': 891, 'n_words': 180}