Dev Tools: Source map decoder

Makefile

@@ -1,8 +1,8 @@
-UNITS = "@unit.abijson or @unit.abijson.byname or @unit.algod or @unit.algod.ledger_refactoring or @unit.applications or @unit.atc_method_args or @unit.atomic_transaction_composer or @unit.dryrun or @unit.dryrun.trace.application or @unit.feetest or @unit.indexer or @unit.indexer.ledger_refactoring 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"
+UNITS = "@unit.abijson or @unit.abijson.byname or @unit.algod or @unit.algod.ledger_refactoring or @unit.applications or @unit.atc_method_args or @unit.atomic_transaction_composer or @unit.dryrun or @unit.dryrun.trace.application or @unit.feetest or @unit.indexer or @unit.indexer.ledger_refactoring 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.sourcemap"
 unit:
 	behave --tags=$(UNITS) tests -f progress2
 
-INTEGRATIONS = "@abi or @algod or @applications or @applications.verified or @assets or @auction or @c2c or @compile or @dryrun or @dryrun.testing or @indexer or @indexer.231 or @indexer.applications or @kmd or @rekey or @send.keyregtxn or @send"
+INTEGRATIONS = "@abi or @algod or @applications or @applications.verified or @assets or @auction or @c2c or @compile or @dryrun or @dryrun.testing or @indexer or @indexer.231 or @indexer.applications or @kmd or @rekey or @send.keyregtxn or @send or @compile.sourcemap"
 integration:
 	behave --tags=$(INTEGRATIONS) tests -f progress2
 

algosdk/__init__.py

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

algosdk/error.py

@@ -222,3 +222,11 @@ def __init__(self, msg):
 class AtomicTransactionComposerError(Exception):
     def __init__(self, msg):
         super().__init__(msg)
+
+
+class SourceMapVersionError(Exception):
+    def __init__(self, version):
+        Exception.__init__(
+            self,
+            "Only SourceMap version 3 is supported, got: {}".format(version),
+        )

algosdk/source_map.py

@@ -0,0 +1,85 @@
+from typing import Dict, Any, List, Tuple
+
+from algosdk.error import SourceMapVersionError
+
+
+class SourceMap:
+    """
+    Decodes a VLQ-encoded source mapping between PC values and TEAL source code lines.
+    Spec available here: https://sourcemaps.info/spec.html
+
+    Args:
+        source_map (dict(str, Any)): source map JSON from algod
+    """
+
+    def __init__(self, source_map: Dict[str, Any]):
+
+        self.version: int = source_map["version"]
+
+        if self.version != 3:
+            raise SourceMapVersionError(self.version)
+
+        self.sources: List[str] = source_map["sources"]
+
+        self.mappings: str = source_map["mappings"]
+
+        pc_list = [
+            _decode_int_value(raw_val) for raw_val in self.mappings.split(";")
+        ]
+
+        self.pc_to_line: Dict[int, int] = {}
+        self.line_to_pc: Dict[int, List[int]] = {}
+
+        last_line = 0
+        for index, line_delta in enumerate(pc_list):
+            # line_delta is None if the line number has not changed
+            # or if the line is empty
+            if line_delta is not None:
+                last_line = last_line + line_delta
+
+            if last_line not in self.line_to_pc:
+                self.line_to_pc[last_line] = []
+
+            self.line_to_pc[last_line].append(index)
+            self.pc_to_line[index] = last_line
+
+    def get_line_for_pc(self, pc: int) -> int:
+        return self.pc_to_line.get(pc, None)
+
+    def get_pcs_for_line(self, line: int) -> List[int]:
+        return self.line_to_pc.get(line, None)
+
+
+def _decode_int_value(value: str) -> int:
+    # Mappings may have up to 5 segments:
+    # Third segment represents the zero-based starting line in the original source represented.
+    decoded_value = _base64vlq_decode(value)
+    return decoded_value[2] if decoded_value else None
+
+
+"""
+Source taken from: https://gist.github.com/mjpieters/86b0d152bb51d5f5979346d11005588b
+"""
+
+_b64chars = b"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/"
+_b64table = [None] * (max(_b64chars) + 1)
+for i, b in enumerate(_b64chars):
+    _b64table[b] = i
+
+shiftsize, flag, mask = 5, 1 << 5, (1 << 5) - 1
+
+
+def _base64vlq_decode(vlqval: str) -> Tuple[int]:
+    """Decode Base64 VLQ value"""
+    results = []
+    shift = value = 0
+    # use byte values and a table to go from base64 characters to integers
+    for v in map(_b64table.__getitem__, vlqval.encode("ascii")):
+        value += (v & mask) << shift
+        if v & flag:
+            shift += shiftsize
+            continue
+        # determine sign and add to results
+        results.append((value >> 1) * (-1 if value & 1 else 1))
+        shift = value = 0
+    return results

algosdk/v2client/algod.py

@@ -347,7 +347,7 @@ def suggested_params(self, **kwargs):
             res["min-fee"],
         )
 
-    def compile(self, source, **kwargs):
+    def compile(self, source, source_map=False, **kwargs):
         """
         Compile TEAL source with remote algod.
 
@@ -365,9 +365,9 @@ def compile(self, source, **kwargs):
             {"Content-Type": "application/x-binary"},
         )
         kwargs["headers"] = headers
-
+        params = {"sourcemap": source_map}
         return self.algod_request(
-            "POST", req, data=source.encode("utf-8"), **kwargs
+            "POST", req, params=params, data=source.encode("utf-8"), **kwargs
         )
 
     def dryrun(self, drr, **kwargs):

tests/steps/other_v2_steps.py

@@ -18,12 +18,7 @@
 from glom import glom
 import parse
 
-from algosdk import (
-    dryrun_results,
-    encoding,
-    error,
-    mnemonic,
-)
+from algosdk import dryrun_results, encoding, error, mnemonic, source_map
 from algosdk.error import AlgodHTTPError
 from algosdk.future import transaction
 from algosdk.v2client import *
@@ -1707,3 +1702,47 @@ def glom_app_eval_delta(context, i, path, field):
     assert field == str(
         actual_field
     ), f"path [{path}] expected value [{field}] but got [{actual_field}] instead"
+
+
+@given('a source map json file "{sourcemap_file}"')
+def parse_source_map(context, sourcemap_file):
+    jsmap = json.loads(load_resource(sourcemap_file, is_binary=False))
+    context.source_map = source_map.SourceMap(jsmap)
+
+
+@then('the string composed of pc:line number equals "{pc_to_line}"')
+def check_source_map(context, pc_to_line):
+    buff = [
+        f"{pc}:{line}" for pc, line in context.source_map.pc_to_line.items()
+    ]
+    actual = ";".join(buff)
+    assert actual == pc_to_line, f"expected {pc_to_line} got {actual}"
+
+
+@then('getting the line associated with a pc "{pc}" equals "{line}"')
+def check_pc_to_line(context, pc, line):
+
+    actual_line = context.source_map.get_line_for_pc(int(pc))
+    assert actual_line == int(line), f"expected line {line} got {actual_line}"
+
+
+@then('getting the last pc associated with a line "{line}" equals "{pc}"')
+def check_line_to_pc(context, line, pc):
+    actual_pcs = context.source_map.get_pcs_for_line(int(line))
+    assert actual_pcs[-1] == int(pc), f"expected pc {pc} got {actual_pcs[-1]}"
+
+
+@when('I compile a teal program "{teal}" with mapping enabled')
+def check_compile_mapping(context, teal):
+    data = load_resource(teal)
+    source = data.decode("utf-8")
+    response = context.app_acl.compile(source, source_map=True)
+    context.raw_source_map = json.dumps(
+        response["sourcemap"], separators=(",", ":")
+    )
+
+
+@then('the resulting source map is the same as the json "{sourcemap}"')
+def check_mapping_equal(context, sourcemap):
+    expected = load_resource(sourcemap).decode("utf-8").strip()
+    assert context.raw_source_map == expected