Coverage for wasmtime/_module.py: 89%

1from . import _ffi as ffi 

2from ctypes import * 

3import ctypes 

4from wasmtime import Engine, wat2wasm, ImportType, ExportType, WasmtimeError, Managed 

5import typing 

6from os import PathLike 

7 

8 

9class Module(Managed["ctypes._Pointer[ffi.wasmtime_module_t]"]): 

10 

11 @classmethod 

12 def from_file(cls, engine: Engine, path: typing.Union[str, bytes, PathLike]) -> "Module": 

13 """ 

14 Compiles and creates a new `Module` by reading the file at `path` and 

15 then delegating to the `Module` constructor. 

16 """ 

17 

18 with open(path, "rb") as f: 

19 contents = f.read() 

20 return cls(engine, contents) 

21 

22 def __init__(self, engine: Engine, wasm: typing.Union[str, bytes]): 

23 

24 if not isinstance(engine, Engine): 

25 raise TypeError("expected an Engine") 

26 

27 # If this looks like a string, parse it as the text format. Note that 

28 # in python 2 strings and bytes are basically the same, so we skip this 

29 # if the first byte in the string is 0, meaning this is actually a wasm 

30 # module. 

31 if isinstance(wasm, str) and len(wasm) > 0 and ord(wasm[0]) != 0: 

32 wasm = wat2wasm(wasm) 

33 if isinstance(wasm, bytes) and len(wasm) > 0 and wasm[0] != 0: 

34 wasm = wat2wasm(wasm) 

35 

36 if not isinstance(wasm, (bytes, bytearray)): 

37 raise TypeError("expected wasm bytes") 

38 

39 # TODO: can the copy be avoided here? I can't for the life of me 

40 # figure this out. 

41 binary = (c_uint8 * len(wasm)).from_buffer_copy(wasm) 

42 ptr = POINTER(ffi.wasmtime_module_t)() 

43 error = ffi.wasmtime_module_new(engine.ptr(), binary, len(wasm), byref(ptr)) 

44 if error: 

45 raise WasmtimeError._from_ptr(error) 

46 self._set_ptr(ptr) 

47 

48 def _delete(self, ptr: "ctypes._Pointer[ffi.wasmtime_module_t]") -> None: 

49 ffi.wasmtime_module_delete(ptr) 

50 

51 @classmethod 

52 def _from_ptr(cls, ptr: "ctypes._Pointer[ffi.wasmtime_module_t]") -> "Module": 

53 if not isinstance(ptr, POINTER(ffi.wasmtime_module_t)): 

54 raise TypeError("wrong pointer type") 

55 ty: "Module" = cls.__new__(cls) 

56 ty._set_ptr(ptr) 

57 return ty 

58 

59 @classmethod 

60 def deserialize(cls, engine: Engine, encoded: typing.Union[bytes, bytearray]) -> 'Module': 

61 """ 

62 Deserializes bytes previously created by `Module.serialize`. 

63 

64 This constructor for `Module` will deserialize bytes previously created 

65 by a serialized module. This will only succeed if the bytes were 

66 previously created by the same version of `wasmtime` as well as the 

67 same configuration within `Engine`. 

68 """ 

69 

70 if not isinstance(encoded, (bytes, bytearray)): 

71 raise TypeError("expected bytes") 

72 

73 ptr = POINTER(ffi.wasmtime_module_t)() 

74 

75 # TODO: can the copy be avoided here? I can't for the life of me 

76 # figure this out. 

77 error = ffi.wasmtime_module_deserialize( 

78 engine.ptr(), 

79 (c_uint8 * len(encoded)).from_buffer_copy(encoded), 

80 len(encoded), 

81 byref(ptr)) 

82 if error: 

83 raise WasmtimeError._from_ptr(error) 

84 ret: "Module" = cls.__new__(cls) 

85 ret._set_ptr(ptr) 

86 return ret 

87 

88 @classmethod 

89 def deserialize_file(cls, engine: Engine, path: str) -> 'Module': 

90 """ 

91 Deserializes bytes previously created by `Module.serialize` that are 

92 stored in a file on the filesystem. 

93 

94 Otherwise this function is the same as `Module.deserialize`. 

95 """ 

96 

97 ptr = POINTER(ffi.wasmtime_module_t)() 

98 path_bytes = path.encode('utf-8') 

99 error = ffi.wasmtime_module_deserialize_file( 

100 engine.ptr(), 

101 path_bytes, 

102 byref(ptr)) 

103 if error: 

104 raise WasmtimeError._from_ptr(error) 

105 ret: "Module" = cls.__new__(cls) 

106 ret._set_ptr(ptr) 

107 return ret 

108 

109 @classmethod 

110 def validate(cls, engine: Engine, wasm: typing.Union[bytes, bytearray]) -> None: 

111 """ 

112 Validates whether the list of bytes `wasm` provided is a valid 

113 WebAssembly binary given the configuration in `store` 

114 

115 Raises a `WasmtimeError` if the wasm isn't valid. 

116 """ 

117 

118 if not isinstance(wasm, (bytes, bytearray)): 

119 raise TypeError("expected wasm bytes") 

120 

121 # TODO: can the copy be avoided here? I can't for the life of me 

122 # figure this out. 

123 buf = (c_uint8 * len(wasm)).from_buffer_copy(wasm) 

124 error = ffi.wasmtime_module_validate(engine.ptr(), buf, len(wasm)) 

125 

126 if error: 

127 raise WasmtimeError._from_ptr(error) 

128 

129# @property 

130# def type(self) -> ModuleType: 

131# """ 

132# Gets the type of this module as a `ModuleType` 

133# """ 

134 

135# ptr = ffi.wasmtime_module_type(self.ptr()) 

136# return ModuleType._from_ptr(ptr, None) 

137 

138 @property 

139 def imports(self) -> typing.List[ImportType]: 

140 """ 

141 Returns the types of imports that this module has 

142 """ 

143 

144 imports = ImportTypeList() 

145 ffi.wasmtime_module_imports(self.ptr(), byref(imports.vec)) 

146 ret = [] 

147 for i in range(0, imports.vec.size): 

148 ret.append(ImportType._from_ptr(imports.vec.data[i], imports)) 

149 return ret 

150 

151 @property 

152 def exports(self) -> typing.List[ExportType]: 

153 """ 

154 Returns the types of the exports that this module has 

155 """ 

156 

157 exports = ExportTypeList() 

158 ffi.wasmtime_module_exports(self.ptr(), byref(exports.vec)) 

159 ret = [] 

160 for i in range(0, exports.vec.size): 

161 ret.append(ExportType._from_ptr(exports.vec.data[i], exports)) 

162 return ret 

163 

164 def serialize(self) -> bytearray: 

165 """ 

166 Serializes this module to a binary representation. 

167 

168 This method will serialize this module to an in-memory byte array which 

169 can be cached and later passed to `Module.deserialize` to recreate this 

170 module. 

171 """ 

172 raw = ffi.wasm_byte_vec_t() 

173 err = ffi.wasmtime_module_serialize(self.ptr(), byref(raw)) 

174 if err: 

175 raise WasmtimeError._from_ptr(err) 

176 ret = ffi.to_bytes(raw) 

177 ffi.wasm_byte_vec_delete(byref(raw)) 

178 return ret 

179 

180 def _as_extern(self) -> ffi.wasmtime_extern_t: 

181 union = ffi.wasmtime_extern_union(module=self.ptr()) 

182 return ffi.wasmtime_extern_t(ffi.WASMTIME_EXTERN_MODULE, union) 

183 

184 

185class ImportTypeList: 

186 def __init__(self) -> None: 

187 self.vec = ffi.wasm_importtype_vec_t(0, None) 

188 

189 def __del__(self) -> None: 

190 ffi.wasm_importtype_vec_delete(byref(self.vec)) 

191 

192 

193class ExportTypeList: 

194 def __init__(self) -> None: 

195 self.vec = ffi.wasm_exporttype_vec_t(0, None) 

196 

197 def __del__(self) -> None: 

198 ffi.wasm_exporttype_vec_delete(byref(self.vec))