New Feature: Adding methods to use the simulate endpoint

.test-env

@@ -1,6 +1,6 @@
 # Configs for testing repo download:
 SDK_TESTING_URL="https://github.com/algorand/algorand-sdk-testing"
-SDK_TESTING_BRANCH="master"
+SDK_TESTING_BRANCH="simulate-endpoint-tests"
 SDK_TESTING_HARNESS="test-harness"
 
 INSTALL_ONLY=0

algosdk/atomic_transaction_composer.py

@@ -479,6 +479,55 @@ def submit(self, client: algod.AlgodClient) -> List[str]:
         self.status = AtomicTransactionComposerStatus.SUBMITTED
         return self.tx_ids
 
+    def simulate(
+        self, client: algod.AlgodClient
+    ) -> "SimulateAtomicTransactionResponse":
+
+        if self.status <= AtomicTransactionComposerStatus.SUBMITTED:
+            self.gather_signatures()
+        else:
+            raise error.AtomicTransactionComposerError(
+                "AtomicTransactionComposerStatus must be submitted or "
+                "lower to simulate a group"
+            )
+
+        simulation_result: Dict[str, Any] = client.simulate_transactions(
+            self.signed_txns, response_format="json"
+        )
+        # Only take the first group in the simulate response
+        txn_group: Dict[str, Any] = simulation_result["txn-groups"][0]
+
+        # Parse out abi results
+        txn_results = [t["txn-result"] for t in txn_group["txn-results"]]
+        method_results = self.parse_response(txn_results)
+
+        # build up data structure with fields we'd want
+        sim_results = []
+        for idx, result in enumerate(method_results):
+            sim_txn: Dict[str, Any] = txn_group["txn-results"][idx]
+
+            sim_results.append(
+                SimulateABIResult(
+                    tx_id=result.tx_id,
+                    raw_value=result.raw_value,
+                    return_value=result.return_value,
+                    decode_error=result.decode_error,
+                    tx_info=result.tx_info,
+                    method=self.method_dict[idx],
+                    missing_signature=sim_txn.get("missing-signature", False),
+                )
+            )
+
+        return SimulateAtomicTransactionResponse(
+            version=simulation_result.get("version", 0),
+            would_succeed=simulation_result.get("would-succeed", False),
+            failure_message=txn_group.get("failure-message", ""),
+            failed_at=txn_group.get("failed-at"),
+            simulate_response=simulation_result,
+            tx_ids=self.tx_ids,
+            results=sim_results,
+        )
+
     def execute(
         self, client: algod.AlgodClient, wait_rounds: int
     ) -> "AtomicTransactionResponse":
@@ -517,20 +566,33 @@ def execute(
         self.status = AtomicTransactionComposerStatus.COMMITTED
 
         confirmed_round = resp["confirmed-round"]
-        method_results = []
 
-        for i, tx_id in enumerate(self.tx_ids):
+        tx_results = [
+            client.pending_transaction_info(tx_id) for tx_id in self.tx_ids
+        ]
+
+        method_results = self.parse_response(tx_results)
+
+        return AtomicTransactionResponse(
+            confirmed_round=confirmed_round,
+            tx_ids=self.tx_ids,
+            results=method_results,
+        )
+
+    def parse_response(self, txns: List[Dict[str, Any]]) -> List["ABIResult"]:
+
+        method_results = []
+        for i, tx_info in enumerate(txns):
+            tx_id = self.tx_ids[i]
             raw_value: Optional[bytes] = None
             return_value = None
             decode_error = None
-            tx_info: Optional[Any] = None
 
             if i not in self.method_dict:
                 continue
 
             # Parse log for ABI method return value
             try:
-                tx_info = client.pending_transaction_info(tx_id)
                 if self.method_dict[i].returns.type == abi.Returns.VOID:
                     method_results.append(
                         ABIResult(
@@ -569,21 +631,18 @@ def execute(
             except Exception as e:
                 decode_error = e
 
-            abi_result = ABIResult(
-                tx_id=tx_id,
-                raw_value=cast(bytes, raw_value),
-                return_value=return_value,
-                decode_error=decode_error,
-                tx_info=cast(Any, tx_info),
-                method=self.method_dict[i],
+            method_results.append(
+                ABIResult(
+                    tx_id=tx_id,
+                    raw_value=cast(bytes, raw_value),
+                    return_value=return_value,
+                    decode_error=decode_error,
+                    tx_info=tx_info,
+                    method=self.method_dict[i],
+                )
             )
-            method_results.append(abi_result)
 
-        return AtomicTransactionResponse(
-            confirmed_round=confirmed_round,
-            tx_ids=self.tx_ids,
-            results=method_results,
-        )
+        return method_results
 
 
 class TransactionSigner(ABC):
@@ -740,3 +799,43 @@ def __init__(
         self.confirmed_round = confirmed_round
         self.tx_ids = tx_ids
         self.abi_results = results
+
+
+class SimulateABIResult(ABIResult):
+    def __init__(
+        self,
+        tx_id: str,
+        raw_value: bytes,
+        return_value: Any,
+        decode_error: Optional[Exception],
+        tx_info: dict,
+        method: abi.Method,
+        missing_signature: bool,
+    ) -> None:
+        self.tx_id = tx_id
+        self.raw_value = raw_value
+        self.return_value = return_value
+        self.decode_error = decode_error
+        self.tx_info = tx_info
+        self.method = method
+        self.missing_signature = missing_signature
+
+
+class SimulateAtomicTransactionResponse:
+    def __init__(
+        self,
+        version: int,
+        would_succeed: bool,
+        failure_message: str,
+        failed_at: Optional[List[int]],
+        simulate_response: Dict[str, Any],
+        tx_ids: List[str],
+        results: List[SimulateABIResult],
+    ) -> None:
+        self.version = version
+        self.would_succeed = would_succeed
+        self.failure_message = failure_message
+        self.failed_at = failed_at
+        self.simulate_response = simulate_response
+        self.tx_ids = tx_ids
+        self.abi_results = results

algosdk/v2client/algod.py

@@ -483,6 +483,47 @@ def get_block_hash(self, round_num, **kwargs):
         req = "/blocks/{}/hash".format(round_num)
         return self.algod_request("GET", req, **kwargs)
 
+    def simulate_transactions(self, txns, **kwargs):
+        """
+        Simulate a list of a signed transaction objects being sent to the network.
+
+        Args:
+            txns (SignedTransaction[] or MultisigTransaction[]):
+                transactions to send
+            request_header (dict, optional): additional header for request
+
+        Returns:
+            Dict[str, Any]: results from simulation of transaction group
+        """
+        serialized = []
+        for txn in txns:
+            serialized.append(base64.b64decode(encoding.msgpack_encode(txn)))
+
+        return self.simulate_raw_transaction(
+            base64.b64encode(b"".join(serialized)), **kwargs
+        )
+
+    def simulate_raw_transaction(self, txn, **kwargs):
+        """
+        Simulate a transaction group
+
+        Args:
+            txn (str): transaction to send, encoded in base64
+            request_header (dict, optional): additional header for request
+
+        Returns:
+            Dict[str, Any]: results from simulation of transaction group
+        """
+        txn = base64.b64decode(txn)
+        req = "/transactions/simulate"
+        headers = util.build_headers_from(
+            kwargs.get("headers", False),
+            {"Content-Type": "application/x-binary"},
+        )
+        kwargs["headers"] = headers
+
+        return self.algod_request("POST", req, data=txn, **kwargs)
+
 
 def _specify_round_string(block, round_num):
     """

tests/integration.tags

@@ -13,3 +13,4 @@
 @rekey_v1
 @send
 @send.keyregtxn
+@simulate

tests/steps/other_v2_steps.py

@@ -19,6 +19,10 @@
     source_map,
     transaction,
 )
+
+from algosdk.atomic_transaction_composer import (
+    SimulateAtomicTransactionResponse,
+)
 from algosdk.error import AlgodHTTPError
 from algosdk.testing.dryrun import DryrunTestCaseMixin
 from algosdk.v2client import *
@@ -1414,3 +1418,33 @@ def transaction_proof(context, round, txid, hashtype):
 @when("we make a Lookup Block Hash call against round {round}")
 def get_block_hash(context, round):
     context.response = context.acl.get_block_hash(round)
+
+
+@when("I simulate the transaction")
+def simulate_transaction(context):
+    context.simulate_response = context.app_acl.simulate_transactions(
+        [context.stx], response_format="json"
+    )
+
+
+@then("the simulation should succeed")
+def simulate_transaction_succeed(context):
+    if hasattr(context, "simulate_response"):
+        assert context.simulate_response["would-succeed"] is True
+    else:
+        assert context.simulate_atc_response.would_succeed is True
+
+
+@then("I simulate the current transaction group with the composer")
+def simulate_atc(context):
+    context.simulate_atc_response = (
+        context.atomic_transaction_composer.simulate(context.app_acl)
+    )
+
+
+@then('the simulation should fail at path "{path}" with message "{message}"')
+def simulate_atc_failure(context, path, message):
+    resp: SimulateAtomicTransactionResponse = context.simulate_atc_response
+    assert resp.would_succeed is False
+    assert resp.failed_at[0] == int(path)
+    assert message in resp.failure_message