Support for reading ScVal contract parameters and fuzzing over them (#17

)

* Add a soroban contract that uses the various SCVals

* Create SCType class that can read the bindings json from soroban CLI

* Start implementing errors. Raise an error when reading an unsupported
SCType.

* SCVals that store the actual values, and hypothesis strategies to
generate them

* to_kast for SCValues

* Fuzz over the test contracts

* Add a failing test

* More explicit naming of test contracts

* Test the various supported scvaltypes

* Set Version: 0.1.12

* pyupgrade

* Set Version: 0.1.13

---------

Co-authored-by: devops <[email protected]>
Co-authored-by: Burak Bilge Yalçınkaya <[email protected]>

package/version

@@ -1 +1 @@
-0.1.12
+0.1.13

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "poetry.core.masonry.api"
 
 [tool.poetry]
 name = "ksoroban"
-version = "0.1.12"
+version = "0.1.13"
 description = "K tooling for the Soroban platform"
 authors = [
     "Runtime Verification, Inc. <[email protected]>",

src/ksoroban/kasmer.py

@@ -8,32 +8,36 @@
 from tempfile import mkdtemp
 from typing import TYPE_CHECKING
 
-from pyk.kast.inner import KSort
+from hypothesis import strategies
+from pyk.kast.inner import KSort, KVariable
 from pyk.kast.manip import Subst, split_config_from
 from pyk.konvert import kast_to_kore, kore_to_kast
 from pyk.kore.parser import KoreParser
+from pyk.kore.syntax import EVar, SortApp
+from pyk.ktool.kfuzz import fuzz
 from pyk.ktool.krun import KRunOutput
 from pyk.utils import run_process
 from pykwasm.wasm2kast import wasm2kast
 
 from .kast.syntax import (
-    SC_VOID,
     account_id,
     call_tx,
     contract_id,
     deploy_contract,
     sc_bool,
-    sc_u32,
     set_account,
     set_exit_code,
     steps_of,
     upload_wasm,
 )
+from .scval import SCType
 
 if TYPE_CHECKING:
     from typing import Any
 
+    from hypothesis.strategies import SearchStrategy
     from pyk.kast.inner import KInner
+    from pyk.kore.syntax import Pattern
 
     from .utils import SorobanDefinitionInfo
 
@@ -70,16 +74,15 @@ def contract_bindings(self, wasm_contract: Path) -> list[ContractBinding]:
         bindings_list = json.loads(proc_res.stdout)
         bindings = []
         for binding_dict in bindings_list:
-            # TODO: Properly read and store the type information in the bindings (ie. type parameters for vecs, tuples, etc.)
             if binding_dict['type'] != 'function':
                 continue
             name = binding_dict['name']
             inputs = []
             for input_dict in binding_dict['inputs']:
-                inputs.append(input_dict['value']['type'])
+                inputs.append(SCType.from_dict(input_dict['value']))
             outputs = []
             for output_dict in binding_dict['outputs']:
-                outputs.append(output_dict['type'])
+                outputs.append(SCType.from_dict(output_dict))
             bindings.append(ContractBinding(name, tuple(inputs), tuple(outputs)))
         return bindings
 
@@ -146,33 +149,31 @@ def deploy_test(self, contract: KInner) -> tuple[KInner, dict[str, KInner]]:
         return conf, subst
 
     def run_test(self, conf: KInner, subst: dict[str, KInner], binding: ContractBinding) -> None:
-        """Given a configuration with a deployed test contract, run the tests for the supplied binding.
+        """Given a configuration with a deployed test contract, fuzz over the tests for the supplied binding.
 
         Raises:
             CalledProcessError if the test fails
         """
 
-        def getarg(arg: str) -> KInner:
-            # TODO: Implement actual argument generation.
-            #      That's every possible ScVal in Soroban.
-            #      Concrete values for fuzzing/variables for proving.
-            if arg == 'u32':
-                return sc_u32(10)
-            return SC_VOID
-
         from_acct = account_id(b'test-account')
         to_acct = contract_id(b'test-contract')
         name = binding.name
-        args = [getarg(arg) for arg in binding.inputs]
         result = sc_bool(True)
 
-        steps = steps_of([set_exit_code(1), call_tx(from_acct, to_acct, name, args, result), set_exit_code(0)])
+        def make_steps(*args: KInner) -> Pattern:
+            steps_kast = steps_of([set_exit_code(1), call_tx(from_acct, to_acct, name, args, result), set_exit_code(0)])
+            return kast_to_kore(self.definition_info.kdefinition, steps_kast, KSort('Steps'))
+
+        subst['PROGRAM_CELL'] = KVariable('STEPS')
+        template_config = Subst(subst).apply(conf)
+        template_config_kore = kast_to_kore(
+            self.definition_info.kdefinition, template_config, KSort('GeneratedTopCell')
+        )
 
-        subst['PROGRAM_CELL'] = steps
-        test_config = Subst(subst).apply(conf)
-        test_config_kore = kast_to_kore(self.definition_info.kdefinition, test_config, KSort('GeneratedTopCell'))
+        steps_strategy = binding.strategy.map(lambda args: make_steps(*args))
+        template_subst = {EVar('VarSTEPS', SortApp('SortSteps')): steps_strategy}
 
-        self.definition_info.krun.run_pattern(test_config_kore, check=True)
+        fuzz(self.definition_info.path, template_config_kore, template_subst, check_exit_code=True)
 
     def deploy_and_run(self, contract_wasm: Path) -> None:
         """Run all of the tests in a soroban test contract.
@@ -199,5 +200,9 @@ class ContractBinding:
     """Represents one of the function bindings for a soroban contract."""
 
     name: str
-    inputs: tuple[str, ...]
-    outputs: tuple[str, ...]
+    inputs: tuple[SCType, ...]
+    outputs: tuple[SCType, ...]
+
+    @cached_property
+    def strategy(self) -> SearchStrategy[tuple[KInner, ...]]:
+        return strategies.tuples(*(arg.strategy().map(lambda x: x.to_kast()) for arg in self.inputs))

src/ksoroban/kast/syntax.py

@@ -3,10 +3,8 @@
 from typing import TYPE_CHECKING
 
 from pyk.kast.inner import KApply, KSort, KToken, build_cons
-from pyk.prelude.bytes import bytesToken
-from pyk.prelude.collections import list_of
-from pyk.prelude.kbool import boolToken
-from pyk.prelude.kint import intToken
+from pyk.prelude.collections import list_of, map_of
+from pyk.prelude.utils import token
 from pykwasm.kwasm_ast import wasm_string
 
 if TYPE_CHECKING:
@@ -21,43 +19,83 @@ def steps_of(steps: Iterable[KInner]) -> KInner:
 
 
 def account_id(acct_id: bytes) -> KApply:
-    return KApply('AccountId', [bytesToken(acct_id)])
+    return KApply('AccountId', [token(acct_id)])
 
 
 def contract_id(contract_id: bytes) -> KApply:
-    return KApply('ContractId', [bytesToken(contract_id)])
+    return KApply('ContractId', [token(contract_id)])
 
 
 def set_exit_code(i: int) -> KInner:
-    return KApply('setExitCode', [intToken(i)])
+    return KApply('setExitCode', [token(i)])
 
 
 def set_account(acct: bytes, i: int) -> KInner:
-    return KApply('setAccount', [account_id(acct), intToken(i)])
+    return KApply('setAccount', [account_id(acct), token(i)])
 
 
 def upload_wasm(name: bytes, contract: KInner) -> KInner:
-    return KApply('uploadWasm', [bytesToken(name), contract])
+    return KApply('uploadWasm', [token(name), contract])
 
 
 def deploy_contract(from_addr: bytes, address: bytes, wasm_hash: bytes, args: list[KInner] | None = None) -> KInner:
     args = args if args is not None else []
-    return KApply('deployContract', [account_id(from_addr), contract_id(address), bytesToken(wasm_hash), list_of(args)])
+    return KApply('deployContract', [account_id(from_addr), contract_id(address), token(wasm_hash), list_of(args)])
 
 
-def call_tx(from_addr: KInner, to_addr: KInner, func: str, args: list[KInner], result: KInner) -> KInner:
+def call_tx(from_addr: KInner, to_addr: KInner, func: str, args: Iterable[KInner], result: KInner) -> KInner:
     return KApply('callTx', [from_addr, to_addr, wasm_string(func), list_of(args), result])
 
 
 # SCVals
 
 
 def sc_bool(b: bool) -> KInner:
-    return KApply('SCVal:Bool', [boolToken(b)])
+    return KApply('SCVal:Bool', [token(b)])
 
 
 def sc_u32(i: int) -> KInner:
-    return KApply('SCVal:U32', [intToken(i)])
+    return KApply('SCVal:U32', [token(i)])
+
+
+def sc_u64(i: int) -> KInner:
+    return KApply('SCVal:U64', [token(i)])
+
+
+def sc_u128(i: int) -> KInner:
+    return KApply('SCVal:U128', [token(i)])
+
+
+def sc_u256(i: int) -> KInner:
+    return KApply('SCVal:U256', [token(i)])
+
+
+def sc_i32(i: int) -> KInner:
+    return KApply('SCVal:I32', [token(i)])
+
+
+def sc_i64(i: int) -> KInner:
+    return KApply('SCVal:I64', [token(i)])
+
+
+def sc_i128(i: int) -> KInner:
+    return KApply('SCVal:I128', [token(i)])
+
+
+def sc_i256(i: int) -> KInner:
+    return KApply('SCVal:I256', [token(i)])
+
+
+def sc_symbol(s: str) -> KInner:
+    return KApply('SCVal:Symbol', [token(s)])
+
+
+def sc_vec(l: Iterable[KInner]) -> KInner:
+    return KApply('SCVal:Vec', list_of(l))
+
+
+def sc_map(m: dict[KInner, KInner] | Iterable[tuple[KInner, KInner]]) -> KInner:
+    return KApply('SCVal:Map', map_of(m))
 
 
 SC_VOID: Final = KToken('Void', KSort('ScVal'))