tests(rpc): Add grpc test for `GetTaddressBalanceStream` and `GetAddressUtxosStream` (#4407)

* add test for `GetTaddressBalanceStream`

* add test for `GetAddressUtxosStream`

Cargo.lock

@@ -6539,6 +6539,7 @@ dependencies = [
  "tempfile",
  "thiserror",
  "tokio",
+ "tokio-stream",
  "toml",
  "tonic",
  "tonic-build",

zebrad/Cargo.toml

@@ -93,6 +93,7 @@ semver = "1.0.9"
 serde_json = { version = "1.0.81", features = ["preserve_order"] }
 tempfile = "3.3.0"
 tokio = { version = "1.18.2", features = ["full", "test-util"] }
+tokio-stream = "0.1.8"
 
 # test feature lightwalletd-grpc-tests
 prost = "0.10.3"

zebrad/tests/common/lightwalletd/wallet_grpc_test.rs

@@ -15,7 +15,7 @@
 //!
 //! - `GetTaddressTxids`: Covered.
 //! - `GetTaddressBalance`: Covered.
-//! - `GetTaddressBalanceStream`: Not covered.
+//! - `GetTaddressBalanceStream`: Covered.
 //!
 //! - `GetMempoolTx`: Not covered.
 //! - `GetMempoolStream`: Not covered.
@@ -23,7 +23,7 @@
 //! - `GetTreeState`: Not covered, Need #3990
 //!
 //! - `GetAddressUtxos` -= Covered.
-//! - `GetAddressUtxosStream`: Not covered.
+//! - `GetAddressUtxosStream`: Covered.
 //!
 //! - `GetLightdInfo`: Covered.
 //! - `Ping`: Not covered and it will never will, ping is only used for testing purposes.
@@ -43,9 +43,9 @@ use crate::common::{
     launch::spawn_zebrad_for_rpc_without_initial_peers,
     lightwalletd::{
         wallet_grpc::{
-            connect_to_lightwalletd, spawn_lightwalletd_with_rpc_server, AddressList, BlockId,
+            connect_to_lightwalletd, spawn_lightwalletd_with_rpc_server, Address, AddressList,
-            BlockRange, ChainSpec, Empty, GetAddressUtxosArg, TransparentAddressBlockFilter,
+            BlockId, BlockRange, ChainSpec, Empty, GetAddressUtxosArg,
-            TxFilter,
+            TransparentAddressBlockFilter, TxFilter,
         },
         zebra_skip_lightwalletd_tests,
         LightwalletdTestType::UpdateCachedState,
@@ -211,7 +211,48 @@ pub async fn run() -> Result<()> {
     // because new coins are created in each block
     assert!(balance.value_zat > 0);
 
-    // TODO: Create call and check for `GetTaddressBalanceStream`
+    // Call `GetTaddressBalanceStream` with the ZF funding stream address as a stream argument
+    let zf_stream_address = Address {
+        address: "t3dvVE3SQEi7kqNzwrfNePxZ1d4hUyztBA1".to_string(),
+    };
+
+    let balance_zf = rpc_client
+        .get_taddress_balance_stream(tokio_stream::iter(vec![zf_stream_address.clone()]))
+        .await?
+        .into_inner();
+
+    // With ZFND funding stream address, the balance will always be greater than zero,
+    // because new coins are created in each block
+    assert!(balance_zf.value_zat > 0);
+
+    // Call `GetTaddressBalanceStream` with the MG funding stream address as a stream argument
+    let mg_stream_address = Address {
+        address: "t3XyYW8yBFRuMnfvm5KLGFbEVz25kckZXym".to_string(),
+    };
+
+    let balance_mg = rpc_client
+        .get_taddress_balance_stream(tokio_stream::iter(vec![mg_stream_address.clone()]))
+        .await?
+        .into_inner();
+
+    // With Major Grants funding stream address, the balance will always be greater than zero,
+    // because new coins are created in each block
+    assert!(balance_mg.value_zat > 0);
+
+    // Call `GetTaddressBalanceStream` with both, the ZFND and the MG funding stream addresses as a stream argument
+    let balance_both = rpc_client
+        .get_taddress_balance_stream(tokio_stream::iter(vec![
+            zf_stream_address,
+            mg_stream_address,
+        ]))
+        .await?
+        .into_inner();
+
+    // The result is the sum of the values in both addresses
+    assert_eq!(
+        balance_both.value_zat,
+        balance_zf.value_zat + balance_mg.value_zat
+    );
 
     // TODO: Create call and checks for `GetMempoolTx` and `GetMempoolTxStream`?
 
@@ -240,7 +281,23 @@ pub async fn run() -> Result<()> {
     // As we requested one entry we should get a response of length 1
     assert_eq!(utxos.address_utxos.len(), 1);
 
-    // TODO: Create call and check for `GetAddressUtxosStream`
+    // Call `GetAddressUtxosStream` with the ZF funding stream address that will always have utxos
+    let mut utxos_zf = rpc_client
+        .get_address_utxos_stream(GetAddressUtxosArg {
+            addresses: vec!["t3dvVE3SQEi7kqNzwrfNePxZ1d4hUyztBA1".to_string()],
+            start_height: 1,
+            max_entries: 2,
+        })
+        .await?
+        .into_inner();
+
+    let mut counter = 0;
+    while let Some(_utxos) = utxos_zf.message().await? {
+        counter += 1;
+    }
+    // As we are in a "in sync" chain we know there are more than 2 utxos for this address
+    // but we will receive the max of 2 from the stream response because we used a limit of 2 `max_entries`.
+    assert_eq!(2, counter);
 
     // Call `GetLightdInfo`
     let lightd_info = rpc_client.get_lightd_info(Empty {}).await?.into_inner();