added multiple checks (sqs src/trg, kinesis trg)

Author: TiganeteaRobert

Diff for: pkg/source/kinesis/kinesis_source_test.go

@@ -290,10 +290,6 @@ func TestGetSource_ConfigErrorLeaderAction(t *testing.T) {
 
 	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) {

Diff for: pkg/source/sqs/sqs_source.go

@@ -115,6 +115,11 @@ var ConfigPair = sourceconfig.ConfigPair{
 // newSQSSourceWithInterfaces allows you to provide an SQS client directly to allow
 // for mocking and localstack usage
 func newSQSSourceWithInterfaces(client sqsiface.SQSAPI, awsAccountID string, concurrentWrites int, region string, queueName string) (*sqsSource, error) {
+	_, err := client.GetQueueUrl(&sqs.GetQueueUrlInput{QueueName: &queueName})
+	if err != nil {
+		return nil, errors.Wrap(err, `Could not connect to SQS`)
+	}
+
 	return &sqsSource{
 		client:             client,
 		queueName:          queueName,

Diff for: pkg/source/sqs/sqs_source_test.go

@@ -11,6 +11,7 @@ import (
 	"os"
 	"path/filepath"
 	"reflect"
+	"strings"
 	"testing"
 	"time"
 
@@ -30,8 +31,8 @@ func TestMain(m *testing.M) {
 	os.Exit(exitVal)
 }
 
-// TestNewSqsSource_ConnectionCheck tests that the SQS source fails on start-up if the connection to SQS fails
-func TestNewSqsSource_ConnectionCheck(t *testing.T) {
+// TestNewSqsSource_AWSConnectionCheck tests that the SQS source fails on start-up if the connection to AWS fails
+func TestNewSqsSource_AWSConnectionCheck(t *testing.T) {
 	assert := assert.New(t)
 
 	target, err := configFunction(&configuration{
@@ -66,20 +67,18 @@ func TestNewSQSSourceWithInterfaces_Success(t *testing.T) {
 }
 
 // newSQSSourceWithInterfaces should fail if we can't reach SQS
-func TestNewSQSSourceWithInterfaces_Failure(t *testing.T) {
+func TestNewSQSSourceWithInterfaces_SQSConnectioNFailure(t *testing.T) {
 	// Unlike the success test, we don't require anything to exist for this one
 	assert := assert.New(t)
 
 	client := testutil.GetAWSLocalstackSQSClient()
 
 	source, err := newSQSSourceWithInterfaces(client, "00000000000", 10, testutil.AWSLocalstackRegion, "nonexistent-queue")
 
-	assert.NotNil(source)
-	assert.Nil(err)
-
-	err = source.Read(nil)
+	assert.Nil(source)
+	assert.NotNil(err)
 	if err != nil {
-		assert.Equal(err.Error(), "Failed to get SQS queue URL: AWS.SimpleQueueService.NonExistentQueue: AWS.SimpleQueueService.NonExistentQueue; see the SQS docs.\n\tstatus code: 400, request id: 00000000-0000-0000-0000-000000000000")
+		assert.True(strings.HasPrefix(err.Error(), `Could not connect to SQS: RequestError: send request failed`))
 	}
 }
 

Diff for: pkg/target/kinesis.go

@@ -62,6 +62,12 @@ func newKinesisTarget(region string, streamName string, roleARN string) (*Kinesi
 // newKinesisTargetWithInterfaces allows you to provide a Kinesis client directly to allow
 // for mocking and localstack usage
 func newKinesisTargetWithInterfaces(client kinesisiface.KinesisAPI, awsAccountID string, region string, streamName string) (*KinesisTarget, error) {
+	// test the connection to kinesis by trying to make an API call
+	_, err := client.DescribeStream(&kinesis.DescribeStreamInput{StreamName: &streamName})
+	if err != nil {
+		return nil, err
+	}
+
 	return &KinesisTarget{
 		client:     client,
 		streamName: streamName,

Diff for: pkg/target/kinesis_test.go

@@ -7,6 +7,7 @@
 package target
 
 import (
+	"strings"
 	"sync/atomic"
 	"testing"
 
@@ -28,7 +29,7 @@ func TestNewKinesisTarget_ConnectionCheck(t *testing.T) {
 	}
 }
 
-func TestKinesisTarget_WriteFailure(t *testing.T) {
+func TestKinesisTarget_KinesisConnectionFailure(t *testing.T) {
 	if testing.Short() {
 		t.Skip("skipping integration test")
 	}
@@ -38,25 +39,11 @@ func TestKinesisTarget_WriteFailure(t *testing.T) {
 	client := testutil.GetAWSLocalstackKinesisClient()
 
 	target, err := newKinesisTargetWithInterfaces(client, "00000000000", testutil.AWSLocalstackRegion, "not-exists")
-	assert.Nil(err)
-	assert.NotNil(target)
-	assert.Equal("arn:aws:kinesis:us-east-1:00000000000:stream/not-exists", target.GetID())
-
-	defer target.Close()
-	target.Open()
-
-	messages := testutil.GetTestMessages(1, "Hello Kinesis!!", nil)
-
-	writeRes, err := target.Write(messages)
 	assert.NotNil(err)
+	assert.Nil(target)
 	if err != nil {
-		assert.Equal("Error writing messages to Kinesis stream: 1 error occurred:\n\t* Failed to send message batch to Kinesis stream: ResourceNotFoundException: Stream not-exists under account 000000000000 not found.\n\n", err.Error())
+		assert.True(strings.HasPrefix(err.Error(), `RequestError`))
 	}
-	assert.NotNil(writeRes)
-
-	// Check results
-	assert.Equal(int64(0), writeRes.SentCount)
-	assert.Equal(int64(1), writeRes.FailedCount)
 }
 
 func TestKinesisTarget_WriteSuccess(t *testing.T) {

Diff for: pkg/target/sqs.go

@@ -64,6 +64,11 @@ func newSQSTarget(region string, queueName string, roleARN string) (*SQSTarget,
 // newSQSTargetWithInterfaces allows you to provide an SQS client directly to allow
 // for mocking and localstack usage
 func newSQSTargetWithInterfaces(client sqsiface.SQSAPI, awsAccountID string, region string, queueName string) (*SQSTarget, error) {
+	_, err := client.GetQueueUrl(&sqs.GetQueueUrlInput{QueueName: &queueName})
+	if err != nil {
+		return nil, errors.Wrap(err, `Could not connect to SQS`)
+	}
+
 	return &SQSTarget{
 		client:    client,
 		queueName: queueName,

Diff for: pkg/target/sqs_test.go

@@ -7,6 +7,7 @@
 package target
 
 import (
+	"strings"
 	"sync/atomic"
 	"testing"
 
@@ -15,16 +16,16 @@ import (
 	"github.com/snowplow-devops/stream-replicator/pkg/testutil"
 )
 
-// TestNewSqsTarget_ConnectionCheck tests that the SQS target fails on start-up if the connection to SQS fails
-func TestNewSqsTarget_ConnectionCheck(t *testing.T) {
+// TestNewSqsTarget_AWSConnectionCheck tests that the SQS target fails on start-up if the connection to AWS fails
+func TestNewSqsTarget_AWSConnectionCheck(t *testing.T) {
 	assert := assert.New(t)
 
 	target, err := newSQSTarget(testutil.AWSLocalstackRegion, "not-exists", `arn:aws:sqs:us-east-1:00000000000:not-exists`)
 	assert.Nil(target)
 	assert.EqualError(err, "NoCredentialProviders: no valid providers in chain. Deprecated.\n\tFor verbose messaging see aws.Config.CredentialsChainVerboseErrors")
 }
 
-func TestSQSTarget_WriteFailure(t *testing.T) {
+func TestSQSTarget_SQSConnectionFailure(t *testing.T) {
 	if testing.Short() {
 		t.Skip("skipping integration test")
 	}
@@ -34,13 +35,11 @@ func TestSQSTarget_WriteFailure(t *testing.T) {
 	client := testutil.GetAWSLocalstackSQSClient()
 
 	target, err := newSQSTargetWithInterfaces(client, "00000000000", testutil.AWSLocalstackRegion, "not-exists")
-	assert.Nil(err)
-	assert.NotNil(target)
-	assert.Equal("arn:aws:sqs:us-east-1:00000000000:not-exists", target.GetID())
-
-	res, err := target.Write(nil)
-	assert.Nil(err)
-	assert.NotNil(res)
+	assert.Nil(target)
+	assert.NotNil(err)
+	if err != nil {
+		assert.True(strings.HasPrefix(err.Error(), `Could not connect to SQS: RequestError: send request failed`))
+	}
 }
 
 func TestSQSTarget_WriteSuccess(t *testing.T) {