diff --git a/docs/superpowers/specs/2026-08-24-python-module-resolver-design.md b/docs/superpowers/specs/2026-08-24-python-module-resolver-design.md new file mode 100644 index 00000000..b8622785 --- /dev/null +++ b/docs/superpowers/specs/2026-08-24-python-module-resolver-design.md @@ -0,0 +1,376 @@ +# Python External DataWeave Module Resolver + +**Date:** 2026-08-24 +**Module:** `native-lib` Python binding +**Related implementation:** Node external-module support in PR #154 + +## Problem + +The Python binding cannot resolve reusable DataWeave modules supplied by an +application. Scripts using imports such as `org::company::lib` therefore fail +unless the module is built into `dwlib`. This also leaves Python TCK scenarios +excluded even though the Node binding executes equivalent scenarios through its +module resolver. + +`dwlib` already exports `run_script_with_resolver`, and its callback ABI is +host-neutral. The Python binding can use that existing export directly through +`ctypes`; this feature does not require Java, native-image, or Node binding +changes. + +## Goals + +1. Give the Python synchronous `DataWeave.run()` API external-module parity + with Node. +2. Support a user-provided synchronous resolver callable. +3. Provide Python equivalents of Node's map, directory, JAR, and composition + resolver factories. +4. Re-enable Python TCK scenarios that become runnable through the same + committed module fixture used by Node. +5. Preserve the existing native ABI and leave every file under + `native-lib/node` unchanged. + +## Non-goals + +- Resolver support for `run_streaming`, `run_transform`, or callback streaming. +- Configuring the module-level `dataweave.run()` singleton with a resolver. +- Running the 17 structurally skipped TCK cases that bundle adjacent `.dwl` + files beside `transform.dwl`. +- Changing the native resolver model. Each Python `DataWeave` keeps its current + dedicated Graal isolate; the first resolver-backed run in that isolate + installs its configured resolver. +- Changes to Java, GraalVM entry points, generated headers, or Node sources. + +## Public API + +Add `dataweave/resolver.py` with snake_case Python APIs: + +```python +from collections.abc import Callable, Mapping, Sequence +from pathlib import Path +from typing import Optional, Union + +ModuleResolver = Callable[[str], Optional[str]] + +def modules_from_map(modules: Mapping[str, str]) -> ModuleResolver: ... +def modules_from_directory(base_dir: Union[str, Path]) -> ModuleResolver: ... +def modules_from_jars(jar_paths: Sequence[Union[str, Path]]) -> ModuleResolver: ... +def compose_resolvers(*resolvers: ModuleResolver) -> ModuleResolver: ... +``` + +Extend explicit runtime construction: + +```python +class DataWeave: + def __init__( + self, + lib_path: Optional[str] = None, + *, + resolve_module: Optional[ModuleResolver] = None, + ): ... +``` + +Example: + +```python +from dataweave import DataWeave, modules_from_map + +resolver = modules_from_map({ + "org/company/lib.dwl": '%dw 2.0\nfun greet(name) = "Hello " ++ name', +}) + +with DataWeave(resolve_module=resolver) as dw: + result = dw.run(""" + %dw 2.0 + import org::company::lib + output application/json + --- + lib::greet("World") + """) +``` + +Export `ModuleResolver`, all four factories, and the existing public symbols +from `dataweave.__init__`. + +The module-level convenience functions remain unchanged. Like Node, callers +must construct a `DataWeave` instance to provide `resolve_module`. + +## Resolver Factories + +### `modules_from_map` + +Copy the input mapping at factory construction and perform exact path lookup. +Return the mapped source or `None`. The copy prevents later caller mutation +from changing resolver behavior unexpectedly. + +### `modules_from_directory` + +Capture both the absolute lexical root and canonical root when constructing the +resolver. Fail construction if the base directory does not exist. + +For every lookup: + +1. Resolve the requested module path beneath the captured lexical root. +2. Reject `..`, absolute-path, and different-root escapes before filesystem I/O. +3. Canonicalize the candidate and return `None` when it does not exist. +4. Reject symlinks whose canonical target escapes the canonical root. +5. Read the module as UTF-8 on each lookup. + +This follows Node's security and current-working-directory stability behavior. +Missing files return `None`; permission, directory, decoding, and other I/O +failures raise a contextual exception. + +### `modules_from_jars` + +Use the Python standard library's `zipfile` module, so no runtime dependency is +added. Read every non-directory `.dwl` entry into an in-memory map and return a +map-backed synchronous resolver. Process JARs in caller order; a later archive +overwrites a duplicate path from an earlier archive, matching Node behavior. +Malformed or unreadable archives raise an exception naming the archive. + +Unlike Node, this factory itself is synchronous because Python's standard ZIP +API is synchronous. The returned resolver has the same synchronous contract. + +### `compose_resolvers` + +Call resolvers in order and return the first non-`None` source. Return `None` +when none resolve the path. Resolver exceptions propagate to the ctypes bridge, +where they follow the callback error policy below. + +## Execution Flow + +```mermaid +flowchart TD + A[Explicit DataWeave instance] --> B{resolve_module configured?} + B -->|no| C[run_script] + B -->|yes| D[run_script_with_resolver] + D --> E[Existing dwlib resolver callback] + E --> F[Python ModuleResolver] + F -->|source| G[UTF-8 callback buffer] + F -->|None or error| H[NULL] + G --> E + H --> E +``` + +`DataWeave.run()` keeps its current input encoding, result parsing, +`raise_on_error`, and exception-wrapping behavior. It selects only the native +entry point: + +- no configured resolver: `run_script`; +- configured resolver: `run_script_with_resolver`. + +Streaming methods deliberately remain on their resolver-less native entry +points and therefore have access only to built-in modules. + +## ctypes ABI Bridge + +Define a Python callback type matching the existing ABI: + +```c +char *resolve_module(void *isolate_thread, const char *module_path); +``` + +Configure `run_script_with_resolver` with these arguments: + +1. `GraalIsolateThreadPointer` +2. script `c_char_p` +3. inputs JSON `c_char_p` +4. resolver callback + +Its result remains an unmanaged C string decoded and released through the +existing `decode_and_free` path. + +The callback bridge: + +1. Decodes `module_path` as UTF-8. +2. Removes exactly one leading `/`, matching the Node adapter and public + separator-less keys. +3. Invokes the configured Python resolver synchronously. +4. Accepts only `str` or `None`. +5. Encodes a returned string as UTF-8 into a `ctypes` buffer. +6. Keeps every returned buffer strongly referenced until the enclosing + `run_script_with_resolver` call returns. +7. Clears those references in `finally`, after Java has copied callback output + into a managed string. + +The `ctypes` callback object itself is retained by the `DataWeave` instance +until its isolate is torn down. No Python exception is allowed to unwind +through the C callback. + +## Capability Detection + +`NativeRuntime._setup_functions()` detects `run_script_with_resolver` and +configures its signature when present. It records a capability flag analogous +to the existing streaming flags. + +Constructing or initializing a `DataWeave` with a resolver remains possible so +normal lifecycle behavior is unchanged. The first resolver-backed `run()` +against a library lacking the export raises `DataWeaveError` with operation +context and the missing symbol name. Resolver-less execution remains compatible +with older libraries. + +## Error and Security Policy + +Resolver outcomes are translated as follows: + +| Outcome | Callback result | Visible behavior | +|---|---|---| +| Source string | UTF-8 C pointer | Module compiles normally | +| `None` | `NULL` | DataWeave module-not-found result | +| Non-string value | `NULL` | Invalid resolver result treated as not found | +| Path UTF-8 decode failure | `NULL` | Module not found | +| Resolver exception | `NULL` | Module not found | + +By default, callback errors write only a fixed, content-free diagnostic to +stderr. Resolver exceptions may contain module source, credentials, or local +paths, so their details must not be logged automatically. When +`DATAWEAVE_RESOLVER_DEBUG=1`, the bridge may include exception type, message, +and traceback for trusted debugging environments. + +The resolver executes arbitrary user Python with full process permissions. +Documentation must instruct callers to use only trusted resolver functions and +trusted module sources. + +## Resolver Lifetime and Isolation + +`ScriptRuntime.setResolver` accepts only the first resolver in a Graal isolate. +The Python binding differs from Node here: each explicit Python `DataWeave` +currently creates and owns a dedicated isolate, while Node shares one isolate +across its process-level native addon. Therefore: + +- the first resolver-backed `run()` on a Python instance installs that + instance's configured resolver in its isolate; +- `initialize()` does not invoke or install the resolver; +- repeated runs on the same instance reuse the installed resolver; +- two live Python `DataWeave` instances may use different resolvers because + they own different isolates; +- `cleanup()` tears down the isolate before releasing that instance's callback + and resolver references. + +The callback and resolver must remain strongly reachable from the instance +until isolate teardown completes. This prevents the native isolate from +retaining a dangling Python function pointer. `compose_resolvers()` remains the +recommended way to build fallback resolution within one instance, not a +workaround for a process-wide Python limitation. + +## TCK Integration + +The session-scoped Python TCK runtime uses: + +```python +DataWeave(resolve_module=modules_from_directory(shared_fixture_directory)) +``` + +The directory is the existing committed fixture used by Node: +`native-lib/node/tests/tck/fixtures`. Referencing that fixture from Python test +configuration does not require changing any Node source or fixture file. + +After adding resolver support: + +1. Run the complete staged Python TCK. +2. Remove only exclusions proven to pass through the shared fixture resolver. +3. Keep genuine module, Java, classpath-resource, and binding limitations with + direct evidence. +4. Update exclusion counts and summary assertions from observed outcomes. +5. Keep all 17 adjacent-DWL cases as structural skips. The Python loader and + Node loader continue to apply the same transform-shape rule. + +The accounting invariant remains: + +```text +passed + failed + active-exclusions + xfail = selected +unaccounted = 0 +``` + +## Files + +| File | Change | +|---|---| +| `native-lib/python/src/dataweave/resolver.py` | New public resolver type and factories | +| `native-lib/python/src/dataweave/models.py` | Add the ctypes resolver callback signature beside existing callback types | +| `native-lib/python/src/dataweave/native.py` | Detect and call existing resolver-aware ABI; callback ownership bridge | +| `native-lib/python/src/dataweave/runtime.py` | Store resolver and select resolver-aware `run()` path | +| `native-lib/python/src/dataweave/__init__.py` | Export resolver APIs | +| `native-lib/python/tests/unit/test_resolver.py` | Resolver factory tests | +| `native-lib/python/tests/unit/test_native.py` | ABI capability and callback bridge tests | +| `native-lib/python/tests/unit/test_facade.py` | Constructor and dispatch behavior tests | +| `native-lib/python/tests/integration/test_module_resolver.py` | Real native module-resolution tests | +| `native-lib/python/tests/conftest.py` | Configure TCK runtime with shared fixture resolver | +| `native-lib/python/tests/tck/ignore_list.py` | Remove empirically recovered exclusions | +| `native-lib/python/tests/tck/test_conformance.py` | Update policy totals/assertions from observed results | +| `native-lib/python/README.md` | Public API, limitations, security, and examples | + +No file under `native-lib/node` is modified. + +## Testing + +### Unit + +- map lookup, defensive copy, exact-key behavior, and missing path; +- directory lookup, stable root after `chdir`, lexical traversal rejection, + symlink escape rejection, missing file, permissions, and invalid UTF-8; +- JAR extraction, nested paths, ignored non-DWL entries, duplicate precedence, + and malformed archives; +- resolver composition order and fallback; +- leading-slash normalization; +- source-buffer lifetime through the native call; +- `None`, invalid return, decode failure, and resolver exception handling; +- content-free default diagnostics and debug opt-in; +- missing `run_script_with_resolver` capability; +- resolver-less versus resolver-aware `DataWeave.run()` dispatch; +- module-level convenience API remains resolver-less. + +### Native Integration + +- resolve a module from a map; +- resolve a module from a directory; +- resolve a module from a JAR; +- resolve a transitive module import; +- return a normal unsuccessful result for a missing module; +- verify `raise_on_error=True` promotes module compilation failure; +- verify repeated runs on one instance reuse its resolver; +- verify two simultaneous instances resolve against different module maps; +- verify streaming APIs do not invoke the custom resolver; +- clean up one resolver-backed instance without invalidating another isolate's + resolver. + +### TCK and Packaging + +- `./gradlew native-lib:pythonTest`; +- `./gradlew native-lib:pythonTck` after staging the corpus; +- `./gradlew native-lib:test -PskipNodeTests=true -PskipPythonTests=true` for + the existing resolver ABI tests; +- `./gradlew native-lib:buildPythonWheel` and install/import smoke testing; +- platform CI on macOS, Linux, and Windows. + +## Risks and Mitigations + +### Dangling ctypes callback + +The native isolate retains the callback after the originating run returns. +Retain the callback and resolver on the owning `DataWeave` instance until +isolate teardown finishes, and cover simultaneous instances plus independent +cleanup with native integration tests. + +### Callback result lifetime + +Java copies the source immediately, but returning temporary Python bytes would +leave an invalid pointer. Use explicit `ctypes` buffers retained through the +entire native call and clear them afterward. + +### Python callback concurrency + +`ctypes` acquires the GIL before invoking Python callbacks. The resolver itself +must remain synchronous. The design does not add resolver support to background +streaming workers, avoiding new cross-thread callback behavior. + +### Filesystem escape + +Directory resolvers could otherwise expose arbitrary files through `..` or +symlink traversal. Apply both lexical and canonical containment checks before +reading a module. + +### TCK overclaiming + +Do not remove exclusions based solely on their category. Re-enable only cases +that pass the full Python TCK with the shared fixture resolver, and preserve the +strict accounting gate. diff --git a/native-lib/python/README.md b/native-lib/python/README.md index 09e3dfd2..1b4294c7 100644 --- a/native-lib/python/README.md +++ b/native-lib/python/README.md @@ -123,6 +123,93 @@ with dataweave.DataWeave() as dw: print(r2.get_string()) # "42" ``` +### External DataWeave Modules + +Custom module resolution is available to synchronous `DataWeave.run()` calls on +an explicit `DataWeave` instance. Resolver keys use `/` separators and include +the `.dwl` suffix; for example, the DataWeave import `org::company::lib` requests +the module key `org/company/lib.dwl` without a leading path separator. + +```python +from dataweave import DataWeave, modules_from_map + +resolver = modules_from_map({ + "org/company/lib.dwl": '%dw 2.0\nfun greet(name) = "Hello " ++ name', +}) + +with DataWeave(resolve_module=resolver) as dw: + result = dw.run(""" + %dw 2.0 + import org::company::lib + output application/json + --- + lib::greet("World") + """) + +assert result.get_string() == '"Hello World"' +``` + +The `ModuleResolver` contract is a synchronous callable from a module key to +the module source string or `None`. Custom resolver configuration is available +only on an explicit `DataWeave` instance; the module-level `dataweave.run()` +singleton does not accept `resolve_module`. `run_streaming()`, +`run_transform()`, and the low-level callback streaming API do not use custom +resolvers and can import only built-in modules. + +Use `modules_from_directory()` for a source tree: + +```python +from dataweave import DataWeave, modules_from_directory + +with DataWeave(resolve_module=modules_from_directory("modules")) as dw: + result = dw.run(script) +``` + +Use `modules_from_jars()` to load `.dwl` entries from one or more JAR or ZIP +archives. Later archives replace duplicate module keys from earlier archives: + +```python +from dataweave import DataWeave, modules_from_jars + +resolver = modules_from_jars(["base-modules.jar", "application-modules.jar"]) +with DataWeave(resolve_module=resolver) as dw: + result = dw.run(script) +``` + +Use `compose_resolvers()` for ordered fallback. The first resolver returning a +source wins: + +```python +from dataweave import ( + DataWeave, + compose_resolvers, + modules_from_directory, + modules_from_jars, + modules_from_map, +) + +resolver = compose_resolvers( + modules_from_map({"org/company/config.dwl": config_source}), + modules_from_directory("modules"), + modules_from_jars(["dependencies.jar"]), +) +with DataWeave(resolve_module=resolver) as dw: + result = dw.run(script) +``` + +Resolvers execute arbitrary Python with the application's process permissions. +Use only trusted resolver functions and trusted module sources. By default, +callback failures write a fixed, content-free diagnostic to stderr so module +source, credentials, and local paths are not exposed. Set +`DATAWEAVE_RESOLVER_DEBUG=1` only in a trusted debugging environment to include +the exception type, message, and traceback. + +Each initialized explicit Python `DataWeave` instance owns a dedicated Graal +isolate. Its first resolver-backed run installs that instance's resolver; later +runs reuse it. The instance retains the resolver callback until successful +isolate teardown, then releases callback references during `cleanup()`. +Different live instances can therefore use different resolvers. + ### Error Handling **Option A: Use `raise_on_error=True` (recommended)** @@ -296,13 +383,22 @@ To stage and run the Python conformance suite, use: ``` `pythonTck` is intentionally separate from normal testing and runs only in the -master-only CI lane. It reuses the corpus staged for Node TCK. It excludes -only binding/environment capability gaps, such as unavailable module resolution, -Java modules, and classpath test resources. Accepted runtime/output baseline -mismatches are strict xfails: a new mismatch fails the lane and a repaired -baseline mismatch XPASSes and also fails. The deferred-writer TCK scenario runs -in a subprocess because that runtime's isolate teardown may block; the main TCK -session runtime is always cleaned up. +master-only CI lane. It reuses the corpus and shared module fixture staged for +Node TCK. The fixture resolver recovered six import scenarios. The +`runtime/module-singleton-out.json` exclusion remains because the shared fixture +does not contain its three singleton modules. All 17 cases that bundle adjacent +`.dwl` files beside `transform.dwl` remain structural skips rather than active +exclusions. + +The observed conformance accounting is 729 selected scenarios, 193 structural +skips, 17 structural module cases, 679 executed passes, 31 active exclusions, +19 strict xfails, 0 failures, and 0 unaccounted scenarios. Active exclusions +cover only directly observed binding or environment capability gaps, including +unavailable modules, Java modules, and classpath test resources. Accepted +runtime/output baseline mismatches are strict xfails: a new mismatch fails the +lane and a repaired baseline mismatch XPASSes and also fails. The deferred-writer +TCK scenario runs in a subprocess because that runtime's isolate teardown may +block; the main TCK session runtime is always cleaned up. ## Running Examples @@ -352,9 +448,10 @@ Low-level callback API for advanced use cases. ### `DataWeave` Class -#### `DataWeave(lib_path=None)` +#### `DataWeave(lib_path=None, *, resolve_module=None)` -Context manager for explicit lifecycle control. +Context manager for explicit lifecycle control. `resolve_module` accepts a +synchronous `ModuleResolver` for `run()` calls. **Methods:** - `run(...)` - Same as module-level `run()` @@ -368,6 +465,17 @@ with DataWeave() as dw: result = dw.run("2 + 2") ``` +### Module Resolvers + +- `ModuleResolver` - Synchronous callable receiving a module key and returning + source text or `None` +- `modules_from_map` - Copy a mapping and resolve exact module keys +- `modules_from_directory` - Resolve UTF-8 `.dwl` files beneath a + traversal-protected directory root +- `modules_from_jars` - Load `.dwl` entries from JAR or ZIP archives + in caller order +- `compose_resolvers` - Return the first non-`None` resolver result + ### `ExecutionResult` ```python @@ -522,6 +630,8 @@ if not stream.metadata.success: - `DW_HOME` - DataWeave home directory (default: `~/.dw`) - `DW_DEFAULT_INPUT_MIMETYPE` - Default input MIME type (default: `application/json`) - `DW_DEFAULT_OUTPUT_MIMETYPE` - Default output MIME type (default: `application/json`) +- `DATAWEAVE_RESOLVER_DEBUG` - Set to `1` to include resolver exception details + in callback diagnostics; use only in trusted debugging environments ## See Also diff --git a/native-lib/python/src/dataweave/__init__.py b/native-lib/python/src/dataweave/__init__.py index f1db4641..a446621a 100644 --- a/native-lib/python/src/dataweave/__init__.py +++ b/native-lib/python/src/dataweave/__init__.py @@ -9,6 +9,7 @@ from .encoding import parse_streaming_result as _parse_streaming_result from .models import ( READ_CALLBACK, + RESOLVE_MODULE_CALLBACK, WRITE_CALLBACK, DataWeaveError, DataWeaveLibraryNotFoundError, @@ -22,6 +23,13 @@ ) from .native import candidate_library_paths as _candidate_library_paths from .native import find_library as _find_library +from .resolver import ( + ModuleResolver, + compose_resolvers, + modules_from_directory, + modules_from_jars, + modules_from_map, +) from .runtime import DataWeave @@ -68,6 +76,7 @@ def cleanup() -> None: __all__ = [ "DataWeave", "DataWeaveError", "DataWeaveLibraryNotFoundError", "DataWeaveScriptError", "ExecutionResult", "InputValue", "ReadCallback", "Stream", "StreamingResult", "WriteCallback", - "READ_CALLBACK", "WRITE_CALLBACK", "run", "run_callback", "run_input_output_callback", - "run_streaming", "run_transform", "cleanup", + "READ_CALLBACK", "RESOLVE_MODULE_CALLBACK", "WRITE_CALLBACK", "run", "run_callback", "run_input_output_callback", + "run_streaming", "run_transform", "cleanup", "ModuleResolver", "compose_resolvers", + "modules_from_directory", "modules_from_jars", "modules_from_map", ] diff --git a/native-lib/python/src/dataweave/models.py b/native-lib/python/src/dataweave/models.py index ae33ceac..d5e90c9a 100644 --- a/native-lib/python/src/dataweave/models.py +++ b/native-lib/python/src/dataweave/models.py @@ -30,6 +30,12 @@ class DataWeaveLibraryNotFoundError(Exception): WRITE_CALLBACK = ctypes.CFUNCTYPE(ctypes.c_int, ctypes.c_void_p, ctypes.c_void_p, ctypes.c_int) # int (*ReadCallback)(void *ctx, char *buffer, int bufferSize) READ_CALLBACK = ctypes.CFUNCTYPE(ctypes.c_int, ctypes.c_void_p, ctypes.c_void_p, ctypes.c_int) +# char *resolve_module(void *isolate_thread, const char *module_path) +RESOLVE_MODULE_CALLBACK = ctypes.CFUNCTYPE( + ctypes.c_void_p, + ctypes.c_void_p, + ctypes.c_char_p, +) WriteCallback = Callable[[bytes], int] diff --git a/native-lib/python/src/dataweave/native.py b/native-lib/python/src/dataweave/native.py index 98f771a1..d9d010d8 100644 --- a/native-lib/python/src/dataweave/native.py +++ b/native-lib/python/src/dataweave/native.py @@ -1,9 +1,14 @@ import ctypes +from contextlib import contextmanager import os from pathlib import Path +import sys +from threading import current_thread, get_ident, Lock +import traceback from typing import Optional -from .models import DataWeaveError, DataWeaveLibraryNotFoundError, READ_CALLBACK, WRITE_CALLBACK +from .models import DataWeaveError, DataWeaveLibraryNotFoundError, READ_CALLBACK, RESOLVE_MODULE_CALLBACK, WRITE_CALLBACK +from .resolver import ModuleResolver _ENV_NATIVE_LIB = "DATAWEAVE_NATIVE_LIB" @@ -62,6 +67,14 @@ def __init__(self, lib_path: Optional[str] = None): self.initialized = False self.has_callback_streaming = False self.has_callback_input_output = False + self.has_module_resolver = False + self._module_resolver = None + self._module_resolver_callback = None + self._resolver_buffers = [] + self._resolver_active = False + self._resolver_lock = Lock() + self._execution_owner = None + self._owner_thread = None def initialize(self) -> None: if self.initialized: @@ -74,6 +87,7 @@ def initialize(self) -> None: try: self._create_isolate() isolate_created = True + self._owner_thread = current_thread() self._setup_functions() self.initialized = True except Exception: @@ -109,6 +123,16 @@ def _setup_functions(self) -> None: self.lib.free_cstring.restype = None self.lib.graal_tear_down_isolate.argtypes = [GraalIsolateThreadPointer] self.lib.graal_tear_down_isolate.restype = ctypes.c_int + self._setup_thread_lifecycle_functions() + if hasattr(self.lib, "run_script_with_resolver"): + self.lib.run_script_with_resolver.argtypes = [ + GraalIsolateThreadPointer, + ctypes.c_char_p, + ctypes.c_char_p, + RESOLVE_MODULE_CALLBACK, + ] + self.lib.run_script_with_resolver.restype = ctypes.c_void_p + self.has_module_resolver = True if hasattr(self.lib, "run_script_callback"): self._require_streaming_lifecycle_exports("run_script_callback") self.lib.run_script_callback.argtypes = [GraalIsolateThreadPointer, ctypes.c_char_p, ctypes.c_char_p, WRITE_CALLBACK, ctypes.c_void_p] @@ -128,6 +152,10 @@ def _require_streaming_lifecycle_exports(self, callback_name: str) -> None: for name in ("free_cstring", "graal_attach_thread", "graal_detach_thread"): if not hasattr(self.lib, name): raise DataWeaveError(f"{callback_name} requires native export {name}") + + def _setup_thread_lifecycle_functions(self) -> None: + self._require_export("graal_attach_thread") + self._require_export("graal_detach_thread") self.lib.graal_attach_thread.argtypes = [GraalIsolatePointer, ctypes.POINTER(GraalIsolateThreadPointer)] self.lib.graal_attach_thread.restype = ctypes.c_int self.lib.graal_detach_thread.argtypes = [GraalIsolateThreadPointer] @@ -169,29 +197,166 @@ def decode_and_free(self, ptr, thread=None) -> str: raise def run_script(self, thread, script: bytes, inputs: bytes): - return self.lib.run_script(thread, script, inputs) + with self._serialized_native_operation(): + return self.lib.run_script(thread, script, inputs) + + def run_script_and_decode(self, thread, script: bytes, inputs: bytes) -> str: + with self._serialized_native_operation(): + with self._current_thread_attachment(thread) as current_thread: + return self.decode_and_free( + self.lib.run_script(current_thread, script, inputs), + current_thread, + ) + + def run_script_with_resolver(self, thread, script: bytes, inputs: bytes, resolver: ModuleResolver): + with self._serialized_native_operation(): + return self._run_script_with_resolver(thread, script, inputs, resolver) + + def run_script_with_resolver_and_decode(self, thread, script: bytes, inputs: bytes, resolver: ModuleResolver) -> str: + with self._serialized_native_operation(): + with self._current_thread_attachment(thread) as current_thread: + return self.decode_and_free( + self._run_script_with_resolver(current_thread, script, inputs, resolver), + current_thread, + ) + + def _run_script_with_resolver(self, thread, script: bytes, inputs: bytes, resolver: ModuleResolver): + if not self.has_module_resolver: + raise DataWeaveError( + "Native library does not support module resolver API " + "(run_script_with_resolver not found)." + ) + if self._module_resolver is None: + self._module_resolver = resolver + self._module_resolver_callback = self._create_module_resolver_callback(resolver) + elif self._module_resolver is not resolver: + raise DataWeaveError("Native runtime already has a different module resolver") + + self._resolver_buffers.clear() + self._resolver_active = True + try: + return self.lib.run_script_with_resolver( + thread, script, inputs, self._module_resolver_callback + ) + finally: + self._resolver_active = False + self._resolver_buffers.clear() + + def _create_module_resolver_callback(self, resolver: ModuleResolver): + def resolve(_thread, module_path): + try: + if not self._resolver_active: + return None + path = module_path.decode("utf-8") + if path.startswith("/"): + path = path[1:] + source = resolver(path) + if not isinstance(source, str): + return None + buffer = ctypes.create_string_buffer(source.encode("utf-8")) + self._resolver_buffers.append(buffer) + return ctypes.addressof(buffer) + except BaseException: + try: + if os.environ.get("DATAWEAVE_RESOLVER_DEBUG") == "1": + traceback.print_exc() + else: + print("DataWeave module resolver callback failed.", file=sys.stderr) + except BaseException: + pass + return None + + return RESOLVE_MODULE_CALLBACK(resolve) def run_script_callback(self, thread, script: bytes, inputs: bytes, write_callback): - return self.lib.run_script_callback(thread, script, inputs, write_callback, None) + with self._serialized_native_operation(): + return self.lib.run_script_callback(thread, script, inputs, write_callback, None) + + def run_script_callback_and_decode(self, thread, script: bytes, inputs: bytes, write_callback) -> str: + with self._serialized_native_operation(): + with self._current_thread_attachment(thread) as current_thread: + return self.decode_and_free( + self.lib.run_script_callback(current_thread, script, inputs, write_callback, None), + current_thread, + ) def run_script_input_output_callback(self, thread, script: bytes, inputs: bytes, input_name: bytes, input_mime_type: bytes, input_charset: Optional[bytes], read_callback, write_callback): - return self.lib.run_script_input_output_callback( - thread, script, inputs, input_name, input_mime_type, input_charset, read_callback, write_callback, None, - ) + with self._serialized_native_operation(): + return self.lib.run_script_input_output_callback( + thread, script, inputs, input_name, input_mime_type, input_charset, read_callback, write_callback, None, + ) + + def run_script_input_output_callback_and_decode(self, thread, script: bytes, inputs: bytes, input_name: bytes, input_mime_type: bytes, input_charset: Optional[bytes], read_callback, write_callback) -> str: + with self._serialized_native_operation(): + with self._current_thread_attachment(thread) as current_thread: + return self.decode_and_free( + self.lib.run_script_input_output_callback( + current_thread, script, inputs, input_name, input_mime_type, input_charset, read_callback, write_callback, None, + ), + current_thread, + ) def cleanup(self) -> None: - if not self.initialized: + with self._serialized_native_operation(): + if not self.initialized: + return + if current_thread() is getattr(self, "_owner_thread", current_thread()): + self._tear_down_isolate() + self._reset() + return + + attached_thread = self.attach_thread() + try: + self._tear_down_isolate(attached_thread) + except Exception: + try: + self.detach_thread(attached_thread) + except Exception: + pass + raise + self._reset() + + @contextmanager + def _serialized_native_operation(self): + owner = get_ident() + if getattr(self, "_execution_owner", None) == owner: + raise DataWeaveError("Reentrant DataWeave execution is not supported.") + if not hasattr(self, "_resolver_lock"): + self._resolver_lock = Lock() + with self._resolver_lock: + self._execution_owner = owner + try: + yield + finally: + self._execution_owner = None + + @contextmanager + def _current_thread_attachment(self, thread): + owner = getattr(self, "_owner_thread", current_thread()) + if current_thread() is owner or thread is not self.thread: + yield thread return + + attached_thread = self.attach_thread() + primary_error = None try: - self._tear_down_isolate() + yield attached_thread + except BaseException as error: + primary_error = error + raise finally: - self._reset() + try: + self.detach_thread(attached_thread) + except Exception: + if primary_error is None: + raise - def _tear_down_isolate(self, suppress_errors: bool = False) -> None: - if self.thread is None: + def _tear_down_isolate(self, thread=None, suppress_errors: bool = False) -> None: + isolate_thread = thread or self.thread + if isolate_thread is None: return try: - result = self.lib.graal_tear_down_isolate(self.thread) + result = self.lib.graal_tear_down_isolate(isolate_thread) if result != 0: raise DataWeaveError(f"Failed to tear down GraalVM isolate. Error code: {result}") except DataWeaveError: @@ -203,8 +368,14 @@ def _tear_down_isolate(self, suppress_errors: bool = False) -> None: def _reset(self) -> None: self.initialized = False + self._owner_thread = None self.thread = None self.isolate = None self.lib = None self.has_callback_streaming = False self.has_callback_input_output = False + self.has_module_resolver = False + self._module_resolver = None + self._module_resolver_callback = None + self._resolver_buffers = [] + self._resolver_active = False diff --git a/native-lib/python/src/dataweave/resolver.py b/native-lib/python/src/dataweave/resolver.py new file mode 100644 index 00000000..0d74418b --- /dev/null +++ b/native-lib/python/src/dataweave/resolver.py @@ -0,0 +1,112 @@ +"""Synchronous external DataWeave module resolver factories.""" + +from collections.abc import Callable, Mapping, Sequence +import os +from pathlib import Path +from typing import Optional, Union +from zipfile import BadZipFile, ZipFile + + +ModuleResolver = Callable[[str], Optional[str]] + + +def modules_from_map(modules: Mapping[str, str]) -> ModuleResolver: + """Create a resolver backed by an immutable snapshot of module sources.""" + copied_modules = dict(modules) + + def resolve(module_path: str) -> Optional[str]: + if module_path in copied_modules: + return copied_modules.get(module_path) + return None + + return resolve + + +def modules_from_directory(base_dir: Union[str, Path]) -> ModuleResolver: + """Create a resolver that reads modules beneath a directory.""" + lexical_root = Path(os.path.abspath(base_dir)) + canonical_root = lexical_root.resolve(strict=True) + + if not canonical_root.is_dir(): + raise NotADirectoryError(f"Module root is not a directory: {base_dir}") + + def resolve(module_path: str) -> Optional[str]: + requested_path = Path(module_path) + if requested_path.is_absolute(): + return None + + candidate = Path(os.path.abspath(lexical_root / requested_path)) + try: + candidate.relative_to(lexical_root) + except ValueError: + return None + + try: + canonical_candidate = candidate.resolve(strict=True) + except FileNotFoundError: + return None + except OSError as error: + raise OSError( + f"Failed to resolve DataWeave module {module_path!r}: {error}" + ) from error + + try: + canonical_candidate.relative_to(canonical_root) + except ValueError: + return None + + try: + return canonical_candidate.read_text(encoding="utf-8") + except UnicodeDecodeError as error: + raise UnicodeDecodeError( + error.encoding, + error.object, + error.start, + error.end, + f"{error.reason} while reading DataWeave module {module_path!r}", + ) from error + except OSError as error: + raise OSError( + f"Failed to read DataWeave module {module_path!r}: {error}" + ) from error + + return resolve + + +def modules_from_jars( + jar_paths: Sequence[Union[str, Path]], +) -> ModuleResolver: + """Create a resolver from DataWeave modules stored in JAR files.""" + modules = {} + for jar_path in jar_paths: + try: + with ZipFile(jar_path) as jar: + for entry in jar.infolist(): + if entry.is_dir() or not entry.filename.endswith(".dwl"): + continue + modules[entry.filename] = jar.read(entry).decode("utf-8") + except ( + OSError, + BadZipFile, + UnicodeDecodeError, + RuntimeError, + NotImplementedError, + ) as error: + raise ValueError( + f"Failed to load DataWeave modules from JAR {jar_path!s}: {error}" + ) from error + + return modules_from_map(modules) + + +def compose_resolvers(*resolvers: ModuleResolver) -> ModuleResolver: + """Create a resolver that returns the first resolved module source.""" + + def resolve(module_path: str) -> Optional[str]: + for resolver in resolvers: + source = resolver(module_path) + if source is not None: + return source + return None + + return resolve diff --git a/native-lib/python/src/dataweave/runtime.py b/native-lib/python/src/dataweave/runtime.py index 8d57d918..cb66b1a1 100644 --- a/native-lib/python/src/dataweave/runtime.py +++ b/native-lib/python/src/dataweave/runtime.py @@ -17,6 +17,7 @@ WriteCallback, ) from .native import NativeRuntime +from .resolver import ModuleResolver _OUTPUT_QUEUE_MAXSIZE = 512 @@ -27,8 +28,14 @@ class DataWeave: """High-level execution API backed by a :class:`NativeRuntime`.""" - def __init__(self, lib_path: Optional[str] = None): + def __init__( + self, + lib_path: Optional[str] = None, + *, + resolve_module: Optional[ModuleResolver] = None, + ): self._native = NativeRuntime(lib_path) + self._resolve_module = resolve_module self._stream_workers = set() self._stream_workers_lock = Lock() self._cleaning_up = False @@ -79,7 +86,17 @@ def _inputs_json(inputs: Optional[Dict[str, Any]]) -> bytes: def run(self, script: str, inputs: Optional[Dict[str, Any]] = None, raise_on_error: bool = False) -> ExecutionResult: self._require_initialized(True, "script execution") try: - raw = self._native.decode_and_free(self._native.run_script(self._native.thread, script.encode("utf-8"), self._inputs_json(inputs))) + encoded_script = script.encode("utf-8") + encoded_inputs = self._inputs_json(inputs) + if self._resolve_module is None: + raw = self._native.run_script_and_decode(self._native.thread, encoded_script, encoded_inputs) + else: + raw = self._native.run_script_with_resolver_and_decode( + self._native.thread, + encoded_script, + encoded_inputs, + self._resolve_module, + ) result = parse_native_encoded_response(raw) except Exception as error: raise DataWeaveError(f"Failed to execute script: {error}") @@ -96,8 +113,7 @@ def write_cb(_context, buffer, length): except Exception: return -1 try: - ptr = self._native.run_script_callback(self._native.thread, script.encode("utf-8"), self._inputs_json(inputs), write_cb) - raw = self._native.decode_and_free(ptr) + raw = self._native.run_script_callback_and_decode(self._native.thread, script.encode("utf-8"), self._inputs_json(inputs), write_cb) return parse_streaming_result(json.loads(raw) if raw else {"success": False, "error": "Empty response"}) except Exception as error: raise DataWeaveError(f"Failed to execute callback streaming: {error}") @@ -134,7 +150,7 @@ def worker_main(): primary_outcome = False try: worker_thread = self._native.attach_thread() - raw = self._native.decode_and_free(invoke(worker_thread, write_cb), worker_thread) + raw = invoke(worker_thread, write_cb) metadata = json.loads(raw) if raw else {"success": False, "error": "Empty response"} primary_outcome = not metadata.get("success", False) publish(metadata) @@ -186,7 +202,7 @@ def run_streaming(self, script: str, inputs: Optional[Dict[str, Any]] = None) -> self._require_initialized(self._native.has_callback_streaming, "callback streaming API (run_script_callback not found)") cancelled = Event() encoded_inputs = self._inputs_json(inputs) - stream = Stream(self._stream_worker(lambda thread, write_cb: self._native.run_script_callback(thread, script.encode("utf-8"), encoded_inputs, write_cb), cancelled)) + stream = Stream(self._stream_worker(lambda thread, write_cb: self._native.run_script_callback_and_decode(thread, script.encode("utf-8"), encoded_inputs, write_cb), cancelled)) stream._on_close = cancelled.set stream._cancelled = cancelled return stream @@ -223,7 +239,7 @@ def run_transform(self, script: str, input_stream: Iterable[bytes], input_name: read_cb = self._chunk_reader(input_stream) encoded_inputs = self._inputs_json(inputs) def invoke(thread, write_cb): - return self._native.run_script_input_output_callback(thread, script.encode("utf-8"), encoded_inputs, input_name.encode("utf-8"), input_mime_type.encode("utf-8"), input_charset.encode("utf-8") if input_charset else None, read_cb, write_cb) + return self._native.run_script_input_output_callback_and_decode(thread, script.encode("utf-8"), encoded_inputs, input_name.encode("utf-8"), input_mime_type.encode("utf-8"), input_charset.encode("utf-8") if input_charset else None, read_cb, write_cb) stream = Stream(self._stream_worker(invoke, cancelled)) stream._on_close = cancelled.set stream._cancelled = cancelled @@ -250,8 +266,7 @@ def write_cb(_context, buffer, length): except Exception: return -1 try: - ptr = self._native.run_script_input_output_callback(self._native.thread, script.encode("utf-8"), self._inputs_json(inputs), input_name.encode("utf-8"), input_mime_type.encode("utf-8"), input_charset.encode("utf-8") if input_charset else None, read_cb, write_cb) - raw = self._native.decode_and_free(ptr) + raw = self._native.run_script_input_output_callback_and_decode(self._native.thread, script.encode("utf-8"), self._inputs_json(inputs), input_name.encode("utf-8"), input_mime_type.encode("utf-8"), input_charset.encode("utf-8") if input_charset else None, read_cb, write_cb) return parse_streaming_result(json.loads(raw) if raw else {"success": False, "error": "Empty response"}) except Exception as error: raise DataWeaveError(f"Failed to execute callback input/output streaming: {error}") diff --git a/native-lib/python/tests/conftest.py b/native-lib/python/tests/conftest.py index d476b48e..c2eccc43 100644 --- a/native-lib/python/tests/conftest.py +++ b/native-lib/python/tests/conftest.py @@ -168,10 +168,17 @@ def clean_dataweave_runtime(request): dataweave.cleanup() +def _tck_runtime(): + fixtures_dir = Path(__file__).resolve().parents[2] / "node" / "tests" / "tck" / "fixtures" + return dataweave.DataWeave( + resolve_module=dataweave.modules_from_directory(fixtures_dir), + ) + + @pytest.fixture(scope="session") def tck_runtime(): """Own one isolate for the TCK session and release it after the lane.""" - runtime = dataweave.DataWeave() + runtime = _tck_runtime() runtime.initialize() try: yield runtime diff --git a/native-lib/python/tests/integration/test_module_resolver.py b/native-lib/python/tests/integration/test_module_resolver.py new file mode 100644 index 00000000..8f9a510c --- /dev/null +++ b/native-lib/python/tests/integration/test_module_resolver.py @@ -0,0 +1,320 @@ +import io +import json +import os +from pathlib import Path +import subprocess +import sys +from zipfile import ZipFile + +import pytest + +import dataweave + + +IMPORT_LIB_SCRIPT = """%dw 2.0 +import org::test::lib +output application/json +--- +lib::answer() +""" + + +def _import_script(module_name): + return f"""%dw 2.0 +import org::test::{module_name} +output application/json +--- +{module_name}::answer() +""" + + +@pytest.mark.integration +def test_run_resolves_module_from_map(): + resolver = dataweave.modules_from_map({ + "org/test/lib.dwl": "%dw 2.0\nfun answer() = 42", + }) + + with dataweave.DataWeave(resolve_module=resolver) as dw: + result = dw.run(IMPORT_LIB_SCRIPT) + + assert result.success is True + assert result.get_string() == "42" + + +@pytest.mark.integration +def test_missing_module_returns_unsuccessful_result(): + with dataweave.DataWeave( + resolve_module=dataweave.modules_from_map({}), + ) as dw: + result = dw.run(IMPORT_LIB_SCRIPT) + + assert result.success is False + assert "resolve" in (result.error or "").lower() + + +@pytest.mark.integration +def test_raise_on_error_promotes_missing_module_result(): + with dataweave.DataWeave( + resolve_module=dataweave.modules_from_map({}), + ) as dw: + with pytest.raises(dataweave.DataWeaveScriptError) as error: + dw.run(IMPORT_LIB_SCRIPT, raise_on_error=True) + + assert error.value.result.success is False + assert "resolve" in (error.value.result.error or "").lower() + + +def _write_transitive_modules(module_root): + module_dir = module_root / "org" / "test" + module_dir.mkdir(parents=True) + (module_dir / "base.dwl").write_text( + "%dw 2.0\nfun value() = 40", + encoding="utf-8", + ) + (module_dir / "lib.dwl").write_text( + "%dw 2.0\nimport org::test::base\nfun answer() = base::value() + 2", + encoding="utf-8", + ) + + +@pytest.mark.integration +def test_directory_resolver_supports_transitive_imports(tmp_path): + _write_transitive_modules(tmp_path) + + with dataweave.DataWeave( + resolve_module=dataweave.modules_from_directory(tmp_path), + ) as dw: + result = dw.run(IMPORT_LIB_SCRIPT) + + assert result.success is True + assert result.get_string() == "42" + + +@pytest.mark.integration +def test_jar_resolver_supports_transitive_imports(tmp_path): + module_root = tmp_path / "modules" + _write_transitive_modules(module_root) + jar_path = tmp_path / "modules.jar" + with ZipFile(jar_path, "w") as jar: + jar.write(module_root / "org" / "test" / "base.dwl", "org/test/base.dwl") + jar.write(module_root / "org" / "test" / "lib.dwl", "org/test/lib.dwl") + + with dataweave.DataWeave( + resolve_module=dataweave.modules_from_jars([jar_path]), + ) as dw: + result = dw.run(IMPORT_LIB_SCRIPT) + + assert result.success is True + assert result.get_string() == "42" + + +@pytest.mark.integration +def test_repeated_runs_reuse_the_instances_resolver(): + resolver = dataweave.modules_from_map({ + "org/test/lib.dwl": "%dw 2.0\nfun answer() = 42", + }) + + with dataweave.DataWeave(resolve_module=resolver) as dw: + first = dw.run(IMPORT_LIB_SCRIPT) + second = dw.run(IMPORT_LIB_SCRIPT) + + assert first.success is True + assert first.get_string() == "42" + assert second.success is True + assert second.get_string() == "42" + + +@pytest.mark.integration +def test_cleanup_of_one_instance_preserves_another_instances_resolver(): + first = dataweave.DataWeave(resolve_module=dataweave.modules_from_map({ + "org/test/lib.dwl": "%dw 2.0\nfun answer() = 41", + })) + second = dataweave.DataWeave(resolve_module=dataweave.modules_from_map({ + "org/test/lib.dwl": "%dw 2.0\nfun answer() = 42", + })) + first_initialized = False + second_initialized = False + try: + first.initialize() + first_initialized = True + second.initialize() + second_initialized = True + + first_result = first.run(IMPORT_LIB_SCRIPT) + second_result = second.run(IMPORT_LIB_SCRIPT) + assert first_result.success is True + assert first_result.get_string() == "41" + assert second_result.success is True + assert second_result.get_string() == "42" + + first.cleanup() + first_initialized = False + + result = second.run(IMPORT_LIB_SCRIPT) + assert result.success is True + assert result.get_string() == "42" + finally: + try: + if first_initialized: + first.cleanup() + finally: + if second_initialized: + second.cleanup() + + +@pytest.mark.integration +def test_streaming_builtin_import_does_not_invoke_unsupported_external_resolver( + collect_stream, +): + calls = [] + + def resolver(module_path): + calls.append(module_path) + return None + + script = """%dw 2.0 +import fromBase64 from dw::core::Binaries +output application/json +--- +sizeOf(fromBase64("aGk=")) +""" + with dataweave.DataWeave(resolve_module=resolver) as dw: + output, metadata = collect_stream(dw.run_streaming(script)) + + assert metadata.success is True + assert output.decode(metadata.charset or "utf-8") == "2" + assert calls == [] + + +@pytest.mark.integration +def test_resolver_is_inactive_for_resolver_less_apis_after_synchronous_install( + collect_stream, +): + calls = [] + + def resolver(module_path): + calls.append(module_path) + return "%dw 2.0\nfun answer() = 42" + + with dataweave.DataWeave(resolve_module=resolver) as dw: + installed = dw.run(IMPORT_LIB_SCRIPT) + calls_after_install = list(calls) + + _output, streaming = collect_stream(dw.run_streaming(_import_script("streaming"))) + _output, transform = collect_stream( + dw.run_transform( + _import_script("transform"), + [b"null"], + input_mime_type="application/json", + ) + ) + + callback_chunks = [] + callback = dw.run_callback( + _import_script("callback"), + lambda chunk: callback_chunks.append(chunk) or 0, + ) + + source = io.BytesIO(b"null") + input_output_chunks = [] + input_output = dw.run_input_output_callback( + _import_script("inputOutput"), + input_name="payload", + input_mime_type="application/json", + read_callback=source.read, + write_callback=lambda chunk: input_output_chunks.append(chunk) or 0, + ) + + assert installed.success is True + assert calls_after_install + assert streaming.success is False + assert transform.success is False + assert callback.success is False + assert input_output.success is False + assert callback_chunks == [] + assert input_output_chunks == [] + assert calls == calls_after_install + + +@pytest.mark.integration +def test_overlapping_resolver_aware_runs_are_serialized(): + source_dir = Path(__file__).resolve().parents[2] / "src" + code = f""" +import json +from threading import Event, Lock, Thread + +import dataweave + +script = {IMPORT_LIB_SCRIPT!r} +first_resolver_call = Event() +release_first = Event() +second_resolver_call = Event() +results = [] +errors = [] +calls = 0 +calls_lock = Lock() + +def resolver(_module_path): + global calls + with calls_lock: + calls += 1 + current_call = calls + if current_call == 1: + first_resolver_call.set() + if not release_first.wait(2): + raise RuntimeError("first resolver call was not released") + else: + second_resolver_call.set() + return "%dw 2.0\\nfun answer() = 42" + +with dataweave.DataWeave(resolve_module=resolver) as dw: + def run(): + try: + result = dw.run(script) + results.append({{"success": result.success, "value": result.get_string()}}) + except Exception as error: + errors.append(str(error)) + + first = Thread(target=run) + second = Thread(target=run) + first.start() + if not first_resolver_call.wait(2): + raise RuntimeError("first resolver was not called") + second.start() + serialized = not second_resolver_call.wait(0.1) + release_first.set() + first.join(2) + second.join(2) + if first.is_alive() or second.is_alive(): + raise RuntimeError("resolver worker did not finish") + +print(json.dumps({{ + "serialized": serialized, + "second_called": second_resolver_call.is_set(), + "results": results, + "errors": errors, +}})) +""" + environment = os.environ.copy() + environment["PYTHONPATH"] = str(source_dir) + os.pathsep + environment.get("PYTHONPATH", "") + + completed = subprocess.run( + [sys.executable, "-c", code], + capture_output=True, + check=False, + env=environment, + text=True, + timeout=30, + ) + + assert completed.returncode == 0, completed.stderr + response = json.loads(completed.stdout) + assert response == { + "serialized": True, + "second_called": True, + "results": [ + {"success": True, "value": "42"}, + {"success": True, "value": "42"}, + ], + "errors": [], + } diff --git a/native-lib/python/tests/tck/ignore_list.py b/native-lib/python/tests/tck/ignore_list.py index df72411d..327303a0 100644 --- a/native-lib/python/tests/tck/ignore_list.py +++ b/native-lib/python/tests/tck/ignore_list.py @@ -57,35 +57,15 @@ def _exclusion(case_identifier: str, category: str, reason: str) -> Exclusion: # Each entry has a full case identifier and direct runtime evidence. Categories # describe only an observed, unsupported limitation; they never match patterns. EXCLUDED_CASES: Dict[str, Exclusion] = { - "runtime/import-component-alias-lib-out.json": _exclusion( - "runtime/import-component-alias-lib-out.json", - UNSUPPORTED_DW_MODULE_RESOLUTION, - "imports a test-only DW module; the Python binding has no module resolver", - ), - "runtime/import-lib-out.json": _exclusion( - "runtime/import-lib-out.json", - UNSUPPORTED_DW_MODULE_RESOLUTION, - "imports a test-only DW module; the Python binding has no module resolver", - ), - "runtime/import-lib-with-alias-out.json": _exclusion( - "runtime/import-lib-with-alias-out.json", - UNSUPPORTED_DW_MODULE_RESOLUTION, - "imports a test-only DW module; the Python binding has no module resolver", - ), - "runtime/import-named-lib-out.json": _exclusion( - "runtime/import-named-lib-out.json", - UNSUPPORTED_DW_MODULE_RESOLUTION, - "imports a test-only DW module; the Python binding has no module resolver", - ), - "runtime/import-star-out.json": _exclusion( - "runtime/import-star-out.json", - UNSUPPORTED_DW_MODULE_RESOLUTION, - "imports a test-only DW module; the Python binding has no module resolver", - ), "runtime/module-singleton-out.json": _exclusion( "runtime/module-singleton-out.json", UNSUPPORTED_DW_MODULE_RESOLUTION, - "imports a test-only DW module; the Python binding has no module resolver", + "runtime cannot resolve org::mule::weave::v2::libs::singleton::libA; " + "runtime cannot resolve org::mule::weave::v2::libs::singleton::libB; " + "runtime cannot resolve org::mule::weave::v2::libs::singleton::libSource; " + "shared fixture lacks org::mule::weave::v2::libs::singleton::libA; " + "shared fixture lacks org::mule::weave::v2::libs::singleton::libB; " + "shared fixture lacks org::mule::weave::v2::libs::singleton::libSource", ), "runtime/is-empty-using-empty-stream-out.json": _exclusion( "runtime/is-empty-using-empty-stream-out.json", @@ -157,11 +137,6 @@ def _exclusion(case_identifier: str, category: str, reason: str) -> Exclusion: UNSUPPORTED_DW_MODULE_RESOLUTION, "cannot resolve dw::core::Assertions before reading the binary fixture", ), - "runtime/full-qualified-name-ref-out.json": _exclusion( - "runtime/full-qualified-name-ref-out.json", - UNSUPPORTED_DW_MODULE_RESOLUTION, - "cannot resolve org::mule::weave::v2::libs::lib test modules", - ), "runtime/private_scope_directives-out.xml": _exclusion( "runtime/private_scope_directives-out.xml", UNSUPPORTED_DW_MODULE_RESOLUTION, diff --git a/native-lib/python/tests/tck/test_conformance.py b/native-lib/python/tests/tck/test_conformance.py index fbcdc2d2..0ffc84c4 100644 --- a/native-lib/python/tests/tck/test_conformance.py +++ b/native-lib/python/tests/tck/test_conformance.py @@ -10,6 +10,7 @@ import pytest import dataweave +import conftest sys.path.insert(0, str(Path(__file__).parent)) @@ -68,6 +69,49 @@ } DEFERRED_WRITER_CASE = "core-modules/deferred-write-should-terminate-out.json:out.json" +RECOVERED_MODULE_CASES = { + "runtime/full-qualified-name-ref-out.json", + "runtime/import-component-alias-lib-out.json", + "runtime/import-lib-out.json", + "runtime/import-lib-with-alias-out.json", + "runtime/import-named-lib-out.json", + "runtime/import-star-out.json", +} + + +@pytest.mark.unit +def test_tck_runtime_uses_shared_module_fixture_resolver(monkeypatch): + fixtures_dir = Path(__file__).resolve().parents[3] / "node" / "tests" / "tck" / "fixtures" + captured = {"fixture_directories": []} + resolver = lambda _path: "module source" + + class FakeRuntime: + def __init__(self, **kwargs): + captured.update(kwargs) + + monkeypatch.setattr(conftest.dataweave, "DataWeave", FakeRuntime) + monkeypatch.setattr( + conftest.dataweave, + "modules_from_directory", + lambda directory: captured["fixture_directories"].append(directory) or resolver, + ) + + runtime = conftest._tck_runtime() + + assert (fixtures_dir / "org" / "mule" / "weave" / "v2" / "libs" / "lib.dwl").is_file() + assert captured["fixture_directories"] == [fixtures_dir] + assert captured["resolve_module"] is resolver + assert isinstance(runtime, FakeRuntime) + + +@pytest.mark.unit +def test_module_singleton_exclusion_preserves_direct_runtime_evidence(): + reason = EXCLUDED_CASES["runtime/module-singleton-out.json"].reason + + for module in ("libA", "libB", "libSource"): + path = f"org::mule::weave::v2::libs::singleton::{module}" + assert f"runtime cannot resolve {path}" in reason + assert f"shared fixture lacks {path}" in reason def tck_params(): @@ -452,12 +496,11 @@ def test_only_declared_case_identifiers_are_excluded(): """Catches broad exclusion matching that can skip unrelated failures.""" assert validate_exclusions(EXCLUDED_CASES, SCENARIOS) == [] assert exclusion_for("unknown-case") is None - exclusion = exclusion_for("runtime/import-lib-out.json") - assert exclusion.case_identifier == "runtime/import-lib-out.json" - assert exclusion.category == "unsupported-dw-module-resolution" - assert len(EXCLUDED_CASES) == 37 + assert RECOVERED_MODULE_CASES.isdisjoint(EXCLUDED_CASES) + assert len(EXCLUDED_CASES) == 31 +@pytest.mark.unit def test_exclusion_registry_uses_the_inventory_categories(): """Catches category collapse that would conceal the unsupported boundary.""" categories = {} @@ -467,7 +510,7 @@ def test_exclusion_registry_uses_the_inventory_categories(): assert categories == { "unavailable-classpath-test-resource": 2, "unavailable-java-module": 11, - "unsupported-dw-module-resolution": 24, + "unsupported-dw-module-resolution": 18, } diff --git a/native-lib/python/tests/unit/test_ci_structure.py b/native-lib/python/tests/unit/test_ci_structure.py index 560f8687..c2d32dec 100644 --- a/native-lib/python/tests/unit/test_ci_structure.py +++ b/native-lib/python/tests/unit/test_ci_structure.py @@ -18,6 +18,24 @@ def named_step_if(document: str, name: str) -> str: return guard.group("guard") +def assert_resolver_restrictions(document: str) -> None: + normalized = " ".join(document.split()) + assert ( + "The `ModuleResolver` contract is a synchronous callable from a module " + "key to the module source string or `None`." + ) in normalized + assert ( + "Custom resolver configuration is available only on an explicit " + "`DataWeave` instance; the module-level `dataweave.run()` singleton " + "does not accept `resolve_module`." + ) in normalized + assert ( + "`run_streaming()`, `run_transform()`, and the low-level callback " + "streaming API do not use custom resolvers and can import only built-in " + "modules." + ) in normalized + + @pytest.mark.unit def test_python_artifact_runs_python_test_before_building_wheel(): action = (Path(__file__).resolve().parents[4] / ".github/actions/python/action.yml").read_text() @@ -55,6 +73,55 @@ def test_python_artifact_owns_test_dependencies_and_tck_junit_upload(): assert "native-lib/build/test-results/pythonTck.xml" in action +@pytest.mark.unit +def test_python_readme_documents_module_resolver_contract(): + readme = (Path(__file__).resolve().parents[2] / "README.md").read_text() + + for public_name in ( + "ModuleResolver", + "modules_from_map", + "modules_from_directory", + "modules_from_jars", + "compose_resolvers", + "resolve_module", + "DATAWEAVE_RESOLVER_DEBUG", + ): + assert f"`{public_name}`" in readme + + assert_resolver_restrictions(readme) + normalized = " ".join(readme.split()) + assert "without a leading path separator" in normalized + assert "without a leading slash or separator" not in normalized + assert "Each initialized explicit Python `DataWeave` instance owns a dedicated Graal isolate." in normalized + assert "retains the resolver callback until successful isolate teardown" in normalized + + +@pytest.mark.unit +@pytest.mark.parametrize( + "contract,negated", + ( + ( + "The `ModuleResolver` contract is a synchronous callable", + "The `ModuleResolver` contract is an asynchronous callable", + ), + ( + "singleton does not accept `resolve_module`", + "singleton does accept `resolve_module`", + ), + ( + "streaming API do not use custom resolvers and can import only built-in modules", + "streaming API do use custom resolvers and can import external modules", + ), + ), +) +def test_python_readme_resolver_restrictions_reject_negation(contract, negated): + readme = (Path(__file__).resolve().parents[2] / "README.md").read_text() + mutated = " ".join(readme.split()).replace(contract, negated) + + with pytest.raises(AssertionError): + assert_resolver_restrictions(mutated) + + @pytest.mark.unit def test_master_tck_stages_the_shared_corpus_once_before_python_and_node(): root = Path(__file__).resolve().parents[4] @@ -77,6 +144,14 @@ def test_tck_metadata_validation_is_not_selected_by_the_pr_python_test_lane(): assert "@pytest.mark.unit\ndef test_accepted_baseline_mismatches" not in conformance +@pytest.mark.unit +def test_tck_corpus_inventory_policy_is_not_selected_by_the_pr_python_test_lane(): + root = Path(__file__).resolve().parents[4] + conformance = (root / "native-lib/python/tests/tck/test_conformance.py").read_text() + + assert "@pytest.mark.unit\ndef test_only_declared_case_identifiers_are_excluded" not in conformance + + @pytest.mark.unit def test_foundation_skips_python_tests_and_master_aggregates_binding_failures(): root = Path(__file__).resolve().parents[4] diff --git a/native-lib/python/tests/unit/test_facade.py b/native-lib/python/tests/unit/test_facade.py index c281495e..3191e032 100644 --- a/native-lib/python/tests/unit/test_facade.py +++ b/native-lib/python/tests/unit/test_facade.py @@ -1,6 +1,35 @@ +import inspect + import pytest import dataweave +from dataweave import runtime + + +class FakeNativeRuntime: + def __init__(self): + self.initialized = True + self.thread = "thread" + self.calls = [] + + def run_script_and_decode(self, *args): + self.calls.append(("run_script_and_decode", args)) + return self._result() + + def run_script_with_resolver_and_decode(self, *args): + self.calls.append(("run_script_with_resolver_and_decode", args)) + return self._result() + + @staticmethod + def _result(): + return '{"success": true, "result": "SGVsbG8=", "binary": false, "mimeType": "text/plain", "charset": "utf-8"}' + + +def configured_runtime(resolve_module=None): + instance = dataweave.DataWeave.__new__(dataweave.DataWeave) + instance._native = FakeNativeRuntime() + instance._resolve_module = resolve_module + return instance @pytest.mark.unit @@ -31,6 +60,78 @@ def test_facade_preserves_fixed_legacy_public_exports(): getattr(dataweave, name) +@pytest.mark.unit +def test_facade_exports_module_resolver_factories(): + resolver_exports = [ + "ModuleResolver", + "RESOLVE_MODULE_CALLBACK", + "compose_resolvers", + "modules_from_directory", + "modules_from_jars", + "modules_from_map", + ] + + for name in resolver_exports: + assert name in dataweave.__all__ + getattr(dataweave, name) + + +@pytest.mark.unit +def test_dataweave_constructor_stores_keyword_only_module_resolver(monkeypatch): + resolver = lambda _path: "source" + native_runtime = FakeNativeRuntime() + monkeypatch.setattr(runtime, "NativeRuntime", lambda _lib_path: native_runtime) + + instance = dataweave.DataWeave(resolve_module=resolver) + + assert instance._resolve_module is resolver + assert inspect.signature(dataweave.DataWeave).parameters[ + "resolve_module" + ].kind is inspect.Parameter.KEYWORD_ONLY + + +@pytest.mark.unit +def test_run_dispatches_to_resolver_aware_native_execution(): + resolver = lambda _path: "source" + instance = configured_runtime(resolver) + + result = instance.run("payload", {"value": 1}) + + assert result == dataweave.ExecutionResult( + True, "SGVsbG8=", None, False, "text/plain", "utf-8" + ) + assert instance._native.calls == [ + ( + "run_script_with_resolver_and_decode", + ( + "thread", + b"payload", + b'{"value": {"content": "MQ==", "mimeType": "application/json", "charset": "utf-8"}}', + resolver, + ), + ) + ] + + +@pytest.mark.unit +def test_run_without_resolver_preserves_native_execution_path(): + instance = configured_runtime() + + result = instance.run("payload") + + assert result == dataweave.ExecutionResult( + True, "SGVsbG8=", None, False, "text/plain", "utf-8" + ) + assert instance._native.calls == [ + ("run_script_and_decode", ("thread", b"payload", b"{}")) + ] + + +@pytest.mark.unit +def test_module_level_run_does_not_accept_module_resolver(): + assert "resolve_module" not in inspect.signature(dataweave.run).parameters + + @pytest.mark.unit def test_global_facade_initializes_once_and_cleanup_allows_recreation(monkeypatch): created = [] diff --git a/native-lib/python/tests/unit/test_native.py b/native-lib/python/tests/unit/test_native.py index edaeb5f0..78c0f80d 100644 --- a/native-lib/python/tests/unit/test_native.py +++ b/native-lib/python/tests/unit/test_native.py @@ -1,4 +1,6 @@ from pathlib import Path +import ctypes +from threading import current_thread, Event, get_ident, Thread import pytest @@ -6,6 +8,47 @@ from dataweave import native +class Function: + pass + + +class CallableFunction(Function): + def __init__(self, callback): + self.callback = callback + + def __call__(self, *args): + return self.callback(*args) + + +class FakeLibrary: + run_script = Function() + free_cstring = Function() + + def __init__(self, *, resolver_export=False): + self.attach_calls = [] + self.detach_calls = [] + self.tear_down_threads = [] + self.graal_create_isolate = CallableFunction(lambda _params, _isolate, _thread: 0) + self.graal_attach_thread = CallableFunction(self._attach_thread) + self.graal_detach_thread = CallableFunction( + lambda thread: self.detach_calls.append((get_ident(), thread)) or 0 + ) + self.graal_tear_down_isolate = CallableFunction( + lambda thread: self.tear_down_threads.append(thread) or 0 + ) + if resolver_export: + self.run_script_with_resolver = Function() + + def _attach_thread(self, _isolate, thread): + worker_thread = native.GraalIsolateThreadPointer() + ctypes.cast( + thread, + ctypes.POINTER(native.GraalIsolateThreadPointer), + )[0] = worker_thread + self.attach_calls.append((get_ident(), worker_thread)) + return 0 + + @pytest.mark.unit def test_parse_native_response_rejects_malformed_json(): result = dataweave._parse_native_encoded_response("not json") @@ -68,45 +111,840 @@ def test_decode_and_free_preserves_decode_failure_when_free_also_fails(monkeypat @pytest.mark.unit def test_native_runtime_registers_abi_and_cleans_up_idempotently(monkeypatch): - class Function: + library = FakeLibrary() + + monkeypatch.setattr(native.ctypes, "CDLL", lambda _path: library) + runtime = native.NativeRuntime("/tmp/dwlib") + runtime.initialize() + runtime.cleanup() + runtime.cleanup() + + assert library.run_script.argtypes[1:] == [native.ctypes.c_char_p, native.ctypes.c_char_p] + assert library.free_cstring.argtypes[1] is native.ctypes.c_void_p + assert len(library.tear_down_threads) == 1 + assert runtime.initialized is False + + +@pytest.mark.unit +def test_buffered_worker_execution_uses_one_current_thread_attachment_for_run_decode_and_free(monkeypatch): + calls = [] + buffer = ctypes.create_string_buffer(b"result") + library = FakeLibrary() + library.run_script = CallableFunction( + lambda thread, _script, _inputs: calls.append(("run", get_ident(), thread)) + or ctypes.addressof(buffer) + ) + library.free_cstring = CallableFunction( + lambda thread, _ptr: calls.append(("free", get_ident(), thread)) + ) + monkeypatch.setattr(native.ctypes, "CDLL", lambda _path: library) + runtime = native.NativeRuntime("/tmp/dwlib") + runtime.initialize() + owner_ident = get_ident() + outcomes = [] + + worker = Thread( + target=lambda: outcomes.append( + (get_ident(), runtime.run_script_and_decode(runtime.thread, b"script", b"{}")) + ) + ) + worker.start() + worker.join(1) + + assert not worker.is_alive() + worker_ident, result = outcomes[0] + assert worker_ident != owner_ident + assert result == "result" + assert runtime._owner_thread is current_thread() + assert len(library.attach_calls) == 1 + assert library.attach_calls[0][0] == worker_ident + worker_pointer = ctypes.cast(calls[0][2], ctypes.c_void_p).value + assert [(name, ident) for name, ident, _thread in calls] == [ + ("run", worker_ident), + ("free", worker_ident), + ] + assert all( + ctypes.cast(thread, ctypes.c_void_p).value == worker_pointer + for _name, _ident, thread in calls + ) + assert library.detach_calls[0][0] == worker_ident + assert ctypes.cast(library.detach_calls[0][1], ctypes.c_void_p).value == worker_pointer + + +@pytest.mark.unit +def test_distinct_thread_object_attaches_when_python_thread_ident_is_reused(monkeypatch): + buffer = ctypes.create_string_buffer(b"result") + library = FakeLibrary() + library.run_script = CallableFunction( + lambda _thread, _script, _inputs: ctypes.addressof(buffer) + ) + library.free_cstring = CallableFunction(lambda _thread, _ptr: None) + monkeypatch.setattr(native.ctypes, "CDLL", lambda _path: library) + monkeypatch.setattr(native, "get_ident", lambda: 7) + runtime = native.NativeRuntime("/tmp/dwlib") + runtime.initialize() + owner_thread = current_thread() + observed_threads = [] + + worker = Thread( + target=lambda: ( + observed_threads.append(current_thread()), + runtime.run_script_and_decode(runtime.thread, b"script", b"{}"), + ) + ) + worker.start() + worker.join(1) + + assert not worker.is_alive() + assert observed_threads == [worker] + assert observed_threads[0] is not owner_thread + assert runtime._owner_thread is owner_thread + assert len(library.attach_calls) == 1 + assert len(library.detach_calls) == 1 + + +@pytest.mark.unit +def test_cleanup_clears_owner_thread_reference(monkeypatch): + library = FakeLibrary() + monkeypatch.setattr(native.ctypes, "CDLL", lambda _path: library) + runtime = native.NativeRuntime("/tmp/dwlib") + runtime.initialize() + + assert runtime._owner_thread is current_thread() + + runtime.cleanup() + + assert runtime._owner_thread is None + + +@pytest.mark.unit +@pytest.mark.parametrize("failure", ["run", "decode", "free", "detach"]) +def test_buffered_worker_execution_detaches_current_thread_after_failure(monkeypatch, failure): + buffer = ctypes.create_string_buffer(b"result") + library = FakeLibrary() + + def run_script(_thread, _script, _inputs): + if failure == "run": + raise RuntimeError("run failed") + return ctypes.addressof(buffer) + + def free_cstring(_thread, _ptr): + if failure == "free": + raise RuntimeError("free failed") + + library.run_script = CallableFunction(run_script) + library.free_cstring = CallableFunction(free_cstring) + if failure == "detach": + library.graal_detach_thread = CallableFunction( + lambda _thread: (_ for _ in ()).throw(RuntimeError("detach failed")) + ) + monkeypatch.setattr(native.ctypes, "CDLL", lambda _path: library) + if failure == "decode": + monkeypatch.setattr(native.ctypes, "string_at", lambda _ptr: b"\xff") + runtime = native.NativeRuntime("/tmp/dwlib") + runtime.initialize() + errors = [] + + worker = Thread( + target=lambda: _capture_error( + errors, + lambda: runtime.run_script_and_decode(runtime.thread, b"script", b"{}"), + ) + ) + worker.start() + worker.join(1) + + assert not worker.is_alive() + assert len(errors) == 1 + if failure == "detach": + assert "detach failed" in str(errors[0]) + assert len(library.attach_calls) == 1 + if failure != "detach": + assert len(library.detach_calls) == 1 + assert library.detach_calls[0][0] == library.attach_calls[0][0] + + +@pytest.mark.unit +@pytest.mark.parametrize( + ("method_name", "native_name", "extra_args"), + [ + ("run_script", "run_script", ()), + ( + "run_script_with_resolver", + "run_script_with_resolver", + (lambda _path: "module source",), + ), + ("run_script_callback", "run_script_callback", (object(),)), + ( + "run_script_input_output_callback", + "run_script_input_output_callback", + (b"payload", b"application/json", None, object(), object()), + ), + ], +) +def test_raw_pointer_calls_use_supplied_thread_without_automatic_attachment( + monkeypatch, method_name, native_name, extra_args +): + observed_threads = [] + library = FakeLibrary(resolver_export=method_name == "run_script_with_resolver") + setattr( + library, + native_name, + CallableFunction( + lambda thread, *_args: observed_threads.append(thread) or 123 + ), + ) + monkeypatch.setattr(native.ctypes, "CDLL", lambda _path: library) + runtime = native.NativeRuntime("/tmp/dwlib") + runtime.initialize() + runtime.has_callback_streaming = True + runtime.has_callback_input_output = True + supplied_thread = runtime.thread + outcomes = [] + + worker = Thread( + target=lambda: outcomes.append( + getattr(runtime, method_name)( + supplied_thread, b"script", b"{}", *extra_args + ) + ) + ) + worker.start() + worker.join(1) + + assert not worker.is_alive() + assert outcomes == [123] + assert observed_threads == [supplied_thread] + assert library.attach_calls == [] + assert library.detach_calls == [] + + +def _capture_error(errors, invoke): + try: + invoke() + except Exception as error: + errors.append(error) + + +@pytest.mark.unit +def test_native_runtime_registers_optional_module_resolver_export(monkeypatch): + library = FakeLibrary(resolver_export=True) + monkeypatch.setattr(native.ctypes, "CDLL", lambda _path: library) + + runtime = native.NativeRuntime("/tmp/dwlib") + runtime.initialize() + + assert runtime.has_module_resolver is True + assert library.run_script_with_resolver.argtypes == [ + native.GraalIsolateThreadPointer, + native.ctypes.c_char_p, + native.ctypes.c_char_p, + dataweave.RESOLVE_MODULE_CALLBACK, + ] + assert library.run_script_with_resolver.restype is native.ctypes.c_void_p + + +@pytest.mark.unit +def test_native_runtime_initializes_without_optional_module_resolver_export(monkeypatch): + library = FakeLibrary() + monkeypatch.setattr(native.ctypes, "CDLL", lambda _path: library) + + runtime = native.NativeRuntime("/tmp/dwlib") + runtime.initialize() + + assert runtime.has_module_resolver is False + + +@pytest.mark.unit +def test_run_script_with_resolver_adapts_path_and_retains_source_buffer(monkeypatch): + observed = [] + resolver_paths = [] + library = FakeLibrary(resolver_export=True) + + def invoke(_thread, _script, _inputs, callback): + address = callback(None, b"/org/test/lib.dwl") + observed.append(ctypes.string_at(address).decode("utf-8")) + assert library.runtime._resolver_buffers + return 0 + + library.run_script_with_resolver = CallableFunction(invoke) + monkeypatch.setattr(native.ctypes, "CDLL", lambda _path: library) + runtime = native.NativeRuntime("/tmp/dwlib") + library.runtime = runtime + runtime.initialize() + stale_buffer = ctypes.create_string_buffer(b"stale") + runtime._resolver_buffers.append(stale_buffer) + + resolver = lambda path: resolver_paths.append(path) or "module source" + result = runtime.run_script_with_resolver("thread", b"script", b"{}", resolver) + + assert result == 0 + assert resolver_paths == ["org/test/lib.dwl"] + assert observed == ["module source"] + assert runtime._resolver_buffers == [] + + +@pytest.mark.unit +@pytest.mark.parametrize( + ("module_path", "resolver"), + [ + (b"/missing.dwl", lambda _path: None), + (b"/invalid.dwl", lambda _path: 42), + (b"\xff", lambda _path: "unreachable"), + ], +) +def test_resolver_callback_returns_null_for_unresolved_or_invalid_values( + monkeypatch, module_path, resolver +): + addresses = [] + library = FakeLibrary(resolver_export=True) + library.run_script_with_resolver = CallableFunction( + lambda _thread, _script, _inputs, callback: addresses.append( + callback(None, module_path) + ) or 0 + ) + monkeypatch.setattr(native.ctypes, "CDLL", lambda _path: library) + runtime = native.NativeRuntime("/tmp/dwlib") + runtime.initialize() + + runtime.run_script_with_resolver("thread", b"script", b"{}", resolver) + + assert addresses == [None] + assert runtime._resolver_buffers == [] + + +@pytest.mark.unit +def test_resolver_callback_contains_exceptions_and_hides_details_by_default( + monkeypatch, capsys +): + addresses = [] + library = FakeLibrary(resolver_export=True) + library.run_script_with_resolver = CallableFunction( + lambda _thread, _script, _inputs, callback: addresses.append( + callback(None, b"/org/test/lib.dwl") + ) or 0 + ) + monkeypatch.setattr(native.ctypes, "CDLL", lambda _path: library) + monkeypatch.delenv("DATAWEAVE_RESOLVER_DEBUG", raising=False) + runtime = native.NativeRuntime("/tmp/dwlib") + runtime.initialize() + + def resolver(_path): + raise RuntimeError("secret /private/path") + + runtime.run_script_with_resolver("thread", b"script", b"{}", resolver) + + captured = capsys.readouterr() + assert addresses == [None] + assert "DataWeave module resolver callback failed." in captured.err + assert "secret" not in captured.err + assert "/private/path" not in captured.err + + +@pytest.mark.unit +def test_resolver_callback_prints_exception_details_in_debug_mode( + monkeypatch, capsys +): + library = FakeLibrary(resolver_export=True) + library.run_script_with_resolver = CallableFunction( + lambda _thread, _script, _inputs, callback: callback( + None, b"/org/test/lib.dwl" + ) or 0 + ) + monkeypatch.setattr(native.ctypes, "CDLL", lambda _path: library) + monkeypatch.setenv("DATAWEAVE_RESOLVER_DEBUG", "1") + runtime = native.NativeRuntime("/tmp/dwlib") + runtime.initialize() + + def resolver(_path): + raise RuntimeError("secret /private/path") + + runtime.run_script_with_resolver("thread", b"script", b"{}", resolver) + + captured = capsys.readouterr() + assert "RuntimeError: secret /private/path" in captured.err + + +@pytest.mark.unit +def test_resolver_callback_contains_base_exceptions(monkeypatch, capsys): + class ResolverExit(BaseException): pass - class FakeLibrary: - run_script = Function() - free_cstring = Function() - graal_attach_thread = Function() - graal_detach_thread = Function() + addresses = [] + library = FakeLibrary(resolver_export=True) + library.run_script_with_resolver = CallableFunction( + lambda _thread, _script, _inputs, callback: addresses.append( + callback(None, b"/org/test/lib.dwl") + ) or 0 + ) + monkeypatch.setattr(native.ctypes, "CDLL", lambda _path: library) + monkeypatch.delenv("DATAWEAVE_RESOLVER_DEBUG", raising=False) + runtime = native.NativeRuntime("/tmp/dwlib") + runtime.initialize() - def __init__(self): - self.tear_down_threads = [] - self.graal_create_isolate = Function() - self.graal_create_isolate.__call__ = lambda _params, _isolate, _thread: 0 - self.graal_tear_down_isolate = Function() - self.graal_tear_down_isolate.__call__ = lambda thread: self.tear_down_threads.append(thread) or 0 + def resolver(_path): + raise ResolverExit("secret /private/path") - class CallableFunction(Function): - def __init__(self, callback): - self.callback = callback + runtime.run_script_with_resolver("thread", b"script", b"{}", resolver) + + captured = capsys.readouterr() + assert addresses == [None] + assert "DataWeave module resolver callback failed." in captured.err + assert "secret" not in captured.err + assert "/private/path" not in captured.err - def __call__(self, *args): - return self.callback(*args) +@pytest.mark.unit +def test_resolver_callback_contains_diagnostic_writer_failures(monkeypatch): + addresses = [] + library = FakeLibrary(resolver_export=True) + library.run_script_with_resolver = CallableFunction( + lambda _thread, _script, _inputs, callback: addresses.append( + callback(None, b"/org/test/lib.dwl") + ) or 0 + ) + monkeypatch.setattr(native.ctypes, "CDLL", lambda _path: library) + monkeypatch.delenv("DATAWEAVE_RESOLVER_DEBUG", raising=False) + monkeypatch.setattr( + native.sys, + "stderr", + type( + "FailingStderr", + (), + {"write": lambda _self, _value: (_ for _ in ()).throw(SystemExit(9))}, + )(), + ) + runtime = native.NativeRuntime("/tmp/dwlib") + runtime.initialize() + + def resolver(_path): + raise KeyboardInterrupt("secret /private/path") + + runtime.run_script_with_resolver("thread", b"script", b"{}", resolver) + + assert addresses == [None] + + +@pytest.mark.unit +def test_run_script_with_resolver_clears_buffers_when_native_call_fails(monkeypatch): + library = FakeLibrary(resolver_export=True) + + def invoke(_thread, _script, _inputs, callback): + assert callback(None, b"/org/test/lib.dwl") + raise RuntimeError("native failure") + + library.run_script_with_resolver = CallableFunction(invoke) + monkeypatch.setattr(native.ctypes, "CDLL", lambda _path: library) + runtime = native.NativeRuntime("/tmp/dwlib") + runtime.initialize() + + with pytest.raises(RuntimeError, match="native failure"): + runtime.run_script_with_resolver( + "thread", b"script", b"{}", lambda _path: "module source" + ) + + assert runtime._resolver_buffers == [] + + +@pytest.mark.unit +def test_run_script_with_resolver_serializes_calls_and_buffer_cleanup(monkeypatch): + first_entered = Event() + release_first = Event() + second_entered = Event() + errors = [] + library = FakeLibrary(resolver_export=True) + + def invoke(_thread, script, _inputs, callback): + address = callback(None, b"/org/test/lib.dwl") + if script == b"first": + first_entered.set() + if not release_first.wait(1): + raise AssertionError("first invocation was not released") + assert ctypes.string_at(address) == b"module source" + assert len(library.runtime._resolver_buffers) == 1 + else: + second_entered.set() + return 0 + + library.run_script_with_resolver = CallableFunction(invoke) + monkeypatch.setattr(native.ctypes, "CDLL", lambda _path: library) + runtime = native.NativeRuntime("/tmp/dwlib") + library.runtime = runtime + runtime.initialize() + resolver = lambda _path: "module source" + + def run(script): + try: + runtime.run_script_with_resolver("thread", script, b"{}", resolver) + except Exception as error: + errors.append(error) + + first = Thread(target=run, args=(b"first",)) + second = Thread(target=run, args=(b"second",)) + first.start() + assert first_entered.wait(1) + second.start() + + assert not second_entered.wait(0.1) + release_first.set() + first.join(1) + second.join(1) + + assert not first.is_alive() + assert not second.is_alive() + assert second_entered.is_set() + assert errors == [] + assert runtime._resolver_buffers == [] + + +@pytest.mark.unit +def test_native_runtime_reentrant_execution_fails_without_deadlocking(monkeypatch): + completed = Event() + nested_errors = [] library = FakeLibrary() - library.graal_create_isolate = CallableFunction(lambda _params, _isolate, _thread: 0) - library.graal_tear_down_isolate = CallableFunction(lambda thread: library.tear_down_threads.append(thread) or 0) + def invoke(thread, script, inputs): + if script == b"outer": + try: + library.runtime.run_script(thread, b"nested", inputs) + except Exception as error: + nested_errors.append(error) + return 0 + + library.run_script = CallableFunction(invoke) monkeypatch.setattr(native.ctypes, "CDLL", lambda _path: library) runtime = native.NativeRuntime("/tmp/dwlib") + library.runtime = runtime runtime.initialize() + + worker = Thread( + target=lambda: (runtime.run_script("thread", b"outer", b"{}"), completed.set()), + daemon=True, + ) + worker.start() + + assert completed.wait(1), "reentrant native execution deadlocked" + assert len(nested_errors) == 1 + assert isinstance(nested_errors[0], dataweave.DataWeaveError) + assert "reentrant" in str(nested_errors[0]).lower() + + +@pytest.mark.unit +def test_resolver_callback_translates_reentrant_execution_to_null(monkeypatch): + completed = Event() + callback_results = [] + library = FakeLibrary(resolver_export=True) + library.run_script = CallableFunction(lambda _thread, _script, _inputs: 0) + library.run_script_with_resolver = CallableFunction( + lambda thread, _script, inputs, callback: callback_results.append( + callback(thread, b"/org/test/lib.dwl") + ) or 0 + ) + monkeypatch.setattr(native.ctypes, "CDLL", lambda _path: library) + runtime = native.NativeRuntime("/tmp/dwlib") + runtime.initialize() + + def resolver(_path): + runtime.run_script("thread", b"nested", b"{}") + return "unreachable" + + worker = Thread( + target=lambda: ( + runtime.run_script_with_resolver("thread", b"outer", b"{}", resolver), + completed.set(), + ), + daemon=True, + ) + worker.start() + + assert completed.wait(1), "resolver callback re-entry deadlocked" + assert callback_results == [None] + + +@pytest.mark.unit +def test_cleanup_waits_for_resolver_aware_call(monkeypatch): + run_entered = Event() + release_run = Event() + teardown_entered = Event() + library = FakeLibrary(resolver_export=True) + + def invoke(_thread, _script, _inputs, callback): + assert callback(None, b"/org/test/lib.dwl") + run_entered.set() + assert release_run.wait(1) + return 0 + + library.run_script_with_resolver = CallableFunction(invoke) + library.graal_tear_down_isolate = CallableFunction( + lambda _thread: teardown_entered.set() or 0 + ) + monkeypatch.setattr(native.ctypes, "CDLL", lambda _path: library) + runtime = native.NativeRuntime("/tmp/dwlib") + runtime.initialize() + + run_thread = Thread( + target=runtime.run_script_with_resolver, + args=("thread", b"script", b"{}", lambda _path: "module source"), + ) + cleanup_thread = Thread(target=runtime.cleanup) + run_thread.start() + assert run_entered.wait(1) + cleanup_thread.start() + + assert not teardown_entered.wait(0.1) + release_run.set() + run_thread.join(1) + cleanup_thread.join(1) + + assert not run_thread.is_alive() + assert not cleanup_thread.is_alive() + assert teardown_entered.is_set() + + +@pytest.mark.unit +@pytest.mark.parametrize( + "invoke", + [ + lambda runtime: runtime.run_script_and_decode("thread", b"script", b"{}"), + lambda runtime: runtime.run_script_with_resolver_and_decode( + "thread", b"script", b"{}", lambda _path: "module source" + ), + lambda runtime: runtime.run_script_callback_and_decode( + "thread", b"script", b"{}", object() + ), + lambda runtime: runtime.run_script_input_output_callback_and_decode( + "thread", + b"script", + b"{}", + b"payload", + b"application/json", + None, + object(), + object(), + ), + ], +) +def test_cleanup_waits_until_native_result_is_decoded_and_freed(monkeypatch, invoke): + native_returned = Event() + release_decode = Event() + freed = Event() + teardown_entered = Event() + errors = [] + buffer = ctypes.create_string_buffer(b"result") + pointer = ctypes.addressof(buffer) + library = FakeLibrary(resolver_export=True) + return_pointer = lambda *_args: native_returned.set() or pointer + library.run_script = CallableFunction(return_pointer) + library.run_script_with_resolver = CallableFunction(return_pointer) + library.run_script_callback = CallableFunction(return_pointer) + library.run_script_input_output_callback = CallableFunction(return_pointer) + library.graal_attach_thread = CallableFunction(lambda _isolate, _thread: 0) + library.graal_detach_thread = CallableFunction(lambda _thread: 0) + library.free_cstring = CallableFunction( + lambda _thread, _ptr: release_decode.wait(1) and freed.set() + ) + library.graal_tear_down_isolate = CallableFunction( + lambda _thread: teardown_entered.set() or 0 + ) + monkeypatch.setattr(native.ctypes, "CDLL", lambda _path: library) + runtime = native.NativeRuntime("/tmp/dwlib") + runtime.initialize() + runtime.has_callback_streaming = True + runtime.has_callback_input_output = True + + def run(): + try: + invoke(runtime) + except Exception as error: + errors.append(error) + + run_thread = Thread(target=run) + cleanup_thread = Thread(target=runtime.cleanup) + run_thread.start() + assert native_returned.wait(1) + cleanup_thread.start() + + assert not teardown_entered.wait(0.1) + release_decode.set() + assert freed.wait(1) + run_thread.join(1) + cleanup_thread.join(1) + + assert not run_thread.is_alive() + assert not cleanup_thread.is_alive() + assert teardown_entered.is_set() + assert errors == [] + + +@pytest.mark.unit +def test_run_script_with_resolver_rejects_missing_native_export(monkeypatch): + library = FakeLibrary() + monkeypatch.setattr(native.ctypes, "CDLL", lambda _path: library) + runtime = native.NativeRuntime("/tmp/dwlib") + runtime.initialize() + + with pytest.raises( + dataweave.DataWeaveError, + match=r"Native library does not support module resolver API \(run_script_with_resolver not found\)\.", + ): + runtime.run_script_with_resolver( + "thread", b"script", b"{}", lambda _path: "module source" + ) + + +@pytest.mark.unit +def test_native_runtime_retains_one_resolver_callback_until_teardown(monkeypatch): + retained_during_teardown = [] + library = FakeLibrary(resolver_export=True) + library.run_script_with_resolver = CallableFunction( + lambda _thread, _script, _inputs, _callback: 0 + ) + monkeypatch.setattr(native.ctypes, "CDLL", lambda _path: library) + runtime = native.NativeRuntime("/tmp/dwlib") + library.graal_tear_down_isolate = CallableFunction( + lambda _thread: retained_during_teardown.append( + runtime._module_resolver_callback is not None + ) or 0 + ) + runtime.initialize() + resolver = lambda _path: "module source" + + runtime.run_script_with_resolver("thread", b"script", b"{}", resolver) + callback = runtime._module_resolver_callback + runtime.run_script_with_resolver("thread", b"script", b"{}", resolver) + + assert runtime._module_resolver_callback is callback + with pytest.raises(dataweave.DataWeaveError): + runtime.run_script_with_resolver( + "thread", b"script", b"{}", lambda _path: "other source" + ) + runtime.cleanup() + + assert retained_during_teardown == [True] + assert runtime._module_resolver_callback is None + assert runtime._module_resolver is None + + +@pytest.mark.unit +def test_cleanup_failure_preserves_runtime_state_for_successful_retry(monkeypatch): + library = FakeLibrary(resolver_export=True) + library.run_script_with_resolver = CallableFunction( + lambda _thread, _script, _inputs, _callback: 0 + ) + tear_down_results = iter((7, 0)) + library.graal_tear_down_isolate = CallableFunction( + lambda _thread: next(tear_down_results) + ) + monkeypatch.setattr(native.ctypes, "CDLL", lambda _path: library) + runtime = native.NativeRuntime("/tmp/dwlib") + runtime.initialize() + isolate = runtime.isolate + thread = runtime.thread + resolver = lambda _path: "module source" + runtime.run_script_with_resolver("thread", b"script", b"{}", resolver) + callback = runtime._module_resolver_callback + + with pytest.raises( + dataweave.DataWeaveError, + match="Failed to tear down GraalVM isolate. Error code: 7", + ): + runtime.cleanup() + + assert runtime.initialized is True + assert runtime.lib is library + assert runtime.isolate is isolate + assert runtime.thread is thread + assert runtime.has_module_resolver is True + assert runtime._module_resolver is resolver + assert runtime._module_resolver_callback is callback + runtime.cleanup() - assert library.run_script.argtypes[1:] == [native.ctypes.c_char_p, native.ctypes.c_char_p] - assert library.free_cstring.argtypes[1] is native.ctypes.c_void_p - assert len(library.tear_down_threads) == 1 + assert runtime.initialized is False + assert runtime.lib is None + assert runtime.isolate is None + assert runtime.thread is None + assert runtime._module_resolver is None + assert runtime._module_resolver_callback is None + + +@pytest.mark.unit +def test_cleanup_from_worker_uses_current_thread_for_isolate_teardown(monkeypatch): + library = FakeLibrary() + teardown_calls = [] + library.graal_tear_down_isolate = CallableFunction( + lambda thread: teardown_calls.append((get_ident(), thread)) or 0 + ) + monkeypatch.setattr(native.ctypes, "CDLL", lambda _path: library) + runtime = native.NativeRuntime("/tmp/dwlib") + runtime.initialize() + owner_ident = get_ident() + + worker = Thread(target=runtime.cleanup) + worker.start() + worker.join(1) + + assert not worker.is_alive() + worker_ident, teardown_thread = teardown_calls[0] + assert worker_ident != owner_ident + assert library.attach_calls[0][0] == worker_ident + assert ctypes.cast(teardown_thread, ctypes.c_void_p).value == ctypes.cast( + library.attach_calls[0][1], ctypes.c_void_p + ).value + assert library.detach_calls == [] assert runtime.initialized is False +@pytest.mark.unit +def test_failed_cleanup_from_worker_detaches_and_preserves_state_for_owner_retry(monkeypatch): + library = FakeLibrary() + tear_down_results = iter((7, 0)) + library.graal_tear_down_isolate = CallableFunction( + lambda _thread: next(tear_down_results) + ) + monkeypatch.setattr(native.ctypes, "CDLL", lambda _path: library) + runtime = native.NativeRuntime("/tmp/dwlib") + runtime.initialize() + errors = [] + + worker = Thread(target=lambda: _capture_error(errors, runtime.cleanup)) + worker.start() + worker.join(1) + + assert not worker.is_alive() + assert len(errors) == 1 + assert runtime.initialized is True + assert len(library.attach_calls) == 1 + assert len(library.detach_calls) == 1 + + runtime.cleanup() + + assert runtime.initialized is False + + +@pytest.mark.unit +def test_failed_worker_cleanup_preserves_teardown_error_when_detach_also_fails(monkeypatch): + library = FakeLibrary() + library.graal_tear_down_isolate = CallableFunction(lambda _thread: 7) + library.graal_detach_thread = CallableFunction( + lambda _thread: (_ for _ in ()).throw(RuntimeError("detach failed")) + ) + monkeypatch.setattr(native.ctypes, "CDLL", lambda _path: library) + runtime = native.NativeRuntime("/tmp/dwlib") + runtime.initialize() + errors = [] + + worker = Thread(target=lambda: _capture_error(errors, runtime.cleanup)) + worker.start() + worker.join(1) + + assert not worker.is_alive() + assert len(errors) == 1 + assert str(errors[0]) == "Failed to tear down GraalVM isolate. Error code: 7" + assert runtime.initialized is True + + @pytest.mark.unit def test_native_runtime_wraps_library_load_errors(monkeypatch): monkeypatch.setattr(native.ctypes, "CDLL", lambda _path: (_ for _ in ()).throw(OSError("bad image"))) @@ -233,7 +1071,7 @@ def __call__(self, *_args): setattr(library, symbol, Function()) monkeypatch.setattr(native.ctypes, "CDLL", lambda _path: library) - with pytest.raises(dataweave.DataWeaveError, match=f"run_script_callback requires native export {missing_symbol}"): + with pytest.raises(dataweave.DataWeaveError, match=f"Native library does not export {missing_symbol}"): native.NativeRuntime("/tmp/dwlib").initialize() @@ -248,9 +1086,10 @@ def test_cleanup_surfaces_native_teardown_error_code(monkeypatch): with pytest.raises(dataweave.DataWeaveError, match="Failed to tear down GraalVM isolate. Error code: 7"): runtime.cleanup() - assert runtime.lib is None - assert runtime.thread is None - assert runtime.isolate is None + assert runtime.initialized is True + assert runtime.lib is not None + assert runtime.thread is not None + assert runtime.isolate is not None @pytest.mark.unit diff --git a/native-lib/python/tests/unit/test_resolver.py b/native-lib/python/tests/unit/test_resolver.py new file mode 100644 index 00000000..e4d9a6d5 --- /dev/null +++ b/native-lib/python/tests/unit/test_resolver.py @@ -0,0 +1,208 @@ +from zipfile import ZipFile + +import pytest + +import dataweave.resolver as resolver_module +from dataweave import ( + compose_resolvers, + modules_from_directory, + modules_from_jars, + modules_from_map, +) + + +@pytest.mark.unit +def test_modules_from_map_copies_input_and_matches_exact_paths(): + modules = {"org/test/lib.dwl": "original"} + resolver = modules_from_map(modules) + modules["org/test/lib.dwl"] = "changed" + + assert resolver("org/test/lib.dwl") == "original" + assert resolver("/org/test/lib.dwl") is None + assert resolver("org/test/missing.dwl") is None + + +@pytest.mark.unit +def test_compose_resolvers_uses_first_match_and_falls_back(): + calls = [] + + def first(path): + calls.append(("first", path)) + return None + + def second(path): + calls.append(("second", path)) + return "source" + + def unused(path): + raise AssertionError(f"unexpected lookup: {path}") + + resolver = compose_resolvers(first, second, unused) + + assert resolver("lib.dwl") == "source" + assert calls == [("first", "lib.dwl"), ("second", "lib.dwl")] + + +@pytest.mark.unit +def test_compose_resolvers_propagates_resolver_errors(): + def failing(_path): + raise PermissionError("denied") + + with pytest.raises(PermissionError, match="denied"): + compose_resolvers(failing)("lib.dwl") + + +@pytest.mark.unit +def test_modules_from_directory_reads_nested_utf8_module(tmp_path): + module = tmp_path / "org" / "test" / "lib.dwl" + module.parent.mkdir(parents=True) + module.write_text("fun answer() = 42", encoding="utf-8") + + resolver = modules_from_directory(tmp_path) + + assert resolver("org/test/lib.dwl") == "fun answer() = 42" + assert resolver("org/test/missing.dwl") is None + + +@pytest.mark.unit +def test_modules_from_directory_keeps_root_after_chdir(tmp_path, monkeypatch): + base = tmp_path / "modules" + base.mkdir() + (base / "lib.dwl").write_text("source", encoding="utf-8") + monkeypatch.chdir(tmp_path) + resolver = modules_from_directory("modules") + elsewhere = tmp_path / "elsewhere" + elsewhere.mkdir() + monkeypatch.chdir(elsewhere) + + assert resolver("lib.dwl") == "source" + + +@pytest.mark.unit +def test_modules_from_directory_normalizes_parent_segments_in_root(tmp_path): + base = tmp_path / "modules" + child = base / "child" + child.mkdir(parents=True) + (base / "lib.dwl").write_text("source", encoding="utf-8") + + resolver = modules_from_directory(child / "..") + + assert resolver("lib.dwl") == "source" + + +@pytest.mark.unit +def test_modules_from_directory_rejects_lexical_escape(tmp_path): + base = tmp_path / "modules" + base.mkdir() + (tmp_path / "secret.dwl").write_text("secret", encoding="utf-8") + + assert modules_from_directory(base)("../secret.dwl") is None + + +@pytest.mark.unit +def test_modules_from_directory_rejects_absolute_path(tmp_path): + base = tmp_path / "modules" + base.mkdir() + secret = tmp_path / "secret.dwl" + secret.write_text("secret", encoding="utf-8") + + assert modules_from_directory(base)(str(secret)) is None + + +@pytest.mark.unit +def test_modules_from_directory_rejects_symlink_escape(tmp_path): + base = tmp_path / "modules" + base.mkdir() + secret = tmp_path / "secret.dwl" + secret.write_text("secret", encoding="utf-8") + link = base / "link.dwl" + try: + link.symlink_to(secret) + except (NotImplementedError, OSError): + pytest.skip("symlink creation is unavailable") + + assert modules_from_directory(base)("link.dwl") is None + + +@pytest.mark.unit +def test_modules_from_directory_fails_for_missing_root(tmp_path): + missing = tmp_path / "missing" + + with pytest.raises(FileNotFoundError, match="missing"): + modules_from_directory(missing) + + +@pytest.mark.unit +def test_modules_from_directory_names_invalid_utf8_module(tmp_path): + module = tmp_path / "invalid.dwl" + module.write_bytes(b"\xff") + + with pytest.raises(Exception, match="invalid[.]dwl"): + modules_from_directory(tmp_path)("invalid.dwl") + + +@pytest.mark.unit +def test_modules_from_jars_loads_dwl_entries_and_later_jars_win(tmp_path): + first = tmp_path / "first.jar" + second = tmp_path / "second.jar" + with ZipFile(first, "w") as jar: + jar.writestr("org/test/lib.dwl", "first") + jar.writestr("ignored.txt", "ignored") + with ZipFile(second, "w") as jar: + jar.writestr("org/test/lib.dwl", "second") + jar.writestr("org/test/other.dwl", "other") + + resolver = modules_from_jars([first, second]) + + assert resolver("org/test/lib.dwl") == "second" + assert resolver("org/test/other.dwl") == "other" + assert resolver("ignored.txt") is None + + +@pytest.mark.unit +def test_modules_from_jars_names_malformed_archive(tmp_path): + malformed = tmp_path / "malformed.jar" + malformed.write_text("not a zip archive", encoding="utf-8") + + with pytest.raises(Exception, match="malformed[.]jar"): + modules_from_jars([malformed]) + + +@pytest.mark.unit +@pytest.mark.parametrize("error", [RuntimeError("read failed"), NotImplementedError("unsupported")]) +def test_modules_from_jars_names_archive_when_entry_read_fails( + monkeypatch, tmp_path, error +): + archive = tmp_path / "modules.jar" + + class Entry: + filename = "org/test/lib.dwl" + + @staticmethod + def is_dir(): + return False + + class FailingZipFile: + def __init__(self, path): + assert path == archive + + def __enter__(self): + return self + + def __exit__(self, *_args): + return False + + @staticmethod + def infolist(): + return [Entry()] + + @staticmethod + def read(_entry): + raise error + + monkeypatch.setattr(resolver_module, "ZipFile", FailingZipFile) + + with pytest.raises(ValueError, match="modules[.]jar") as raised: + modules_from_jars([archive]) + + assert raised.value.__cause__ is error diff --git a/native-lib/python/tests/unit/test_streaming.py b/native-lib/python/tests/unit/test_streaming.py index b489da0e..8f68a2ee 100644 --- a/native-lib/python/tests/unit/test_streaming.py +++ b/native-lib/python/tests/unit/test_streaming.py @@ -1,6 +1,6 @@ import ctypes from queue import Full, Queue -from threading import Event, Thread +from threading import current_thread, Event, Thread from time import sleep import pytest @@ -15,12 +15,14 @@ def __init__(self, metadata=None, attach_code=0, emit=b"", consume_input=False): self.attach_code = attach_code self.emit = emit self.consume_input = consume_input + self.attach_count = 0 self.detached = [] self.freed = [] self._buffers = [] self.detached_event = Event() def graal_attach_thread(self, _isolate, _thread): + self.attach_count += 1 return self.attach_code def graal_detach_thread(self, thread): @@ -74,6 +76,7 @@ def configured_runtime(native): native_runtime.lib = native native_runtime.isolate = object() native_runtime.thread = object() + native_runtime._owner_thread = current_thread() runtime._native = native_runtime return runtime @@ -102,6 +105,61 @@ def test_run_input_output_callback_converts_read_exception_to_abort_result(): assert native.read_status == -1 +@pytest.mark.unit +def test_write_callback_reentry_is_translated_to_abort_without_deadlocking(): + completed = Event() + outcomes = [] + native = FakeNative('{"success": false, "error": "write aborted"}', emit=b"chunk") + runtime = configured_runtime(native) + + worker = Thread( + target=lambda: ( + outcomes.append( + runtime.run_callback( + "outer", + lambda _chunk: runtime.run_callback("nested", lambda _data: 0), + ) + ), + completed.set(), + ), + daemon=True, + ) + worker.start() + + assert completed.wait(1), "write callback re-entry deadlocked" + assert native.write_status == -1 + assert outcomes == [dataweave.StreamingResult(False, "write aborted", None, None, False)] + + +@pytest.mark.unit +def test_read_callback_reentry_is_translated_to_abort_without_deadlocking(): + completed = Event() + outcomes = [] + native = FakeNative('{"success": false, "error": "read aborted"}', consume_input=True) + runtime = configured_runtime(native) + + worker = Thread( + target=lambda: ( + outcomes.append( + runtime.run_input_output_callback( + "outer", + "payload", + "application/json", + lambda _size: runtime.run("nested").get_bytes(), + lambda _data: 0, + ) + ), + completed.set(), + ), + daemon=True, + ) + worker.start() + + assert completed.wait(1), "read callback re-entry deadlocked" + assert native.read_status == -1 + assert outcomes == [dataweave.StreamingResult(False, "read aborted", None, None, False)] + + @pytest.mark.unit @pytest.mark.parametrize( "invoke", @@ -130,6 +188,17 @@ def test_run_transform_preserves_remainder_of_large_input_chunk(): assert stream.metadata == dataweave.StreamingResult(True, None, "application/json", "utf-8", False) +@pytest.mark.unit +def test_streaming_explicit_worker_thread_is_not_attached_twice(): + native = FakeNative('{"success": true}') + runtime = configured_runtime(native) + + assert list(runtime.run_streaming("script")) == [] + + assert native.attach_count == 1 + assert len(native.detached) == 1 + + @pytest.mark.unit def test_run_streaming_returns_failure_metadata_when_worker_produces_no_metadata(monkeypatch): class MetadataDroppingQueue(Queue):