test_helper.go 24 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682
  1. // Copyright 2019 The go-ethereum Authors
  2. // This file is part of the go-ethereum library.
  3. //
  4. // The go-ethereum library is free software: you can redistribute it and/or modify
  5. // it under the terms of the GNU Lesser General Public License as published by
  6. // the Free Software Foundation, either version 3 of the License, or
  7. // (at your option) any later version.
  8. //
  9. // The go-ethereum library is distributed in the hope that it will be useful,
  10. // but WITHOUT ANY WARRANTY; without even the implied warranty of
  11. // MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
  12. // GNU Lesser General Public License for more details.
  13. //
  14. // You should have received a copy of the GNU Lesser General Public License
  15. // along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
  16. // This file contains some shares testing functionality, common to multiple
  17. // different files and modules being tested. Client based network and Server
  18. // based network can be created easily with available APIs.
  19. package les
  20. import (
  21. "context"
  22. "crypto/rand"
  23. "fmt"
  24. "math/big"
  25. "sync/atomic"
  26. "testing"
  27. "time"
  28. "github.com/ethereum/go-ethereum/accounts/abi/bind"
  29. "github.com/ethereum/go-ethereum/accounts/abi/bind/backends"
  30. "github.com/ethereum/go-ethereum/common"
  31. "github.com/ethereum/go-ethereum/common/mclock"
  32. "github.com/ethereum/go-ethereum/consensus"
  33. "github.com/ethereum/go-ethereum/consensus/ethash"
  34. "github.com/ethereum/go-ethereum/contracts/checkpointoracle/contract"
  35. "github.com/ethereum/go-ethereum/core"
  36. "github.com/ethereum/go-ethereum/core/forkid"
  37. "github.com/ethereum/go-ethereum/core/rawdb"
  38. "github.com/ethereum/go-ethereum/core/types"
  39. "github.com/ethereum/go-ethereum/crypto"
  40. "github.com/ethereum/go-ethereum/eth/ethconfig"
  41. "github.com/ethereum/go-ethereum/ethdb"
  42. "github.com/ethereum/go-ethereum/event"
  43. "github.com/ethereum/go-ethereum/les/checkpointoracle"
  44. "github.com/ethereum/go-ethereum/les/flowcontrol"
  45. vfs "github.com/ethereum/go-ethereum/les/vflux/server"
  46. "github.com/ethereum/go-ethereum/light"
  47. "github.com/ethereum/go-ethereum/p2p"
  48. "github.com/ethereum/go-ethereum/p2p/enode"
  49. "github.com/ethereum/go-ethereum/params"
  50. )
  51. var (
  52. bankKey, _ = crypto.GenerateKey()
  53. bankAddr = crypto.PubkeyToAddress(bankKey.PublicKey)
  54. bankFunds = big.NewInt(1_000_000_000_000_000_000)
  55. userKey1, _ = crypto.GenerateKey()
  56. userKey2, _ = crypto.GenerateKey()
  57. userAddr1 = crypto.PubkeyToAddress(userKey1.PublicKey)
  58. userAddr2 = crypto.PubkeyToAddress(userKey2.PublicKey)
  59. testContractAddr common.Address
  60. testContractCode = common.Hex2Bytes("606060405260cc8060106000396000f360606040526000357c01000000000000000000000000000000000000000000000000000000009004806360cd2685146041578063c16431b914606b57603f565b005b6055600480803590602001909190505060a9565b6040518082815260200191505060405180910390f35b60886004808035906020019091908035906020019091905050608a565b005b80600060005083606481101560025790900160005b50819055505b5050565b6000600060005082606481101560025790900160005b5054905060c7565b91905056")
  61. testContractCodeDeployed = testContractCode[16:]
  62. testContractDeployed = uint64(2)
  63. testEventEmitterCode = common.Hex2Bytes("60606040523415600e57600080fd5b7f57050ab73f6b9ebdd9f76b8d4997793f48cf956e965ee070551b9ca0bb71584e60405160405180910390a160358060476000396000f3006060604052600080fd00a165627a7a723058203f727efcad8b5811f8cb1fc2620ce5e8c63570d697aef968172de296ea3994140029")
  64. // Checkpoint oracle relative fields
  65. oracleAddr common.Address
  66. signerKey, _ = crypto.GenerateKey()
  67. signerAddr = crypto.PubkeyToAddress(signerKey.PublicKey)
  68. )
  69. var (
  70. // The block frequency for creating checkpoint(only used in test)
  71. sectionSize = big.NewInt(128)
  72. // The number of confirmations needed to generate a checkpoint(only used in test).
  73. processConfirms = big.NewInt(1)
  74. // The token bucket buffer limit for testing purpose.
  75. testBufLimit = uint64(1000000)
  76. // The buffer recharging speed for testing purpose.
  77. testBufRecharge = uint64(1000)
  78. )
  79. /*
  80. contract test {
  81. uint256[100] data;
  82. function Put(uint256 addr, uint256 value) {
  83. data[addr] = value;
  84. }
  85. function Get(uint256 addr) constant returns (uint256 value) {
  86. return data[addr];
  87. }
  88. }
  89. */
  90. // prepare pre-commits specified number customized blocks into chain.
  91. func prepare(n int, backend *backends.SimulatedBackend) {
  92. var (
  93. ctx = context.Background()
  94. signer = types.HomesteadSigner{}
  95. )
  96. for i := 0; i < n; i++ {
  97. switch i {
  98. case 0:
  99. // Builtin-block
  100. // number: 1
  101. // txs: 2
  102. // deploy checkpoint contract
  103. auth, _ := bind.NewKeyedTransactorWithChainID(bankKey, big.NewInt(1337))
  104. oracleAddr, _, _, _ = contract.DeployCheckpointOracle(auth, backend, []common.Address{signerAddr}, sectionSize, processConfirms, big.NewInt(1))
  105. // bankUser transfers some ether to user1
  106. nonce, _ := backend.PendingNonceAt(ctx, bankAddr)
  107. tx, _ := types.SignTx(types.NewTransaction(nonce, userAddr1, big.NewInt(10_000_000_000_000_000), params.TxGas, big.NewInt(params.InitialBaseFee), nil), signer, bankKey)
  108. backend.SendTransaction(ctx, tx)
  109. case 1:
  110. // Builtin-block
  111. // number: 2
  112. // txs: 4
  113. bankNonce, _ := backend.PendingNonceAt(ctx, bankAddr)
  114. userNonce1, _ := backend.PendingNonceAt(ctx, userAddr1)
  115. // bankUser transfers more ether to user1
  116. tx1, _ := types.SignTx(types.NewTransaction(bankNonce, userAddr1, big.NewInt(1_000_000_000_000_000), params.TxGas, big.NewInt(params.InitialBaseFee), nil), signer, bankKey)
  117. backend.SendTransaction(ctx, tx1)
  118. // user1 relays ether to user2
  119. tx2, _ := types.SignTx(types.NewTransaction(userNonce1, userAddr2, big.NewInt(1_000_000_000_000_000), params.TxGas, big.NewInt(params.InitialBaseFee), nil), signer, userKey1)
  120. backend.SendTransaction(ctx, tx2)
  121. // user1 deploys a test contract
  122. tx3, _ := types.SignTx(types.NewContractCreation(userNonce1+1, big.NewInt(0), 200000, big.NewInt(params.InitialBaseFee), testContractCode), signer, userKey1)
  123. backend.SendTransaction(ctx, tx3)
  124. testContractAddr = crypto.CreateAddress(userAddr1, userNonce1+1)
  125. // user1 deploys a event contract
  126. tx4, _ := types.SignTx(types.NewContractCreation(userNonce1+2, big.NewInt(0), 200000, big.NewInt(params.InitialBaseFee), testEventEmitterCode), signer, userKey1)
  127. backend.SendTransaction(ctx, tx4)
  128. case 2:
  129. // Builtin-block
  130. // number: 3
  131. // txs: 2
  132. // bankUser transfer some ether to signer
  133. bankNonce, _ := backend.PendingNonceAt(ctx, bankAddr)
  134. tx1, _ := types.SignTx(types.NewTransaction(bankNonce, signerAddr, big.NewInt(1000000000), params.TxGas, big.NewInt(params.InitialBaseFee), nil), signer, bankKey)
  135. backend.SendTransaction(ctx, tx1)
  136. // invoke test contract
  137. data := common.Hex2Bytes("C16431B900000000000000000000000000000000000000000000000000000000000000010000000000000000000000000000000000000000000000000000000000000001")
  138. tx2, _ := types.SignTx(types.NewTransaction(bankNonce+1, testContractAddr, big.NewInt(0), 100000, big.NewInt(params.InitialBaseFee), data), signer, bankKey)
  139. backend.SendTransaction(ctx, tx2)
  140. case 3:
  141. // Builtin-block
  142. // number: 4
  143. // txs: 1
  144. // invoke test contract
  145. bankNonce, _ := backend.PendingNonceAt(ctx, bankAddr)
  146. data := common.Hex2Bytes("C16431B900000000000000000000000000000000000000000000000000000000000000020000000000000000000000000000000000000000000000000000000000000002")
  147. tx, _ := types.SignTx(types.NewTransaction(bankNonce, testContractAddr, big.NewInt(0), 100000, big.NewInt(params.InitialBaseFee), data), signer, bankKey)
  148. backend.SendTransaction(ctx, tx)
  149. }
  150. backend.Commit()
  151. }
  152. }
  153. // testIndexers creates a set of indexers with specified params for testing purpose.
  154. func testIndexers(db ethdb.Database, odr light.OdrBackend, config *light.IndexerConfig, disablePruning bool) []*core.ChainIndexer {
  155. var indexers [3]*core.ChainIndexer
  156. indexers[0] = light.NewChtIndexer(db, odr, config.ChtSize, config.ChtConfirms, disablePruning)
  157. indexers[1] = core.NewBloomIndexer(db, config.BloomSize, config.BloomConfirms)
  158. indexers[2] = light.NewBloomTrieIndexer(db, odr, config.BloomSize, config.BloomTrieSize, disablePruning)
  159. // make bloomTrieIndexer as a child indexer of bloom indexer.
  160. indexers[1].AddChildIndexer(indexers[2])
  161. return indexers[:]
  162. }
  163. func newTestClientHandler(backend *backends.SimulatedBackend, odr *LesOdr, indexers []*core.ChainIndexer, db ethdb.Database, peers *serverPeerSet, ulcServers []string, ulcFraction int) (*clientHandler, func()) {
  164. var (
  165. evmux = new(event.TypeMux)
  166. engine = ethash.NewFaker()
  167. gspec = core.Genesis{
  168. Config: params.AllEthashProtocolChanges,
  169. Alloc: core.GenesisAlloc{bankAddr: {Balance: bankFunds}},
  170. GasLimit: 100000000,
  171. BaseFee: big.NewInt(params.InitialBaseFee),
  172. }
  173. oracle *checkpointoracle.CheckpointOracle
  174. )
  175. genesis := gspec.MustCommit(db)
  176. chain, _ := light.NewLightChain(odr, gspec.Config, engine, nil)
  177. if indexers != nil {
  178. checkpointConfig := &params.CheckpointOracleConfig{
  179. Address: crypto.CreateAddress(bankAddr, 0),
  180. Signers: []common.Address{signerAddr},
  181. Threshold: 1,
  182. }
  183. getLocal := func(index uint64) params.TrustedCheckpoint {
  184. chtIndexer := indexers[0]
  185. sectionHead := chtIndexer.SectionHead(index)
  186. return params.TrustedCheckpoint{
  187. SectionIndex: index,
  188. SectionHead: sectionHead,
  189. CHTRoot: light.GetChtRoot(db, index, sectionHead),
  190. BloomRoot: light.GetBloomTrieRoot(db, index, sectionHead),
  191. }
  192. }
  193. oracle = checkpointoracle.New(checkpointConfig, getLocal)
  194. }
  195. client := &LightEthereum{
  196. lesCommons: lesCommons{
  197. genesis: genesis.Hash(),
  198. config: &ethconfig.Config{LightPeers: 100, NetworkId: NetworkId},
  199. chainConfig: params.AllEthashProtocolChanges,
  200. iConfig: light.TestClientIndexerConfig,
  201. chainDb: db,
  202. oracle: oracle,
  203. chainReader: chain,
  204. closeCh: make(chan struct{}),
  205. },
  206. peers: peers,
  207. reqDist: odr.retriever.dist,
  208. retriever: odr.retriever,
  209. odr: odr,
  210. engine: engine,
  211. blockchain: chain,
  212. eventMux: evmux,
  213. merger: consensus.NewMerger(rawdb.NewMemoryDatabase()),
  214. }
  215. client.handler = newClientHandler(ulcServers, ulcFraction, nil, client)
  216. if client.oracle != nil {
  217. client.oracle.Start(backend)
  218. }
  219. client.handler.start()
  220. return client.handler, func() {
  221. client.handler.stop()
  222. }
  223. }
  224. func newTestServerHandler(blocks int, indexers []*core.ChainIndexer, db ethdb.Database, clock mclock.Clock) (*serverHandler, *backends.SimulatedBackend, func()) {
  225. var (
  226. gspec = core.Genesis{
  227. Config: params.AllEthashProtocolChanges,
  228. Alloc: core.GenesisAlloc{bankAddr: {Balance: bankFunds}},
  229. GasLimit: 100000000,
  230. BaseFee: big.NewInt(params.InitialBaseFee),
  231. }
  232. oracle *checkpointoracle.CheckpointOracle
  233. )
  234. genesis := gspec.MustCommit(db)
  235. // create a simulation backend and pre-commit several customized block to the database.
  236. simulation := backends.NewSimulatedBackendWithDatabase(db, gspec.Alloc, 100000000)
  237. prepare(blocks, simulation)
  238. txpoolConfig := core.DefaultTxPoolConfig
  239. txpoolConfig.Journal = ""
  240. txpool := core.NewTxPool(txpoolConfig, gspec.Config, simulation.Blockchain())
  241. if indexers != nil {
  242. checkpointConfig := &params.CheckpointOracleConfig{
  243. Address: crypto.CreateAddress(bankAddr, 0),
  244. Signers: []common.Address{signerAddr},
  245. Threshold: 1,
  246. }
  247. getLocal := func(index uint64) params.TrustedCheckpoint {
  248. chtIndexer := indexers[0]
  249. sectionHead := chtIndexer.SectionHead(index)
  250. return params.TrustedCheckpoint{
  251. SectionIndex: index,
  252. SectionHead: sectionHead,
  253. CHTRoot: light.GetChtRoot(db, index, sectionHead),
  254. BloomRoot: light.GetBloomTrieRoot(db, index, sectionHead),
  255. }
  256. }
  257. oracle = checkpointoracle.New(checkpointConfig, getLocal)
  258. }
  259. server := &LesServer{
  260. lesCommons: lesCommons{
  261. genesis: genesis.Hash(),
  262. config: &ethconfig.Config{LightPeers: 100, NetworkId: NetworkId},
  263. chainConfig: params.AllEthashProtocolChanges,
  264. iConfig: light.TestServerIndexerConfig,
  265. chainDb: db,
  266. chainReader: simulation.Blockchain(),
  267. oracle: oracle,
  268. closeCh: make(chan struct{}),
  269. },
  270. peers: newClientPeerSet(),
  271. servingQueue: newServingQueue(int64(time.Millisecond*10), 1),
  272. defParams: flowcontrol.ServerParams{
  273. BufLimit: testBufLimit,
  274. MinRecharge: testBufRecharge,
  275. },
  276. fcManager: flowcontrol.NewClientManager(nil, clock),
  277. }
  278. server.costTracker, server.minCapacity = newCostTracker(db, server.config)
  279. server.costTracker.testCostList = testCostList(0) // Disable flow control mechanism.
  280. server.clientPool = vfs.NewClientPool(db, testBufRecharge, defaultConnectedBias, clock, alwaysTrueFn)
  281. server.clientPool.Start()
  282. server.clientPool.SetLimits(10000, 10000) // Assign enough capacity for clientpool
  283. server.handler = newServerHandler(server, simulation.Blockchain(), db, txpool, func() bool { return true })
  284. if server.oracle != nil {
  285. server.oracle.Start(simulation)
  286. }
  287. server.servingQueue.setThreads(4)
  288. server.handler.start()
  289. closer := func() { server.Stop() }
  290. return server.handler, simulation, closer
  291. }
  292. func alwaysTrueFn() bool {
  293. return true
  294. }
  295. // testPeer is a simulated peer to allow testing direct network calls.
  296. type testPeer struct {
  297. cpeer *clientPeer
  298. speer *serverPeer
  299. net p2p.MsgReadWriter // Network layer reader/writer to simulate remote messaging
  300. app *p2p.MsgPipeRW // Application layer reader/writer to simulate the local side
  301. }
  302. // handshakeWithServer executes the handshake with the remote server peer.
  303. func (p *testPeer) handshakeWithServer(t *testing.T, td *big.Int, head common.Hash, headNum uint64, genesis common.Hash, forkID forkid.ID) {
  304. // It only works for the simulated client peer
  305. if p.cpeer == nil {
  306. t.Fatal("handshake for client peer only")
  307. }
  308. var sendList keyValueList
  309. sendList = sendList.add("protocolVersion", uint64(p.cpeer.version))
  310. sendList = sendList.add("networkId", uint64(NetworkId))
  311. sendList = sendList.add("headTd", td)
  312. sendList = sendList.add("headHash", head)
  313. sendList = sendList.add("headNum", headNum)
  314. sendList = sendList.add("genesisHash", genesis)
  315. if p.cpeer.version >= lpv4 {
  316. sendList = sendList.add("forkID", &forkID)
  317. }
  318. if err := p2p.ExpectMsg(p.app, StatusMsg, nil); err != nil {
  319. t.Fatalf("status recv: %v", err)
  320. }
  321. if err := p2p.Send(p.app, StatusMsg, &sendList); err != nil {
  322. t.Fatalf("status send: %v", err)
  323. }
  324. }
  325. // handshakeWithClient executes the handshake with the remote client peer.
  326. func (p *testPeer) handshakeWithClient(t *testing.T, td *big.Int, head common.Hash, headNum uint64, genesis common.Hash, forkID forkid.ID, costList RequestCostList, recentTxLookup uint64) {
  327. // It only works for the simulated client peer
  328. if p.speer == nil {
  329. t.Fatal("handshake for server peer only")
  330. }
  331. var sendList keyValueList
  332. sendList = sendList.add("protocolVersion", uint64(p.speer.version))
  333. sendList = sendList.add("networkId", uint64(NetworkId))
  334. sendList = sendList.add("headTd", td)
  335. sendList = sendList.add("headHash", head)
  336. sendList = sendList.add("headNum", headNum)
  337. sendList = sendList.add("genesisHash", genesis)
  338. sendList = sendList.add("serveHeaders", nil)
  339. sendList = sendList.add("serveChainSince", uint64(0))
  340. sendList = sendList.add("serveStateSince", uint64(0))
  341. sendList = sendList.add("serveRecentState", uint64(core.TriesInMemory-4))
  342. sendList = sendList.add("txRelay", nil)
  343. sendList = sendList.add("flowControl/BL", testBufLimit)
  344. sendList = sendList.add("flowControl/MRR", testBufRecharge)
  345. sendList = sendList.add("flowControl/MRC", costList)
  346. if p.speer.version >= lpv4 {
  347. sendList = sendList.add("forkID", &forkID)
  348. sendList = sendList.add("recentTxLookup", recentTxLookup)
  349. }
  350. if err := p2p.ExpectMsg(p.app, StatusMsg, nil); err != nil {
  351. t.Fatalf("status recv: %v", err)
  352. }
  353. if err := p2p.Send(p.app, StatusMsg, &sendList); err != nil {
  354. t.Fatalf("status send: %v", err)
  355. }
  356. }
  357. // close terminates the local side of the peer, notifying the remote protocol
  358. // manager of termination.
  359. func (p *testPeer) close() {
  360. p.app.Close()
  361. }
  362. func newTestPeerPair(name string, version int, server *serverHandler, client *clientHandler, noInitAnnounce bool) (*testPeer, *testPeer, error) {
  363. // Create a message pipe to communicate through
  364. app, net := p2p.MsgPipe()
  365. // Generate a random id and create the peer
  366. var id enode.ID
  367. rand.Read(id[:])
  368. peer1 := newClientPeer(version, NetworkId, p2p.NewPeer(id, name, nil), net)
  369. peer2 := newServerPeer(version, NetworkId, false, p2p.NewPeer(id, name, nil), app)
  370. // Start the peer on a new thread
  371. errc1 := make(chan error, 1)
  372. errc2 := make(chan error, 1)
  373. go func() {
  374. select {
  375. case <-server.closeCh:
  376. errc1 <- p2p.DiscQuitting
  377. case errc1 <- server.handle(peer1):
  378. }
  379. }()
  380. go func() {
  381. select {
  382. case <-client.closeCh:
  383. errc2 <- p2p.DiscQuitting
  384. case errc2 <- client.handle(peer2, noInitAnnounce):
  385. }
  386. }()
  387. // Ensure the connection is established or exits when any error occurs
  388. for {
  389. select {
  390. case err := <-errc1:
  391. return nil, nil, fmt.Errorf("failed to establish protocol connection %v", err)
  392. case err := <-errc2:
  393. return nil, nil, fmt.Errorf("failed to establish protocol connection %v", err)
  394. default:
  395. }
  396. if atomic.LoadUint32(&peer1.serving) == 1 && atomic.LoadUint32(&peer2.serving) == 1 {
  397. break
  398. }
  399. time.Sleep(50 * time.Millisecond)
  400. }
  401. return &testPeer{cpeer: peer1, net: net, app: app}, &testPeer{speer: peer2, net: app, app: net}, nil
  402. }
  403. type indexerCallback func(*core.ChainIndexer, *core.ChainIndexer, *core.ChainIndexer)
  404. // testClient represents a client object for testing with necessary auxiliary fields.
  405. type testClient struct {
  406. clock mclock.Clock
  407. db ethdb.Database
  408. peer *testPeer
  409. handler *clientHandler
  410. chtIndexer *core.ChainIndexer
  411. bloomIndexer *core.ChainIndexer
  412. bloomTrieIndexer *core.ChainIndexer
  413. }
  414. // newRawPeer creates a new server peer connects to the server and do the handshake.
  415. func (client *testClient) newRawPeer(t *testing.T, name string, version int, recentTxLookup uint64) (*testPeer, func(), <-chan error) {
  416. // Create a message pipe to communicate through
  417. app, net := p2p.MsgPipe()
  418. // Generate a random id and create the peer
  419. var id enode.ID
  420. rand.Read(id[:])
  421. peer := newServerPeer(version, NetworkId, false, p2p.NewPeer(id, name, nil), net)
  422. // Start the peer on a new thread
  423. errCh := make(chan error, 1)
  424. go func() {
  425. select {
  426. case <-client.handler.closeCh:
  427. errCh <- p2p.DiscQuitting
  428. case errCh <- client.handler.handle(peer, false):
  429. }
  430. }()
  431. tp := &testPeer{
  432. app: app,
  433. net: net,
  434. speer: peer,
  435. }
  436. var (
  437. genesis = client.handler.backend.blockchain.Genesis()
  438. head = client.handler.backend.blockchain.CurrentHeader()
  439. td = client.handler.backend.blockchain.GetTd(head.Hash(), head.Number.Uint64())
  440. )
  441. forkID := forkid.NewID(client.handler.backend.blockchain.Config(), genesis.Hash(), head.Number.Uint64())
  442. tp.handshakeWithClient(t, td, head.Hash(), head.Number.Uint64(), genesis.Hash(), forkID, testCostList(0), recentTxLookup) // disable flow control by default
  443. // Ensure the connection is established or exits when any error occurs
  444. for {
  445. select {
  446. case <-errCh:
  447. return nil, nil, nil
  448. default:
  449. }
  450. if atomic.LoadUint32(&peer.serving) == 1 {
  451. break
  452. }
  453. time.Sleep(50 * time.Millisecond)
  454. }
  455. closePeer := func() {
  456. tp.speer.close()
  457. tp.close()
  458. }
  459. return tp, closePeer, errCh
  460. }
  461. // testServer represents a server object for testing with necessary auxiliary fields.
  462. type testServer struct {
  463. clock mclock.Clock
  464. backend *backends.SimulatedBackend
  465. db ethdb.Database
  466. peer *testPeer
  467. handler *serverHandler
  468. chtIndexer *core.ChainIndexer
  469. bloomIndexer *core.ChainIndexer
  470. bloomTrieIndexer *core.ChainIndexer
  471. }
  472. // newRawPeer creates a new client peer connects to the server and do the handshake.
  473. func (server *testServer) newRawPeer(t *testing.T, name string, version int) (*testPeer, func(), <-chan error) {
  474. // Create a message pipe to communicate through
  475. app, net := p2p.MsgPipe()
  476. // Generate a random id and create the peer
  477. var id enode.ID
  478. rand.Read(id[:])
  479. peer := newClientPeer(version, NetworkId, p2p.NewPeer(id, name, nil), net)
  480. // Start the peer on a new thread
  481. errCh := make(chan error, 1)
  482. go func() {
  483. select {
  484. case <-server.handler.closeCh:
  485. errCh <- p2p.DiscQuitting
  486. case errCh <- server.handler.handle(peer):
  487. }
  488. }()
  489. tp := &testPeer{
  490. app: app,
  491. net: net,
  492. cpeer: peer,
  493. }
  494. var (
  495. genesis = server.handler.blockchain.Genesis()
  496. head = server.handler.blockchain.CurrentHeader()
  497. td = server.handler.blockchain.GetTd(head.Hash(), head.Number.Uint64())
  498. )
  499. forkID := forkid.NewID(server.handler.blockchain.Config(), genesis.Hash(), head.Number.Uint64())
  500. tp.handshakeWithServer(t, td, head.Hash(), head.Number.Uint64(), genesis.Hash(), forkID)
  501. // Ensure the connection is established or exits when any error occurs
  502. for {
  503. select {
  504. case <-errCh:
  505. return nil, nil, nil
  506. default:
  507. }
  508. if atomic.LoadUint32(&peer.serving) == 1 {
  509. break
  510. }
  511. time.Sleep(50 * time.Millisecond)
  512. }
  513. closePeer := func() {
  514. tp.cpeer.close()
  515. tp.close()
  516. }
  517. return tp, closePeer, errCh
  518. }
  519. // testnetConfig wraps all the configurations for testing network.
  520. type testnetConfig struct {
  521. blocks int
  522. protocol int
  523. indexFn indexerCallback
  524. ulcServers []string
  525. ulcFraction int
  526. simClock bool
  527. connect bool
  528. nopruning bool
  529. }
  530. func newClientServerEnv(t *testing.T, config testnetConfig) (*testServer, *testClient, func()) {
  531. var (
  532. sdb = rawdb.NewMemoryDatabase()
  533. cdb = rawdb.NewMemoryDatabase()
  534. speers = newServerPeerSet()
  535. )
  536. var clock mclock.Clock = &mclock.System{}
  537. if config.simClock {
  538. clock = &mclock.Simulated{}
  539. }
  540. dist := newRequestDistributor(speers, clock)
  541. rm := newRetrieveManager(speers, dist, func() time.Duration { return time.Millisecond * 500 })
  542. odr := NewLesOdr(cdb, light.TestClientIndexerConfig, speers, rm)
  543. sindexers := testIndexers(sdb, nil, light.TestServerIndexerConfig, true)
  544. cIndexers := testIndexers(cdb, odr, light.TestClientIndexerConfig, config.nopruning)
  545. scIndexer, sbIndexer, sbtIndexer := sindexers[0], sindexers[1], sindexers[2]
  546. ccIndexer, cbIndexer, cbtIndexer := cIndexers[0], cIndexers[1], cIndexers[2]
  547. odr.SetIndexers(ccIndexer, cbIndexer, cbtIndexer)
  548. server, b, serverClose := newTestServerHandler(config.blocks, sindexers, sdb, clock)
  549. client, clientClose := newTestClientHandler(b, odr, cIndexers, cdb, speers, config.ulcServers, config.ulcFraction)
  550. scIndexer.Start(server.blockchain)
  551. sbIndexer.Start(server.blockchain)
  552. ccIndexer.Start(client.backend.blockchain)
  553. cbIndexer.Start(client.backend.blockchain)
  554. if config.indexFn != nil {
  555. config.indexFn(scIndexer, sbIndexer, sbtIndexer)
  556. }
  557. var (
  558. err error
  559. speer, cpeer *testPeer
  560. )
  561. if config.connect {
  562. done := make(chan struct{})
  563. client.syncEnd = func(_ *types.Header) { close(done) }
  564. cpeer, speer, err = newTestPeerPair("peer", config.protocol, server, client, false)
  565. if err != nil {
  566. t.Fatalf("Failed to connect testing peers %v", err)
  567. }
  568. select {
  569. case <-done:
  570. case <-time.After(10 * time.Second):
  571. t.Fatal("test peer did not connect and sync within 3s")
  572. }
  573. }
  574. s := &testServer{
  575. clock: clock,
  576. backend: b,
  577. db: sdb,
  578. peer: cpeer,
  579. handler: server,
  580. chtIndexer: scIndexer,
  581. bloomIndexer: sbIndexer,
  582. bloomTrieIndexer: sbtIndexer,
  583. }
  584. c := &testClient{
  585. clock: clock,
  586. db: cdb,
  587. peer: speer,
  588. handler: client,
  589. chtIndexer: ccIndexer,
  590. bloomIndexer: cbIndexer,
  591. bloomTrieIndexer: cbtIndexer,
  592. }
  593. teardown := func() {
  594. if config.connect {
  595. speer.close()
  596. cpeer.close()
  597. cpeer.cpeer.close()
  598. speer.speer.close()
  599. }
  600. ccIndexer.Close()
  601. cbIndexer.Close()
  602. scIndexer.Close()
  603. sbIndexer.Close()
  604. dist.close()
  605. serverClose()
  606. b.Close()
  607. clientClose()
  608. }
  609. return s, c, teardown
  610. }
  611. // NewFuzzerPeer creates a client peer for test purposes, and also returns
  612. // a function to close the peer: this is needed to avoid goroutine leaks in the
  613. // exec queue.
  614. func NewFuzzerPeer(version int) (p *clientPeer, closer func()) {
  615. p = newClientPeer(version, 0, p2p.NewPeer(enode.ID{}, "", nil), nil)
  616. return p, func() { p.peerCommons.close() }
  617. }