[FAB-3383] Increase gossip unit test coverage

Gossip unit test coverage is pretty good, but can be even better.

This change set makes all our code (except protos extension that are
found in protos/gossip/) have code coverage of above 85%.

github.com/hyperledger/fabric/core/deliverservice 29.143s	: 92.5%
github.com/hyperledger/fabric/core/deliverservice/blocksprovider 7.047s	: 86.1%
github.com/hyperledger/fabric/core/deliverservice/mocks	1.018s	: 93.8%
github.com/hyperledger/fabric/gossip/api	0.003s	: 100.0%
github.com/hyperledger/fabric/gossip/comm	11.078s	: 85.9%
github.com/hyperledger/fabric/gossip/comm/mock	0.031s	: 85.3%
github.com/hyperledger/fabric/gossip/common	0.002s	: 100.0%
github.com/hyperledger/fabric/gossip/discovery	25.064s	: 86.5%
github.com/hyperledger/fabric/gossip/election	7.130s	: 94.2%
github.com/hyperledger/fabric/gossip/filter	0.040s	: 100.0%
github.com/hyperledger/fabric/gossip/gossip	72.131s	: 86.7%
github.com/hyperledger/fabric/gossip/gossip/algo	6.552s	: 97.1%
github.com/hyperledger/fabric/gossip/gossip/channel	21.811s	: 92.1%
github.com/hyperledger/fabric/gossip/gossip/msgstore	8.014s	: 100.0%
github.com/hyperledger/fabric/gossip/gossip/pull	2.688s	: 96.3%
github.com/hyperledger/fabric/gossip/identity	0.012s	: 90.7%
github.com/hyperledger/fabric/gossip/integration	0.078s	: 85.7%
github.com/hyperledger/fabric/gossip/service	83.191s	: 86.8%
github.com/hyperledger/fabric/gossip/state	37.891s	: 86.3%
github.com/hyperledger/fabric/gossip/state/mocks	0.020s	: 90.0%
github.com/hyperledger/fabric/gossip/util	0.029s	: 100.0%

Change-Id: Ifcbf4fa20cd5829f64cdd9b5d1e0507b38118271
Signed-off-by: Yacov Manevich <yacovm@il.ibm.com>

core/deliverservice/blocksprovider/blocksprovider_test.go

@@ -16,7 +16,9 @@ limitations under the License.
 package blocksprovider
 
 import (
+	"errors"
 	"sync"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -26,16 +28,22 @@ import (
 	"github.com/hyperledger/fabric/protos/common"
 	"github.com/hyperledger/fabric/protos/orderer"
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/mock"
 )
 
 type mockMCS struct {
+	mock.Mock
 }
 
 func (*mockMCS) GetPKIidOfCert(peerIdentity api.PeerIdentityType) common2.PKIidType {
 	return common2.PKIidType("pkiID")
 }
 
-func (*mockMCS) VerifyBlock(chainID common2.ChainID, signedBlock []byte) error {
+func (m *mockMCS) VerifyBlock(chainID common2.ChainID, signedBlock []byte) error {
+	args := m.Called()
+	if args.Get(0) != nil {
+		return args.Get(0).(error)
+	}
 	return nil
 }
 
@@ -55,47 +63,42 @@ func (*mockMCS) ValidateIdentity(peerIdentity api.PeerIdentityType) error {
 	return nil
 }
 
+type rcvFunc func(mock *mocks.MockBlocksDeliverer) (*orderer.DeliverResponse, error)
+
 // Used to generate a simple test case to initialize delivery
 // from given block sequence number.
-func makeTestCase(ledgerHeight uint64) func(*testing.T) {
+func makeTestCase(ledgerHeight uint64, mcs api.MessageCryptoService, shouldSucceed bool, rcv rcvFunc) func(*testing.T) {
 	return func(t *testing.T) {
 		gossipServiceAdapter := &mocks.MockGossipServiceAdapter{GossipBlockDisseminations: make(chan uint64)}
 		deliverer := &mocks.MockBlocksDeliverer{Pos: ledgerHeight}
-		deliverer.MockRecv = mocks.MockRecv
-
-		provider := &blocksProviderImpl{
-			chainID: "***TEST_CHAINID***",
-			gossip:  gossipServiceAdapter,
-			client:  deliverer,
-			mcs:     &mockMCS{},
-		}
-
+		deliverer.MockRecv = rcv
+		provider := NewBlocksProvider("***TEST_CHAINID***", deliverer, gossipServiceAdapter, mcs)
+		defer provider.Stop()
 		ready := make(chan struct{})
 		go func() {
 			go provider.DeliverBlocks()
 			// Send notification
 			ready <- struct{}{}
 		}()
 
-		time.Sleep(time.Duration(10) * time.Millisecond)
-		provider.Stop()
-
-		select {
-		case <-ready:
-			{
-				// Check that all blocks received eventually get gossiped and locally committed
-				assert.True(t, deliverer.RecvCnt == gossipServiceAdapter.AddPayloadsCnt)
-				select {
-				case <-gossipServiceAdapter.GossipBlockDisseminations:
-				case <-time.After(time.Second):
-					assert.Fail(t, "Didn't gossip a block within a timely manner")
-				}
-				return
-			}
-		case <-time.After(time.Duration(1) * time.Second):
-			{
-				t.Fatal("Test hasn't finished in timely manner, failing.")
-			}
+		time.Sleep(time.Second)
+
+		assertDelivery(t, gossipServiceAdapter, deliverer, shouldSucceed)
+	}
+}
+
+func assertDelivery(t *testing.T, ga *mocks.MockGossipServiceAdapter, deliverer *mocks.MockBlocksDeliverer, shouldSucceed bool) {
+	// Check that all blocks received eventually get gossiped and locally committed
+
+	select {
+	case <-ga.GossipBlockDisseminations:
+		if !shouldSucceed {
+			assert.Fail(t, "Should not have succeede")
+		}
+		assert.True(t, deliverer.RecvCnt == ga.AddPayloadsCnt)
+	case <-time.After(time.Second):
+		if shouldSucceed {
+			assert.Fail(t, "Didn't gossip a block within a timely manner")
 		}
 	}
 }
@@ -105,19 +108,22 @@ func makeTestCase(ledgerHeight uint64) func(*testing.T) {
    oldest and that eventually it terminates after the Stop method has been called.
 */
 func TestBlocksProviderImpl_GetBlockFromTheOldest(t *testing.T) {
-	makeTestCase(uint64(0))(t)
+	mcs := &mockMCS{}
+	mcs.On("VerifyBlock", mock.Anything).Return(nil)
+	makeTestCase(uint64(0), mcs, true, mocks.MockRecv)(t)
 }
 
 /*
    Test to check whenever blocks provider starts calling new blocks from the
    oldest and that eventually it terminates after the Stop method has been called.
 */
 func TestBlocksProviderImpl_GetBlockFromSpecified(t *testing.T) {
-	makeTestCase(uint64(101))(t)
+	mcs := &mockMCS{}
+	mcs.On("VerifyBlock", mock.Anything).Return(nil)
+	makeTestCase(uint64(101), mcs, true, mocks.MockRecv)(t)
 }
 
 func TestBlocksProvider_CheckTerminationDeliveryResponseStatus(t *testing.T) {
-
 	tmp := struct{ mocks.MockBlocksDeliverer }{}
 
 	// Making mocked Recv() function to return DeliverResponse_Status to force block
@@ -158,7 +164,7 @@ func TestBlocksProvider_CheckTerminationDeliveryResponseStatus(t *testing.T) {
 			assert.Equal(t, int32(1), tmp.RecvCnt)
 			// No payload should commit locally
 			assert.Equal(t, int32(0), gossipServiceAdapter.AddPayloadsCnt)
-			// No payload should be transfered to other peers
+			// No payload should be transferred to other peers
 			select {
 			case <-gossipServiceAdapter.GossipBlockDisseminations:
 				assert.Fail(t, "Gossiped block but shouldn't have")
@@ -172,3 +178,42 @@ func TestBlocksProvider_CheckTerminationDeliveryResponseStatus(t *testing.T) {
 		}
 	}
 }
+
+func TestBlockFetchFailure(t *testing.T) {
+	rcvr := func(mock *mocks.MockBlocksDeliverer) (*orderer.DeliverResponse, error) {
+		return nil, errors.New("Failed fetching block")
+	}
+	mcs := &mockMCS{}
+	mcs.On("VerifyBlock", mock.Anything).Return(nil)
+	makeTestCase(uint64(0), mcs, false, rcvr)(t)
+}
+
+func TestBlockVerificationFailure(t *testing.T) {
+	attempts := int32(0)
+	rcvr := func(mock *mocks.MockBlocksDeliverer) (*orderer.DeliverResponse, error) {
+		if atomic.LoadInt32(&attempts) == int32(1) {
+			return &orderer.DeliverResponse{
+				Type: &orderer.DeliverResponse_Status{
+					Status: common.Status_SUCCESS,
+				},
+			}, nil
+		}
+		atomic.AddInt32(&attempts, int32(1))
+		return &orderer.DeliverResponse{
+			Type: &orderer.DeliverResponse_Block{
+				Block: &common.Block{
+					Header: &common.BlockHeader{
+						Number:       0,
+						DataHash:     []byte{},
+						PreviousHash: []byte{},
+					},
+					Data: &common.BlockData{
+						Data: [][]byte{},
+					},
+				}},
+		}, nil
+	}
+	mcs := &mockMCS{}
+	mcs.On("VerifyBlock", mock.Anything).Return(errors.New("Invalid signature"))
+	makeTestCase(uint64(0), mcs, false, rcvr)(t)
+}

core/deliverservice/mocks/blocksprovider.go

@@ -114,13 +114,9 @@ func (mock *MockBlocksDeliverer) Send(env *common.Envelope) error {
 	// Read starting position
 	switch t := seekInfo.Start.Type.(type) {
 	case *orderer.SeekPosition_Oldest:
-		{
-			mock.Pos = 0
-		}
+		mock.Pos = 0
 	case *orderer.SeekPosition_Specified:
-		{
-			mock.Pos = t.Specified.Number
-		}
+		mock.Pos = t.Specified.Number
 	}
 	return nil
 }

core/deliverservice/mocks/blocksprovider_test.go

@@ -17,26 +17,114 @@ limitations under the License.
 package mocks
 
 import (
+	"context"
+	"math"
+	"sync/atomic"
 	"testing"
+	"time"
 
+	pb "github.com/golang/protobuf/proto"
 	"github.com/hyperledger/fabric/core/deliverservice/blocksprovider"
+	"github.com/hyperledger/fabric/protos/common"
+	proto "github.com/hyperledger/fabric/protos/gossip"
+	"github.com/hyperledger/fabric/protos/orderer"
+	"github.com/stretchr/testify/assert"
 )
 
 func TestMockBlocksDeliverer(t *testing.T) {
+	// Make sure it implements BlocksDeliverer
 	var bd blocksprovider.BlocksDeliverer
 	bd = &MockBlocksDeliverer{}
 	_ = bd
+
+	assert.Panics(t, func() {
+		bd.Recv()
+	})
+	bd.(*MockBlocksDeliverer).MockRecv = func(mock *MockBlocksDeliverer) (*orderer.DeliverResponse, error) {
+		return &orderer.DeliverResponse{
+			Type: &orderer.DeliverResponse_Status{
+				Status: common.Status_FORBIDDEN,
+			},
+		}, nil
+	}
+	status, err := bd.Recv()
+	assert.Nil(t, err)
+	assert.Equal(t, common.Status_FORBIDDEN, status.GetStatus())
+	bd.(*MockBlocksDeliverer).MockRecv = MockRecv
+	block, err := bd.Recv()
+	assert.Nil(t, err)
+	assert.Equal(t, uint64(0), block.GetBlock().Header.Number)
+
+	bd.(*MockBlocksDeliverer).Close()
+
+	seekInfo := &orderer.SeekInfo{
+		Start:    &orderer.SeekPosition{Type: &orderer.SeekPosition_Oldest{Oldest: &orderer.SeekOldest{}}},
+		Stop:     &orderer.SeekPosition{Type: &orderer.SeekPosition_Specified{Specified: &orderer.SeekSpecified{Number: math.MaxUint64}}},
+		Behavior: orderer.SeekInfo_BLOCK_UNTIL_READY,
+	}
+	si, err := pb.Marshal(seekInfo)
+	assert.NoError(t, err)
+
+	payload := &common.Payload{}
+
+	payload.Data = si
+	b, err := pb.Marshal(payload)
+	assert.NoError(t, err)
+	assert.Nil(t, bd.Send(&common.Envelope{Payload: b}))
 }
 
 func TestMockGossipServiceAdapter(t *testing.T) {
+	// Make sure it implements GossipServiceAdapter
 	var gsa blocksprovider.GossipServiceAdapter
-	gsa = &MockGossipServiceAdapter{}
+	seqNums := make(chan uint64, 1)
+	gsa = &MockGossipServiceAdapter{GossipBlockDisseminations: seqNums}
 	_ = gsa
 
+	// Test gossip
+	msg := &proto.GossipMessage{
+		Content: &proto.GossipMessage_DataMsg{
+			DataMsg: &proto.DataMessage{
+				Payload: &proto.Payload{
+					SeqNum: uint64(100),
+				},
+			},
+		},
+	}
+	gsa.Gossip(msg)
+	select {
+	case seq := <-seqNums:
+		assert.Equal(t, uint64(100), seq)
+	case <-time.After(time.Second):
+		assert.Fail(t, "Didn't gossip within a timely manner")
+	}
+
+	// Test AddPayload
+	gsa.AddPayload("TEST", msg.GetDataMsg().Payload)
+	assert.Equal(t, int32(1), atomic.LoadInt32(&(gsa.(*MockGossipServiceAdapter).AddPayloadsCnt)))
+
+	// Test PeersOfChannel
+	assert.Len(t, gsa.PeersOfChannel(nil), 0)
+}
+
+func TestMockAtomicBroadcastClient(t *testing.T) {
+	// Make sure it implements MockAtomicBroadcastClient
+	var abc orderer.AtomicBroadcastClient
+	abc = &MockAtomicBroadcastClient{BD: &MockBlocksDeliverer{}}
+
+	assert.Panics(t, func() {
+		abc.Broadcast(context.Background())
+	})
+	c, err := abc.Deliver(context.Background())
+	assert.Nil(t, err)
+	assert.NotNil(t, c)
 }
 
 func TestMockLedgerInfo(t *testing.T) {
 	var li blocksprovider.LedgerInfo
-	li = &MockLedgerInfo{}
+	li = &MockLedgerInfo{uint64(8)}
 	_ = li
+
+	height, err := li.LedgerHeight()
+	assert.Equal(t, uint64(8), height)
+	assert.NoError(t, err)
 }

core/deliverservice/mocks/orderer.go

@@ -77,7 +77,6 @@ func (o *Orderer) SendBlock(seq uint64) {
 func (o *Orderer) Deliver(stream orderer.AtomicBroadcast_DeliverServer) error {
 	envlp, err := stream.Recv()
 	if err != nil {
-		fmt.Println(err)
 		return nil
 	}
 	payload := &common.Payload{}

core/deliverservice/mocks/orderer_test.go

@@ -0,0 +1,69 @@
+/*
+Copyright IBM Corp. 2017 All Rights Reserved.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+                 http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package mocks
+
+import (
+	"math"
+	"testing"
+	"time"
+
+	pb "github.com/golang/protobuf/proto"
+	"github.com/hyperledger/fabric/protos/common"
+	"github.com/hyperledger/fabric/protos/orderer"
+	"github.com/stretchr/testify/assert"
+	"google.golang.org/grpc"
+)
+
+type clStream struct {
+	grpc.ServerStream
+}
+
+func (cs *clStream) Send(*orderer.DeliverResponse) error {
+	return nil
+}
+func (cs *clStream) Recv() (*common.Envelope, error) {
+	seekInfo := &orderer.SeekInfo{
+		Start:    &orderer.SeekPosition{Type: &orderer.SeekPosition_Specified{Specified: &orderer.SeekSpecified{Number: 0}}},
+		Stop:     &orderer.SeekPosition{Type: &orderer.SeekPosition_Specified{Specified: &orderer.SeekSpecified{Number: math.MaxUint64}}},
+		Behavior: orderer.SeekInfo_BLOCK_UNTIL_READY,
+	}
+	si, _ := pb.Marshal(seekInfo)
+	payload := &common.Payload{}
+	payload.Data = si
+	b, err := pb.Marshal(payload)
+	if err != nil {
+		panic(err)
+	}
+	e := &common.Envelope{Payload: b}
+	return e, nil
+}
+
+func TestOrderer(t *testing.T) {
+	o := NewOrderer(8000, t)
+
+	go func() {
+		time.Sleep(time.Second)
+		o.SendBlock(uint64(0))
+		o.Shutdown()
+	}()
+
+	assert.Panics(t, func() {
+		o.Broadcast(nil)
+	})
+	o.SetNextExpectedSeek(uint64(0))
+	o.Deliver(&clStream{})
+}