Skip to content

Commit d75c04c

Browse files
rideshnath-scoutrideshnath-scout
committed
Revert changes from '/common/authentication/aws'
Signed-off-by: rideshnath-scout <ridesh.nath@scoutmotors.com>
1 parent bd2639a commit d75c04c

File tree

6 files changed

+187
-0
lines changed

6 files changed

+187
-0
lines changed

‎common/authentication/aws/client.go‎

Lines changed: 34 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,8 @@ import (
2323
"github.com/aws/aws-sdk-go/aws/session"
2424
"github.com/aws/aws-sdk-go/service/dynamodb"
2525
"github.com/aws/aws-sdk-go/service/dynamodb/dynamodbiface"
26+
"github.com/aws/aws-sdk-go/service/kinesis"
27+
"github.com/aws/aws-sdk-go/service/kinesis/kinesisiface"
2628
"github.com/aws/aws-sdk-go/service/s3"
2729
"github.com/aws/aws-sdk-go/service/s3/s3manager"
2830
"github.com/aws/aws-sdk-go/service/secretsmanager"
@@ -34,6 +36,7 @@ import (
3436
"github.com/aws/aws-sdk-go/service/ssm"
3537
"github.com/aws/aws-sdk-go/service/ssm/ssmiface"
3638
"github.com/aws/aws-sdk-go/service/sts"
39+
"github.com/vmware/vmware-go-kcl/clientlibrary/config"
3740
)
3841

3942
type Clients struct {
@@ -113,6 +116,7 @@ type ParameterStoreClients struct {
113116
}
114117

115118
type KinesisClients struct {
119+
Kinesis kinesisiface.KinesisAPI
116120
Region string
117121
Credentials *credentials.Credentials
118122
}
@@ -167,10 +171,40 @@ func (c *ParameterStoreClients) New(session *session.Session) {
167171
}
168172

169173
func (c *KinesisClients) New(session *session.Session) {
174+
c.Kinesis = kinesis.New(session, session.Config)
170175
c.Region = *session.Config.Region
171176
c.Credentials = session.Config.Credentials
172177
}
173178

179+
func (c *KinesisClients) Stream(ctx context.Context, streamName string) (*string, error) {
180+
if c.Kinesis != nil {
181+
stream, err := c.Kinesis.DescribeStreamWithContext(ctx, &kinesis.DescribeStreamInput{
182+
StreamName: aws.String(streamName),
183+
})
184+
if stream != nil {
185+
return stream.StreamDescription.StreamARN, err
186+
}
187+
}
188+
189+
return nil, errors.New("unable to get stream arn due to empty client")
190+
}
191+
192+
func (c *KinesisClients) WorkerCfg(ctx context.Context, stream, consumer, mode string) *config.KinesisClientLibConfiguration {
193+
const sharedMode = "shared"
194+
if c.Kinesis != nil {
195+
if mode == sharedMode {
196+
if c.Credentials != nil {
197+
kclConfig := config.NewKinesisClientLibConfigWithCredential(consumer,
198+
stream, c.Region, consumer,
199+
c.Credentials)
200+
return kclConfig
201+
}
202+
}
203+
}
204+
205+
return nil
206+
}
207+
174208
func (c *SesClients) New(session *session.Session) {
175209
c.Ses = ses.New(session, session.Config)
176210
}

‎common/authentication/aws/client_test.go‎

Lines changed: 146 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -19,12 +19,16 @@ import (
1919
"testing"
2020

2121
"github.com/aws/aws-sdk-go/aws"
22+
"github.com/aws/aws-sdk-go/aws/credentials"
2223
"github.com/aws/aws-sdk-go/aws/request"
2324
"github.com/aws/aws-sdk-go/aws/session"
25+
"github.com/aws/aws-sdk-go/service/kinesis"
26+
"github.com/aws/aws-sdk-go/service/kinesis/kinesisiface"
2427
"github.com/aws/aws-sdk-go/service/sqs"
2528
"github.com/aws/aws-sdk-go/service/sqs/sqsiface"
2629
"github.com/stretchr/testify/assert"
2730
"github.com/stretchr/testify/require"
31+
"github.com/vmware/vmware-go-kcl/clientlibrary/config"
2832
)
2933

3034
type mockedSQS struct {
@@ -36,7 +40,14 @@ func (m *mockedSQS) GetQueueUrlWithContext(ctx context.Context, input *sqs.GetQu
3640
return m.GetQueueURLFn(ctx, input)
3741
}
3842

43+
type mockedKinesis struct {
44+
kinesisiface.KinesisAPI
45+
DescribeStreamFn func(ctx context.Context, input *kinesis.DescribeStreamInput) (*kinesis.DescribeStreamOutput, error)
46+
}
3947

48+
func (m *mockedKinesis) DescribeStreamWithContext(ctx context.Context, input *kinesis.DescribeStreamInput, opts ...request.Option) (*kinesis.DescribeStreamOutput, error) {
49+
return m.DescribeStreamFn(ctx, input)
50+
}
4051

4152
func TestS3Clients_New(t *testing.T) {
4253
tests := []struct {
@@ -116,4 +127,139 @@ func TestSqsClients_QueueURL(t *testing.T) {
116127
}
117128
}
118129

130+
func TestKinesisClients_Stream(t *testing.T) {
131+
tests := []struct {
132+
name string
133+
kinesisClient *KinesisClients
134+
streamName string
135+
mockStreamARN *string
136+
mockError error
137+
expectedStream *string
138+
expectedErr error
139+
}{
140+
{
141+
name: "successfully retrieves stream ARN",
142+
kinesisClient: &KinesisClients{
143+
Kinesis: &mockedKinesis{DescribeStreamFn: func(ctx context.Context, input *kinesis.DescribeStreamInput) (*kinesis.DescribeStreamOutput, error) {
144+
return &kinesis.DescribeStreamOutput{
145+
StreamDescription: &kinesis.StreamDescription{
146+
StreamARN: aws.String("arn:aws:kinesis:some-region:123456789012:stream/some-stream"),
147+
},
148+
}, nil
149+
}},
150+
Region: "us-west-1",
151+
Credentials: credentials.NewStaticCredentials("accessKey", "secretKey", ""),
152+
},
153+
streamName: "some-stream",
154+
expectedStream: aws.String("arn:aws:kinesis:some-region:123456789012:stream/some-stream"),
155+
expectedErr: nil,
156+
},
157+
{
158+
name: "returns error when stream not found",
159+
kinesisClient: &KinesisClients{
160+
Kinesis: &mockedKinesis{DescribeStreamFn: func(ctx context.Context, input *kinesis.DescribeStreamInput) (*kinesis.DescribeStreamOutput, error) {
161+
return nil, errors.New("stream not found")
162+
}},
163+
Region: "us-west-1",
164+
Credentials: credentials.NewStaticCredentials("accessKey", "secretKey", ""),
165+
},
166+
streamName: "nonexistent-stream",
167+
expectedStream: nil,
168+
expectedErr: errors.New("unable to get stream arn due to empty client"),
169+
},
170+
}
171+
172+
for _, tt := range tests {
173+
t.Run(tt.name, func(t *testing.T) {
174+
got, err := tt.kinesisClient.Stream(t.Context(), tt.streamName)
175+
if tt.expectedErr != nil {
176+
require.Error(t, err)
177+
assert.Equal(t, tt.expectedErr.Error(), err.Error())
178+
} else {
179+
require.NoError(t, err)
180+
assert.Equal(t, tt.expectedStream, got)
181+
}
182+
})
183+
}
184+
}
185+
186+
func TestKinesisClients_WorkerCfg(t *testing.T) {
187+
testCreds := credentials.NewStaticCredentials("accessKey", "secretKey", "")
188+
tests := []struct {
189+
name string
190+
kinesisClient *KinesisClients
191+
streamName string
192+
consumer string
193+
mode string
194+
expectedConfig *config.KinesisClientLibConfiguration
195+
}{
196+
{
197+
name: "successfully creates shared mode worker config",
198+
kinesisClient: &KinesisClients{
199+
Kinesis: &mockedKinesis{
200+
DescribeStreamFn: func(ctx context.Context, input *kinesis.DescribeStreamInput) (*kinesis.DescribeStreamOutput, error) {
201+
return &kinesis.DescribeStreamOutput{
202+
StreamDescription: &kinesis.StreamDescription{
203+
StreamARN: aws.String("arn:aws:kinesis:us-east-1:123456789012:stream/existing-stream"),
204+
},
205+
}, nil
206+
},
207+
},
208+
Region: "us-west-1",
209+
Credentials: testCreds,
210+
},
211+
streamName: "existing-stream",
212+
consumer: "consumer1",
213+
mode: "shared",
214+
expectedConfig: config.NewKinesisClientLibConfigWithCredential(
215+
"consumer1", "existing-stream", "us-west-1", "consumer1", testCreds,
216+
),
217+
},
218+
{
219+
name: "returns nil when mode is not shared",
220+
kinesisClient: &KinesisClients{
221+
Kinesis: &mockedKinesis{
222+
DescribeStreamFn: func(ctx context.Context, input *kinesis.DescribeStreamInput) (*kinesis.DescribeStreamOutput, error) {
223+
return &kinesis.DescribeStreamOutput{
224+
StreamDescription: &kinesis.StreamDescription{
225+
StreamARN: aws.String("arn:aws:kinesis:us-east-1:123456789012:stream/existing-stream"),
226+
},
227+
}, nil
228+
},
229+
},
230+
Region: "us-west-1",
231+
Credentials: testCreds,
232+
},
233+
streamName: "existing-stream",
234+
consumer: "consumer1",
235+
mode: "exclusive",
236+
expectedConfig: nil,
237+
},
238+
{
239+
name: "returns nil when client is nil",
240+
kinesisClient: &KinesisClients{
241+
Kinesis: nil,
242+
Region: "us-west-1",
243+
Credentials: credentials.NewStaticCredentials("accessKey", "secretKey", ""),
244+
},
245+
streamName: "existing-stream",
246+
consumer: "consumer1",
247+
mode: "shared",
248+
expectedConfig: nil,
249+
},
250+
}
119251

252+
for _, tt := range tests {
253+
t.Run(tt.name, func(t *testing.T) {
254+
cfg := tt.kinesisClient.WorkerCfg(t.Context(), tt.streamName, tt.consumer, tt.mode)
255+
if tt.expectedConfig == nil {
256+
assert.Equal(t, tt.expectedConfig, cfg)
257+
return
258+
}
259+
assert.Equal(t, tt.expectedConfig.StreamName, cfg.StreamName)
260+
assert.Equal(t, tt.expectedConfig.EnhancedFanOutConsumerName, cfg.EnhancedFanOutConsumerName)
261+
assert.Equal(t, tt.expectedConfig.EnableEnhancedFanOutConsumer, cfg.EnableEnhancedFanOutConsumer)
262+
assert.Equal(t, tt.expectedConfig.RegionName, cfg.RegionName)
263+
})
264+
}
265+
}

‎go.mod‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -127,6 +127,7 @@ require (
127127
github.com/tetratelabs/wazero v1.7.0
128128
github.com/tmc/langchaingo v0.1.13
129129
github.com/valyala/fasthttp v1.53.0
130+
github.com/vmware/vmware-go-kcl v1.5.1
130131
github.com/vmware/vmware-go-kcl-v2 v1.0.0
131132
github.com/xdg-go/scram v1.1.2
132133
go.etcd.io/etcd/client/v3 v3.5.21

‎go.sum‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1717,6 +1717,8 @@ github.com/valyala/fasthttp v1.53.0 h1:lW/+SUkOxCx2vlIu0iaImv4JLrVRnbbkpCoaawvA4
17171717
github.com/valyala/fasthttp v1.53.0/go.mod h1:6dt4/8olwq9QARP/TDuPmWyWcl4byhpvTJ4AAtcz+QM=
17181718
github.com/valyala/fasttemplate v1.2.1/go.mod h1:KHLXt3tVN2HBp8eijSv/kGJopbvo7S+qRAEEKiv+SiQ=
17191719
github.com/valyala/tcplisten v0.0.0-20161114210144-ceec8f93295a/go.mod h1:v3UYOV9WzVtRmSR+PDvWpU/qWl4Wa5LApYYX4ZtKbio=
1720+
github.com/vmware/vmware-go-kcl v1.5.1 h1:1rJLfAX4sDnCyatNoD/WJzVafkwST6u/cgY/Uf2VgHk=
1721+
github.com/vmware/vmware-go-kcl v1.5.1/go.mod h1:kXJmQ6h0dRMRrp1uWU9XbIXvwelDpTxSPquvQUBdpbo=
17201722
github.com/vmware/vmware-go-kcl-v2 v1.0.0 h1:HPT5vu+khRmGspBSc/+AilEWbRGoTZhjlYqdrBbRMZs=
17211723
github.com/vmware/vmware-go-kcl-v2 v1.0.0/go.mod h1:GBDu+P4Neo0vwZAk0ZUCEC8GYsUOWvi3XhFwAZR3SjA=
17221724
github.com/x448/float16 v0.8.4 h1:qLwI1I70+NjRFUR3zs1JPUCgaCXSh3SW62uAKT1mSBM=

‎tests/certification/go.mod‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -294,6 +294,7 @@ require (
294294
github.com/tklauser/go-sysconf v0.3.12 // indirect
295295
github.com/tklauser/numcpus v0.6.1 // indirect
296296
github.com/tmc/langchaingo v0.1.13 // indirect
297+
github.com/vmware/vmware-go-kcl v1.5.1 // indirect
297298
github.com/x448/float16 v0.8.4 // indirect
298299
github.com/xdg-go/pbkdf2 v1.0.0 // indirect
299300
github.com/xdg-go/scram v1.1.2 // indirect
@@ -339,6 +340,7 @@ require (
339340
google.golang.org/grpc v1.73.0 // indirect
340341
google.golang.org/protobuf v1.36.6 // indirect
341342
gopkg.in/inf.v0 v0.9.1 // indirect
343+
gopkg.in/natefinch/lumberjack.v2 v2.2.1 // indirect
342344
gopkg.in/yaml.v2 v2.4.0 // indirect
343345
gopkg.in/yaml.v3 v3.0.1 // indirect
344346
k8s.io/api v0.31.0 // indirect

‎tests/certification/go.sum‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1432,6 +1432,8 @@ github.com/ugorji/go v1.2.6/go.mod h1:anCg0y61KIhDlPZmnH+so+RQbysYVyDko0IMgJv0Nn
14321432
github.com/ugorji/go/codec v1.2.6/go.mod h1:V6TCNZ4PHqoHGFZuSG1W8nrCzzdgA2DozYxWFFpvxTw=
14331433
github.com/urfave/cli v1.20.0/go.mod h1:70zkFmudgCuE/ngEzBv17Jvp/497gISqfk5gWijbERA=
14341434
github.com/urfave/cli v1.22.1/go.mod h1:Gos4lmkARVdJ6EkW0WaNv/tZAAMe9V7XWyB60NtXRu0=
1435+
github.com/vmware/vmware-go-kcl v1.5.1 h1:1rJLfAX4sDnCyatNoD/WJzVafkwST6u/cgY/Uf2VgHk=
1436+
github.com/vmware/vmware-go-kcl v1.5.1/go.mod h1:kXJmQ6h0dRMRrp1uWU9XbIXvwelDpTxSPquvQUBdpbo=
14351437
github.com/x448/float16 v0.8.4 h1:qLwI1I70+NjRFUR3zs1JPUCgaCXSh3SW62uAKT1mSBM=
14361438
github.com/x448/float16 v0.8.4/go.mod h1:14CWIYCyZA/cWjXOioeEpHeN/83MdbZDRQHoFcYsOfg=
14371439
github.com/xdg-go/pbkdf2 v1.0.0 h1:Su7DPu48wXMwC3bs7MCNG+z4FhcyEuz5dlvchbq0B0c=

0 commit comments

Comments
 (0)