Dryrun response

Makefile

@@ -1,4 +1,4 @@
-UNITS = "@unit.abijson or @unit.algod or @unit.applications or @unit.atomic_transaction_composer or @unit.dryrun or @unit.feetest or @unit.indexer or @unit.indexer.logs or @unit.offline or @unit.rekey or @unit.transactions.keyreg or @unit.responses or @unit.responses.231 or @unit.tealsign or @unit.transactions or @unit.transactions.payment or @unit.responses.unlimited_assets or @unit.indexer.ledger_refactoring or @unit.algod.ledger_refactoring"
+UNITS = "@unit.abijson or @unit.algod or @unit.applications or @unit.atomic_transaction_composer or @unit.dryrun or @unit.feetest or @unit.indexer or @unit.indexer.logs or @unit.offline or @unit.rekey or @unit.transactions.keyreg or @unit.responses or @unit.responses.231 or @unit.tealsign or @unit.transactions or @unit.transactions.payment or @unit.responses.unlimited_assets or @unit.indexer.ledger_refactoring or @unit.algod.ledger_refactoring or @unit.dryrun.trace.application"
 unit:
 	behave --tags=$(UNITS) test -f progress2
 

algosdk/__init__.py

@@ -15,5 +15,6 @@
 from . import v2client
 from . import wallet
 from . import wordlist
+from . import dryrun_results
 
 name = "algosdk"

algosdk/dryrun_results.py

@@ -0,0 +1,227 @@
+import base64
+from typing import List
+import tabulate as tlib
+from tabulate import tabulate, TableFormat, DataRow
+
+
+tlib.MIN_PADDING = 0
+
+
+class StackPrinterConfig:
+    DEFAULT_MAX_WIDTH: int = 30
+
+    def __init__(self, max_width=DEFAULT_MAX_WIDTH, top_of_stack_first=True):
+        self.max_width = max_width
+        self.top_of_stack_first = top_of_stack_first
+
+
+class DryrunResponse:
+    def __init__(self, drrjson: dict):
+        for param in ["error", "protocol-version", "txns"]:
+            assert (
+                param in drrjson
+            ), f"expecting dryrun response object to have key '{param}' but it is missing"
+
+        # These are all required fields
+        self.error = drrjson["error"]
+        self.protocol = drrjson["protocol-version"]
+        self.txns = [DryrunTransactionResult(txn) for txn in drrjson["txns"]]
+
+
+class DryrunTransactionResult:
+    def __init__(self, dr):
+        assert (
+            "disassembly" in dr
+        ), "expecting dryrun transaction result to have key 'disassembly' but its missing"
+
+        self.disassembly = dr["disassembly"]
+
+        optionals = [
+            "app-call-messages",
+            "local-deltas",
+            "global-delta",
+            "cost",
+            "logic-sig-messages",
+            "logic-sig-disassembly",
+            "logs",
+        ]
+        for field in optionals:
+            if field in dr:
+                setattr(self, field.replace("-", "_"), dr[field])
+            else:
+                setattr(self, field.replace("-", "_"), None)
+
+        traces = ["app-call-trace", "logic-sig-trace"]
+        for trace_field in traces:
+            if trace_field in dr:
+                setattr(
+                    self,
+                    trace_field.replace("-", "_"),
+                    DryrunTrace(dr[trace_field]),
+                )
+
+    def app_call_rejected(self) -> bool:
+        if self.app_call_messages is not None:
+            return "REJECT" in self.app_call_messages
+        return False
+
+    def logic_sig_rejected(self) -> bool:
+        if self.logic_sig_messages is not None:
+            return "REJECT" in self.logic_sig_messages
+        return False
+
+    @classmethod
+    def trace(
+        cls,
+        dr_trace: "DryrunTrace",
+        disassembly: List[str],
+        spc: StackPrinterConfig,
+    ) -> str:
+
+        # 16 for length of the header up to spaces
+        headers = ["pc#", "ln#", "source", "scratch", "stack"]
+        lines = []
+        for idx in range(len(dr_trace.trace)):
+
+            trace_line = dr_trace.trace[idx]
+
+            src = disassembly[trace_line.line]
+            if trace_line.error != "":
+                src = "!! {} !!".format(trace_line.error)
+
+            prev_scratch = []
+            if idx > 0:
+                prev_scratch = dr_trace.trace[idx - 1].scratch
+
+            scratch = scratch_to_string(prev_scratch, trace_line.scratch)
+            stack = stack_to_string(trace_line.stack, spc.top_of_stack_first)
+            lines.append(
+                [
+                    "{}".format(trace_line.pc),
+                    "{}".format(trace_line.line),
+                    truncate(src, spc.max_width),
+                    truncate(scratch, spc.max_width),
+                    truncate(stack, spc.max_width),
+                ]
+            )
+
+        return (
+            tabulate(
+                lines,
+                headers,
+                disable_numparse=True,
+                tablefmt=TableFormat(
+                    headerrow=DataRow("", " |", ""),
+                    datarow=DataRow("", " |", ""),
+                    padding=0,
+                    lineabove=None,
+                    linebelowheader=None,
+                    linebetweenrows=None,
+                    linebelow=None,
+                    with_header_hide=None,
+                ),
+            )
+            + "\n"
+        )
+
+    def app_trace(self, spc: StackPrinterConfig = None) -> str:
+        if not hasattr(self, "app_call_trace"):
+            return ""
+
+        if spc == None:
+            spc = StackPrinterConfig(top_of_stack_first=False)
+
+        return self.trace(self.app_call_trace, self.disassembly, spc=spc)
+
+    def lsig_trace(self, spc: StackPrinterConfig = None) -> str:
+        if not hasattr(self, "logic_sig_trace"):
+            return ""
+
+        if (
+            not hasattr(self, "logic_sig_disassembly")
+            or self.logic_sig_disassembly is None
+        ):
+            return ""
+
+        if spc == None:
+            spc = StackPrinterConfig(top_of_stack_first=False)
+
+        return self.trace(
+            self.logic_sig_trace, self.logic_sig_disassembly, spaces=spc
+        )
+
+
+class DryrunTrace:
+    def __init__(self, trace: List[dict]):
+        self.trace = [DryrunTraceLine(line) for line in trace]
+
+    def get_trace(self) -> List[str]:
+        return [line.trace_line() for line in self.trace]
+
+
+class DryrunTraceLine:
+    def __init__(self, tl):
+        self.line = tl["line"]
+        self.pc = tl["pc"]
+
+        self.error = ""
+        if "error" in tl:
+            self.error = tl["error"]
+
+        self.scratch = []
+        if "scratch" in tl:
+            self.scratch = [DryrunStackValue(sv) for sv in tl["scratch"]]
+
+        self.stack = [DryrunStackValue(sv) for sv in tl["stack"]]
+
+
+class DryrunStackValue:
+    def __init__(self, v):
+        self.type = v["type"]
+        self.bytes = v["bytes"]
+        self.int = v["uint"]
+
+    def __str__(self) -> str:
+        if len(self.bytes) > 0:
+            return "0x" + base64.b64decode(self.bytes).hex()
+        return str(self.int)
+
+    def __eq__(self, other: "DryrunStackValue"):
+        return (
+            self.type == other.type
+            and self.bytes == other.bytes
+            and self.int == other.int
+        )
+
+
+def truncate(s: str, max_width: int) -> str:
+    if len(s) > max_width and max_width > 0:
+        return s[:max_width] + "..."
+    return s
+
+
+def scratch_to_string(
+    prev_scratch: List[DryrunStackValue], curr_scratch: List[DryrunStackValue]
+) -> str:
+    if len(curr_scratch) == 0:
+        return ""
+
+    new_idx = None
+    for idx in range(len(curr_scratch)):
+        if idx >= len(prev_scratch):
+            new_idx = idx
+            continue
+
+        if curr_scratch[idx] != prev_scratch[idx]:
+            new_idx = idx
+
+    if new_idx == None:
+        return ""
+
+    return "{} = {}".format(new_idx, str(curr_scratch[new_idx]))
+
+
+def stack_to_string(stack: List[DryrunStackValue], reverse: bool) -> str:
+    if reverse:
+        stack.reverse()
+    return "[{}]".format(", ".join([str(sv) for sv in stack]))

requirements.txt

@@ -2,4 +2,5 @@
 black==21.9b0
 glom==20.11.0
 pytest==6.2.5
+tabulate==0.8.9
 git+https://github.com/behave/behave

test/steps/v2_steps.py

@@ -27,6 +27,7 @@
     abi,
     account,
     atomic_transaction_composer,
+    dryrun_results,
     encoding,
     error,
     logic,
@@ -3082,6 +3083,51 @@ def deserialize_json_to_contract(context):
     assert actual == context.abi_contract
 
 
+@given(
+    'a dryrun response file "{dryrun_response_file}" and a transaction at index "{txn_id}"'
+)
+def parse_dryrun_response_object(context, dryrun_response_file, txn_id):
+    dir_path = os.path.dirname(os.path.realpath(__file__))
+    dir_path = os.path.dirname(os.path.dirname(dir_path))
+    with open(
+        dir_path + "/test/features/resources/" + dryrun_response_file, "r"
+    ) as f:
+        drr_dict = json.loads(f.read())
+
+    context.dryrun_response_object = dryrun_results.DryrunResponse(drr_dict)
+    context.dryrun_txn_result = context.dryrun_response_object.txns[
+        int(txn_id)
+    ]
+
+
+@then('calling app trace produces "{app_trace_file}"')
+def dryrun_compare_golden(context, app_trace_file):
+    dir_path = os.path.dirname(os.path.realpath(__file__))
+    dir_path = os.path.dirname(os.path.dirname(dir_path))
+    with open(
+        dir_path + "/test/features/resources/" + app_trace_file, "r"
+    ) as f:
+        trace_expected = f.read()
+
+    dryrun_trace = context.dryrun_txn_result.app_trace()
+
+    got_lines = dryrun_trace.split("\n")
+    expected_lines = trace_expected.split("\n")
+
+    print("{} {}".format(len(got_lines), len(expected_lines)))
+    for idx in range(len(got_lines)):
+        if got_lines[idx] != expected_lines[idx]:
+            print(
+                "  {}  \n{}\n{}\n".format(
+                    idx, got_lines[idx], expected_lines[idx]
+                )
+            )
+
+    assert trace_expected == dryrun_trace, "Expected \n{}\ngot\n{}\n".format(
+        trace_expected, dryrun_trace
+    )
+
+
 @then(
     'I dig into the paths "{paths}" of the resulting atomic transaction tree I see group ids and they are all the same'
 )