MYST-20 Mock NATS connection it tests

communication/nats/connection_fake.go

@@ -0,0 +1,154 @@
+package nats
+
+import (
+	"fmt"
+	"github.com/nats-io/go-nats"
+	"github.com/pkg/errors"
+	"time"
+)
+
+func NewConnectionFake() *connectionFake {
+	return &connectionFake{
+		subscriptions: make(map[string][]nats.MsgHandler),
+		queue:         make(chan *nats.Msg),
+		queueShutdown: make(chan bool),
+	}
+}
+
+func StartConnectionFake() *connectionFake {
+	connection := NewConnectionFake()
+	connection.Start()
+
+	return connection
+}
+
+type connectionFake struct {
+	subscriptions map[string][]nats.MsgHandler
+	queue         chan *nats.Msg
+	queueShutdown chan bool
+
+	messageLast *nats.Msg
+	requestLast *nats.Msg
+	errorMock   error
+}
+
+func (conn *connectionFake) GetLastMessage() []byte {
+	if conn.messageLast != nil {
+		return conn.messageLast.Data
+	}
+	return []byte{}
+}
+
+func (conn *connectionFake) GetLastRequest() []byte {
+	if conn.requestLast != nil {
+		return conn.requestLast.Data
+	}
+	return []byte{}
+}
+
+func (conn *connectionFake) MockResponse(subject string, payload []byte) {
+	conn.Subscribe(subject, func(message *nats.Msg) {
+		conn.Publish(message.Reply, payload)
+	})
+}
+
+func (conn *connectionFake) MockError(message string) {
+	conn.errorMock = errors.New(message)
+}
+
+func (conn *connectionFake) MessageWait(waitChannel chan interface{}) (interface{}, error) {
+	select {
+	case message := <-waitChannel:
+		return message, nil
+	case <-time.After(10 * time.Millisecond):
+		return nil, errors.New("Message not received")
+	}
+}
+
+func (conn *connectionFake) Publish(subject string, payload []byte) error {
+	if conn.errorMock != nil {
+		return conn.errorMock
+	}
+
+	conn.messageLast = &nats.Msg{
+		Subject: subject,
+		Data:    payload,
+	}
+	conn.queue <- conn.messageLast
+
+	return nil
+}
+
+func (conn *connectionFake) Subscribe(subject string, handler nats.MsgHandler) (*nats.Subscription, error) {
+	if conn.errorMock != nil {
+		return nil, conn.errorMock
+	}
+
+	conn.subscriptionAdd(subject, handler)
+
+	return &nats.Subscription{}, nil
+}
+
+func (conn *connectionFake) Request(subject string, payload []byte, timeout time.Duration) (*nats.Msg, error) {
+	if conn.errorMock != nil {
+		return nil, conn.errorMock
+	}
+
+	subjectReply := subject + "-reply"
+	responseCh := make(chan *nats.Msg)
+	conn.Subscribe(subjectReply, func(response *nats.Msg) {
+		responseCh <- response
+	})
+
+	conn.requestLast = &nats.Msg{
+		Subject: subject,
+		Reply:   subjectReply,
+		Data:    payload,
+	}
+	conn.queue <- conn.requestLast
+
+	select {
+	case response := <-responseCh:
+		return response, nil
+	case <-time.After(timeout):
+		return nil, fmt.Errorf("Request '%s' timeout", subject)
+	}
+}
+
+func (conn *connectionFake) Start() {
+	go conn.queueProcessing()
+}
+
+func (conn *connectionFake) Close() {
+	conn.queueShutdown <- true
+}
+
+func (conn *connectionFake) subscriptionAdd(subject string, handler nats.MsgHandler) {
+	subscriptions, exist := conn.subscriptions[subject]
+	if exist {
+		subscriptions = append(subscriptions, handler)
+	} else {
+		conn.subscriptions[subject] = []nats.MsgHandler{handler}
+	}
+}
+
+func (conn *connectionFake) subscriptionsGet(subject string) (*[]nats.MsgHandler, bool) {
+	subscriptions, exist := conn.subscriptions[subject]
+	return &subscriptions, exist
+}
+
+func (conn *connectionFake) queueProcessing() {
+	for {
+		select {
+		case <-conn.queueShutdown:
+			break
+
+		case message := <-conn.queue:
+			if subscriptions, exist := conn.subscriptionsGet(message.Subject); exist {
+				for _, handler := range *subscriptions {
+					go handler(message)
+				}
+			}
+		}
+	}
+}

communication/nats/connection_interface.go

@@ -0,0 +1,13 @@
+package nats
+
+import (
+	"github.com/nats-io/go-nats"
+	"time"
+)
+
+type Connection interface {
+	Publish(subject string, payload []byte) error
+	Subscribe(subject string, handler nats.MsgHandler) (*nats.Subscription, error)
+	Request(subject string, payload []byte, timeout time.Duration) (*nats.Msg, error)
+	Close()
+}

communication/nats/message_bytes_test.go

@@ -2,8 +2,6 @@ package nats
 
 import (
 	"github.com/mysterium/node/communication"
-	"github.com/nats-io/go-nats"
-	"github.com/nats-io/go-nats/test"
 	"github.com/stretchr/testify/assert"
 	"testing"
 )
@@ -21,7 +19,7 @@ func (producer *bytesMessageProducer) Produce() (messagePtr interface{}) {
 }
 
 type bytesMessageConsumer struct {
-	Callback func(*[]byte)
+	messageReceived chan interface{}
 }
 
 func (consumer *bytesMessageConsumer) GetMessageType() communication.MessageType {
@@ -34,60 +32,41 @@ func (consumer *bytesMessageConsumer) NewMessage() (messagePtr interface{}) {
 }
 
 func (consumer *bytesMessageConsumer) Consume(messagePtr interface{}) error {
-	consumer.Callback(messagePtr.(*[]byte))
+	consumer.messageReceived <- messagePtr
 	return nil
 }
 
 func TestMessageBytesSend(t *testing.T) {
-	server := test.RunDefaultServer()
-	defer server.Shutdown()
-	connection := test.NewDefaultConnection(t)
+	connection := StartConnectionFake()
 	defer connection.Close()
 
 	sender := &senderNats{
 		connection: connection,
 		codec:      communication.NewCodecBytes(),
 	}
 
-	messageSent := make(chan bool)
-	_, err := connection.Subscribe("bytes-message", func(message *nats.Msg) {
-		assert.Equal(t, []byte("123"), message.Data)
-		messageSent <- true
-	})
-	assert.Nil(t, err)
-
-	err = sender.Send(
+	err := sender.Send(
 		&bytesMessageProducer{[]byte("123")},
 	)
-	assert.Nil(t, err)
-
-	if err := test.Wait(messageSent); err != nil {
-		t.Fatal("Message not sent")
-	}
+	assert.NoError(t, err)
+	assert.Equal(t, []byte("123"), connection.GetLastMessage())
 }
 
 func TestMessageBytesReceive(t *testing.T) {
-	server := test.RunDefaultServer()
-	defer server.Shutdown()
-	connection := test.NewDefaultConnection(t)
+	connection := StartConnectionFake()
 	defer connection.Close()
 
 	receiver := &receiverNats{
 		connection: connection,
 		codec:      communication.NewCodecBytes(),
 	}
 
-	messageReceived := make(chan bool)
-	err := receiver.Receive(&bytesMessageConsumer{func(message *[]byte) {
-		assert.Equal(t, []byte("123"), *message)
-		messageReceived <- true
-	}})
-	assert.Nil(t, err)
+	consumer := &bytesMessageConsumer{messageReceived: make(chan interface{})}
+	err := receiver.Receive(consumer)
+	assert.NoError(t, err)
 
-	err = connection.Publish("bytes-message", []byte("123"))
-	assert.Nil(t, err)
-
-	if err := test.Wait(messageReceived); err != nil {
-		t.Fatal("Message not received")
-	}
+	connection.Publish("bytes-message", []byte("123"))
+	message, err := connection.MessageWait(consumer.messageReceived)
+	assert.NoError(t, err)
+	assert.Equal(t, []byte("123"), *message.(*[]byte))
 }

communication/nats/message_custom_test.go

@@ -2,8 +2,6 @@ package nats
 
 import (
 	"github.com/mysterium/node/communication"
-	"github.com/nats-io/go-nats"
-	"github.com/nats-io/go-nats/test"
 	"github.com/stretchr/testify/assert"
 	"testing"
 )
@@ -25,59 +23,35 @@ func (producer *customMessageProducer) Produce() (messagePtr interface{}) {
 }
 
 func TestMessageCustomSend(t *testing.T) {
-	server := test.RunDefaultServer()
-	defer server.Shutdown()
-	connection := test.NewDefaultConnection(t)
+	connection := StartConnectionFake()
 	defer connection.Close()
 
 	sender := &senderNats{
 		connection: connection,
 		codec:      communication.NewCodecJSON(),
 	}
 
-	messageSent := make(chan bool)
-	_, err := connection.Subscribe("custom-message", func(message *nats.Msg) {
-		assert.JSONEq(t, `{"Field": 123}`, string(message.Data))
-		messageSent <- true
-	})
-	assert.Nil(t, err)
-
-	err = sender.Send(&customMessageProducer{&customMessage{123}})
-	assert.Nil(t, err)
-
-	if err := test.Wait(messageSent); err != nil {
-		t.Fatal("Message not sent")
-	}
+	err := sender.Send(&customMessageProducer{&customMessage{123}})
+	assert.NoError(t, err)
+	assert.JSONEq(t, `{"Field": 123}`, string(connection.GetLastMessage()))
 }
 
 func TestMessageCustomSendNull(t *testing.T) {
-	server := test.RunDefaultServer()
-	defer server.Shutdown()
-	connection := test.NewDefaultConnection(t)
+	connection := StartConnectionFake()
 	defer connection.Close()
 
 	sender := &senderNats{
 		connection: connection,
 		codec:      communication.NewCodecJSON(),
 	}
 
-	messageSent := make(chan bool)
-	_, err := connection.Subscribe("custom-message", func(message *nats.Msg) {
-		assert.JSONEq(t, `null`, string(message.Data))
-		messageSent <- true
-	})
-	assert.Nil(t, err)
-
-	err = sender.Send(&customMessageProducer{})
-	assert.Nil(t, err)
-
-	if err := test.Wait(messageSent); err != nil {
-		t.Fatal("Message not sent")
-	}
+	err := sender.Send(&customMessageProducer{})
+	assert.NoError(t, err)
+	assert.JSONEq(t, `null`, string(connection.GetLastMessage()))
 }
 
 type customMessageConsumer struct {
-	Callback func(message *customMessage)
+	messageReceived chan interface{}
 }
 
 func (consumer *customMessageConsumer) GetMessageType() communication.MessageType {
@@ -89,32 +63,25 @@ func (consumer *customMessageConsumer) NewMessage() (messagePtr interface{}) {
 }
 
 func (consumer *customMessageConsumer) Consume(messagePtr interface{}) error {
-	consumer.Callback(messagePtr.(*customMessage))
+	consumer.messageReceived <- messagePtr
 	return nil
 }
 
 func TestMessageCustomReceive(t *testing.T) {
-	server := test.RunDefaultServer()
-	defer server.Shutdown()
-	connection := test.NewDefaultConnection(t)
+	connection := StartConnectionFake()
 	defer connection.Close()
 
 	receiver := &receiverNats{
 		connection: connection,
 		codec:      communication.NewCodecJSON(),
 	}
 
-	messageReceived := make(chan bool)
-	err := receiver.Receive(&customMessageConsumer{func(message *customMessage) {
-		assert.Exactly(t, customMessage{123}, *message)
-		messageReceived <- true
-	}})
-	assert.Nil(t, err)
+	consumer := &customMessageConsumer{messageReceived: make(chan interface{})}
+	err := receiver.Receive(consumer)
+	assert.NoError(t, err)
 
-	err = connection.Publish("custom-message", []byte(`{"Field":123}`))
-	assert.Nil(t, err)
-
-	if err := test.Wait(messageReceived); err != nil {
-		t.Fatal("Message not received")
-	}
+	connection.Publish("custom-message", []byte(`{"Field":123}`))
+	message, err := connection.MessageWait(consumer.messageReceived)
+	assert.NoError(t, err)
+	assert.Exactly(t, customMessage{123}, *message.(*customMessage))
 }