luciano/vmware-go-kcl-v2
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Update unit tests

Browse source 
Signed-off-by: Tao Jiang <taoj@vmware.com>
This commit is contained in:
Tao Jiang 2021-12-22 22:16:06 -06:00
parent 86cc5a1a64
commit c02b7a85d4
9 changed files with 259 additions and 120 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

10
.github/workflows/vmware-go-kcl-v2-ci.yml vendored
View file

@ -5,7 +5,7 @@ on:
branches: [ main ]
paths-ignore: [ README.md ]
pull_request:
branches: [ master ]
branches: [ main ]
paths-ignore: [ README.md ]
jobs:
@ -27,10 +27,10 @@ jobs:
run: |
make build
# - name: Test
# shell: bash
# run: |
# make test
- name: Test
shell: bash
run: |
make test
scans:
name: Checks, Lints and Scans

13
.gitignore vendored
View file

@ -1,11 +1,6 @@
# Ignore everything in this directory
*
# Except this file
!.gitignore
=======
/src/gen
/src/vendor
!/src/vendor/manifest
/gen
/vendor
!/vendor/manifest
/bin
/pkg
/tmp
@ -22,5 +17,3 @@
filenames
.DS_Store
.scannerwork/
*.sarif

1
README.md
View file

@ -1,5 +1,6 @@
# VMWare Go KCL v2
![technology Go](https://img.shields.io/badge/technology-go-blue.svg)
[![Go Report Card](https://goreportcard.com/badge/github.com/vmware/vmware-go-kcl)](https://goreportcard.com/report/github.com/vmware/vmware-go-kcl)
[![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](https://opensource.org/licenses/MIT)
[![vmware-go-kcl-v2](https://github.com/fafg/vmware-go-kcl/actions/workflows/vmware-go-kcl-v2-ci.yml/badge.svg)](https://github.com/fafg/vmware-go-kcl/actions/workflows/vmware-go-kcl-v2-ci.yml)

106
clientlibrary/checkpoint/dynamodb-api.go Normal file
View file

@ -0,0 +1,106 @@
/*
* Copyright (c) 2021 VMware, Inc.
*
* Permission is hereby granted, free of charge, to any person obtaining a copy of this software and
* associated documentation files (the "Software"), to deal in the Software without restriction, including
* without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
* copies of the Software, and to permit persons to whom the Software is furnished to do
* so, subject to the following conditions:
*
* The above copyright notice and this permission notice shall be included in all copies or substantial
* portions of the Software.
*
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT
* NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.
* IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY,
* WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE
* SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
*/
package checkpoint
import (
"context"
"github.com/aws/aws-sdk-go-v2/service/dynamodb"
)
// DynamoDBAPI provides an interface to enable mocking the
// dynamodb.DynamoDB service client's API operation,
// paginators, and waiters. This make unit testing your code that calls out
// to the SDK's service client's calls easier.
type DynamoDBAPI interface {
// The Scan operation returns one or more items and item attributes by accessing
// every item in a table or a secondary index. To have DynamoDB return fewer items,
// you can provide a FilterExpression operation. If the total number of scanned
// items exceeds the maximum dataset size limit of 1 MB, the scan stops and results
// are returned to the user as a LastEvaluatedKey value to continue the scan in a
// subsequent operation. The results also include the number of items exceeding the
// limit. A scan can result in no table data meeting the filter criteria. A single
// Scan operation reads up to the maximum number of items set (if using the Limit
// parameter) or a maximum of 1 MB of data and then apply any filtering to the
// results using FilterExpression. If LastEvaluatedKey is present in the response,
// you need to paginate the result set.
Scan(ctx context.Context, params *dynamodb.ScanInput, optFns ...func(*dynamodb.Options)) (*dynamodb.ScanOutput, error)
// DescribeTable returns information about the table, including the current status of the table,
// when it was created, the primary key schema, and any indexes on the table. If
// you issue a DescribeTable request immediately after a CreateTable request,
// DynamoDB might return a ResourceNotFoundException. This is because DescribeTable
// uses an eventually consistent query, and the metadata for your table might not
// be available at that moment. Wait for a few seconds, and then try the
// DescribeTable request again.
DescribeTable(ctx context.Context, params *dynamodb.DescribeTableInput, optFns ...func(*dynamodb.Options)) (*dynamodb.DescribeTableOutput, error)
// The CreateTable operation adds a new table to your account. In an AWS account,
// table names must be unique within each Region. That is, you can have two tables
// with same name if you create the tables in different Regions. CreateTable is an
// asynchronous operation. Upon receiving a CreateTable request, DynamoDB
// immediately returns a response with a TableStatus of CREATING. After the table
// is created, DynamoDB sets the TableStatus to ACTIVE. You can perform read and
// write operations only on an ACTIVE table. You can optionally define secondary
// indexes on the new table, as part of the CreateTable operation. If you want to
// create multiple tables with secondary indexes on them, you must create the
// tables sequentially. Only one table with secondary indexes can be in the
// CREATING state at any given time. You can use the DescribeTable action to check
// the table status.
CreateTable(ctx context.Context, params *dynamodb.CreateTableInput, optFns ...func(*dynamodb.Options)) (*dynamodb.CreateTableOutput, error)
// PutItem creates a new item, or replaces an old item with a new item. If an item that has
// the same primary key as the new item already exists in the specified table, the
// new item completely replaces the existing item. You can perform a conditional
// put operation (add a new item if one with the specified primary key doesn't
// exist), or replace an existing item if it has certain attribute values. You can
// return the item's attribute values in the same operation, using the ReturnValues
// parameter.
PutItem(ctx context.Context, params *dynamodb.PutItemInput, optFns ...func(*dynamodb.Options)) (*dynamodb.PutItemOutput, error)
// The GetItem operation returns a set of attributes for the item with the given
// primary key. If there is no matching item, GetItem does not return any data and
// there will be no Item element in the response. GetItem provides an eventually
// consistent read by default. If your application requires a strongly consistent
// read, set ConsistentRead to true. Although a strongly consistent read might take
// more time than an eventually consistent read, it always returns the last updated
// value.
GetItem(ctx context.Context, params *dynamodb.GetItemInput, optFns ...func(*dynamodb.Options)) (*dynamodb.GetItemOutput, error)
// UpdateItem edits an existing item's attributes, or adds a new item to the table if it does
// not already exist. You can put, delete, or add attribute values. You can also
// perform a conditional update on an existing item (insert a new attribute
// name-value pair if it doesn't exist, or replace an existing name-value pair if
// it has certain expected attribute values). You can also return the item's
// attribute values in the same UpdateItem operation using the ReturnValues
// parameter.
UpdateItem(ctx context.Context, params *dynamodb.UpdateItemInput, optFns ...func(*dynamodb.Options)) (*dynamodb.UpdateItemOutput, error)
// DeleteItem deletes a single item in a table by primary key. You can perform a conditional
// delete operation that deletes the item if it exists, or if it has an expected
// attribute value. In addition to deleting an item, you can also return the item's
// attribute values in the same operation, using the ReturnValues parameter. Unless
// you specify conditions, the DeleteItem is an idempotent operation; running it
// multiple times on the same item or attribute does not result in an error
// response. Conditional deletes are useful for deleting items only if specific
// conditions are met. If those conditions are met, DynamoDB performs the delete.
// Otherwise, the item is not deleted.
DeleteItem(ctx context.Context, params *dynamodb.DeleteItemInput, optFns ...func(*dynamodb.Options)) (*dynamodb.DeleteItemOutput, error)
}

29
clientlibrary/checkpoint/dynamodb-checkpointer.go
View file

@ -38,7 +38,6 @@ import (
"github.com/aws/aws-sdk-go-v2/aws"
"github.com/aws/aws-sdk-go-v2/aws/retry"
awsConfig "github.com/aws/aws-sdk-go-v2/config"
"github.com/aws/aws-sdk-go-v2/credentials"
"github.com/aws/aws-sdk-go-v2/service/dynamodb"
"github.com/aws/aws-sdk-go-v2/service/dynamodb/types"
@ -60,7 +59,7 @@ type DynamoCheckpoint struct {
leaseTableWriteCapacity int64
LeaseDuration int
svc *dynamodb.Client
svc DynamoDBAPI
kclConfig *config.KinesisClientLibConfiguration
Retries int
lastLeaseSync time.Time
@ -81,7 +80,7 @@ func NewDynamoCheckpoint(kclConfig *config.KinesisClientLibConfiguration) *Dynam
}
// WithDynamoDB is used to provide DynamoDB service
func (checkpointer *DynamoCheckpoint) WithDynamoDB(svc *dynamodb.Client) *DynamoCheckpoint {
func (checkpointer *DynamoCheckpoint) WithDynamoDB(svc DynamoDBAPI) *DynamoCheckpoint {
checkpointer.svc = svc
return checkpointer
}
@ -91,23 +90,23 @@ func (checkpointer *DynamoCheckpoint) Init() error {
checkpointer.log.Infof("Creating DynamoDB session")
if checkpointer.svc == nil {
resolver := aws.EndpointResolverFunc(func(service, region string) (aws.Endpoint, error) {
return aws.Endpoint{
PartitionID: "aws",
URL: checkpointer.kclConfig.DynamoDBEndpoint,
SigningRegion: checkpointer.kclConfig.RegionName,
}, nil
er := aws.EndpointResolverFunc(func(service, region string) (aws.Endpoint, error) {
if service == dynamodb.ServiceID && len(checkpointer.kclConfig.DynamoDBEndpoint) > 0 {
return aws.Endpoint{
PartitionID: "aws",
URL: checkpointer.kclConfig.DynamoDBEndpoint,
SigningRegion: checkpointer.kclConfig.RegionName,
}, nil
}
// returning EndpointNotFoundError will allow the service to fallback to it's default resolution
return aws.Endpoint{}, &aws.EndpointNotFoundError{}
})
cfg, err := awsConfig.LoadDefaultConfig(
context.TODO(),
awsConfig.WithRegion(checkpointer.kclConfig.RegionName),
awsConfig.WithCredentialsProvider(
credentials.NewStaticCredentialsProvider(
checkpointer.kclConfig.DynamoDBCredentials.Value.AccessKeyID,
checkpointer.kclConfig.DynamoDBCredentials.Value.SecretAccessKey,
checkpointer.kclConfig.DynamoDBCredentials.Value.SessionToken)),
awsConfig.WithEndpointResolver(resolver),
awsConfig.WithCredentialsProvider(checkpointer.kclConfig.DynamoDBCredentials),
awsConfig.WithEndpointResolver(er),
awsConfig.WithRetryer(func() aws.Retryer {
return retry.AddWithMaxBackoffDelay(retry.NewStandard(), retry.DefaultMaxBackoff)
}),

100
clientlibrary/checkpoint/dynamodb-checkpointer_test.go
View file

@ -47,14 +47,14 @@ func TestDoesTableExist(t *testing.T) {
svc := &mockDynamoDB{client: nil, tableExist: true, item: map[string]types.AttributeValue{}}
checkpoint := &DynamoCheckpoint{
TableName: "TableName",
svc: svc.client,
svc: svc,
}
if !checkpoint.doesTableExist() {
t.Error("Table exists but returned false")
}
svc = &mockDynamoDB{tableExist: false}
checkpoint.svc = svc.client
checkpoint.svc = svc
if checkpoint.doesTableExist() {
t.Error("Table does not exist but returned true")
}
@ -69,7 +69,7 @@ func TestGetLeaseNotAcquired(t *testing.T) {
WithShardSyncIntervalMillis(5000).
WithFailoverTimeMillis(300000)
checkpoint := NewDynamoCheckpoint(kclConfig).WithDynamoDB(svc.client)
checkpoint := NewDynamoCheckpoint(kclConfig).WithDynamoDB(svc)
_ = checkpoint.Init()
err := checkpoint.GetLease(&par.ShardStatus{
ID: "0001",
@ -100,7 +100,7 @@ func TestGetLeaseAquired(t *testing.T) {
WithShardSyncIntervalMillis(5000).
WithFailoverTimeMillis(300000)
checkpoint := NewDynamoCheckpoint(kclConfig).WithDynamoDB(svc.client)
checkpoint := NewDynamoCheckpoint(kclConfig).WithDynamoDB(svc)
_ = checkpoint.Init()
marshalledCheckpoint := map[string]types.AttributeValue{
LeaseKeyKey: &types.AttributeValueMemberS{
@ -175,7 +175,7 @@ func TestGetLeaseShardClaimed(t *testing.T) {
WithFailoverTimeMillis(300000).
WithLeaseStealing(true)
checkpoint := NewDynamoCheckpoint(kclConfig).WithDynamoDB(svc.client)
checkpoint := NewDynamoCheckpoint(kclConfig).WithDynamoDB(svc)
_ = checkpoint.Init()
err := checkpoint.GetLease(&par.ShardStatus{
ID: "0001",
@ -222,7 +222,7 @@ func TestGetLeaseClaimRequestExpiredOwner(t *testing.T) {
},
}
checkpoint := NewDynamoCheckpoint(kclConfig).WithDynamoDB(svc.client)
checkpoint := NewDynamoCheckpoint(kclConfig).WithDynamoDB(svc)
_ = checkpoint.Init()
err := checkpoint.GetLease(&par.ShardStatus{
ID: "0001",
@ -259,7 +259,7 @@ func TestGetLeaseClaimRequestExpiredClaimer(t *testing.T) {
},
}
checkpoint := NewDynamoCheckpoint(kclConfig).WithDynamoDB(svc.client)
checkpoint := NewDynamoCheckpoint(kclConfig).WithDynamoDB(svc)
_ = checkpoint.Init()
err := checkpoint.GetLease(&par.ShardStatus{
ID: "0001",
@ -294,7 +294,7 @@ func TestFetchCheckpointWithStealing(t *testing.T) {
WithFailoverTimeMillis(300000).
WithLeaseStealing(true)
checkpoint := NewDynamoCheckpoint(kclConfig).WithDynamoDB(svc.client)
checkpoint := NewDynamoCheckpoint(kclConfig).WithDynamoDB(svc)
_ = checkpoint.Init()
status := &par.ShardStatus{
@ -320,7 +320,7 @@ func TestGetLeaseConditional(t *testing.T) {
WithFailoverTimeMillis(300000).
WithLeaseStealing(true)
checkpoint := NewDynamoCheckpoint(kclConfig).WithDynamoDB(svc.client)
checkpoint := NewDynamoCheckpoint(kclConfig).WithDynamoDB(svc)
_ = checkpoint.Init()
marshalledCheckpoint := map[string]types.AttributeValue{
LeaseKeyKey: &types.AttributeValueMemberS{
@ -369,7 +369,7 @@ func TestListActiveWorkers(t *testing.T) {
kclConfig := cfg.NewKinesisClientLibConfig("appName", "test", "us-west-2", "abc").
WithLeaseStealing(true)
checkpoint := NewDynamoCheckpoint(kclConfig).WithDynamoDB(svc.client)
checkpoint := NewDynamoCheckpoint(kclConfig).WithDynamoDB(svc)
err := checkpoint.Init()
if err != nil {
t.Errorf("Checkpoint initialization failed: %+v", err)
@ -407,7 +407,7 @@ func TestListActiveWorkersErrShardNotAssigned(t *testing.T) {
kclConfig := cfg.NewKinesisClientLibConfig("appName", "test", "us-west-2", "abc").
WithLeaseStealing(true)
checkpoint := NewDynamoCheckpoint(kclConfig).WithDynamoDB(svc.client)
checkpoint := NewDynamoCheckpoint(kclConfig).WithDynamoDB(svc)
err := checkpoint.Init()
if err != nil {
t.Errorf("Checkpoint initialization failed: %+v", err)
@ -433,7 +433,7 @@ func TestClaimShard(t *testing.T) {
WithFailoverTimeMillis(300000).
WithLeaseStealing(true)
checkpoint := NewDynamoCheckpoint(kclConfig).WithDynamoDB(svc.client)
checkpoint := NewDynamoCheckpoint(kclConfig).WithDynamoDB(svc)
_ = checkpoint.Init()
marshalledCheckpoint := map[string]types.AttributeValue{
@ -485,79 +485,3 @@ func TestClaimShard(t *testing.T) {
assert.Equal(t, shard.Checkpoint, status.Checkpoint)
assert.Equal(t, shard.ParentShardId, status.ParentShardId)
}
type mockDynamoDB struct {
client *dynamodb.Client
tableExist bool
item map[string]types.AttributeValue
conditionalExpression string
expressionAttributeValues map[string]types.AttributeValue
}
func (m *mockDynamoDB) ScanPages(_ *dynamodb.ScanInput, _ func(*dynamodb.ScanOutput, bool) bool) error {
return nil
}
func (m *mockDynamoDB) DescribeTable(_ *dynamodb.DescribeTableInput) (*dynamodb.DescribeTableOutput, error) {
if !m.tableExist {
return &dynamodb.DescribeTableOutput{}, &types.ResourceNotFoundException{Message: aws.String("doesNotExist")}
}
return &dynamodb.DescribeTableOutput{}, nil
}
func (m *mockDynamoDB) PutItem(input *dynamodb.PutItemInput) (*dynamodb.PutItemOutput, error) {
item := input.Item
if shardID, ok := item[LeaseKeyKey]; ok {
m.item[LeaseKeyKey] = shardID
}
if owner, ok := item[LeaseOwnerKey]; ok {
m.item[LeaseOwnerKey] = owner
}
if timeout, ok := item[LeaseTimeoutKey]; ok {
m.item[LeaseTimeoutKey] = timeout
}
if checkpoint, ok := item[SequenceNumberKey]; ok {
m.item[SequenceNumberKey] = checkpoint
}
if parent, ok := item[ParentShardIdKey]; ok {
m.item[ParentShardIdKey] = parent
}
if claimRequest, ok := item[ClaimRequestKey]; ok {
m.item[ClaimRequestKey] = claimRequest
}
if input.ConditionExpression != nil {
m.conditionalExpression = *input.ConditionExpression
}
m.expressionAttributeValues = input.ExpressionAttributeValues
return nil, nil
}
func (m *mockDynamoDB) GetItem(_ *dynamodb.GetItemInput) (*dynamodb.GetItemOutput, error) {
return &dynamodb.GetItemOutput{
Item: m.item,
}, nil
}
func (m *mockDynamoDB) UpdateItem(input *dynamodb.UpdateItemInput) (*dynamodb.UpdateItemOutput, error) {
exp := input.UpdateExpression
if aws.ToString(exp) == "remove "+LeaseOwnerKey {
delete(m.item, LeaseOwnerKey)
}
return nil, nil
}
func (m *mockDynamoDB) CreateTable(_ *dynamodb.CreateTableInput) (*dynamodb.CreateTableOutput, error) {
return &dynamodb.CreateTableOutput{}, nil
}

116
clientlibrary/checkpoint/mock-dynamodb_test.go Normal file
View file

@ -0,0 +1,116 @@
/*
* Copyright (c) 2021 VMware, Inc.
*
* Permission is hereby granted, free of charge, to any person obtaining a copy of this software and
* associated documentation files (the "Software"), to deal in the Software without restriction, including
* without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
* copies of the Software, and to permit persons to whom the Software is furnished to do
* so, subject to the following conditions:
*
* The above copyright notice and this permission notice shall be included in all copies or substantial
* portions of the Software.
*
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT
* NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.
* IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY,
* WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE
* SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
*/
// The implementation is derived from https://github.com/patrobinson/gokini
//
// Copyright 2018 Patrick robinson
//
// Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions:
//
// The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software.
//
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
package checkpoint
import (
"context"
"github.com/aws/aws-sdk-go-v2/aws"
"github.com/aws/aws-sdk-go-v2/service/dynamodb"
"github.com/aws/aws-sdk-go-v2/service/dynamodb/types"
)
type mockDynamoDB struct {
client *dynamodb.Client
tableExist bool
item map[string]types.AttributeValue
conditionalExpression string
expressionAttributeValues map[string]types.AttributeValue
}
func (m *mockDynamoDB) Scan(ctx context.Context, params *dynamodb.ScanInput, optFns ...func(*dynamodb.Options)) (*dynamodb.ScanOutput, error) {
return &dynamodb.ScanOutput{}, nil
}
func (m *mockDynamoDB) DescribeTable(ctx context.Context, params *dynamodb.DescribeTableInput, optFns ...func(*dynamodb.Options)) (*dynamodb.DescribeTableOutput, error) {
if !m.tableExist {
return &dynamodb.DescribeTableOutput{}, &types.ResourceNotFoundException{Message: aws.String("doesNotExist")}
}
return &dynamodb.DescribeTableOutput{}, nil
}
func (m *mockDynamoDB) CreateTable(ctx context.Context, params *dynamodb.CreateTableInput, optFns ...func(*dynamodb.Options)) (*dynamodb.CreateTableOutput, error) {
return &dynamodb.CreateTableOutput{}, nil
}
func (m *mockDynamoDB) PutItem(ctx context.Context, params *dynamodb.PutItemInput, optFns ...func(*dynamodb.Options)) (*dynamodb.PutItemOutput, error) {
item := params.Item
if shardID, ok := item[LeaseKeyKey]; ok {
m.item[LeaseKeyKey] = shardID
}
if owner, ok := item[LeaseOwnerKey]; ok {
m.item[LeaseOwnerKey] = owner
}
if timeout, ok := item[LeaseTimeoutKey]; ok {
m.item[LeaseTimeoutKey] = timeout
}
if checkpoint, ok := item[SequenceNumberKey]; ok {
m.item[SequenceNumberKey] = checkpoint
}
if parent, ok := item[ParentShardIdKey]; ok {
m.item[ParentShardIdKey] = parent
}
if claimRequest, ok := item[ClaimRequestKey]; ok {
m.item[ClaimRequestKey] = claimRequest
}
if params.ConditionExpression != nil {
m.conditionalExpression = *params.ConditionExpression
}
m.expressionAttributeValues = params.ExpressionAttributeValues
return nil, nil
}
func (m *mockDynamoDB) GetItem(ctx context.Context, params *dynamodb.GetItemInput, optFns ...func(*dynamodb.Options)) (*dynamodb.GetItemOutput, error) {
return &dynamodb.GetItemOutput{
Item: m.item,
}, nil
}
func (m *mockDynamoDB) UpdateItem(ctx context.Context, params *dynamodb.UpdateItemInput, optFns ...func(*dynamodb.Options)) (*dynamodb.UpdateItemOutput, error) {
exp := params.UpdateExpression
if aws.ToString(exp) == "remove "+LeaseOwnerKey {
delete(m.item, LeaseOwnerKey)
}
return nil, nil
}
func (m *mockDynamoDB) DeleteItem(ctx context.Context, params *dynamodb.DeleteItemInput, optFns ...func(*dynamodb.Options)) (*dynamodb.DeleteItemOutput, error) {
return &dynamodb.DeleteItemOutput{}, nil
}

2
test/worker_custom_test.go
View file

@ -20,6 +20,7 @@ package test
import (
"context"
chk "github.com/vmware/vmware-go-kcl-v2/clientlibrary/checkpoint"
"os"
"sync"
"testing"
@ -30,7 +31,6 @@ import (
log "github.com/sirupsen/logrus"
"github.com/stretchr/testify/assert"
chk "github.com/vmware/vmware-go-kcl-v2/clientlibrary/checkpoint"
cfg "github.com/vmware/vmware-go-kcl-v2/clientlibrary/config"
par "github.com/vmware/vmware-go-kcl-v2/clientlibrary/partition"
wk "github.com/vmware/vmware-go-kcl-v2/clientlibrary/worker"

2
test/worker_lease_stealing_test.go
View file

@ -1,9 +1,9 @@
package test
import (
chk "github.com/vmware/vmware-go-kcl-v2/clientlibrary/checkpoint"
"testing"
chk "github.com/vmware/vmware-go-kcl-v2/clientlibrary/checkpoint"
cfg "github.com/vmware/vmware-go-kcl-v2/clientlibrary/config"
wk "github.com/vmware/vmware-go-kcl-v2/clientlibrary/worker"
"github.com/vmware/vmware-go-kcl-v2/logger"