from copy import deepcopy
from typing import Any
from typing import List
from typing import Optional
from typing import Tuple
from typing import cast
from attr import dataclass
from pytezos.context.impl import ExecutionContext
from pytezos.michelson.micheline import MichelineSequence
from pytezos.michelson.micheline import MichelsonRuntimeError
from pytezos.michelson.parse import MichelsonParser
from pytezos.michelson.parse import MichelsonParserError
from pytezos.michelson.parse import michelson_to_micheline
from pytezos.michelson.program import MichelsonProgram
from pytezos.michelson.program import TztMichelsonProgram
from pytezos.michelson.sections import CodeSection
from pytezos.michelson.stack import MichelsonStack
from pytezos.michelson.types import OperationType
[docs]@dataclass(kw_only=True)
class InterpreterResult:
"""Result of running contract in interpreter"""
operations = None
storage = None
lazy_diff = None
stdout: List[str]
error: Optional[Exception] = None
instructions: Optional[MichelineSequence] = None
stack: Optional[MichelsonStack] = None
[docs]class Interpreter:
"""Michelson interpreter reimplemented in Python.
Based on the following reference: https://tezos.gitlab.io/michelson-reference/
"""
def __init__(
self,
extra_primitives: Optional[List[str]] = None,
debug: bool = False,
) -> None:
self.stack = MichelsonStack()
self.context = ExecutionContext()
self.context.debug = debug
self.parser = MichelsonParser(debug=debug, extra_primitives=extra_primitives)
[docs] def execute(self, code: str) -> InterpreterResult:
"""Execute some code preserving current context and stack
:param code: Michelson code
"""
result = InterpreterResult(stdout=[])
stack_backup = deepcopy(self.stack)
context_backup = deepcopy(self.context)
try:
code_section = CodeSection.match(michelson_to_micheline(code))
instructions = code_section.args[0].execute(self.stack, result.stdout, self.context)
result.instructions = MichelineSequence([instructions])
result.stack = self.stack
except (MichelsonParserError, MichelsonRuntimeError) as e:
if self.context.debug:
raise
self.stack = stack_backup
self.context = context_backup
result.stdout.append(e.format_stdout())
result.error = e
return result
[docs] def reset(self) -> None:
"""Reset interpreter's stack and context"""
self.stack = MichelsonStack()
self.context = ExecutionContext()
[docs] @staticmethod
def run_code(
parameter,
storage,
script: str,
entrypoint='default',
output_mode='readable',
amount=None,
chain_id=None,
source=None,
sender=None,
balance=None,
block_id=None,
**kwargs,
) -> Tuple[List[dict], Any, List[dict], List[str], Optional[Exception]]:
"""Execute contract in interpreter
:param parameter: parameter expression
:param storage: storage expression
:param script: contract's Michelson code
:param entrypoint: contract entrypoint
:param output_mode: one of readable/optimized/legacy_optimized
:param amount: patch AMOUNT
:param chain_id: patch CHAIN_ID
:param source: patch SOURCE
:param sender: patch SENDER
:param balance: patch BALANCE
:param block_id: set block ID
"""
context = ExecutionContext(
amount=amount,
chain_id=chain_id,
source=source,
sender=sender,
balance=balance,
block_id=block_id,
script={'code': script, 'storage': storage},
**kwargs,
)
stack = MichelsonStack()
stdout = [] # type: ignore
try:
program = MichelsonProgram.load(context, with_code=True)
res = program.instantiate(
entrypoint=entrypoint,
parameter=parameter,
storage=storage,
)
res.begin(stack, stdout, context)
res.execute(stack, stdout, context)
operations, storage, lazy_diff, _ = res.end(stack, stdout, output_mode=output_mode)
return operations, storage, lazy_diff, stdout, None
except MichelsonRuntimeError as e:
stdout.append(e.format_stdout())
return [], None, [], stdout, e
[docs] @staticmethod
def run_callback(
entrypoint: str,
parameter,
storage,
context: ExecutionContext,
) -> Tuple[Any, Any, List[str], Optional[Exception]]:
"""Execute view entrypoint of the contract loaded into the context
:param entrypoint: contract entrypoint
:param parameter: parameter section
:param storage: storage section
:param context: execution context
:returns: [operations, storage, stdout, error]
"""
ctx = ExecutionContext(
shell=context.shell,
key=context.key,
block_id=context.block_id,
script=context.script,
address=context.address,
)
stack = MichelsonStack()
stdout = [] # type: ignore
try:
program = MichelsonProgram.load(ctx, with_code=True)
res = program.instantiate(entrypoint=entrypoint, parameter=parameter, storage=storage)
res.begin(stack, stdout, context)
res.execute(stack, stdout, context)
_, _, _, pair = res.end(stack, stdout)
operations = cast(List[OperationType], list(pair.items[0]))
storage = pair.items[1]
# Note: the `storage` returned by the Michelson interpreter above is not
# required to include the full annotations specified in the contract's storage.
# The lack of annotations affects calls to `to_python_object()`, causing the storage
# you get back from the view to not always be converted to the same object
# as if you called ContractInterface.storage() directly.
# Re-parsing using the contract's storage section here to recover the annotations.
storage = program.storage.from_micheline_value(storage.to_micheline_value())
return [op.to_python_object() for op in operations], storage.to_python_object(), stdout, None
except MichelsonRuntimeError as e:
stdout.append(e.format_stdout())
return None, None, stdout, e
[docs] @staticmethod
def run_view(name: str, parameter, storage, context: ExecutionContext) -> Tuple[Any, Any, Optional[Exception]]:
ctx = ExecutionContext(
shell=context.shell,
key=context.key,
block_id=context.block_id,
script=context.script,
address=context.address,
view_results=context.view_results,
)
stack = MichelsonStack()
stdout = [] # type: ignore
try:
program = MichelsonProgram.load(ctx, with_code=True)
res = program.instantiate_view(name=name, parameter=parameter, storage=storage)
res.begin(stack, stdout, context)
res.execute_view(stack, stdout, context)
ret_value = res.ret(stack, stdout)
return ret_value.to_python_object(), stdout, None
except MichelsonRuntimeError as e:
stdout.append(e.format_stdout())
return None, stdout, e
[docs] @staticmethod
def run_tzt(
script: str,
amount=None,
chain_id=None,
source=None,
sender=None,
balance=None,
block_id=None,
**kwargs,
) -> None:
"""Execute TZT test suite code
:param script: test contract's Michelson code
:param amount: patch AMOUNT
:param chain_id: patch CHAIN_ID
:param source: patch SOURCE
:param sender: patch SENDER
:param balance: patch BALANCE
:param block_id: set block ID
"""
context = ExecutionContext(
amount=amount,
chain_id=chain_id,
source=source,
sender=sender,
balance=balance,
block_id=block_id,
script={'code': script},
tzt=True,
**kwargs,
)
stack = MichelsonStack()
stdout: List[str] = []
program = TztMichelsonProgram.load(context, with_code=True)
res = program.instantiate()
res.fill_context(script, context)
res.register_bigmaps(stack, stdout, context)
res.begin(stack, stdout, context)
res.execute(stack, stdout, context)
res.end(stack, stdout, context)