added unit tests

process/blocksPool_test.go

@@ -0,0 +1,3 @@
+package process_test
+
+// TODO: add tests

process/dataAggregator.go

@@ -17,7 +17,7 @@ func NewDataAggregator(
 	blocksPool BlocksPool,
 ) (*dataAggregator, error) {
 	if check.IfNil(blocksPool) {
-		return nil, errNilBlocksPool
+		return nil, ErrNilBlocksPool
 	}
 
 	return &dataAggregator{

process/dataAggregator_test.go

@@ -0,0 +1,57 @@
+package process_test
+
+import (
+	"encoding/hex"
+	"testing"
+
+	"github.com/multiversx/mx-chain-core-go/data/outport"
+	"github.com/multiversx/mx-chain-ws-connector-template-go/process"
+	"github.com/multiversx/mx-chain-ws-connector-template-go/testscommon"
+	"github.com/stretchr/testify/require"
+)
+
+func TestNewDataAggregator(t *testing.T) {
+	t.Parallel()
+
+	t.Run("nil blocks pool", func(t *testing.T) {
+		t.Parallel()
+
+		da, err := process.NewDataAggregator(nil)
+		require.Nil(t, da)
+		require.Equal(t, process.ErrNilBlocksPool, err)
+	})
+
+	t.Run("should work", func(t *testing.T) {
+		t.Parallel()
+
+		da, err := process.NewDataAggregator(&testscommon.BlocksPoolStub{})
+		require.Nil(t, err)
+		require.False(t, da.IsInterfaceNil())
+	})
+}
+
+func TestDataAggregator_ProcessHyperBlock(t *testing.T) {
+	t.Parallel()
+
+	headerHash := []byte("headerHash1")
+
+	shardOutportBlock := createOutportBlock()
+	shardOutportBlock.BlockData.HeaderHash = headerHash
+
+	blocksPoolStub := &testscommon.BlocksPoolStub{
+		GetBlockCalled: func(hash []byte) (*outport.OutportBlock, error) {
+
+			return shardOutportBlock, nil
+		},
+	}
+
+	da, err := process.NewDataAggregator(blocksPoolStub)
+	require.Nil(t, err)
+
+	outportBlock := createMetaOutportBlock()
+	outportBlock.NotarizedHeadersHashes = []string{hex.EncodeToString(headerHash)}
+
+	hyperOutportBlock, err := da.ProcessHyperBlock(outportBlock)
+	require.Nil(t, err)
+	require.Equal(t, shardOutportBlock, hyperOutportBlock.NotarizedHeadersOutportData[0].OutportBlock)
+}

process/dataProcessor.go

@@ -23,16 +23,16 @@ func NewDataProcessor(
 	dataAggregator DataAggregator,
 ) (DataProcessor, error) {
 	if publisher == nil {
-		return nil, errNilWriter
+		return nil, ErrNilPublisher
 	}
 	if check.IfNil(blocksPool) {
-		return nil, errNilBlocksPool
+		return nil, ErrNilBlocksPool
 	}
 	if check.IfNil(marshaller) {
-		return nil, errNilMarshaller
+		return nil, ErrNilMarshaller
 	}
 	if check.IfNil(dataAggregator) {
-		return nil, errNilDataAggregator
+		return nil, ErrNilDataAggregator
 	}
 
 	dp := &dataProcessor{
@@ -67,7 +67,7 @@ func (dp *dataProcessor) saveBlock(marshalledData []byte) error {
 	}
 
 	if outportBlock == nil || outportBlock.BlockData == nil {
-		return errNilOutportBlockData
+		return ErrNilOutportBlockData
 	}
 
 	log.Info("saving block", "hash", outportBlock.BlockData.GetHeaderHash(), "shardID", outportBlock.ShardID)

process/dataProcessor_test.go

@@ -0,0 +1,253 @@
+package process_test
+
+import (
+	"errors"
+	"testing"
+
+	"github.com/multiversx/mx-chain-core-go/core"
+	"github.com/multiversx/mx-chain-core-go/data/block"
+	outportcore "github.com/multiversx/mx-chain-core-go/data/outport"
+	"github.com/multiversx/mx-chain-ws-connector-template-go/data"
+	"github.com/multiversx/mx-chain-ws-connector-template-go/process"
+	"github.com/multiversx/mx-chain-ws-connector-template-go/testscommon"
+	"github.com/stretchr/testify/require"
+)
+
+func createOutportBlock() *outportcore.OutportBlock {
+	header := &block.Header{
+		Nonce:     1,
+		PrevHash:  []byte("prev hash"),
+		TimeStamp: 100,
+	}
+	headerBytes, _ := protoMarshaller.Marshal(header)
+
+	return &outportcore.OutportBlock{
+		ShardID: 1,
+		BlockData: &outportcore.BlockData{
+			HeaderBytes: headerBytes,
+			HeaderType:  string(core.ShardHeaderV1),
+			HeaderHash:  []byte("hash"),
+		},
+	}
+}
+
+func createMetaOutportBlock() *outportcore.OutportBlock {
+	header := &block.MetaBlock{
+		Nonce:     1,
+		PrevHash:  []byte("prev hash"),
+		TimeStamp: 100,
+	}
+	headerBytes, _ := protoMarshaller.Marshal(header)
+
+	return &outportcore.OutportBlock{
+		ShardID: core.MetachainShardId,
+		BlockData: &outportcore.BlockData{
+			HeaderBytes: headerBytes,
+			HeaderType:  string(core.MetaHeader),
+			HeaderHash:  []byte("hash"),
+		},
+	}
+}
+
+func TestNewDataProcessor(t *testing.T) {
+	t.Parallel()
+
+	t.Run("nil publisher", func(t *testing.T) {
+		t.Parallel()
+
+		dp, err := process.NewDataProcessor(
+			nil,
+			&testscommon.MarshallerStub{},
+			&testscommon.BlocksPoolStub{},
+			&testscommon.DataAggregatorStub{},
+		)
+		require.Nil(t, dp)
+		require.Equal(t, process.ErrNilPublisher, err)
+	})
+
+	t.Run("nil marshaller", func(t *testing.T) {
+		t.Parallel()
+
+		dp, err := process.NewDataProcessor(
+			&testscommon.PublisherStub{},
+			nil,
+			&testscommon.BlocksPoolStub{},
+			&testscommon.DataAggregatorStub{},
+		)
+		require.Nil(t, dp)
+		require.Equal(t, process.ErrNilMarshaller, err)
+	})
+
+	t.Run("nil blocks pool", func(t *testing.T) {
+		t.Parallel()
+
+		dp, err := process.NewDataProcessor(
+			&testscommon.PublisherStub{},
+			&testscommon.MarshallerStub{},
+			nil,
+			&testscommon.DataAggregatorStub{},
+		)
+		require.Nil(t, dp)
+		require.Equal(t, process.ErrNilBlocksPool, err)
+	})
+
+	t.Run("nil data aggregator", func(t *testing.T) {
+		t.Parallel()
+
+		dp, err := process.NewDataProcessor(
+			&testscommon.PublisherStub{},
+			&testscommon.MarshallerStub{},
+			&testscommon.BlocksPoolStub{},
+			nil,
+		)
+		require.Nil(t, dp)
+		require.Equal(t, process.ErrNilDataAggregator, err)
+	})
+
+	t.Run("should work", func(t *testing.T) {
+		t.Parallel()
+
+		dp, err := process.NewDataProcessor(
+			&testscommon.PublisherStub{},
+			&testscommon.MarshallerStub{},
+			&testscommon.BlocksPoolStub{},
+			&testscommon.DataAggregatorStub{},
+		)
+		require.Nil(t, err)
+		require.False(t, dp.IsInterfaceNil())
+	})
+}
+
+func TestDataProcessor_ProcessPayload_NotImplementedTopics(t *testing.T) {
+	t.Parallel()
+
+	dp, _ := process.NewDataProcessor(
+		&testscommon.PublisherStub{},
+		&testscommon.MarshallerStub{},
+		&testscommon.BlocksPoolStub{},
+		&testscommon.DataAggregatorStub{},
+	)
+
+	require.Nil(t, dp.ProcessPayload([]byte("payload"), "random topic", 1))
+	require.Nil(t, dp.ProcessPayload([]byte("payload"), outportcore.TopicSaveRoundsInfo, 1))
+	require.Nil(t, dp.ProcessPayload([]byte("payload"), outportcore.TopicSaveValidatorsRating, 1))
+	require.Nil(t, dp.ProcessPayload([]byte("payload"), outportcore.TopicSaveValidatorsPubKeys, 1))
+	require.Nil(t, dp.ProcessPayload([]byte("payload"), outportcore.TopicSaveAccounts, 1))
+	require.Nil(t, dp.ProcessPayload([]byte("payload"), outportcore.TopicFinalizedBlock, 1))
+}
+
+func TestDataProcessor_ProcessPayload(t *testing.T) {
+	t.Parallel()
+
+	t.Run("nil outport block, should return error", func(t *testing.T) {
+		t.Parallel()
+
+		dp, _ := process.NewDataProcessor(
+			&testscommon.PublisherStub{},
+			protoMarshaller,
+			&testscommon.BlocksPoolStub{},
+			&testscommon.DataAggregatorStub{},
+		)
+
+		err := dp.ProcessPayload(nil, outportcore.TopicSaveBlock, 1)
+		require.Equal(t, process.ErrNilOutportBlockData, err)
+
+		outportBlock := createOutportBlock()
+		outportBlock.BlockData = nil
+		outportBlockBytes, _ := protoMarshaller.Marshal(outportBlock)
+
+		err = dp.ProcessPayload(outportBlockBytes, outportcore.TopicSaveBlock, 1)
+		require.Equal(t, process.ErrNilOutportBlockData, err)
+	})
+
+	t.Run("invalid payload, cannot unmarshall, should return error", func(t *testing.T) {
+		t.Parallel()
+
+		dp, _ := process.NewDataProcessor(
+			&testscommon.PublisherStub{},
+			protoMarshaller,
+			&testscommon.BlocksPoolStub{},
+			&testscommon.DataAggregatorStub{},
+		)
+
+		err := dp.ProcessPayload([]byte("invalid payload"), outportcore.TopicSaveBlock, 1)
+		require.NotNil(t, err)
+	})
+
+	t.Run("shard outport block, should work", func(t *testing.T) {
+		t.Parallel()
+
+		outportBlock := createOutportBlock()
+		outportBlockBytes, _ := protoMarshaller.Marshal(outportBlock)
+
+		putBlockWasCalled := false
+		dp, _ := process.NewDataProcessor(
+			&testscommon.PublisherStub{},
+			protoMarshaller,
+			&testscommon.BlocksPoolStub{
+				PutBlockCalled: func(hash []byte, outportBlock *outportcore.OutportBlock) error {
+					putBlockWasCalled = true
+					return nil
+				},
+			},
+			&testscommon.DataAggregatorStub{},
+		)
+
+		err := dp.ProcessPayload(outportBlockBytes, outportcore.TopicSaveBlock, 1)
+		require.Nil(t, err)
+
+		require.True(t, putBlockWasCalled)
+	})
+
+	t.Run("meta outport block, should work", func(t *testing.T) {
+		t.Parallel()
+
+		outportBlock := createMetaOutportBlock()
+		outportBlockBytes, _ := protoMarshaller.Marshal(outportBlock)
+
+		publishWasCalled := false
+		dp, _ := process.NewDataProcessor(
+			&testscommon.PublisherStub{
+				PublishHyperBlockCalled: func(hyperOutportBlock *data.HyperOutportBlock) error {
+					publishWasCalled = true
+					return nil
+				},
+			},
+			protoMarshaller,
+			&testscommon.BlocksPoolStub{},
+			&testscommon.DataAggregatorStub{
+				ProcessHyperBlockCalled: func(outportBlock *outportcore.OutportBlock) (*data.HyperOutportBlock, error) {
+					return &data.HyperOutportBlock{
+						MetaOutportBlock: outportBlock,
+					}, nil
+				},
+			},
+		)
+
+		err := dp.ProcessPayload(outportBlockBytes, outportcore.TopicSaveBlock, 1)
+		require.Nil(t, err)
+
+		require.True(t, publishWasCalled)
+	})
+}
+
+func TestDataProcessor_Close(t *testing.T) {
+	t.Parallel()
+
+	expectedErr := errors.New("expected error")
+
+	dp, err := process.NewDataProcessor(
+		&testscommon.PublisherStub{
+			CloseCalled: func() error {
+				return expectedErr
+			},
+		},
+		&testscommon.MarshallerStub{},
+		&testscommon.BlocksPoolStub{},
+		&testscommon.DataAggregatorStub{},
+	)
+	require.Nil(t, err)
+
+	err = dp.Close()
+	require.Equal(t, expectedErr, err)
+}

process/errors.go

@@ -2,19 +2,26 @@ package process
 
 import "errors"
 
-var errNilMarshaller = errors.New("nil marshaller provided")
+// ErrNilMarshaller signals that a nil marshaller was provided
+var ErrNilMarshaller = errors.New("nil marshaller provided")
 
-var errNilOutportBlockData = errors.New("nil outport block data")
+// ErrNilOutportBlockData signals that a nil outport block data was provided
+var ErrNilOutportBlockData = errors.New("nil outport block data")
 
-var errNilWriter = errors.New("nil writer provided")
+// ErrNilWriter signals that a nil write was provided
+var ErrNilWriter = errors.New("nil writer provided")
 
-var errNilBlockCreator = errors.New("nil block creator provided")
+// ErrNilBlockCreator signals that a nil block creator was provided
+var ErrNilBlockCreator = errors.New("nil block creator provided")
 
-var errNilPublisher = errors.New("nil publisher provided")
+// ErrNilPublisher signals that a nil publisher was provided
+var ErrNilPublisher = errors.New("nil publisher provided")
 
-var errNilBlocksPool = errors.New("nil blocks pool provided")
+// ErrNilBlocksPool signals that a nil blocks pool was provided
+var ErrNilBlocksPool = errors.New("nil blocks pool provided")
 
-var errNilDataAggregator = errors.New("nil data aggregator provided")
+// ErrNilDataAggregator signals that a nil data aggregator was provided
+var ErrNilDataAggregator = errors.New("nil data aggregator provided")
 
 // ErrWrongTypeAssertion signals that a type assertion faled
 var ErrWrongTypeAssertion = errors.New("type assertion failed")

process/export_test.go

@@ -0,0 +1,6 @@
+package process
+
+const (
+	FirehosePrefix = firehosePrefix
+	BlockPrefix    = blockPrefix
+)