netsync/chainmgr/tx_keeper_test.go

   1 package chainmgr
   2
   3 import (
   4         "io/ioutil"
   5         "os"
   6         "reflect"
   7         "testing"
   8         "time"
   9
  10         "github.com/davecgh/go-spew/spew"
  11
  12         "github.com/bytom/vapor/consensus"
  13         dbm "github.com/bytom/vapor/database/leveldb"
  14         "github.com/bytom/vapor/protocol"
  15         core "github.com/bytom/vapor/protocol"
  16         "github.com/bytom/vapor/protocol/bc"
  17         "github.com/bytom/vapor/protocol/bc/types"
  18         "github.com/bytom/vapor/test/mock"
  19 )
  20
  21 const txsNumber = 2000
  22
  23 type mempool struct {
  24 }
  25
  26 func (m *mempool) GetTransactions() []*core.TxDesc {
  27         txs := []*core.TxDesc{}
  28         for i := 0; i < txsNumber; i++ {
  29                 txInput := types.NewSpendInput(nil, bc.NewHash([32]byte{0x01}), *consensus.BTMAssetID, uint64(i), 1, []byte{0x51})
  30                 txInput.CommitmentSuffix = []byte{0, 1, 2}
  31                 txInput.WitnessSuffix = []byte{0, 1, 2}
  32
  33                 tx := &types.Tx{
  34
  35                         TxData: types.TxData{
  36                                 //SerializedSize: uint64(i * 10),
  37                                 Inputs: []*types.TxInput{
  38                                         txInput,
  39                                 },
  40                                 Outputs: []*types.TxOutput{
  41                                         types.NewIntraChainOutput(*consensus.BTMAssetID, uint64(i), []byte{0x6a}),
  42                                 },
  43                                 SerializedSize: 1000,
  44                         },
  45                         Tx: &bc.Tx{
  46                                 ID: bc.Hash{V0: uint64(i), V1: uint64(i), V2: uint64(i), V3: uint64(i)},
  47                         },
  48                 }
  49                 txs = append(txs, &core.TxDesc{Tx: tx})
  50         }
  51         return txs
  52 }
  53
  54 func (m *mempool) IsDust(tx *types.Tx) bool {
  55         return false
  56 }
  57
  58 func TestSyncMempool(t *testing.T) {
  59         tmpDir, err := ioutil.TempDir(".", "")
  60         if err != nil {
  61                 t.Fatalf("failed to create temporary data folder: %v", err)
  62         }
  63         defer os.RemoveAll(tmpDir)
  64         testDBA := dbm.NewDB("testdba", "leveldb", tmpDir)
  65         testDBB := dbm.NewDB("testdbb", "leveldb", tmpDir)
  66
  67         blocks := mockBlocks(nil, 5)
  68         a := mockSync(blocks, &mock.Mempool{}, testDBA)
  69         b := mockSync(blocks, &mock.Mempool{}, testDBB)
  70         a.mempool = &mempool{}
  71         netWork := NewNetWork()
  72         netWork.Register(a, "192.168.0.1", "test node A", consensus.SFFullNode)
  73         netWork.Register(b, "192.168.0.2", "test node B", consensus.SFFullNode)
  74         if B2A, A2B, err := netWork.HandsShake(a, b); err != nil {
  75                 t.Errorf("fail on peer hands shake %v", err)
  76         } else {
  77                 go B2A.postMan()
  78                 go A2B.postMan()
  79         }
  80
  81         go a.syncMempoolLoop()
  82         a.syncMempool("test node B")
  83         wantTxs := a.mempool.GetTransactions()
  84
  85         timeout := time.NewTimer(2 * time.Second)
  86         defer timeout.Stop()
  87         ticker := time.NewTicker(500 * time.Millisecond)
  88         defer ticker.Stop()
  89
  90         gotTxs := []*protocol.TxDesc{}
  91         for {
  92                 select {
  93                 case <-ticker.C:
  94                         gotTxs = b.mempool.GetTransactions()
  95                         if len(gotTxs) >= txsNumber {
  96                                 goto out
  97                         }
  98                 case <-timeout.C:
  99                         t.Fatalf("mempool sync timeout")
 100                 }
 101         }
 102
 103 out:
 104         if len(gotTxs) != txsNumber {
 105                 t.Fatalf("mempool sync txs num err. got:%d want:%d", len(gotTxs), txsNumber)
 106         }
 107
 108         for i, gotTx := range gotTxs {
 109                 index := gotTx.Tx.Inputs[0].Amount()
 110                 if !reflect.DeepEqual(gotTx.Tx.Inputs[0].Amount(), wantTxs[index].Tx.Inputs[0].Amount()) {
 111                         t.Fatalf("mempool tx err. index:%d\n,gotTx:%s\n,wantTx:%s", i, spew.Sdump(gotTx.Tx.Inputs), spew.Sdump(wantTxs[0].Tx.Inputs))
 112                 }
 113
 114                 if !reflect.DeepEqual(gotTx.Tx.Outputs[0].AssetAmount(), wantTxs[index].Tx.Outputs[0].AssetAmount()) {
 115                         t.Fatalf("mempool tx err. index:%d\n,gotTx:%s\n,wantTx:%s", i, spew.Sdump(gotTx.Tx.Outputs), spew.Sdump(wantTxs[0].Tx.Outputs))
 116                 }
 117         }
 118 }
 119
 120 func TestBroadcastTxsLoop(t *testing.T) {
 121         tmpDir, err := ioutil.TempDir(".", "")
 122         if err != nil {
 123                 t.Fatalf("failed to create temporary data folder: %v", err)
 124         }
 125         defer os.RemoveAll(tmpDir)
 126         testDBA := dbm.NewDB("testdba", "leveldb", tmpDir)
 127         testDBB := dbm.NewDB("testdbb", "leveldb", tmpDir)
 128
 129         blocks := mockBlocks(nil, 5)
 130         a := mockSync(blocks, &mock.Mempool{}, testDBA)
 131         b := mockSync(blocks, &mock.Mempool{}, testDBB)
 132         a.mempool = &mempool{}
 133         netWork := NewNetWork()
 134         netWork.Register(a, "192.168.0.1", "test node A", consensus.SFFullNode)
 135         netWork.Register(b, "192.168.0.2", "test node B", consensus.SFFullNode)
 136         if B2A, A2B, err := netWork.HandsShake(a, b); err != nil {
 137                 t.Errorf("fail on peer hands shake %v", err)
 138         } else {
 139                 go B2A.postMan()
 140                 go A2B.postMan()
 141         }
 142         a.txMsgSub, err = a.eventDispatcher.Subscribe(core.TxMsgEvent{})
 143         if err != nil {
 144                 t.Fatal("txMsgSub subscribe err", err)
 145         }
 146         go a.broadcastTxsLoop()
 147         wantTxs := a.mempool.GetTransactions()
 148         txsNum := 50
 149         for i, txD := range wantTxs {
 150                 if i >= txsNum {
 151                         break
 152                 }
 153                 a.eventDispatcher.Post(core.TxMsgEvent{TxMsg: &core.TxPoolMsg{TxDesc: txD, MsgType: core.MsgNewTx}})
 154         }
 155         timeout := time.NewTimer(2 * time.Second)
 156         defer timeout.Stop()
 157         ticker := time.NewTicker(500 * time.Millisecond)
 158         defer ticker.Stop()
 159
 160         gotTxs := []*protocol.TxDesc{}
 161         for {
 162                 select {
 163                 case <-ticker.C:
 164                         gotTxs = b.mempool.GetTransactions()
 165                         if len(gotTxs) >= txsNum {
 166                                 goto out
 167                         }
 168                 case <-timeout.C:
 169                         t.Fatalf("mempool sync timeout")
 170                 }
 171         }
 172
 173 out:
 174         if len(gotTxs) != txsNum {
 175                 t.Fatalf("mempool sync txs num err. got:%d want:%d", len(gotTxs), txsNumber)
 176         }
 177
 178         for i, gotTx := range gotTxs {
 179                 index := gotTx.Tx.Inputs[0].Amount()
 180                 if !reflect.DeepEqual(gotTx.Tx.Inputs[0].Amount(), wantTxs[index].Tx.Inputs[0].Amount()) {
 181                         t.Fatalf("mempool tx err. index:%d\n,gotTx:%s\n,wantTx:%s", i, spew.Sdump(gotTx.Tx.Inputs), spew.Sdump(wantTxs[0].Tx.Inputs))
 182                 }
 183
 184                 if !reflect.DeepEqual(gotTx.Tx.Outputs[0].AssetAmount(), wantTxs[index].Tx.Outputs[0].AssetAmount()) {
 185                         t.Fatalf("mempool tx err. index:%d\n,gotTx:%s\n,wantTx:%s", i, spew.Sdump(gotTx.Tx.Outputs), spew.Sdump(wantTxs[0].Tx.Outputs))
 186                 }
 187         }
 188 }