added tests

internal/helloworld/greeter_server.go

@@ -15,7 +15,7 @@ import (
 // unary, client streaming, server streaming, bidi
 type CallType string
 
-// Unary is a uniry call
+// Unary is a unary call
 var Unary CallType = "unary"
 
 // ClientStream is a client streaming call
@@ -35,31 +35,48 @@ type Greeter struct {
 
 	mutex      *sync.RWMutex
 	callCounts map[CallType]int
+	calls map[CallType][][]*HelloRequest
 }
 
-func RandomSleep() {
+func randomSleep() {
 	msCount := rand.Intn(4) + 1
 	time.Sleep(time.Millisecond * time.Duration(msCount))
 }
 
+func (s *Greeter) recordCall(ct CallType) int {
+	s.mutex.Lock()
+	defer s.mutex.Unlock()
+
+	s.callCounts[ct]++
+	var messages []*HelloRequest
+	s.calls[ct] = append(s.calls[ct], messages)
+
+	return len(s.calls[ct]) - 1
+}
+
+func (s *Greeter) recordMessage(ct CallType, callIdx int, msg *HelloRequest) {
+	s.mutex.Lock()
+	defer s.mutex.Unlock()
+
+	s.calls[ct][callIdx] = append(s.calls[ct][callIdx], msg)
+}
+
 // SayHello implements helloworld.GreeterServer
 func (s *Greeter) SayHello(ctx context.Context, in *HelloRequest) (*HelloReply, error) {
-	s.mutex.Lock()
-	s.callCounts[Unary]++
-	s.mutex.Unlock()
+	callIdx := s.recordCall(Unary)
+	s.recordMessage(Unary, callIdx, in)
 
-	RandomSleep()
+	randomSleep()
 
 	return &HelloReply{Message: "Hello " + in.Name}, nil
 }
 
 // SayHellos lists all hellos
 func (s *Greeter) SayHellos(req *HelloRequest, stream Greeter_SayHellosServer) error {
-	s.mutex.Lock()
-	s.callCounts[ServerStream]++
-	s.mutex.Unlock()
+	callIdx := s.recordCall(ServerStream)
+	s.recordMessage(ServerStream, callIdx, req)
 
-	RandomSleep()
+	randomSleep()
 
 	for _, msg := range s.streamData {
 		if err := stream.Send(msg); err != nil {
@@ -72,34 +89,31 @@ func (s *Greeter) SayHellos(req *HelloRequest, stream Greeter_SayHellosServer) e
 
 // SayHelloCS is client streaming handler
 func (s *Greeter) SayHelloCS(stream Greeter_SayHelloCSServer) error {
-	s.mutex.Lock()
-	s.callCounts[ClientStream]++
-	s.mutex.Unlock()
+	callIdx := s.recordCall(ClientStream)
 
-	RandomSleep()
+	randomSleep()
 
 	msgCount := 0
 
 	for {
-		_, err := stream.Recv()
+		in, err := stream.Recv()
 		if err == io.EOF {
 			msgStr := fmt.Sprintf("Hello count: %d", msgCount)
 			return stream.SendAndClose(&HelloReply{Message: msgStr})
 		}
 		if err != nil {
 			return err
 		}
+		s.recordMessage(ClientStream, callIdx, in)
 		msgCount++
 	}
 }
 
 // SayHelloBidi duplex call handler
 func (s *Greeter) SayHelloBidi(stream Greeter_SayHelloBidiServer) error {
-	s.mutex.Lock()
-	s.callCounts[Bidi]++
-	s.mutex.Unlock()
+	callIdx := s.recordCall(Bidi)
 
-	RandomSleep()
+	randomSleep()
 
 	for {
 		in, err := stream.Recv()
@@ -110,6 +124,7 @@ func (s *Greeter) SayHelloBidi(stream Greeter_SayHelloBidiServer) error {
 			return err
 		}
 
+		s.recordMessage(Bidi, callIdx, in)
 		msg := "Hello " + in.Name
 		if err := stream.Send(&HelloReply{Message: msg}); err != nil {
 			return err
@@ -120,10 +135,19 @@ func (s *Greeter) SayHelloBidi(stream Greeter_SayHelloBidiServer) error {
 // ResetCounters resets the call counts
 func (s *Greeter) ResetCounters() {
 	s.mutex.Lock()
+
+	s.callCounts = make(map[CallType]int)
 	s.callCounts[Unary] = 0
 	s.callCounts[ServerStream] = 0
 	s.callCounts[ClientStream] = 0
 	s.callCounts[Bidi] = 0
+
+	s.calls = make(map[CallType][][]*HelloRequest)
+	s.calls[Unary] = make([][]*HelloRequest, 0)
+	s.calls[ServerStream] = make([][]*HelloRequest, 0)
+	s.calls[ClientStream] = make([][]*HelloRequest, 0)
+	s.calls[Bidi] = make([][]*HelloRequest, 0)
+
 	s.mutex.Unlock()
 
 	if s.Stats != nil {
@@ -144,6 +168,17 @@ func (s *Greeter) GetCount(key CallType) int {
 	return -1
 }
 
+func (s *Greeter) GetCalls(key CallType) [][]*HelloRequest {
+	s.mutex.Lock()
+	val, ok := s.calls[key]
+	s.mutex.Unlock()
+
+	if ok {
+		return val
+	}
+	return nil
+}
+
 // GetConnectionCount gets the connection count
 func (s *Greeter) GetConnectionCount() int {
 	return s.Stats.GetConnectionCount()
@@ -158,13 +193,10 @@ func NewGreeter() *Greeter {
 		&HelloReply{Message: "Hello Sara"},
 	}
 
-	m := make(map[CallType]int)
-	m[Unary] = 0
-	m[ServerStream] = 0
-	m[ClientStream] = 0
-	m[Bidi] = 0
+	greeter := &Greeter{streamData: streamData, mutex: &sync.RWMutex{}}
+	greeter.ResetCounters()
 
-	return &Greeter{streamData: streamData, callCounts: m, mutex: &sync.RWMutex{}}
+	return greeter
 }
 
 // NewHWStats creates new stats handler

runner/data_test.go

@@ -42,7 +42,7 @@ func TestData_createPayloads(t *testing.T) {
 	assert.NoError(t, err)
 	assert.NotNil(t, mtdTestUnaryTwo)
 
-	t.Run("get nil, empty when empty", func(t *testing.T) {
+	t.Run("get empty when empty", func(t *testing.T) {
 		inputs, err := createPayloads("", mtdUnary)
 		assert.NoError(t, err)
 		assert.Empty(t, inputs)
@@ -60,9 +60,7 @@ func TestData_createPayloads(t *testing.T) {
 		assert.Nil(t, inputs)
 	})
 
-	// TODO: update tests below that comment
-
-	t.Run("create single object from map for unary", func(t *testing.T) {
+	t.Run("create slice with single element from map for unary", func(t *testing.T) {
 		m1 := make(map[string]interface{})
 		m1["name"] = "bob"
 
@@ -75,7 +73,7 @@ func TestData_createPayloads(t *testing.T) {
 		assert.NotNil(t, (*inputs)[0])
 	})
 
-	t.Run("create array from map for client streaming", func(t *testing.T) {
+	t.Run("create slice with single element from map for client streaming", func(t *testing.T) {
 		m1 := make(map[string]interface{})
 		m1["name"] = "bob"
 
@@ -125,7 +123,7 @@ func TestData_createPayloads(t *testing.T) {
 		assert.Nil(t, inputs)
 	})
 
-	t.Run("get object for slice and unary", func(t *testing.T) {
+	t.Run("create slice of messages from slice for unary", func(t *testing.T) {
 		m1 := make(map[string]interface{})
 		m1["name"] = "bob"
 
@@ -145,7 +143,7 @@ func TestData_createPayloads(t *testing.T) {
 		assert.Len(t, *inputs, 3)
 	})
 
-	t.Run("create single object from map for unary with camelCase property", func(t *testing.T) {
+	t.Run("create slice with single object from map for unary with camelCase property", func(t *testing.T) {
 		m1 := make(map[string]interface{})
 		m1["paramOne"] = "bob"
 
@@ -158,7 +156,7 @@ func TestData_createPayloads(t *testing.T) {
 		assert.NotNil(t, (*inputs)[0])
 	})
 
-	t.Run("create single object from map for unary with snake_case property", func(t *testing.T) {
+	t.Run("create slice with single object from map for unary with snake_case property", func(t *testing.T) {
 		m1 := make(map[string]interface{})
 		m1["param_one"] = "bob"
 
@@ -171,7 +169,7 @@ func TestData_createPayloads(t *testing.T) {
 		assert.NotNil(t, (*inputs)[0])
 	})
 
-	t.Run("create single object from map for unary with nested camelCase property", func(t *testing.T) {
+	t.Run("create slice with single object from map for unary with nested camelCase property", func(t *testing.T) {
 		inner := make(map[string]interface{})
 		inner["paramOne"] = "bob"
 
@@ -187,7 +185,7 @@ func TestData_createPayloads(t *testing.T) {
 		assert.NotNil(t, (*inputs)[0])
 	})
 
-	t.Run("create single object from map for unary with nested snake_case property", func(t *testing.T) {
+	t.Run("create slice with single object from map for unary with nested snake_case property", func(t *testing.T) {
 		inner := make(map[string]interface{})
 		inner["param_one"] = "bob"
 

runner/run_test.go

@@ -1,6 +1,7 @@
 package runner
 
 import (
+	"strconv"
 	"sync"
 	"testing"
 	"time"
@@ -266,6 +267,88 @@ func TestRunUnary(t *testing.T) {
 		connCount := gs.GetConnectionCount()
 		assert.Equal(t, 5, connCount)
 	})
+
+	t.Run("test round-robin c = 2", func(t *testing.T) {
+		gs.ResetCounters()
+
+		data := make([]map[string]interface{}, 3)
+		for i:= 0; i < 3; i++ {
+			data[i] = make(map[string]interface{})
+			data[i]["name"] = strconv.Itoa(i)
+		}
+
+		report, err := Run(
+			"helloworld.Greeter.SayHello",
+			internal.TestLocalhost,
+			WithProtoFile("../testdata/greeter.proto", []string{}),
+			WithTotalRequests(6),
+			WithConcurrency(2),
+			WithTimeout(time.Duration(20*time.Second)),
+			WithDialTimeout(time.Duration(20*time.Second)),
+			WithInsecure(true),
+			WithData(data),
+			)
+
+		assert.NoError(t, err)
+		assert.NotNil(t, report)
+
+		count := gs.GetCount(callType)
+		assert.Equal(t, 6, count)
+
+		calls := gs.GetCalls(callType)
+		assert.NotNil(t, calls)
+		assert.Len(t, calls, 6)
+		names := make([]string, 0)
+		for _, msgs := range calls {
+			for _, msg := range msgs {
+				names = append(names, msg.GetName())
+			}
+		}
+
+		// we don't expect to have the same order of elements since requests are concurrent
+		assert.ElementsMatch(t, []string {"0", "1", "2", "0", "1", "2"}, names)
+	})
+
+	t.Run("test round-robin c = 1", func(t *testing.T) {
+		gs.ResetCounters()
+
+		data := make([]map[string]interface{}, 3)
+		for i:= 0; i < 3; i++ {
+			data[i] = make(map[string]interface{})
+			data[i]["name"] = strconv.Itoa(i)
+		}
+
+		report, err := Run(
+			"helloworld.Greeter.SayHello",
+			internal.TestLocalhost,
+			WithProtoFile("../testdata/greeter.proto", []string{}),
+			WithTotalRequests(6),
+			WithConcurrency(1),
+			WithTimeout(time.Duration(20*time.Second)),
+			WithDialTimeout(time.Duration(20*time.Second)),
+			WithInsecure(true),
+			WithData(data),
+		)
+
+		assert.NoError(t, err)
+		assert.NotNil(t, report)
+
+		count := gs.GetCount(callType)
+		assert.Equal(t, 6, count)
+
+		calls := gs.GetCalls(callType)
+		assert.NotNil(t, calls)
+		assert.Len(t, calls, 6)
+		names := make([]string, 0)
+		for _, msgs := range calls {
+			for _, msg := range msgs {
+				names = append(names, msg.GetName())
+			}
+		}
+
+		// we expect the same order of messages with single worker
+		assert.Equal(t, []string {"0", "1", "2", "0", "1", "2"}, names)
+	})
 }
 
 func TestRunServerStreaming(t *testing.T) {

runner/worker.go

@@ -116,7 +116,7 @@ func (w *Worker) makeRequest() error {
 	if inputsLen == 0 {
 		return fmt.Errorf("Error: can't create a request without payload. Check your data");
 	}
-	inputIdx := int(reqNum % int64(inputsLen))
+	inputIdx := int((reqNum - 1) % int64(inputsLen)) // we want to start from inputs[0] so dec reqNum
 
 	if w.mtd.IsServerStreaming() {
 		_ = w.makeServerStreamingRequest(&ctx, (*inputs)[inputIdx])