Add tests that NodeCallbacks are executed (#1582)

* Add tests that NodeCallbacks are executed

* Respond to review

* Rebase fixes

* Formatting changes

* Formatting

core/src/main/scala/org/bitcoins/core/protocol/CompactSizeUInt.scala

@@ -46,6 +46,8 @@ object CompactSizeUInt extends Factory[CompactSizeUInt] {
 
   val zero: CompactSizeUInt = CompactSizeUInt(UInt64.zero)
 
+  lazy val one: CompactSizeUInt = CompactSizeUInt(UInt64.one)
+
   override def fromBytes(bytes: ByteVector): CompactSizeUInt = {
     parseCompactSizeUInt(bytes)
   }

node-test/src/test/scala/org/bitcoins/node/BroadcastTransactionTest.scala

@@ -32,9 +32,6 @@ class BroadcastTransactionTest extends NodeUnitTest {
 
   private val sendAmount = 1.bitcoin
 
-  private val junkAddress: BitcoinAddress =
-    BitcoinAddress("2NFyxovf6MyxfHqtVjstGzs6HeLqv92Nq4U")
-
   it must "broadcast a transaction" in { param =>
     val SpvNodeFundedWalletBitcoind(node, wallet, rpc, _) = param
 

node-test/src/test/scala/org/bitcoins/node/UpdateBloomFilterTest.scala

@@ -1,11 +1,9 @@
 package org.bitcoins.node
 
 import org.bitcoins.core.currency._
-import org.bitcoins.core.protocol.BitcoinAddress
 import org.bitcoins.server.BitcoinSAppConfig
 import org.bitcoins.testkit.BitcoinSTestAppConfig
-import org.bitcoins.testkit.node.NodeUnitTest
+import org.bitcoins.testkit.node.{NodeUnitTest, SpvNodeFundedWalletBitcoind}
-import org.bitcoins.testkit.node.SpvNodeFundedWalletBitcoind
 import org.scalatest.{BeforeAndAfter, FutureOutcome}
 
 class UpdateBloomFilterTest extends NodeUnitTest with BeforeAndAfter {
@@ -20,9 +18,6 @@ class UpdateBloomFilterTest extends NodeUnitTest with BeforeAndAfter {
     withSpvNodeFundedWalletBitcoind(test, NodeCallbacks.empty, None)
   }
 
-  private val junkAddress: BitcoinAddress =
-    BitcoinAddress("2NFyxovf6MyxfHqtVjstGzs6HeLqv92Nq4U")
-
   it must "update the bloom filter with a TX" in { param =>
     val SpvNodeFundedWalletBitcoind(spv, wallet, rpc, _) = param
 

node-test/src/test/scala/org/bitcoins/node/networking/peer/DataMessageHandlerTest.scala

@@ -0,0 +1,153 @@
+package org.bitcoins.node.networking.peer
+
+import org.bitcoins.core.currency._
+import org.bitcoins.core.gcs.{FilterType, GolombFilter}
+import org.bitcoins.core.p2p._
+import org.bitcoins.core.protocol.CompactSizeUInt
+import org.bitcoins.core.protocol.blockchain.{Block, BlockHeader, MerkleBlock}
+import org.bitcoins.core.protocol.transaction.Transaction
+import org.bitcoins.crypto.DoubleSha256Digest
+import org.bitcoins.node._
+import org.bitcoins.server.BitcoinSAppConfig
+import org.bitcoins.testkit.BitcoinSTestAppConfig
+import org.bitcoins.testkit.node.NodeUnitTest
+import org.bitcoins.testkit.node.fixture.SpvNodeConnectedWithBitcoindV19
+import org.scalatest.FutureOutcome
+
+import scala.concurrent.{Future, Promise}
+
+class DataMessageHandlerTest extends NodeUnitTest {
+
+  /** Wallet config with data directory set to user temp directory */
+  implicit override protected def config: BitcoinSAppConfig =
+    BitcoinSTestAppConfig.getSpvWithEmbeddedDbTestConfig(pgUrl)
+
+  override type FixtureParam = SpvNodeConnectedWithBitcoindV19
+
+  override def withFixture(test: OneArgAsyncTest): FutureOutcome =
+    withSpvNodeConnectedToBitcoindV19(test)
+
+  it must "verify OnMerkleBlock callbacks are executed" in {
+    param: FixtureParam =>
+      val SpvNodeConnectedWithBitcoindV19(spv, bitcoind) = param
+
+      val resultP: Promise[(MerkleBlock, Vector[Transaction])] = Promise()
+
+      for {
+        sender <- spv.peerMsgSenderF
+
+        txId <- bitcoind.sendToAddress(junkAddress, 1.bitcoin)
+        tx <- bitcoind.getRawTransactionRaw(txId)
+        _ <- bitcoind.generateToAddress(blocks = 1, junkAddress)
+        merkleBlock <- bitcoind.getTxOutProof(Vector(txId))
+
+        payload1 = MerkleBlockMessage(merkleBlock)
+        payload2 = TransactionMessage(tx)
+
+        callback: OnMerkleBlockReceived =
+          (merkle: MerkleBlock, txs: Vector[Transaction]) => {
+            Future {
+              resultP.success((merkle, txs))
+              ()
+            }
+          }
+
+        callbacks = NodeCallbacks.onMerkleBlockReceived(callback)
+
+        dataMessageHandler = DataMessageHandler(genesisChainApi, callbacks)
+        _ <- dataMessageHandler.handleDataPayload(payload1, sender)
+        _ <- dataMessageHandler.handleDataPayload(payload2, sender)
+        result <- resultP.future
+      } yield assert(result == (merkleBlock, Vector(tx)))
+  }
+
+  it must "verify OnBlockReceived callbacks are executed" in {
+    param: FixtureParam =>
+      val SpvNodeConnectedWithBitcoindV19(spv, bitcoind) = param
+
+      val resultP: Promise[Block] = Promise()
+
+      for {
+        sender <- spv.peerMsgSenderF
+
+        hash <- bitcoind.generateToAddress(blocks = 1, junkAddress).map(_.head)
+        block <- bitcoind.getBlockRaw(hash)
+
+        payload = BlockMessage(block)
+
+        callback: OnBlockReceived = (block: Block) => {
+          Future {
+            resultP.success(block)
+            ()
+          }
+        }
+
+        callbacks = NodeCallbacks.onBlockReceived(callback)
+
+        dataMessageHandler = DataMessageHandler(genesisChainApi, callbacks)
+        _ <- dataMessageHandler.handleDataPayload(payload, sender)
+        result <- resultP.future
+      } yield assert(result == block)
+  }
+
+  it must "verify OnBlockHeadersReceived callbacks are executed" in {
+    param: FixtureParam =>
+      val SpvNodeConnectedWithBitcoindV19(spv, bitcoind) = param
+
+      val resultP: Promise[Vector[BlockHeader]] = Promise()
+
+      for {
+        sender <- spv.peerMsgSenderF
+
+        hash <- bitcoind.generateToAddress(blocks = 1, junkAddress).map(_.head)
+        header <- bitcoind.getBlockHeaderRaw(hash)
+
+        payload = HeadersMessage(CompactSizeUInt.one, Vector(header))
+
+        callback: OnBlockHeadersReceived = (headers: Vector[BlockHeader]) => {
+          Future {
+            resultP.success(headers)
+            ()
+          }
+        }
+
+        callbacks = NodeCallbacks.onBlockHeadersReceived(callback)
+
+        dataMessageHandler = DataMessageHandler(genesisChainApi, callbacks)
+        _ <- dataMessageHandler.handleDataPayload(payload, sender)
+        result <- resultP.future
+      } yield assert(result == Vector(header))
+  }
+
+  it must "verify OnCompactFilterReceived callbacks are executed" in {
+    param: FixtureParam =>
+      val SpvNodeConnectedWithBitcoindV19(spv, bitcoind) = param
+
+      val resultP: Promise[Vector[(DoubleSha256Digest, GolombFilter)]] =
+        Promise()
+
+      for {
+        sender <- spv.peerMsgSenderF
+
+        hash <- bitcoind.generateToAddress(blocks = 1, junkAddress).map(_.head)
+        filter <- bitcoind.getBlockFilter(hash, FilterType.Basic)
+
+        payload =
+          CompactFilterMessage(FilterType.Basic, hash.flip, filter.filter.bytes)
+
+        callback: OnCompactFiltersReceived =
+          (filters: Vector[(DoubleSha256Digest, GolombFilter)]) => {
+            Future {
+              resultP.success(filters)
+              ()
+            }
+          }
+
+        callbacks = NodeCallbacks.onCompactFilterReceived(callback)
+
+        dataMessageHandler = DataMessageHandler(genesisChainApi, callbacks)
+        _ <- dataMessageHandler.handleDataPayload(payload, sender)
+        result <- resultP.future
+      } yield assert(result == Vector((hash.flip, filter.filter)))
+  }
+}

testkit/src/main/scala/org/bitcoins/testkit/node/NodeUnitTest.scala

@@ -5,7 +5,18 @@ import java.net.InetSocketAddress
 import akka.actor.ActorSystem
 import org.bitcoins.chain.api.ChainApi
 import org.bitcoins.chain.config.ChainAppConfig
+import org.bitcoins.chain.models.{
+  BlockHeaderDb,
+  CompactFilterDb,
+  CompactFilterHeaderDb
+}
+import org.bitcoins.core.api.ChainQueryApi
 import org.bitcoins.core.config.NetworkParameters
+import org.bitcoins.core.gcs.FilterHeader
+import org.bitcoins.core.p2p.CompactFilterMessage
+import org.bitcoins.core.protocol.blockchain.BlockHeader
+import org.bitcoins.core.protocol.{BitcoinAddress, BlockStamp}
+import org.bitcoins.crypto.{DoubleSha256Digest, DoubleSha256DigestBE}
 import org.bitcoins.db.AppConfig
 import org.bitcoins.node._
 import org.bitcoins.node.config.NodeAppConfig
@@ -16,8 +27,9 @@ import org.bitcoins.node.networking.peer.{
   PeerMessageReceiverState,
   PeerMessageSender
 }
-import org.bitcoins.rpc.client.common.BitcoindVersion.V18
+import org.bitcoins.rpc.client.common.BitcoindVersion.{V18, V19}
 import org.bitcoins.rpc.client.common.{BitcoindRpcClient, BitcoindVersion}
+import org.bitcoins.rpc.client.v19.BitcoindV19RpcClient
 import org.bitcoins.server.BitcoinSAppConfig
 import org.bitcoins.server.BitcoinSAppConfig._
 import org.bitcoins.testkit.EmbeddedPg
@@ -27,7 +39,8 @@ import org.bitcoins.testkit.keymanager.KeyManagerTestUtil
 import org.bitcoins.testkit.node.fixture.{
   NeutrinoNodeConnectedWithBitcoind,
   NodeConnectedWithBitcoind,
-  SpvNodeConnectedWithBitcoind
+  SpvNodeConnectedWithBitcoind,
+  SpvNodeConnectedWithBitcoindV19
 }
 import org.bitcoins.testkit.rpc.BitcoindRpcTestUtil
 import org.bitcoins.testkit.wallet.{BitcoinSWalletTest, WalletWithBitcoindRpc}
@@ -60,6 +73,102 @@ trait NodeUnitTest extends BitcoinSFixture with EmbeddedPg {
 
   lazy val bitcoindPeerF = startedBitcoindF.map(NodeTestUtil.getBitcoindPeer)
 
+  lazy val junkAddress: BitcoinAddress =
+    BitcoinAddress("2NFyxovf6MyxfHqtVjstGzs6HeLqv92Nq4U")
+
+  val genesisChainApi: ChainApi = new ChainApi {
+
+    override def processHeaders(
+        headers: Vector[BlockHeader]): Future[ChainApi] =
+      Future.successful(this)
+
+    override def getHeader(
+        hash: DoubleSha256DigestBE): Future[Option[BlockHeaderDb]] =
+      Future.successful(None)
+
+    override def getHeadersAtHeight(
+        height: Int): Future[Vector[BlockHeaderDb]] =
+      Future.successful(Vector.empty)
+
+    override def getBlockCount(): Future[Int] = Future.successful(0)
+
+    override def getBestBlockHeader(): Future[BlockHeaderDb] =
+      Future.successful(ChainUnitTest.genesisHeaderDb)
+
+    override def processFilterHeaders(
+        filterHeaders: Vector[FilterHeader],
+        stopHash: DoubleSha256DigestBE): Future[ChainApi] =
+      Future.successful(this)
+
+    override def nextHeaderBatchRange(
+        stopHash: DoubleSha256DigestBE,
+        batchSize: Int): Future[Option[(Int, DoubleSha256Digest)]] =
+      Future.successful(None)
+
+    override def nextFilterHeaderBatchRange(
+        stopHash: DoubleSha256DigestBE,
+        batchSize: Int): Future[Option[(Int, DoubleSha256Digest)]] =
+      Future.successful(None)
+
+    override def processFilters(
+        message: Vector[CompactFilterMessage]): Future[ChainApi] =
+      Future.successful(this)
+
+    override def processCheckpoints(
+        checkpoints: Vector[DoubleSha256DigestBE],
+        blockHash: DoubleSha256DigestBE): Future[ChainApi] =
+      Future.successful(this)
+
+    override def getFilterHeaderCount(): Future[Int] = Future.successful(0)
+
+    override def getFilterHeadersAtHeight(
+        height: Int): Future[Vector[CompactFilterHeaderDb]] =
+      Future.successful(Vector.empty)
+
+    override def getBestFilterHeader(): Future[Option[CompactFilterHeaderDb]] =
+      Future.successful(None)
+
+    override def getFilterHeader(blockHash: DoubleSha256DigestBE): Future[
+      Option[CompactFilterHeaderDb]] = Future.successful(None)
+
+    override def getFilter(
+        hash: DoubleSha256DigestBE): Future[Option[CompactFilterDb]] =
+      Future.successful(None)
+
+    override def getFilterCount(): Future[Int] = Future.successful(0)
+
+    override def getFiltersAtHeight(
+        height: Int): Future[Vector[CompactFilterDb]] =
+      Future.successful(Vector.empty)
+
+    override def getHeightByBlockStamp(blockStamp: BlockStamp): Future[Int] =
+      Future.successful(0)
+
+    override def getHeadersBetween(
+        from: BlockHeaderDb,
+        to: BlockHeaderDb): Future[Vector[BlockHeaderDb]] =
+      Future.successful(Vector.empty)
+
+    override def getBlockHeight(
+        blockHash: DoubleSha256DigestBE): Future[Option[Int]] =
+      Future.successful(None)
+
+    override def getBestBlockHash(): Future[DoubleSha256DigestBE] =
+      Future.successful(DoubleSha256DigestBE.empty)
+
+    override def getNumberOfConfirmations(
+        blockHashOpt: DoubleSha256DigestBE): Future[Option[Int]] =
+      Future.successful(None)
+
+    override def getFiltersBetweenHeights(
+        startHeight: Int,
+        endHeight: Int): Future[Vector[ChainQueryApi.FilterResponse]] =
+      Future.successful(Vector.empty)
+
+    override def epochSecondToBlockHeight(time: Long): Future[Int] =
+      Future.successful(0)
+  }
+
   def withSpvNodeConnectedToBitcoind(
       test: OneArgAsyncTest,
       versionOpt: Option[BitcoindVersion] = None)(implicit
@@ -84,6 +193,31 @@ trait NodeUnitTest extends BitcoinSFixture with EmbeddedPg {
     )(test)
   }
 
+  def withSpvNodeConnectedToBitcoindV19(test: OneArgAsyncTest)(implicit
+      system: ActorSystem,
+      appConfig: BitcoinSAppConfig): FutureOutcome = {
+    val nodeWithBitcoindBuilder: () => Future[
+      SpvNodeConnectedWithBitcoindV19] = { () =>
+      require(appConfig.isSPVEnabled && !appConfig.isNeutrinoEnabled)
+      for {
+        bitcoind <-
+          BitcoinSFixture
+            .createBitcoindWithFunds(Some(V19))
+            .map(_.asInstanceOf[BitcoindV19RpcClient])
+        node <- NodeUnitTest.createSpvNode(bitcoind, NodeCallbacks.empty)(
+          system,
+          appConfig.chainConf,
+          appConfig.nodeConf)
+      } yield SpvNodeConnectedWithBitcoindV19(node, bitcoind)
+    }
+
+    makeDependentFixture(
+      build = nodeWithBitcoindBuilder,
+      destroy = NodeUnitTest.destroyNodeConnectedWithBitcoind(
+        _: NodeConnectedWithBitcoind)(system, appConfig)
+    )(test)
+  }
+
   def withNeutrinoNodeConnectedToBitcoind(
       test: OneArgAsyncTest,
       versionOpt: Option[BitcoindVersion] = None)(implicit

testkit/src/main/scala/org/bitcoins/testkit/node/fixture/NodeConnectedWithBitcoind.scala

@@ -2,6 +2,7 @@ package org.bitcoins.testkit.node.fixture
 
 import org.bitcoins.node.{NeutrinoNode, Node, SpvNode}
 import org.bitcoins.rpc.client.common.BitcoindRpcClient
+import org.bitcoins.rpc.client.v19.BitcoindV19RpcClient
 
 /** Gives us a fixture that has a SPV node connected with the bitcoind instance */
 trait NodeConnectedWithBitcoind {
@@ -14,6 +15,11 @@ case class SpvNodeConnectedWithBitcoind(
     bitcoind: BitcoindRpcClient)
     extends NodeConnectedWithBitcoind
 
+case class SpvNodeConnectedWithBitcoindV19(
+    node: SpvNode,
+    bitcoind: BitcoindV19RpcClient)
+    extends NodeConnectedWithBitcoind
+
 case class NeutrinoNodeConnectedWithBitcoind(
     node: NeutrinoNode,
     bitcoind: BitcoindRpcClient)