[FAB-5661]: Add UT to coordinator of state

Commit add a test case to check the interaction of state transfer with
coordinator interface and also to simulate real replication of missing
blocks between two peers. Additional logic which takes care of
marshalling and unmarshallling of private collection extracted from
state transfer.

Change-Id: I484cf39250fdf5ebb04aedb3f4db605efcb3d7a3
Signed-off-by: Artem Barger <bartem@il.ibm.com>

Author: C0rWin

gossip/state/coordinator.go

@@ -9,26 +9,61 @@ package state
 import (
 	"fmt"
 
+	"github.com/golang/protobuf/proto"
 	"github.com/hyperledger/fabric/core/committer"
 	"github.com/hyperledger/fabric/protos/common"
 	"github.com/hyperledger/fabric/protos/ledger/rwset"
+	"github.com/pkg/errors"
 )
 
 // PvtData a placeholder to represent private data
 type PvtData struct {
 	Payload *rwset.TxPvtReadWriteSet
 }
 
+// PvtDataCollections data type to encapsulate collections
+// of private data
+type PvtDataCollections []*PvtData
+
+// Marshal encodes private collection into bytes array
+func (pvt PvtDataCollections) Marshal() ([][]byte, error) {
+	pvtDataBytes := make([][]byte, 0)
+	for index, each := range pvt {
+		pvtBytes, err := proto.Marshal(each.Payload)
+		if err != nil {
+			errMsg := fmt.Sprintf("Could not marshal private rwset index %d, due to %s", index, err)
+			logger.Errorf(errMsg)
+			return nil, errors.New(errMsg)
+		}
+		pvtDataBytes = append(pvtDataBytes, pvtBytes)
+	}
+	return pvtDataBytes, nil
+}
+
+// Unmarshal read and unmarshal collection of private data
+// from given bytes array
+func (pvt PvtDataCollections) Unmarshal(data [][]byte) error {
+	for _, each := range data {
+		payload := &rwset.TxPvtReadWriteSet{}
+		if err := proto.Unmarshal(each, payload); err != nil {
+			return err
+		}
+		pvt = append(pvt, &PvtData{Payload: payload})
+	}
+
+	return nil
+}
+
 // Coordinator orchestrates the flow of the new
 // blocks arrival and in flight transient data, responsible
 // to complete missing parts of transient data for given block.
 type Coordinator interface {
 	// StoreBlock deliver new block with underlined private data
 	// returns missing transaction ids
-	StoreBlock(block *common.Block, data ...[]*PvtData) ([]string, error)
+	StoreBlock(block *common.Block, data ...PvtDataCollections) ([]string, error)
 
 	// GetBlockByNum returns block and related to the block private data
-	GetBlockByNum(seqNum uint64) (*common.Block, []*PvtData, error)
+	GetBlockByNum(seqNum uint64) (*common.Block, PvtDataCollections, error)
 
 	// Get recent block sequence number
 	LedgerHeight() (uint64, error)
@@ -46,12 +81,12 @@ func NewCoordinator(committer committer.Committer) Coordinator {
 	return &coordinator{Committer: committer}
 }
 
-func (c *coordinator) StoreBlock(block *common.Block, data ...[]*PvtData) ([]string, error) {
+func (c *coordinator) StoreBlock(block *common.Block, data ...PvtDataCollections) ([]string, error) {
 	// Need to check whenever there are missing private rwset
 	return nil, c.Committer.Commit(block)
 }
 
-func (c *coordinator) GetBlockByNum(seqNum uint64) (*common.Block, []*PvtData, error) {
+func (c *coordinator) GetBlockByNum(seqNum uint64) (*common.Block, PvtDataCollections, error) {
 	blocks := c.Committer.GetBlocks([]uint64{seqNum})
 	if len(blocks) == 0 {
 		return nil, nil, fmt.Errorf("Cannot retreive block number %d", seqNum)

gossip/state/mocks/gossip.go

@@ -19,32 +19,33 @@ type GossipMock struct {
 	mock.Mock
 }
 
-func (*GossipMock) SuspectPeers(s api.PeerSuspector) {
-	panic("implement me")
-}
+func (g *GossipMock) SuspectPeers(s api.PeerSuspector) {
+	g.Called(s)
 
-func (*GossipMock) Send(msg *proto.GossipMessage, peers ...*comm.RemotePeer) {
-	panic("implement me")
 }
 
-func (*GossipMock) Peers() []discovery.NetworkMember {
-	panic("implement me")
+func (g *GossipMock) Send(msg *proto.GossipMessage, peers ...*comm.RemotePeer) {
+	g.Called(msg, peers)
 }
 
-func (*GossipMock) PeersOfChannel(common.ChainID) []discovery.NetworkMember {
-	return nil
+func (g *GossipMock) Peers() []discovery.NetworkMember {
+	return g.Called().Get(0).([]discovery.NetworkMember)
 }
 
-func (*GossipMock) UpdateMetadata(metadata []byte) {
-	panic("implement me")
+func (g *GossipMock) PeersOfChannel(chainID common.ChainID) []discovery.NetworkMember {
+	args := g.Called(chainID)
+	return args.Get(0).([]discovery.NetworkMember)
 }
 
-func (*GossipMock) UpdateChannelMetadata(metadata []byte, chainID common.ChainID) {
+func (g *GossipMock) UpdateMetadata(metadata []byte) {
+	g.Called(metadata)
+}
 
+func (g *GossipMock) UpdateChannelMetadata(metadata []byte, chainID common.ChainID) {
 }
 
-func (*GossipMock) Gossip(msg *proto.GossipMessage) {
-	panic("implement me")
+func (g *GossipMock) Gossip(msg *proto.GossipMessage) {
+	g.Called(msg)
 }
 
 func (g *GossipMock) Accept(acceptor common.MessageAcceptor, passThrough bool) (<-chan *proto.GossipMessage, <-chan proto.ReceivedMessage) {
@@ -58,5 +59,6 @@ func (g *GossipMock) Accept(acceptor common.MessageAcceptor, passThrough bool) (
 func (g *GossipMock) JoinChan(joinMsg api.JoinChannelMessage, chainID common.ChainID) {
 }
 
-func (*GossipMock) Stop() {
+func (g *GossipMock) Stop() {
+
 }

gossip/state/mocks/gossip_test.go

@@ -11,6 +11,7 @@ import (
 
 	"github.com/hyperledger/fabric/gossip/api"
 	"github.com/hyperledger/fabric/gossip/common"
+	"github.com/hyperledger/fabric/gossip/discovery"
 	proto "github.com/hyperledger/fabric/protos/gossip"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/mock"
@@ -38,6 +39,7 @@ func TestGossipMock(t *testing.T) {
 	assert.Panics(t, func() {
 		g.Peers()
 	})
+	g.On("PeersOfChannel", mock.Anything).Return([]discovery.NetworkMember{})
 	assert.Empty(t, g.PeersOfChannel(common.ChainID("A")))
 
 	assert.Panics(t, func() {

gossip/state/state.go

@@ -23,7 +23,6 @@ import (
 	"github.com/hyperledger/fabric/gossip/util"
 	"github.com/hyperledger/fabric/protos/common"
 	proto "github.com/hyperledger/fabric/protos/gossip"
-	"github.com/hyperledger/fabric/protos/ledger/rwset"
 	"github.com/op/go-logging"
 )
 
@@ -344,35 +343,30 @@ func (s *GossipStateProviderImpl) handleStateRequest(msg proto.ReceivedMessage)
 
 		blockBytes, err := pb.Marshal(block)
 
-		// Marshal private data
 		if err != nil {
 			logger.Errorf("Could not marshal block: %s", err)
 			continue
 		}
 
-		// TODO: Need to extract orgID of the requester and filter out
-		// private data entries which doesn't belongs to collections
-		// allowed for sender organization based on policies
-		pvtDataBytes := make([][]byte, 0)
-		err = nil
-		for index, each := range pvtData {
-			pvtBytes, err := pb.Marshal(each.Payload)
+		var pvtBytes [][]byte
+		if pvtData != nil {
+			// TODO: Need to extract orgID of the requester and filter out
+			// private data entries which doesn't belongs to collections
+			// allowed for sender organization based on policies
+
+			// Marshal private data
+			pvtBytes, err = pvtData.Marshal()
 			if err != nil {
-				logger.Errorf("Could not marshal private rwset index %d, due to %s", index, err)
-				break
+				logger.Errorf("Failed to marshal private rwset for block %d due to %s", seqNum, err)
+				continue
 			}
-			pvtDataBytes = append(pvtDataBytes, pvtBytes)
-		}
-		if err != nil {
-			logger.Errorf("Failed to marshal private rwset for block %d due to %s", seqNum, err)
-			continue
 		}
 
 		// Appending result to the response
 		response.Payloads = append(response.Payloads, &proto.Payload{
 			SeqNum:      seqNum,
 			Data:        blockBytes,
-			PrivateData: pvtDataBytes,
+			PrivateData: pvtBytes,
 		})
 	}
 	// Sending back response with missing blocks
@@ -470,15 +464,8 @@ func (s *GossipStateProviderImpl) deliverPayloads() {
 
 				// Read all private data into slice
 				pvt := make([]*PvtData, 0)
-				var err error
-				for _, each := range payload.PrivateData {
-					payload := &rwset.TxPvtReadWriteSet{}
-					if err = pb.Unmarshal(each, payload); err != nil {
-						break
-					}
-					pvt = append(pvt, &PvtData{Payload: payload})
-				}
-
+				var p PvtDataCollections
+				err := p.Unmarshal(payload.PrivateData)
 				if err != nil {
 					logger.Errorf("Wasn't able to unmarshal private data for block seqNum = %d due to (%s)...dropping block", payload.SeqNum, err)
 					continue