import io import os import subprocess import sys from tempfile import NamedTemporaryFile import pywasm import wasm3 import wasmer import wasmer_compiler_cranelift import wasmtime from phasm.codestyle import phasm_render from phasm.compiler import phasm_compile from phasm.parser import phasm_parse from . import runners DASHES = '-' * 16 def wat2wasm(code_wat): path = os.environ.get('WAT2WASM', 'wat2wasm') with NamedTemporaryFile('w+t') as input_fp: input_fp.write(code_wat) input_fp.flush() with NamedTemporaryFile('w+b') as output_fp: subprocess.run( [ path, input_fp.name, '-o', output_fp.name, ], check=True, ) output_fp.seek(0) return output_fp.read() class SuiteResult: def __init__(self): self.returned_value = None RUNNER_CLASS_MAP = { 'pywasm': runners.RunnerPywasm, 'pywasm3': runners.RunnerPywasm3, 'wasmtime': runners.RunnerWasmtime, 'wasmer': runners.RunnerWasmer, } class Suite: """ WebAssembly test suite """ def __init__(self, code_py): self.code_py = code_py def run_code(self, *args, runtime='pywasm3', imports=None): """ Compiles the given python code into wasm and then runs it Returned is an object with the results set """ class_ = RUNNER_CLASS_MAP[runtime] runner = class_(self.code_py) runner.parse() runner.compile_ast() runner.compile_wat() runner.compile_wasm() runner.interpreter_setup() runner.interpreter_load(imports) write_header(sys.stderr, 'Phasm') runner.dump_phasm_code(sys.stderr) write_header(sys.stderr, 'Assembly') runner.dump_wasm_wat(sys.stderr) # Check if code formatting works assert self.code_py == '\n' + phasm_render(runner.phasm_ast) # \n for formatting in tests wasm_args = [] if args: write_header(sys.stderr, 'Memory (pre alloc)') runner.interpreter_dump_memory(sys.stderr) for arg in args: if isinstance(arg, (int, float, )): wasm_args.append(arg) continue if isinstance(arg, bytes): adr = runner.call('stdlib.types.__alloc_bytes__', len(arg)) sys.stderr.write(f'Allocation 0x{adr:08x} {repr(arg)}\n') runner.interpreter_write_memory(adr + 4, arg) wasm_args.append(adr) continue raise NotImplementedError(arg) write_header(sys.stderr, 'Memory (pre run)') runner.interpreter_dump_memory(sys.stderr) result = SuiteResult() result.returned_value = runner.call('testEntry', *wasm_args) write_header(sys.stderr, 'Memory (post run)') runner.interpreter_dump_memory(sys.stderr) return result def write_header(textio, msg: str) -> None: textio.write(f'{DASHES} {msg.ljust(16)} {DASHES}\n')