Support local Kinesis e.g. Kinesalite

And add some tests that run against the local Kinesis.

I need this for testing my own app which uses go-kinesis. But I think it’s also useful for testing go-kinesis itself. I have a feature I’d like to add but I don’t think it’d be safe to do so without some functional tests.

Author: avigavin

README.md

@@ -12,3 +12,15 @@ Example you can find in folder `examples`.
 ## Command line interface
 
 You can find a tool for interacting with kinesis from the command line in folder `kinesis-cli`.
+
+## Testing
+
+The tests require a local Kinesis server such as [Kinesalite](https://github.com/mhart/kinesalite)
+to be running and reachable at `http://127.0.0.1:4567`.
+
+To make the tests complete faster, you might want to have Kinesalite perform stream creation and
+deletion faster than the default of 500ms, like so:
+
+    kinesalite --createStreamMs 5 --deleteStreamMs 5 &
+
+The `&` runs Kinesalite in the background, which is probably what you want.

client.go

@@ -141,7 +141,10 @@ func (c *Client) client() *http.Client {
 
 // Do some request, but sign it before sending
 func (c *Client) Do(req *http.Request) (resp *http.Response, err error) {
-	Sign(c.Auth, req)
+	err = Sign(c.Auth, req)
+	if err != nil {
+		return nil, err
+	}
 
 	if !c.Auth.Expiry.IsZero() {
 		if time.Now().After(c.Auth.Expiry) {

kinesis.go

@@ -31,9 +31,10 @@ var (
 
 // Structure for kinesis client
 type Kinesis struct {
-	client  *Client
-	Region  string
-	Version string
+	client   *Client
+	endpoint string
+	Region   string
+	Version  string
 }
 
 // Interface implemented by Kinesis
@@ -50,9 +51,19 @@ type KinesisClient interface {
 	SplitShard(args *RequestArgs) error
 }
 
-// New returns an initialized AWS Kinesis client
+// New returns an initialized AWS Kinesis client using the canonical live “production” endpoint
+// for AWS Kinesis, i.e. https://kinesis.{region}.amazonaws.com
 func New(auth *Auth, region Region) *Kinesis {
-	return &Kinesis{client: NewClient(auth), Version: "20131202", Region: GetRegion(region)}
+	endpoint := fmt.Sprintf("https://kinesis.%s.amazonaws.com", GetRegion(region))
+	return NewWithEndpoint(auth, region, endpoint)
+}
+
+// NewWithEndpoint returns an initialized AWS Kinesis client using the specified endpoint.
+// This is generally useful for testing, so a local Kinesis server can be used.
+func NewWithEndpoint(auth *Auth, region Region, endpoint string) *Kinesis {
+	// TODO: remove trailing slash on endpoint if there is one? does it matter?
+	// TODO: validate endpoint somehow?
+	return &Kinesis{client: NewClient(auth), Version: "20131202", Region: GetRegion(region), endpoint: endpoint}
 }
 
 // Create params object for request
@@ -115,7 +126,7 @@ func buildError(r *http.Response) error {
 	err.Message = errors.Message
 	err.StatusCode = r.StatusCode
 	if err.Message == "" {
-		err.Message = r.Status
+		err.Message = fmt.Sprintf("%s\n%s", r.Status, r.Body)
 	}
 	return &err
 }
@@ -130,7 +141,7 @@ func (kinesis *Kinesis) query(params map[string]string, data interface{}, resp i
 	// request
 	request, err := http.NewRequest(
 		"POST",
-		fmt.Sprintf("https://kinesis.%s.amazonaws.com", kinesis.Region),
+		kinesis.endpoint,
 		bytes.NewReader(jsonData),
 	)
 

kinesis_test.go

@@ -1,6 +1,14 @@
 package kinesis
 
-import "testing"
+import (
+	"errors"
+	"fmt"
+	"strings"
+	"testing"
+	"time"
+)
+
+const localEndpoint = "http://127.0.0.1:4567"
 
 // Trivial test to make sure that Kinesis implements KinesisClient.
 func TestInterfaceIsImplemented(t *testing.T) {
@@ -40,3 +48,140 @@ func TestAddRecord(t *testing.T) {
 		t.Errorf("%q != %q", len(args.Records), 1)
 	}
 }
+
+func TestListStreams(t *testing.T) {
+	auth := &Auth{
+		AccessKey: "BAD_ACCESS_KEY",
+		SecretKey: "BAD_SECRET_KEY",
+	}
+
+	client := NewWithEndpoint(auth, USEast1, localEndpoint)
+	resp, err := client.ListStreams(NewArgs())
+	if resp == nil {
+		t.Error("resp == nil")
+	}
+	if err != nil {
+		t.Errorf("%q != nil", err)
+	}
+}
+
+func TestCreateStream(t *testing.T) {
+	auth := &Auth{
+		AccessKey: "BAD_ACCESS_KEY",
+		SecretKey: "BAD_SECRET_KEY",
+	}
+
+	client := NewWithEndpoint(auth, USEast1, localEndpoint)
+
+	streamName := "test2"
+
+	err := client.CreateStream(streamName, 1)
+	if err != nil {
+		t.Errorf("%q != nil", err)
+	}
+
+	err = waitForStreamStatus(client, streamName, "ACTIVE")
+	if err != nil {
+		t.Errorf("%q != nil", err)
+	}
+
+	client.DeleteStream(streamName)
+	err = waitForStreamDeletion(client, streamName)
+	if err != nil {
+		t.Errorf("%q != nil", err)
+	}
+}
+
+func TestPutRecord(t *testing.T) {
+	auth := &Auth{
+		AccessKey: "BAD_ACCESS_KEY",
+		SecretKey: "BAD_SECRET_KEY",
+	}
+
+	client := NewWithEndpoint(auth, USEast1, localEndpoint)
+
+	streamName := "pizza"
+	err := createStream(client, streamName, 1)
+
+	if err != nil {
+		t.Errorf("%q != nil", err)
+	}
+
+	args := NewArgs()
+	args.Add("StreamName", streamName)
+	args.AddData([]byte("The cheese is old and moldy, where is the bathroom?"))
+	args.Add("PartitionKey", "key-1")
+
+	resp, err := client.PutRecord(args)
+	if resp == nil {
+		t.Error("resp == nil")
+	}
+	if err != nil {
+		t.Errorf("%q != nil", err)
+	}
+
+	client.DeleteStream(streamName)
+	err = waitForStreamDeletion(client, streamName)
+	if err != nil {
+		t.Errorf("%q != nil", err)
+	}
+}
+
+// waitForStreamStatus will poll for a stream status repeatedly, once every MS, for up to 1000 MS,
+// blocking until the stream has the desired status. It will return an error if the stream never
+// achieves the desired status. If a stream doesn’t exist then an error will be returned.
+func waitForStreamStatus(client *Kinesis, streamName string, statusToAwait string) error {
+	args := NewArgs()
+	args.Add("StreamName", streamName)
+	var resp3 *DescribeStreamResp
+	var err error
+
+	for i := 1; i < 1000; i++ {
+		resp3, err = client.DescribeStream(args)
+		if err != nil {
+			return err
+		}
+
+		if resp3.StreamDescription.StreamStatus == statusToAwait {
+			break
+		} else {
+			time.Sleep(1 * time.Millisecond)
+		}
+	}
+
+	if resp3 == nil {
+		return errors.New("Could not get Stream Description")
+	}
+
+	if resp3.StreamDescription.StreamStatus != statusToAwait {
+		return errors.New(fmt.Sprintf("Timed out waiting for stream to enter status %v; last status was %v.", statusToAwait, resp3.StreamDescription.StreamStatus))
+	}
+
+	return nil
+}
+
+// waitForStreamDeletion will poll for a stream status repeatedly, once every MS, for up to 1000 MS,
+// blocking until the stream has been deleted. It will return an error if the stream is never deleted
+// or some other error occurs. If it succeeds then the return value will be nil.
+func waitForStreamDeletion(client *Kinesis, streamName string) error {
+	err := waitForStreamStatus(client, streamName, "FOO")
+	if !strings.Contains(err.Error(), "not found") {
+		return err
+	}
+	return nil
+}
+
+// helper
+func createStream(client *Kinesis, streamName string, partitions int) error {
+	err := client.CreateStream(streamName, partitions)
+	if err != nil {
+		return err
+	}
+
+	err = waitForStreamStatus(client, streamName, "ACTIVE")
+	if err != nil {
+		return err
+	}
+
+	return nil
+}