test: add integration test (#315)

* Rename NonByzantineTestSuite to WithSchedulerTestsuite
* Add a method to query the latest position delivered
* Add integration test for core.Consensus
* Show detailed list for test cases in CI

7 files changed, 197 insertions, 20 deletions

diff --git a/GNUmakefile b/GNUmakefile
index 4988fa3..51c6b34 100644
--- a/GNUmakefile
+++ b/GNUmakefile
@@ -85,7 +85,7 @@ vet:
 
 test-short:
 	@for pkg in `go list ./... | grep -v 'vendor'`; do \
-		if ! go test -race -short $$pkg; then \
+		if ! go test -race -short -v $$pkg; then \
 			echo 'Some test failed, abort'; \
 			exit 1; \
 		fi; \
@@ -93,7 +93,7 @@ test-short:
 
 test:
 	@for pkg in `go list ./... | grep -v 'vendor'`; do \
-		if ! go test -timeout 15m -race $$pkg; then \
+		if ! go test -timeout 15m -race -v $$pkg; then \
 			echo 'Some test failed, abort'; \
 			exit 1; \
 		fi; \
diff --git a/core/consensus.go b/core/consensus.go
index a449701..89f2242 100644
--- a/core/consensus.go
+++ b/core/consensus.go
@@ -719,7 +719,6 @@ func (con *Consensus) initialRound(
 			}
 			con.initialRound(
 				startTime.Add(config.RoundInterval), nextRound, nextConfig)
-			con.round = nextRound
 		})
 }
 
@@ -909,9 +908,6 @@ func (con *Consensus) ProcessAgreementResult(
 	if !con.ccModule.blockRegistered(rand.BlockHash) {
 		return nil
 	}
-	if DiffUint64(con.round, rand.Position.Round) > 1 {
-		return nil
-	}
 	// Sanity check done.
 	if !con.cfgModule.touchTSigHash(rand.BlockHash) {
 		return nil
diff --git a/core/test/app.go b/core/test/app.go
index 7477329..a5d0270 100644
--- a/core/test/app.go
+++ b/core/test/app.go
@@ -76,6 +76,7 @@ type AppDeliveredRecord struct {
 	ConsensusTime   time.Time
 	ConsensusHeight uint64
 	When            time.Time
+	Pos             types.Position
 }
 
 // App implements Application interface for testing purpose.
@@ -89,7 +90,7 @@ type App struct {
 	DeliverSequence    common.Hashes
 	deliveredLock      sync.RWMutex
 	blocks             map[common.Hash]*types.Block
-	blocksLock         sync.Mutex
+	blocksLock         sync.RWMutex
 	state              *State
 }
 
@@ -162,7 +163,7 @@ func (app *App) TotalOrderingDelivered(blockHashes common.Hashes, mode uint32) {
 
 // BlockDelivered implements Application interface.
 func (app *App) BlockDelivered(
-	blockHash common.Hash, _ types.Position, result types.FinalizationResult) {
+	blockHash common.Hash, pos types.Position, result types.FinalizationResult) {
 	func() {
 		app.deliveredLock.Lock()
 		defer app.deliveredLock.Unlock()
@@ -170,6 +171,7 @@ func (app *App) BlockDelivered(
 			ConsensusTime:   result.Timestamp,
 			ConsensusHeight: result.Height,
 			When:            time.Now().UTC(),
+			Pos:             pos,
 		}
 		app.DeliverSequence = append(app.DeliverSequence, blockHash)
 	}()
@@ -178,8 +180,8 @@ func (app *App) BlockDelivered(
 		if app.state == nil {
 			return
 		}
-		app.blocksLock.Lock()
-		defer app.blocksLock.Unlock()
+		app.blocksLock.RLock()
+		defer app.blocksLock.RUnlock()
 		b := app.blocks[blockHash]
 		if err := app.state.Apply(b.Payload); err != nil {
 			if err != ErrDuplicatedChange {
@@ -189,6 +191,16 @@ func (app *App) BlockDelivered(
 	}()
 }
 
+// GetLatestDeliveredPosition would return the latest position of delivered
+// block seen by this application instance.
+func (app *App) GetLatestDeliveredPosition() types.Position {
+	app.deliveredLock.RLock()
+	defer app.deliveredLock.RUnlock()
+	app.blocksLock.RLock()
+	defer app.blocksLock.RUnlock()
+	return app.blocks[app.DeliverSequence[len(app.DeliverSequence)-1]].Position
+}
+
 // Compare performs these checks against another App instance
 // and return erros if not passed:
 // - deliver sequence by comparing block hashes.
diff --git a/core/test/network.go b/core/test/network.go
index 949b158..6aff3d6 100644
--- a/core/test/network.go
+++ b/core/test/network.go
@@ -207,6 +207,8 @@ func (n *Network) BroadcastVote(vote *types.Vote) {
 
 // BroadcastBlock implements core.Network interface.
 func (n *Network) BroadcastBlock(block *types.Block) {
+	// Avoid data race in fake transport.
+	block = n.cloneForFake(block).(*types.Block)
 	if err := n.trans.Broadcast(block); err != nil {
 		panic(err)
 	}
@@ -316,7 +318,8 @@ func (n *Network) Setup(serverEndpoint interface{}) (err error) {
 }
 
 func (n *Network) dispatchMsg(e *TransportEnvelope) {
-	switch v := e.Msg.(type) {
+	msg := n.cloneForFake(e.Msg)
+	switch v := msg.(type) {
 	case *types.Block:
 		n.addBlockToCache(v)
 		// Notify pulling routine about the newly arrived block.
@@ -327,14 +330,14 @@ func (n *Network) dispatchMsg(e *TransportEnvelope) {
 				ch <- v.Hash
 			}
 		}()
-		n.toConsensus <- e.Msg
+		n.toConsensus <- v
 	case *types.Vote:
 		// Add this vote to cache.
 		n.addVoteToCache(v)
-		n.toConsensus <- e.Msg
+		n.toConsensus <- v
 	case *types.AgreementResult, *types.BlockRandomnessResult,
 		*typesDKG.PrivateShare, *typesDKG.PartialSignature:
-		n.toConsensus <- e.Msg
+		n.toConsensus <- v
 	case packedStateChanges:
 		if n.stateModule == nil {
 			panic(errors.New(
@@ -346,7 +349,7 @@ func (n *Network) dispatchMsg(e *TransportEnvelope) {
 	case *PullRequest:
 		go n.handlePullRequest(v)
 	default:
-		n.toNode <- e.Msg
+		n.toNode <- v
 	}
 }
 
@@ -596,3 +599,17 @@ func (n *Network) addVoteToCache(v *types.Vote) {
 	n.voteCache[v.Position][v.VoteHeader] = v
 	n.voteCacheSize++
 }
+
+func (n *Network) cloneForFake(v interface{}) interface{} {
+	if n.config.Type != NetworkTypeFake {
+		return v
+	}
+	switch val := v.(type) {
+	case *types.Block:
+		return val.Clone()
+	case *types.BlockRandomnessResult:
+		// Perform deep copy for randomness result.
+		return cloneBlockRandomnessResult(val)
+	}
+	return v
+}
diff --git a/core/test/utils.go b/core/test/utils.go
index f5f4c36..56c5eac 100644
--- a/core/test/utils.go
+++ b/core/test/utils.go
@@ -157,3 +157,16 @@ func cloneDKGFinalize(final *typesDKG.Finalize) (
 	}
 	return
 }
+
+func cloneBlockRandomnessResult(rand *types.BlockRandomnessResult) (
+	copied *types.BlockRandomnessResult) {
+	b, err := rlp.EncodeToBytes(rand)
+	if err != nil {
+		panic(err)
+	}
+	copied = &types.BlockRandomnessResult{}
+	if err = rlp.DecodeBytes(b, copied); err != nil {
+		panic(err)
+	}
+	return
+}
diff --git a/integration_test/consensus_test.go b/integration_test/consensus_test.go
new file mode 100644
index 0000000..52f7f9f
--- /dev/null
+++ b/integration_test/consensus_test.go
@@ -0,0 +1,139 @@
+// Copyright 2018 The dexon-consensus Authors
+// This file is part of the dexon-consensus library.
+//
+// The dexon-consensus library is free software: you can redistribute it
+// and/or modify it under the terms of the GNU Lesser General Public License as
+// published by the Free Software Foundation, either version 3 of the License,
+// or (at your option) any later version.
+//
+// The dexon-consensus library is distributed in the hope that it will be
+// useful, but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser
+// General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with the dexon-consensus library. If not, see
+// <http://www.gnu.org/licenses/>.
+
+package integration
+
+import (
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/dexon-foundation/dexon-consensus/common"
+	"github.com/dexon-foundation/dexon-consensus/core"
+	"github.com/dexon-foundation/dexon-consensus/core/blockdb"
+	"github.com/dexon-foundation/dexon-consensus/core/crypto"
+	"github.com/dexon-foundation/dexon-consensus/core/test"
+	"github.com/dexon-foundation/dexon-consensus/core/types"
+	"github.com/stretchr/testify/suite"
+)
+
+// There is no scheduler in these tests, we need to wait a long period to make
+// sure these tests are ok.
+type ConsensusTestSuite struct {
+	suite.Suite
+}
+
+type node struct {
+	con *core.Consensus
+	app *test.App
+	gov *test.Governance
+	db  blockdb.BlockDatabase
+}
+
+func (s *ConsensusTestSuite) setupNodes(
+	dMoment time.Time,
+	prvKeys []crypto.PrivateKey,
+	seedGov *test.Governance) map[types.NodeID]*node {
+	var (
+		wg sync.WaitGroup
+	)
+	// Setup peer server at transport layer.
+	server := test.NewFakeTransportServer()
+	serverChannel, err := server.Host()
+	s.Require().NoError(err)
+	// setup nodes.
+	nodes := make(map[types.NodeID]*node)
+	wg.Add(len(prvKeys))
+	for _, k := range prvKeys {
+		db, err := blockdb.NewMemBackedBlockDB()
+		s.Require().NoError(err)
+		// Prepare essential modules: app, gov, db.
+		networkModule := test.NewNetwork(
+			k.PublicKey(),
+			&test.FixedLatencyModel{},
+			test.NewDefaultMarshaller(nil),
+			test.NetworkConfig{Type: test.NetworkTypeFake})
+		gov := seedGov.Clone()
+		gov.SwitchToRemoteMode(networkModule)
+		app := test.NewApp(gov.State())
+		// Now is the consensus module.
+		con := core.NewConsensus(
+			dMoment,
+			app,
+			gov,
+			db,
+			networkModule,
+			k,
+			&common.NullLogger{})
+		nodes[con.ID] = &node{con, app, gov, db}
+		go func() {
+			defer wg.Done()
+			s.Require().NoError(networkModule.Setup(serverChannel))
+			go networkModule.Run()
+		}()
+	}
+	// Make sure transport layer is ready.
+	s.Require().NoError(server.WaitForPeers(uint32(len(prvKeys))))
+	wg.Wait()
+	return nodes
+}
+
+func (s *ConsensusTestSuite) TestSimple() {
+	// The simplest test case:
+	//  - Node set is equals to DKG set and notary set for each chain in each
+	//    round.
+	//  - No configuration change.
+	//  - 4 rounds (0, 1 are genesis rounds, round 2 would be ready when the
+	//    first block delivered. Test until round 3 should be enough.
+	var (
+		req        = s.Require()
+		peerCount  = 4
+		dMoment    = time.Now().UTC()
+		untilRound = uint64(5)
+	)
+	prvKeys, pubKeys, err := test.NewKeys(peerCount)
+	req.NoError(err)
+	// Setup seed governance instance. Give a short latency to make this test
+	// run faster.
+	seedGov, err := test.NewGovernance(pubKeys, 30*time.Millisecond)
+	req.NoError(err)
+	req.NoError(seedGov.State().RequestChange(
+		test.StateChangeRoundInterval, 25*time.Second))
+	// A short round interval.
+	nodes := s.setupNodes(dMoment, prvKeys, seedGov)
+	for _, n := range nodes {
+		go n.con.Run(&types.Block{})
+	}
+Loop:
+	for {
+		<-time.After(5 * time.Second)
+		s.T().Log("check latest position delivered by each node")
+		for _, n := range nodes {
+			latestPos := n.app.GetLatestDeliveredPosition()
+			s.T().Log("latestPos", n.con.ID, &latestPos)
+			if latestPos.Round < untilRound {
+				continue Loop
+			}
+		}
+		// Oh ya.
+		break
+	}
+}
+
+func TestConsensus(t *testing.T) {
+	suite.Run(t, new(ConsensusTestSuite))
+}
diff --git a/integration_test/non-byzantine_test.go b/integration_test/with_scheduler_test.go
index bc016d5..1105805 100644
--- a/integration_test/non-byzantine_test.go
+++ b/integration_test/with_scheduler_test.go
@@ -25,11 +25,11 @@ import (
 	"github.com/stretchr/testify/suite"
 )
 
-type NonByzantineTestSuite struct {
+type WithSchedulerTestSuite struct {
 	suite.Suite
 }
 
-func (s *NonByzantineTestSuite) TestNonByzantine() {
+func (s *WithSchedulerTestSuite) TestNonByzantine() {
 	var (
 		networkLatency = &test.NormalLatencyModel{
 			Sigma: 20,
@@ -67,7 +67,7 @@ func (s *NonByzantineTestSuite) TestNonByzantine() {
 	req.NoError(VerifyApps(apps))
 }
 
-func (s *NonByzantineTestSuite) TestConfigurationChange() {
+func (s *WithSchedulerTestSuite) TestConfigurationChange() {
 	// This test case verify the correctness of core.Lattice when configuration
 	// changes.
 	// - Configuration changes are registered at 'pickedNode', and would carried
@@ -137,6 +137,6 @@ func (s *NonByzantineTestSuite) TestConfigurationChange() {
 	req.NoError(VerifyApps(apps))
 }
 
-func TestNonByzantine(t *testing.T) {
-	suite.Run(t, new(NonByzantineTestSuite))
+func TestWithScheduler(t *testing.T) {
+	suite.Run(t, new(WithSchedulerTestSuite))
 }