fix ut

plugin/consensus/para/para.go

@@ -162,11 +162,8 @@ func New(cfg *types.Consensus, sub []byte) queue.Module {
 		quitCreate:  make(chan struct{}),
 	}
 
-	waitBlocks := int32(3) //最小是2
+	waitBlocks := int32(3) //缺省是3
 	if subcfg.WaitBlocks4CommitMsg > 0 {
-		if subcfg.WaitBlocks4CommitMsg < waitBlocks {
-			panic("config WaitBlocks4CommitMsg should not less 2")
-		}
 		waitBlocks = subcfg.WaitBlocks4CommitMsg
 	}
 
@@ -181,7 +178,7 @@ func New(cfg *types.Consensus, sub []byte) queue.Module {
 		waitConsensStopTimes: waitConsensTimes,
 		commitCh:             make(chan int64, 1),
 		resetCh:              make(chan int64, 1),
-		verifyCh:             make(chan bool, 1),
+		verifyCh:             make(chan []byte, 1),
 		consensHeight:        -2,
 		sendingHeight:        -1,
 		quit:                 make(chan struct{}),

plugin/consensus/para/para_test.go

@@ -212,27 +212,31 @@ func initBlock() {
 	println("initblock")
 }
 
-func TestGetLastBlockInfo(t *testing.T) {
-	para := new(client)
-
+func getMockLastBlock(para *client, returnBlock *types.Block) {
 	baseCli := drivers.NewBaseClient(&types.Consensus{Name: "name"})
 	para.BaseClient = baseCli
-	grpcClient := &typesmocks.Chain33Client{}
+
 	qClient := &qmocks.Client{}
 	para.InitClient(qClient, initBlock)
 
-	api := &apimocks.QueueProtocolAPI{}
-	para.SetAPI(api)
+	msg := queue.NewMessage(0, "", 1, returnBlock)
+
+	qClient.On("NewMessage", "blockchain", int64(types.EventGetLastBlock), mock.Anything).Return(msg)
+	qClient.On("Send", mock.Anything, mock.Anything).Return(nil)
+
+	qClient.On("Wait", mock.Anything).Return(msg, nil)
+}
 
+func TestGetLastBlockInfo(t *testing.T) {
+	para := new(client)
+	grpcClient := &typesmocks.Chain33Client{}
 	para.grpcClient = grpcClient
 
 	block := &types.Block{Height: 0}
-	msg := queue.NewMessage(0, "", 1, block)
+	getMockLastBlock(para, block)
 
-	qClient.On("NewMessage", mock.Anything, mock.Anything, mock.Anything).Return(msg)
-	qClient.On("Send", mock.Anything, mock.Anything).Return(nil)
-
-	qClient.On("Wait", mock.Anything).Return(msg, nil)
+	api := &apimocks.QueueProtocolAPI{}
+	para.SetAPI(api)
 
 	grpcClient.On("GetSequenceByHash", mock.Anything, mock.Anything).Return(&types.Int64{Data: int64(10)}, nil)
 

plugin/consensus/para/paracommitmsg.go

@@ -13,6 +13,8 @@ import (
 	"sync/atomic"
 	"unsafe"
 
+	"bytes"
+
 	"github.com/33cn/chain33/common"
 	"github.com/33cn/chain33/common/crypto"
 	"github.com/33cn/chain33/types"
@@ -32,7 +34,7 @@ type commitMsgClient struct {
 	waitConsensStopTimes uint32 //共识高度低于完成高度， reset高度重发等待的次数
 	commitCh             chan int64
 	resetCh              chan int64
-	verifyCh             chan bool
+	verifyCh             chan []byte
 	sendMsgCh            chan *types.Transaction
 	minerSwitch          int32
 	currentTx            unsafe.Pointer
@@ -112,8 +114,8 @@ func (client *commitMsgClient) resetNotify() {
 	client.resetCh <- 1
 }
 
-func (client *commitMsgClient) verifyNotify(verified bool) {
-	client.verifyCh <- verified
+func (client *commitMsgClient) verifyNotify(verifyTx []byte) {
+	client.verifyCh <- verifyTx
 }
 
 func (client *commitMsgClient) resetSendEnv() {
@@ -169,13 +171,13 @@ func (client *commitMsgClient) procSendTx() {
 
 }
 
-func (client *commitMsgClient) procVerifyTx(verified bool) {
+func (client *commitMsgClient) procVerifyTx(verifyTx []byte) {
 	curTx := client.getCurrentTx()
 	if curTx == nil {
 		return
 	}
 
-	if verified {
+	if bytes.Equal(curTx.Hash(), verifyTx) {
 		client.clearCurrentTx()
 		client.procSendTx()
 		return
@@ -190,6 +192,7 @@ func (client *commitMsgClient) procVerifyTx(verified bool) {
 
 }
 
+//如果共识高度一直没有追上发送高度，且当前发送高度已经上链，说明共识一直没达成，安全起见，超过停止次数后，重发
 func (client *commitMsgClient) checkConsensusStop(consensStopTimes uint32) uint32 {
 	consensHeight := client.getConsensusHeight()
 	if client.sendingHeight > consensHeight && !client.isSendingCommitMsg() {
@@ -649,7 +652,6 @@ func (client *commitMsgClient) getSelfConsensusStatus() (*pt.ParacrossStatus, er
 		if resp.Height > -1 {
 			var statusMainHeight int64
 			if pt.IsParaForkHeight(resp.MainHeight, pt.ForkLoopCheckCommitTxDone) {
-				plog.Error("getSelfConsensusStatus ForkLoopCheckCommitTxDone reach")
 				statusMainHeight = resp.MainHeight
 			} else {
 				block, err := client.paraClient.GetBlockByHeight(resp.Height)

plugin/consensus/para/paracreate.go

@@ -64,7 +64,11 @@ func (client *client) checkCommitTxSuccess(txs []*pt.TxDetail) {
 		return
 	}
 
-	client.commitMsgClient.verifyNotify(txMap[string(curTx.Hash())])
+	if txMap[string(curTx.Hash())] {
+		client.commitMsgClient.verifyNotify(curTx.Hash())
+	} else {
+		client.commitMsgClient.verifyNotify(nil)
+	}
 
 }
 

plugin/consensus/para/paratxrequest_test.go

@@ -14,6 +14,7 @@ import (
 	"github.com/33cn/chain33/common/log"
 	"github.com/stretchr/testify/mock"
 
+	apimocks "github.com/33cn/chain33/client/mocks"
 	_ "github.com/33cn/chain33/system"
 	"github.com/33cn/chain33/types"
 	typesmocks "github.com/33cn/chain33/types/mocks"
@@ -69,32 +70,46 @@ func TestCalcCommitMsgTxs(t *testing.T) {
 
 }
 
-//func TestGetConsensusStatus(t *testing.T) {
-//	para := new(client)
-//	grpcClient := &typesmocks.Chain33Client{}
-//	//grpcClient.On("GetFork", mock.Anything, &types.ReqKey{Key: []byte("ForkBlockHash")}).Return(&types.Int64{Data: 1}, errors.New("err")).Once()
-//	para.grpcClient = grpcClient
-//	commitCli := new(commitMsgClient)
-//	commitCli.paraClient = para
-//
-//	block := &types.Block{
-//		Height:     1,
-//		MainHeight: 10,
-//	}
-//
-//	status := &pt.ParacrossStatus{
-//		Height: 1,
-//	}
-//	reply := &types.Reply{
-//		IsOk: true,
-//		Msg:  types.Encode(status),
-//	}
-//	grpcClient.On("QueryChain", mock.Anything, mock.Anything).Return(reply, nil).Once()
-//	ret, err := commitCli.getSelfConsensusStatus()
-//
-//	assert.Nil(t, err)
-//	assert.Equal(t, int64(1), ret.Height)
-//}
+func TestGetConsensusStatus(t *testing.T) {
+	mainFork := mainParaSelfConsensusForkHeight
+	mainParaSelfConsensusForkHeight = 1
+	para := new(client)
+	grpcClient := &typesmocks.Chain33Client{}
+	//grpcClient.On("GetFork", mock.Anything, &types.ReqKey{Key: []byte("ForkBlockHash")}).Return(&types.Int64{Data: 1}, errors.New("err")).Once()
+	para.grpcClient = grpcClient
+	commitCli := new(commitMsgClient)
+	commitCli.paraClient = para
+
+	block := &types.Block{
+		Height:     1,
+		MainHeight: 10,
+	}
+	getMockLastBlock(para, block)
+
+	api := &apimocks.QueueProtocolAPI{}
+	para.SetAPI(api)
+
+	status := &pt.ParacrossStatus{
+		Height: 1,
+	}
+
+	//msgx := &types.Message{types.Encode(status)}
+	//msg := types.Encode(status)
+	//reply := &types.Reply{
+	//	IsOk: true,
+	//	Msg:  types.Encode(status),
+	//}
+	api.On("QueryChain", mock.Anything, mock.Anything, mock.Anything).Return(status, nil).Once()
+	detail := &types.BlockDetail{Block: block}
+	details := &types.BlockDetails{Items: []*types.BlockDetail{detail}}
+
+	api.On("GetBlocks", mock.Anything).Return(details, nil).Once()
+	ret, err := commitCli.getSelfConsensusStatus()
+
+	assert.Nil(t, err)
+	assert.Equal(t, int64(1), ret.Height)
+	mainParaSelfConsensusForkHeight = mainFork
+}
 
 func TestSendCommitMsg(t *testing.T) {
 	para := new(client)