messageprocessor-stat/statmessageprocessor_test.go at master · asecurityteam/messageprocessor-stat · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
package stats

import (
	"context"
	"testing"
	"time"

	"github.com/rs/xstats"
	"github.com/stretchr/testify/assert"

	"github.com/aws/aws-sdk-go/service/kinesis"

	"github.com/golang/mock/gomock"
)

func TestStatMessageProcessor_ProcessMessageSuccess(t *testing.T) {
	ctrl := gomock.NewController(t)
	defer ctrl.Finish()

	mockMessageProcessor := NewMockMessageProcessor(ctrl)
	mockStater := NewMockStat(ctrl)

	statMessageProcessor := StatMessageProcessor{
		ConsumedCounter:        consumedCounter,
		ConsumerSuccessCounter: consumerSuccessCounter,
		ConsumedSize:           consumedSize,
		ConsumerLag:            consumerLag,
		ConsumerTimingSuccess:  consumerTimingSuccess,
		wrapped:                mockMessageProcessor,
	}

	incomingDataRecord := []byte(`{
		"type":"CHANNEL",
		"rawpayload":"{\"channel\":\"channelbob\",\"text\":\"the message\"}"
	}`)
	currentTime := time.Now()
	sequenceNumber := "12345"
	kinesisRecord := kinesis.Record{
		Data:                        incomingDataRecord,
		ApproximateArrivalTimestamp: &currentTime,
		SequenceNumber:              &sequenceNumber,
	}
	gomock.InOrder(
		mockStater.EXPECT().Timing(consumerLag, gomock.Any()),
		mockStater.EXPECT().Count(consumedCounter, gomock.Any()),
		mockStater.EXPECT().Count(consumedSize, gomock.Any()),
		mockStater.EXPECT().Count(consumerSuccessCounter, gomock.Any()),
		mockStater.EXPECT().Timing(consumerTimingSuccess, gomock.Any()),
	)

	mockMessageProcessor.EXPECT().ProcessMessage(gomock.Any(), gomock.Any()).Return(nil)
	e := statMessageProcessor.ProcessMessage(xstats.NewContext(context.Background(), mockStater), &kinesisRecord)
	assert.Nil(t, e)

}

func TestStatMessageProcessor_ProcessMessageFailure(t *testing.T) {
	ctrl := gomock.NewController(t)
	defer ctrl.Finish()

	mockMessageProcessor := NewMockMessageProcessor(ctrl)
	mockStater := NewMockStat(ctrl)
	mockMessageProcessorError := NewMockMessageProcessorError(ctrl)

	statMessageProcessor := StatMessageProcessor{
		ConsumedCounter:       consumedCounter,
		ConsumerErrorCounter:  consumerErrorCounter,
		ConsumedSize:          consumedSize,
		ConsumerLag:           consumerLag,
		ConsumerTimingFailure: consumerTimingFailure,
		wrapped:               mockMessageProcessor,
	}

	incomingDataRecord := []byte(`{
		"type":"CHANNEL",
		"rawpayload":"{\"channel\":\"channelbob\",\"text\":\"the message\"}"
	}`)
	currentTime := time.Now()
	sequenceNumber := "12345"
	kinesisRecord := kinesis.Record{
		Data:                        incomingDataRecord,
		ApproximateArrivalTimestamp: &currentTime,
		SequenceNumber:              &sequenceNumber,
	}

	gomock.InOrder(
		mockStater.EXPECT().Timing(consumerLag, gomock.Any()),
		mockStater.EXPECT().Count(consumedCounter, gomock.Any()),
		mockStater.EXPECT().Count(consumedSize, gomock.Any()),
		mockStater.EXPECT().Count(consumerErrorCounter, gomock.Any()),
		mockStater.EXPECT().Timing(consumerTimingFailure, gomock.Any()),
	)

	mockMessageProcessor.EXPECT().ProcessMessage(gomock.Any(), gomock.Any()).Return(mockMessageProcessorError)
	e := statMessageProcessor.ProcessMessage(xstats.NewContext(context.Background(), mockStater), &kinesisRecord)
	assert.NotNil(t, e)

}