fixed merge conflict and fixed unit tests

pkg/source/kinesis/kinesis_source_test.go

@@ -79,7 +79,7 @@ func TestNewKinesisSourceWithInterfaces_ConnectionCheck(t *testing.T) {
 	kinesisClient := testutil.GetAWSLocalstackKinesisClient()
 	dynamodbClient := testutil.GetAWSLocalstackDynamoDBClient()
 
-	_, err := newKinesisSourceWithInterfaces(kinesisClient, dynamodbClient, "00000000000", 15, testutil.AWSLocalstackRegion, "nonexistent-stream", "test", nil)
+	_, err := newKinesisSourceWithInterfaces(kinesisClient, dynamodbClient, "00000000000", 15, testutil.AWSLocalstackRegion, "nonexistent-stream", "test", nil, 10, 10, nil)
 
 	if err != nil {
 		assert.Equal("ResourceNotFoundException: Stream nonexistent-stream under account 000000000000 not found.", err.Error())
@@ -259,13 +259,41 @@ func TestGetSource_ConfigErrorLeaderAction(t *testing.T) {
 
 	assert := assert.New(t)
 
+	// Set up localstack resources
 	kinesisClient := testutil.GetAWSLocalstackKinesisClient()
 	dynamodbClient := testutil.GetAWSLocalstackDynamoDBClient()
 
+	streamName := "kinesis-source-integration-2"
+	createErr := testutil.CreateAWSLocalstackKinesisStream(kinesisClient, streamName)
+	if createErr != nil {
+		t.Fatal(createErr)
+	}
+	defer testutil.DeleteAWSLocalstackKinesisStream(kinesisClient, streamName)
+
+	appName := "integration"
+	ddbErr := testutil.CreateAWSLocalstackDynamoDBTables(dynamodbClient, appName)
+	if ddbErr != nil {
+		t.Fatal(ddbErr)
+	}
+	defer testutil.DeleteAWSLocalstackDynamoDBTables(dynamodbClient, appName)
+
+	// Put ten records into kinesis stream
+	putErr := putNRecordsIntoKinesis(kinesisClient, 10, streamName, "Test")
+	if putErr != nil {
+		t.Fatal(putErr)
+	}
+
+	time.Sleep(1 * time.Second)
+
+	// Create the source and assert that it's there
 	source, err := newKinesisSourceWithInterfaces(kinesisClient, dynamodbClient, "00000000000", 15, testutil.AWSLocalstackRegion, "something", "test", nil, 0, 10, &one)
 
 	assert.Nil(source)
 	assert.EqualError(err, `Failed to create Kinsumer client: leaderActionFrequency config value is mandatory and must be at least as long as ShardCheckFrequency`)
+
+	if testing.Short() {
+		t.Skip("skipping integration test")
+	}
 }
 
 func TestGetSource_ConfigErrorMaxAge(t *testing.T) {