fix: fixing integration tests (#1165)

Description
---
Fixing integration tests that are failing.

Motivation and Context
---
Some of the integration tests are failing and uses VN cli instead of the
most up-to-date wallet daemon.

How Has This Been Tested?
---
Run all integration tests via `cargo test --all-features --release
--package integration_tests`

What process can a PR reviewer use to test or verify this change?
---


Breaking Changes
---

- [x] None
- [ ] Requires data directory to be deleted
- [ ] Other - Please specify

dan_layer/state_store_sqlite/src/store.rs

@@ -71,8 +71,12 @@ impl<TAddr> fmt::Debug for SqliteStateStore<TAddr> {
 
 impl<TAddr: NodeAddressable + Serialize + DeserializeOwned> StateStore for SqliteStateStore<TAddr> {
     type Addr = TAddr;
-    type ReadTransaction<'a> = SqliteStateStoreReadTransaction<'a, Self::Addr> where TAddr: 'a;
-    type WriteTransaction<'a> = SqliteStateStoreWriteTransaction<'a, Self::Addr> where TAddr: 'a;
+    type ReadTransaction<'a>
+        = SqliteStateStoreReadTransaction<'a, Self::Addr>
+    where TAddr: 'a;
+    type WriteTransaction<'a>
+        = SqliteStateStoreWriteTransaction<'a, Self::Addr>
+    where TAddr: 'a;
 
     fn create_read_tx(&self) -> Result<Self::ReadTransaction<'_>, StorageError> {
         let tx = SqliteTransaction::begin(self.connection.lock().unwrap())?;

integration_tests/Cargo.toml

@@ -61,6 +61,7 @@ serde_json = { workspace = true }
 time = { workspace = true }
 tokio = { workspace = true, features = ["default", "macros", "time", "sync", "rt-multi-thread", "signal"] }
 tonic = { workspace = true }
+regex = "1.11.0"
 
 [[test]]
 name = "cucumber" # this should be the same as the filename of your test target

integration_tests/src/wallet_daemon_cli.rs

@@ -22,7 +22,7 @@
 
 use std::{collections::HashMap, str::FromStr, time::Duration};
 
-use anyhow::bail;
+use anyhow::{anyhow, bail};
 use base64::{engine::general_purpose::STANDARD as BASE64, Engine};
 use serde_json::json;
 use tari_crypto::{
@@ -35,7 +35,7 @@ use tari_dan_wallet_sdk::{
     apis::confidential_transfer::ConfidentialTransferInputSelection,
     models::{Account, NonFungibleToken},
 };
-use tari_engine_types::instruction::Instruction;
+use tari_engine_types::{instruction::Instruction, substate::SubstateId};
 use tari_template_lib::{
     args,
     constants::CONFIDENTIAL_TARI_RESOURCE_ADDRESS,
@@ -70,7 +70,7 @@ use tari_wallet_daemon_client::{
     ComponentAddressOrName,
     WalletDaemonClient,
 };
-use tokio::time::timeout;
+use tokio::{task::JoinSet, time::timeout};
 
 use crate::{helpers::get_address_from_output, validator_node_cli::add_substate_ids, TariWorld};
 
@@ -692,18 +692,41 @@ pub async fn create_component(
     );
 }
 
+pub fn find_output_version(
+    world: &mut TariWorld,
+    output_ref: &str,
+    output_component_substate_id: SubstateId,
+) -> anyhow::Result<Option<u32>> {
+    let outputs_name = output_ref.split('/').next().ok_or(anyhow!("Output must have a name"))?;
+    Ok(world
+        .outputs
+        .entry(outputs_name.to_string())
+        .or_default()
+        .iter()
+        .filter(|(_, requirement)| requirement.substate_id == output_component_substate_id)
+        .map(|(_, requirement)| requirement.version)
+        .last()
+        .unwrap_or_default())
+}
+
 pub async fn call_component(
     world: &mut TariWorld,
     account_name: String,
     output_ref: String,
     wallet_daemon_name: String,
     function_call: String,
+    new_outputs_name: Option<String>,
 ) -> anyhow::Result<TransactionWaitResultResponse> {
     let mut client = get_auth_wallet_daemon_client(world, &wallet_daemon_name).await;
 
     let source_component_address = get_address_from_output(world, output_ref.clone())
         .as_component_address()
         .expect("Failed to get component address from output");
+    let source_component_name = output_ref
+        .split('/')
+        .next()
+        .ok_or(anyhow!("Output must have a name"))?
+        .to_string();
 
     let account = get_account_from_name(&mut client, account_name).await;
     let account_component_address = account
@@ -714,24 +737,38 @@ pub async fn call_component(
     let tx = Transaction::builder()
         .fee_transaction_pay_from_component(account_component_address, Amount(1000))
         .call_method(source_component_address, &function_call, vec![])
+        .with_inputs(vec![
+            SubstateRequirement::new(
+                account_component_address.into(),
+                find_output_version(world, output_ref.as_str(), account_component_address.into())?,
+            ),
+            SubstateRequirement::new(
+                source_component_address.into(),
+                find_output_version(world, output_ref.as_str(), source_component_address.into())?,
+            ),
+        ])
         .build_unsigned_transaction();
 
-    let resp = submit_unsigned_tx_and_wait_for_response(client, tx, account).await;
+    let resp = submit_unsigned_tx_and_wait_for_response(client, tx, account).await?;
+
+    let final_outputs_name = if let Some(name) = new_outputs_name {
+        name
+    } else {
+        source_component_name
+    };
 
     add_substate_ids(
         world,
-        output_ref,
+        final_outputs_name,
         &resp
-            .as_ref()
-            .unwrap()
             .clone()
             .result
             .expect("Call component transaction has timed out")
             .result
             .expect("Call component transaction has failed"),
     );
 
-    resp
+    Ok(resp)
 }
 
 pub async fn concurrent_call_component(
@@ -754,34 +791,38 @@ pub async fn concurrent_call_component(
         .as_component_address()
         .expect("Failed to get account component address");
 
-    let mut handles = Vec::new();
+    let mut join_set = JoinSet::new();
     for _ in 0..times {
         let acc = account.clone();
         let clt = client.clone();
         let tx = Transaction::builder()
-            .fee_transaction_pay_from_component(account_component_address, Amount(2000))
+            .fee_transaction_pay_from_component(account_component_address, Amount(1000))
             .call_method(source_component_address, &function_call, vec![])
             .build_unsigned_transaction();
-        let handle = tokio::spawn(submit_unsigned_tx_and_wait_for_response(clt, tx, acc));
-        handles.push(handle);
+        join_set.spawn(submit_unsigned_tx_and_wait_for_response(clt, tx, acc));
     }
 
-    let mut last_resp = None;
-    for handle in handles {
-        let result = handle.await.map_err(|e| e.to_string());
-        if result.is_err() {
-            bail!("{}", result.as_ref().unwrap_err());
-        }
+    while let Some(result) = join_set.join_next().await {
+        let result = result.map_err(|e| e.to_string());
         match result {
-            Ok(response) => last_resp = Some(response),
+            Ok(response) => match response {
+                Ok(resp) => {
+                    add_substate_ids(
+                        world,
+                        output_ref.clone(),
+                        &resp
+                            .result
+                            .expect("no finalize result")
+                            .result
+                            .expect("no transaction result"),
+                    );
+                },
+                Err(error) => bail!("Failed to submit transaction: {error:?}"),
+            },
             Err(e) => bail!("Failed to get response from handler: {}", e),
         }
     }
 
-    if last_resp.is_none() {
-        bail!("No responses from any of the wallet daemon concurrent calls");
-    }
-
     Ok(())
 }