// kafkaStubConsumer mock which exposes Channels function returning empty channels
// to be used on tests that are not supposed to receive any kafka message
// var kafkaStubConsumer = &kafkatest.IConsumerGroupMock{
//	ChannelsFunc: func() *kafka.ConsumerGroupChannels {
//		return &kafka.ConsumerGroupChannels{}
//	},
//}
kafkaStubConsumer mock which exposes Channels function returning empty channels
to be used on tests that are not supposed to receive any kafka message
var kafkaStubConsumer = &kafkatest.IConsumerGroupMock{
ChannelsFunc: func() *kafka.ConsumerGroupChannels {
return &kafka.ConsumerGroupChannels{}
},
}
// kafkaStubConsumer mock which exposes Channels function returning empty channels // to be used on tests that are not supposed to receive any kafka message // var kafkaStubConsumer = &kafkatest . IConsumerGroupMock { // ChannelsFunc : func () * kafka . ConsumerGroupChannels { // return &kafka . ConsumerGroupChannels {} // } // }
func TestConsume(t *testing.T) {
	Convey("Given kafka consumer and event handler mocks", t, func() {
		cgChannels := &kafka.ConsumerGroupChannels{Upstream: make(chan kafka.Message, 2)}
		mockConsumer := &kafkatest.IConsumerGroupMock{
			ChannelsFunc: func() *kafka.ConsumerGroupChannels { return cgChannels },
		}
		handlerWg := &sync.WaitGroup{}
		mockEventHandler := &mock.HandlerMock{
			HandleFunc: func(ctx context.Context, event *models.ReindexRequested) error {
				defer handlerWg.Done()
				return nil
			},
		}
		Convey("And a kafka message with the valid schema being sent to the Upstream channel", func() {
			message := kafkatest.NewMessage(marshal(testEvent), 0)
			mockConsumer.Channels().Upstream <- message
			Convey("When consume message is called", func() {
				handlerWg.Add(1)
				event.Consume(testCtx, mockConsumer, mockEventHandler, &config.Config{KafkaConfig: config.KafkaConfig{NumWorkers: 1}})
				handlerWg.Wait()
				Convey("An event is sent to the mockEventHandler ", func() {
					So(len(mockEventHandler.HandleCalls()), ShouldEqual, 1)
					So(*mockEventHandler.HandleCalls()[0].ReindexRequested, ShouldResemble, testEvent)
				})
				Convey("The message is committed and the consumer is released", func() {
					<-message.UpstreamDone()
					So(len(message.CommitCalls()), ShouldEqual, 1)
					So(len(message.ReleaseCalls()), ShouldEqual, 1)
				})
			})
		})
		Convey("And two kafka messages, one with a valid schema and one with an invalid schema", func() {
			validMessage := kafkatest.NewMessage(marshal(testEvent), 1)
			invalidMessage := kafkatest.NewMessage([]byte("invalid schema"), 0)
			mockConsumer.Channels().Upstream <- invalidMessage
			mockConsumer.Channels().Upstream <- validMessage
			Convey("When consume messages is called", func() {
				handlerWg.Add(1)
				event.Consume(testCtx, mockConsumer, mockEventHandler, &config.Config{KafkaConfig: config.KafkaConfig{NumWorkers: 1}})
				handlerWg.Wait()
				Convey("Only the valid event is sent to the mockEventHandler ", func() {
					So(len(mockEventHandler.HandleCalls()), ShouldEqual, 1)
					So(*mockEventHandler.HandleCalls()[0].ReindexRequested, ShouldResemble, testEvent)
				})
				Convey("Only the valid message is committed, but the consumer is released for both messages", func() {
					<-validMessage.UpstreamDone()
					<-invalidMessage.UpstreamDone()
					So(len(validMessage.CommitCalls()), ShouldEqual, 1)
					So(len(invalidMessage.CommitCalls()), ShouldEqual, 1)
					So(len(validMessage.ReleaseCalls()), ShouldEqual, 1)
					So(len(invalidMessage.ReleaseCalls()), ShouldEqual, 1)
				})
			})
		})
		Convey("With a failing handler and a kafka message with the valid schema being sent to the Upstream channel", func() {
			mockEventHandler.HandleFunc = func(ctx context.Context, event *models.ReindexRequested) error {
				defer handlerWg.Done()
				return errHandler
			}
			message := kafkatest.NewMessage(marshal(testEvent), 0)
			mockConsumer.Channels().Upstream <- message
			Convey("When consume message is called", func() {
				handlerWg.Add(1)
				event.Consume(testCtx, mockConsumer, mockEventHandler, &config.Config{KafkaConfig: config.KafkaConfig{NumWorkers: 1}})
				handlerWg.Wait()
				Convey("An event is sent to the mockEventHandler ", func() {
					So(len(mockEventHandler.HandleCalls()), ShouldEqual, 1)
					So(*mockEventHandler.HandleCalls()[0].ReindexRequested, ShouldResemble, testEvent)
				})
				Convey("The message is committed and the consumer is released", func() {
					<-message.UpstreamDone()
					So(len(message.CommitCalls()), ShouldEqual, 1)
					So(len(message.ReleaseCalls()), ShouldEqual, 1)
				})
			})
		})
	})
}
func TestConsume ( t * testing . T ) { Convey ( " " , t , func ( ) { cgChannels := & kafka . ConsumerGroupChannels { Upstream : make ( chan kafka . Message , 2 ) } mockConsumer := & kafkatest . IConsumerGroupMock { ChannelsFunc : func ( ) * kafka . ConsumerGroupChannels { return cgChannels } , } handlerWg := & sync . WaitGroup { } mockEventHandler := & mock . HandlerMock { HandleFunc : func ( ctx context . Context , event * models . ReindexRequested ) error { defer handlerWg . Done ( ) return nil } , } Convey ( " " , func ( ) { message := kafkatest . NewMessage ( marshal ( testEvent ) , 0 ) mockConsumer . Channels ( ) . Upstream <- message Convey ( " " , func ( ) { handlerWg . Add ( 1 ) event . Consume ( testCtx , mockConsumer , mockEventHandler , & config . Config { KafkaConfig : config . KafkaConfig { NumWorkers : 1 } } ) handlerWg . Wait ( ) Convey ( " " , func ( ) { So ( len ( mockEventHandler . HandleCalls ( ) ) , ShouldEqual , 1 ) So ( * mockEventHandler . HandleCalls ( ) [ 0 ] . ReindexRequested , ShouldResemble , testEvent ) } ) Convey ( " " , func ( ) { <- message . UpstreamDone ( ) So ( len ( message . CommitCalls ( ) ) , ShouldEqual , 1 ) So ( len ( message . ReleaseCalls ( ) ) , ShouldEqual , 1 ) } ) } ) } ) Convey ( " " , func ( ) { validMessage := kafkatest . NewMessage ( marshal ( testEvent ) , 1 ) invalidMessage := kafkatest . NewMessage ( [ ] byte ( " " ) , 0 ) mockConsumer . Channels ( ) . Upstream <- invalidMessage mockConsumer . Channels ( ) . Upstream <- validMessage Convey ( " " , func ( ) { handlerWg . Add ( 1 ) event . Consume ( testCtx , mockConsumer , mockEventHandler , & config . Config { KafkaConfig : config . KafkaConfig { NumWorkers : 1 } } ) handlerWg . Wait ( ) Convey ( " " , func ( ) { So ( len ( mockEventHandler . HandleCalls ( ) ) , ShouldEqual , 1 ) So ( * mockEventHandler . HandleCalls ( ) [ 0 ] . ReindexRequested , ShouldResemble , testEvent ) } ) Convey ( " " , func ( ) { <- validMessage . UpstreamDone ( ) <- invalidMessage . UpstreamDone ( ) So ( len ( validMessage . CommitCalls ( ) ) , ShouldEqual , 1 ) So ( len ( invalidMessage . CommitCalls ( ) ) , ShouldEqual , 1 ) So ( len ( validMessage . ReleaseCalls ( ) ) , ShouldEqual , 1 ) So ( len ( invalidMessage . ReleaseCalls ( ) ) , ShouldEqual , 1 ) } ) } ) } ) Convey ( " " , func ( ) { mockEventHandler . HandleFunc = func ( ctx context . Context , event * models . ReindexRequested ) error { defer handlerWg . Done ( ) return errHandler } message := kafkatest . NewMessage ( marshal ( testEvent ) , 0 ) mockConsumer . Channels ( ) . Upstream <- message Convey ( " " , func ( ) { handlerWg . Add ( 1 ) event . Consume ( testCtx , mockConsumer , mockEventHandler , & config . Config { KafkaConfig : config . KafkaConfig { NumWorkers : 1 } } ) handlerWg . Wait ( ) Convey ( " " , func ( ) { So ( len ( mockEventHandler . HandleCalls ( ) ) , ShouldEqual , 1 ) So ( * mockEventHandler . HandleCalls ( ) [ 0 ] . ReindexRequested , ShouldResemble , testEvent ) } ) Convey ( " " , func ( ) { <- message . UpstreamDone ( ) So ( len ( message . CommitCalls ( ) ) , ShouldEqual , 1 ) So ( len ( message . ReleaseCalls ( ) ) , ShouldEqual , 1 ) } ) } ) } ) } ) }
kafkaStubConsumer mock which exposes Channels function returning empty channels
to be used on tests that are not supposed to receive any kafka message
var kafkaStubConsumer = &kafkatest.IConsumerGroupMock{
ChannelsFunc: func() *kafka.ConsumerGroupChannels {
return &kafka.ConsumerGroupChannels{}
},
}
kafkaStubConsumer mock which exposes Channels function returning empty channels to be used on tests that are not supposed to receive any kafka message var kafkaStubConsumer = &kafkatest . IConsumerGroupMock { ChannelsFunc : func () * kafka . ConsumerGroupChannels { return &kafka . ConsumerGroupChannels {} } }