ATC: Refactor Pending Transaction Information in ATC into try block

algosdk/atomic_transaction_composer.py

@@ -718,7 +718,25 @@ def simulate(
 
         # Parse out abi results
         txn_results = [t["txn-result"] for t in txn_group["txn-results"]]
-        method_results = self.parse_response(txn_results)
+        method_results: List[ABIResult] = []
+        for method_index, method in self.method_dict.items():
+            tx_id = self.tx_ids[method_index]
+            result: ABIResult = ABIResult(
+                tx_id=tx_id,
+                raw_value=b"",
+                return_value=None,
+                decode_error=None,
+                tx_info={},
+                method=method,
+            )
+            try:
+                tx_info = txn_results[method_index]
+                result = self.parse_result(
+                    method, self.tx_ids[method_index], tx_info
+                )
+            except Exception as e:
+                result.decode_error = e
+            method_results.append(result)
 
         # build up data structure with fields we'd want
         sim_results = []
@@ -785,80 +803,78 @@ def execute(
         self.status = AtomicTransactionComposerStatus.COMMITTED
 
         confirmed_round = resp["confirmed-round"]
-
-        tx_results = [
-            cast(Dict[str, Any], client.pending_transaction_info(tx_id))
-            for tx_id in self.tx_ids
-        ]
-
-        method_results = self.parse_response(tx_results)
+        method_results: List[ABIResult] = []
+
+        for method_index, method in self.method_dict.items():
+            tx_id = self.tx_ids[method_index]
+            result: ABIResult = ABIResult(
+                tx_id=tx_id,
+                raw_value=b"",
+                return_value=None,
+                decode_error=None,
+                tx_info={},
+                method=method,
+            )
+            try:
+                tx_info = cast(
+                    Dict[str, Any], client.pending_transaction_info(tx_id)
+                )
+                result = self.parse_result(
+                    method, self.tx_ids[method_index], tx_info
+                )
+            except Exception as e:
+                result.decode_error = e
+            method_results.append(result)
 
         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
-
-            if i not in self.method_dict:
-                continue
-
-            # Parse log for ABI method return value
-            try:
-                if self.method_dict[i].returns.type == abi.Returns.VOID:
-                    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],
-                        )
-                    )
-                    continue
-
-                logs = tx_info["logs"] if "logs" in tx_info else []
-
-                # Look for the last returned value in the log
-                if not logs:
-                    raise error.AtomicTransactionComposerError(
-                        "app call transaction did not log a return value"
-                    )
-                result = logs[-1]
-                # Check that the first four bytes is the hash of "return"
-                result_bytes = base64.b64decode(result)
-                if (
-                    len(result_bytes) < 4
-                    or result_bytes[:4] != ABI_RETURN_HASH
-                ):
-                    raise error.AtomicTransactionComposerError(
-                        "app call transaction did not log a return value"
-                    )
-                raw_value = result_bytes[4:]
-                method_return_type = cast(
-                    abi.ABIType, self.method_dict[i].returns.type
-                )
-                return_value = method_return_type.decode(raw_value)
-            except Exception as e:
-                decode_error = e
-
-            method_results.append(
-                ABIResult(
+    def parse_result(
+        self, method: abi.Method, txid: str, txn: Dict[str, Any]
+    ) -> ABIResult:
+        tx_id = txid
+        raw_value = b""
+        return_value = None
+        decode_error = None
+        try:
+            if method.returns.type == abi.Returns.VOID:
+                return ABIResult(
                     tx_id=tx_id,
-                    raw_value=cast(bytes, raw_value),
+                    raw_value=raw_value,
                     return_value=return_value,
                     decode_error=decode_error,
-                    tx_info=tx_info,
-                    method=self.method_dict[i],
+                    tx_info=txn,
+                    method=method,
                 )
-            )
 
-        return method_results
+            logs = txn["logs"] if "logs" in txn else []
+
+            # Look for the last returned value in the log
+            if not logs:
+                raise error.AtomicTransactionComposerError(
+                    "app call transaction did not log a return value"
+                )
+            result = logs[-1]
+            # Check that the first four bytes is the hash of "return"
+            result_bytes = base64.b64decode(result)
+            if len(result_bytes) < 4 or result_bytes[:4] != ABI_RETURN_HASH:
+                raise error.AtomicTransactionComposerError(
+                    "app call transaction did not log a return value"
+                )
+            raw_value = result_bytes[4:]
+            method_return_type = cast(abi.ABIType, method.returns.type)
+            return_value = method_return_type.decode(raw_value)
+        except Exception as e:
+            decode_error = e
+
+        return ABIResult(
+            tx_id=tx_id,
+            raw_value=cast(bytes, raw_value),
+            return_value=return_value,
+            decode_error=decode_error,
+            tx_info=txn,
+            method=method,
+        )

tests/integration.tags

@@ -14,4 +14,4 @@
 @rekey_v1
 @send
 @send.keyregtxn
-@simulate
+@simulate

tests/steps/application_v2_steps.py

@@ -909,7 +909,7 @@ def check_atomic_transaction_composer_return_type(context, abiTypes):
         assert result.decode_error is None
 
         if expected == "void":
-            assert result.raw_value is None
+            assert result.raw_value is bytes()
             with pytest.raises(ABITypeError):
                 abi.ABIType.from_string(expected)
             continue