luciano/kinesis-consumer
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Refactor the consumer tests

Browse source 
A previous PR from @vincent6767 had nicer mock Kinesis client that
simplified setting up data for the tests.

Mock client pulled from:
https://github.com/harlow/kinesis-consumer/pull/64
This commit is contained in:
Harlow Ward 2018-07-29 10:13:03 -07:00
parent 911282363e
commit d6ded158bf
4 changed files with 85 additions and 54 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
checkpoint.go
View file

@ -6,6 +6,7 @@ type Checkpoint interface {
Set(streamName, shardID, sequenceNumber string) error
}
// noopCheckpoint implements the checkpoint interface with discard
type noopCheckpoint struct{}
func (n noopCheckpoint) Set(string, string, string) error { return nil }

171
consumer_test.go
View file

@ -3,8 +3,6 @@ package consumer
import (
"context"
"fmt"
"io/ioutil"
"log"
"sync"
"testing"
@ -21,69 +19,7 @@ func TestNew(t *testing.T) {
}
func TestScanShard(t *testing.T) {
var (
resultData string
ckp = &fakeCheckpoint{cache: map[string]string{}}
ctr = &fakeCounter{}
mockSvc = &mockKinesisClient{}
logger = &noopLogger{
logger: log.New(ioutil.Discard, "", log.LstdFlags),
}
)
c := &Consumer{
streamName: "myStreamName",
client: mockSvc,
checkpoint: ckp,
counter: ctr,
logger: logger,
}
var recordNum = 0
// callback fn simply appends the record data to result string
var fn = func(r *Record) ScanStatus {
resultData += string(r.Data)
recordNum++
stopScan := recordNum == 2
return ScanStatus{
StopScan: stopScan,
SkipCheckpoint: false,
}
}
// scan shard
err := c.ScanShard(context.Background(), "myShard", fn)
if err != nil {
t.Fatalf("scan shard error: %v", err)
}
// increments counter
if val := ctr.counter; val != 2 {
t.Fatalf("counter error expected %d, got %d", 2, val)
}
// sets checkpoint
val, err := ckp.Get("myStreamName", "myShard")
if err != nil && val != "lastSeqNum" {
t.Fatalf("checkout error expected %s, got %s", "lastSeqNum", val)
}
// calls callback func
if resultData != "firstDatalastData" {
t.Fatalf("callback error expected %s, got %s", "firstDatalastData", val)
}
}
type mockKinesisClient struct {
kinesisiface.KinesisAPI
}
func (m *mockKinesisClient) GetRecords(input *kinesis.GetRecordsInput) (*kinesis.GetRecordsOutput, error) {
return &kinesis.GetRecordsOutput{
Records: []*kinesis.Record{
var records = []*kinesis.Record{
&kinesis.Record{
Data: []byte("firstData"),
SequenceNumber: aws.String("firstSeqNum"),
@ -92,16 +28,110 @@ func (m *mockKinesisClient) GetRecords(input *kinesis.GetRecordsInput) (*kinesis
Data: []byte("lastData"),
SequenceNumber: aws.String("lastSeqNum"),
},
},
}, nil
}
}
func (m *mockKinesisClient) GetShardIterator(input *kinesis.GetShardIteratorInput) (*kinesis.GetShardIteratorOutput, error) {
var client = &kinesisClientMock{
getShardIteratorMock: func(input *kinesis.GetShardIteratorInput) (*kinesis.GetShardIteratorOutput, error) {
return &kinesis.GetShardIteratorOutput{
ShardIterator: aws.String("myshard"),
ShardIterator: aws.String("49578481031144599192696750682534686652010819674221576194"),
}, nil
},
getRecordsMock: func(input *kinesis.GetRecordsInput) (*kinesis.GetRecordsOutput, error) {
return &kinesis.GetRecordsOutput{
NextShardIterator: nil,
Records: records,
}, nil
},
}
var (
cp = &fakeCheckpoint{cache: map[string]string{}}
ctr = &fakeCounter{}
)
c, err := New("myStreamName",
WithClient(client),
WithCounter(ctr),
WithCheckpoint(cp),
)
if err != nil {
t.Fatalf("new consumer error: %v", err)
}
var resultData string
// callback fn appends record data
var fn = func(r *Record) ScanStatus {
resultData += string(r.Data)
return ScanStatus{}
}
// scan shard
if err := c.ScanShard(context.Background(), "myShard", fn); err != nil {
t.Fatalf("scan shard error: %v", err)
}
// runs callback func
if resultData != "firstDatalastData" {
t.Fatalf("callback error expected %s, got %s", "firstDatalastData", resultData)
}
// increments counter
if val := ctr.counter; val != 2 {
t.Fatalf("counter error expected %d, got %d", 2, val)
}
// sets checkpoint
val, err := cp.Get("myStreamName", "myShard")
if err != nil && val != "lastSeqNum" {
t.Fatalf("checkout error expected %s, got %s", "lastSeqNum", val)
}
}
func TestScanShard_ShardIsClosed(t *testing.T) {
var client = &kinesisClientMock{
getShardIteratorMock: func(input *kinesis.GetShardIteratorInput) (*kinesis.GetShardIteratorOutput, error) {
return &kinesis.GetShardIteratorOutput{
ShardIterator: aws.String("49578481031144599192696750682534686652010819674221576194"),
}, nil
},
getRecordsMock: func(input *kinesis.GetRecordsInput) (*kinesis.GetRecordsOutput, error) {
return &kinesis.GetRecordsOutput{
NextShardIterator: nil,
Records: make([]*Record, 0),
}, nil
},
}
c, err := New("myStreamName", WithClient(client))
if err != nil {
t.Fatalf("new consumer error: %v", err)
}
var fn = func(r *Record) ScanStatus {
return ScanStatus{}
}
if err := c.ScanShard(context.Background(), "myShard", fn); err != nil {
t.Fatalf("scan shard error: %v", err)
}
}
type kinesisClientMock struct {
kinesisiface.KinesisAPI
getShardIteratorMock func(*kinesis.GetShardIteratorInput) (*kinesis.GetShardIteratorOutput, error)
getRecordsMock func(*kinesis.GetRecordsInput) (*kinesis.GetRecordsOutput, error)
}
func (c *kinesisClientMock) GetRecords(in *kinesis.GetRecordsInput) (*kinesis.GetRecordsOutput, error) {
return c.getRecordsMock(in)
}
func (c *kinesisClientMock) GetShardIterator(in *kinesis.GetShardIteratorInput) (*kinesis.GetShardIteratorOutput, error) {
return c.getShardIteratorMock(in)
}
// implementation of checkpoint
type fakeCheckpoint struct {
cache map[string]string
mu sync.Mutex
@ -124,6 +154,7 @@ func (fc *fakeCheckpoint) Get(streamName, shardID string) (string, error) {
return fc.cache[key], nil
}
// implementation of counter
type fakeCounter struct {
counter int64
}

1
counter.go
View file

@ -5,6 +5,7 @@ type Counter interface {
Add(string, int64)
}
// noopCounter implements counter interface with discard
type noopCounter struct{}
func (n noopCounter) Add(string, int64) {}

2
logger.go
View file

@ -9,8 +9,6 @@ type Logger interface {
Log(...interface{})
}
type LoggerFunc func(...interface{})
// noopLogger implements logger interface with discard
type noopLogger struct {
logger *log.Logger