python_ffi_test.py (8974B)
1 #!/usr/bin/env python3 2 """Python FFI tests for the Arboricx C ABI. 3 4 Tests both the native fast-path bundle loader and the Tricu kernel fallback. 5 """ 6 import ctypes 7 import os 8 import sys 9 import time 10 11 SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__)) 12 ZIG_DIR = os.path.dirname(SCRIPT_DIR) 13 lib_path = os.environ.get( 14 "ARBORICX_LIB", 15 os.path.join(ZIG_DIR, "zig-out", "lib", "libarboricx.so"), 16 ) 17 lib = ctypes.CDLL(lib_path) 18 19 # --- Lifecycle --- 20 lib.arboricx_init.restype = ctypes.c_void_p 21 lib.arboricx_free.argtypes = [ctypes.c_void_p] 22 23 # --- Tree construction --- 24 lib.arb_leaf.argtypes = [ctypes.c_void_p] 25 lib.arb_leaf.restype = ctypes.c_uint32 26 lib.arb_stem.argtypes = [ctypes.c_void_p, ctypes.c_uint32] 27 lib.arb_stem.restype = ctypes.c_uint32 28 lib.arb_fork.argtypes = [ctypes.c_void_p, ctypes.c_uint32, ctypes.c_uint32] 29 lib.arb_fork.restype = ctypes.c_uint32 30 lib.arb_app.argtypes = [ctypes.c_void_p, ctypes.c_uint32, ctypes.c_uint32] 31 lib.arb_app.restype = ctypes.c_uint32 32 33 # --- Reduction --- 34 lib.arb_reduce.argtypes = [ctypes.c_void_p, ctypes.c_uint32, ctypes.c_uint64] 35 lib.arb_reduce.restype = ctypes.c_uint32 36 37 # --- Codecs --- 38 lib.arb_of_number.argtypes = [ctypes.c_void_p, ctypes.c_uint64] 39 lib.arb_of_number.restype = ctypes.c_uint32 40 lib.arb_of_string.argtypes = [ctypes.c_void_p, ctypes.c_char_p] 41 lib.arb_of_string.restype = ctypes.c_uint32 42 lib.arb_of_bytes.argtypes = [ctypes.c_void_p, ctypes.POINTER(ctypes.c_uint8), ctypes.c_size_t] 43 lib.arb_of_bytes.restype = ctypes.c_uint32 44 lib.arb_of_list.argtypes = [ctypes.c_void_p, ctypes.POINTER(ctypes.c_uint32), ctypes.c_size_t] 45 lib.arb_of_list.restype = ctypes.c_uint32 46 lib.arb_to_number.argtypes = [ctypes.c_void_p, ctypes.c_uint32, ctypes.POINTER(ctypes.c_uint64)] 47 lib.arb_to_number.restype = ctypes.c_int 48 lib.arb_to_string.argtypes = [ctypes.c_void_p, ctypes.c_uint32, ctypes.POINTER(ctypes.POINTER(ctypes.c_uint8)), ctypes.POINTER(ctypes.c_size_t)] 49 lib.arb_to_string.restype = ctypes.c_int 50 lib.arb_to_bool.argtypes = [ctypes.c_void_p, ctypes.c_uint32, ctypes.POINTER(ctypes.c_int)] 51 lib.arb_to_bool.restype = ctypes.c_int 52 lib.arboricx_free_buf.argtypes = [ctypes.c_void_p, ctypes.POINTER(ctypes.c_uint8), ctypes.c_size_t] 53 54 # --- Result unwrapping --- 55 lib.arb_unwrap_result.argtypes = [ctypes.c_void_p, ctypes.c_uint32, ctypes.POINTER(ctypes.c_int), ctypes.POINTER(ctypes.c_uint32), ctypes.POINTER(ctypes.c_uint32)] 56 lib.arb_unwrap_result.restype = ctypes.c_int 57 lib.arb_unwrap_host_value.argtypes = [ctypes.c_void_p, ctypes.c_uint32, ctypes.POINTER(ctypes.c_uint64), ctypes.POINTER(ctypes.c_uint32)] 58 lib.arb_unwrap_host_value.restype = ctypes.c_int 59 60 # --- Kernel --- 61 lib.arb_kernel_root.argtypes = [ctypes.c_void_p] 62 lib.arb_kernel_root.restype = ctypes.c_uint32 63 64 # --- Native bundle loading --- 65 lib.arb_load_bundle.argtypes = [ctypes.c_void_p, ctypes.POINTER(ctypes.c_uint8), ctypes.c_size_t, ctypes.c_char_p] 66 lib.arb_load_bundle.restype = ctypes.c_uint32 67 lib.arb_load_bundle_default.argtypes = [ctypes.c_void_p, ctypes.POINTER(ctypes.c_uint8), ctypes.c_size_t] 68 lib.arb_load_bundle_default.restype = ctypes.c_uint32 69 70 71 ctx = lib.arboricx_init() 72 print("ctx init ok") 73 74 fixtures = os.path.join(ZIG_DIR, "..", "..", "test", "fixtures") 75 76 77 def read_bundle(name): 78 path = os.path.join(fixtures, name) 79 with open(path, "rb") as f: 80 return f.read() 81 82 83 def c_bytes(py_bytes): 84 arr = (ctypes.c_uint8 * len(py_bytes))(*py_bytes) 85 return arr 86 87 88 def to_string(ctx, root): 89 ptr = ctypes.POINTER(ctypes.c_uint8)() 90 length = ctypes.c_size_t() 91 if not lib.arb_to_string(ctx, root, ctypes.byref(ptr), ctypes.byref(length)): 92 raise RuntimeError("to_string failed") 93 result = bytes(ptr[i] for i in range(length.value)) 94 lib.arboricx_free_buf(ctx, ptr, length.value) 95 return result.decode("utf-8") 96 97 98 def to_number(ctx, root): 99 out = ctypes.c_uint64() 100 if not lib.arb_to_number(ctx, root, ctypes.byref(out)): 101 raise RuntimeError("to_number failed") 102 return out.value 103 104 105 def to_bool(ctx, root): 106 out = ctypes.c_int() 107 if not lib.arb_to_bool(ctx, root, ctypes.byref(out)): 108 raise RuntimeError("to_bool failed") 109 return bool(out.value) 110 111 112 def kernel_run(bundle_bytes, args): 113 """Run via the Tricu kernel interpreter (slow, ~3s for append).""" 114 buf = c_bytes(bundle_bytes) 115 bundle_tree = lib.arb_of_bytes(ctx, buf, len(bundle_bytes)) 116 tag = lib.arb_of_number(ctx, 1) 117 arg_items = [] 118 for a in args: 119 arg_items.append(lib.arb_of_string(ctx, a.encode("utf-8"))) 120 current = lib.arb_leaf(ctx) 121 for item in reversed(arg_items): 122 current = lib.arb_fork(ctx, item, current) 123 app0 = lib.arb_app(ctx, lib.arb_kernel_root(ctx), tag) 124 app1 = lib.arb_app(ctx, app0, bundle_tree) 125 app2 = lib.arb_app(ctx, app1, current) 126 result = lib.arb_reduce(ctx, app2, 1_000_000_000) 127 ok = ctypes.c_int() 128 value = ctypes.c_uint32() 129 rest = ctypes.c_uint32() 130 if not lib.arb_unwrap_result(ctx, result, ctypes.byref(ok), ctypes.byref(value), ctypes.byref(rest)): 131 raise RuntimeError("unwrap_result failed") 132 tag_num = ctypes.c_uint64() 133 payload = ctypes.c_uint32() 134 if not lib.arb_unwrap_host_value(ctx, value.value, ctypes.byref(tag_num), ctypes.byref(payload)): 135 raise RuntimeError("unwrap_host_value failed") 136 return to_string(ctx, payload.value) 137 138 139 def native_run_default(bundle_bytes, args): 140 """Run via native bundle loader (fast, ~0.01s).""" 141 buf = c_bytes(bundle_bytes) 142 term = lib.arb_load_bundle_default(ctx, buf, len(bundle_bytes)) 143 if term == 0: 144 raise RuntimeError("load_bundle_default failed") 145 current = term 146 for a in args: 147 arg_tree = lib.arb_of_string(ctx, a.encode("utf-8")) 148 current = lib.arb_app(ctx, current, arg_tree) 149 result = lib.arb_reduce(ctx, current, 1_000_000_000) 150 return to_string(ctx, result) 151 152 153 def native_run_named(bundle_bytes, name, args): 154 """Run via native bundle loader with named export (fast).""" 155 buf = c_bytes(bundle_bytes) 156 term = lib.arb_load_bundle(ctx, buf, len(bundle_bytes), name.encode("utf-8")) 157 if term == 0: 158 raise RuntimeError(f"load_bundle({name!r}) failed") 159 current = term 160 for a in args: 161 arg_tree = lib.arb_of_string(ctx, a.encode("utf-8")) 162 current = lib.arb_app(ctx, current, arg_tree) 163 result = lib.arb_reduce(ctx, current, 1_000_000_000) 164 return to_string(ctx, result) 165 166 167 # ============================================================================ 168 # Tests 169 # ============================================================================ 170 171 all_ok = True 172 173 174 def check(label, got, want): 175 global all_ok 176 if got != want: 177 print(f"FAIL {label}: got {got!r}, want {want!r}") 178 all_ok = False 179 else: 180 print(f"PASS {label}: {got!r}") 181 182 183 # Test 1: id via kernel 184 print("\n--- Test 1: id (kernel path) ---") 185 bundle = read_bundle("id.arboricx") 186 t0 = time.time() 187 result = kernel_run(bundle, ["hello"]) 188 t1 = time.time() 189 check("id kernel", result, "hello") 190 print(f" time: {(t1 - t0) * 1000:.1f} ms") 191 192 # Test 2: id via native 193 print("\n--- Test 2: id (native path) ---") 194 t0 = time.time() 195 result = native_run_default(bundle, ["hello"]) 196 t1 = time.time() 197 check("id native", result, "hello") 198 print(f" time: {(t1 - t0) * 1000:.1f} ms") 199 200 # Test 3: append via kernel 201 print("\n--- Test 3: append (kernel path) ---") 202 bundle = read_bundle("append.arboricx") 203 t0 = time.time() 204 result = kernel_run(bundle, ["Hello, ", "world!"]) 205 t1 = time.time() 206 check("append kernel", result, "Hello, world!") 207 print(f" time: {(t1 - t0) * 1000:.1f} ms") 208 209 # Test 4: append via native 210 print("\n--- Test 4: append (native path) ---") 211 t0 = time.time() 212 result = native_run_default(bundle, ["Hello, ", "world!"]) 213 t1 = time.time() 214 check("append native", result, "Hello, world!") 215 print(f" time: {(t1 - t0) * 1000:.1f} ms") 216 217 # Test 5: append via native named export 218 print("\n--- Test 5: append via named export 'root' ---") 219 t0 = time.time() 220 result = native_run_named(bundle, "append", ["Hello, ", "world!"]) 221 t1 = time.time() 222 check("append named", result, "Hello, world!") 223 print(f" time: {(t1 - t0) * 1000:.1f} ms") 224 225 # Test 6: true / false via native 226 print("\n--- Test 6: true / false (native path) ---") 227 for name, expected in [("true.arboricx", True), ("false.arboricx", False)]: 228 bundle = read_bundle(name) 229 buf = c_bytes(bundle) 230 term = lib.arb_load_bundle_default(ctx, buf, len(bundle)) 231 result = lib.arb_reduce(ctx, term, 1_000_000_000) 232 check(f"{name} bool", to_bool(ctx, result), expected) 233 234 # Test 7: number roundtrip 235 print("\n--- Test 7: number roundtrip ---") 236 num_tree = lib.arb_of_number(ctx, 42) 237 check("number 42", to_number(ctx, num_tree), 42) 238 239 # Test 8: string roundtrip 240 print("\n--- Test 8: string roundtrip ---") 241 str_tree = lib.arb_of_string(ctx, b"hello") 242 check("string hello", to_string(ctx, str_tree), "hello") 243 244 lib.arboricx_free(ctx) 245 246 if all_ok: 247 print("\nAll tests passed!") 248 sys.exit(0) 249 else: 250 print("\nSome tests failed!") 251 sys.exit(1)