diff --git a/native-lib/node/src/result.ts b/native-lib/node/src/result.ts index 267fe2f..636baa8 100644 --- a/native-lib/node/src/result.ts +++ b/native-lib/node/src/result.ts @@ -1,5 +1,74 @@ import type { ExecutionResult, StreamingResult } from "./types"; +/** Normalizes a charset name for comparison: lowercased, non-alphanumerics stripped. */ +function normalizeCharset(charset: string): string { + return charset.toLowerCase().replace(/[^a-z0-9]/g, ""); +} + +/** + * Decodes payload bytes to a string using a DataWeave/IANA charset name. + * + * The native runtime reports charsets using IANA-style names (e.g. `UTF-16`, + * `US-ASCII`, `ISO-8859-1`) that Node's `Buffer.toString` does not accept — + * passing them through directly throws `ERR_UNKNOWN_ENCODING`. This maps the + * common cases to a valid Node {@link BufferEncoding} and handles UTF-16 + * byte-order (Node only decodes little-endian): a leading BOM, or a `UTF-16BE` + * label, is honored by stripping the BOM and byte-swapping big-endian input. + * Unrecognized charsets fall back to UTF-8 so decoding degrades gracefully + * instead of throwing. + * + * @param bytes - The decoded payload bytes. + * @param charset - The charset name reported by the runtime, or `null`. + * @returns The decoded string. + */ +export function decodeBytes(bytes: Buffer, charset: string | null): string { + if (!charset) return bytes.toString("utf-8"); + + switch (normalizeCharset(charset)) { + case "utf16": + case "utf16le": + case "utf16be": + case "ucs2": + case "unicode": + return decodeUtf16(bytes, normalizeCharset(charset) === "utf16be"); + case "usascii": + case "ascii": + return bytes.toString("ascii"); + case "latin1": + case "iso88591": + case "cp1252": + case "windows1252": + return bytes.toString("latin1"); + case "base64": + return bytes.toString("base64"); + case "hex": + return bytes.toString("hex"); + case "utf8": + default: + return bytes.toString("utf-8"); + } +} + +/** + * Decodes UTF-16 bytes to a string, honoring a BOM if present (which overrides + * `labelIsBE`) and byte-swapping big-endian input, since Node only has a + * little-endian UTF-16 decoder. + */ +function decodeUtf16(bytes: Buffer, labelIsBE: boolean): string { + let big = labelIsBE; + let start = 0; + if (bytes.length >= 2) { + if (bytes[0] === 0xfe && bytes[1] === 0xff) { big = true; start = 2; } // BE BOM + else if (bytes[0] === 0xff && bytes[1] === 0xfe) { big = false; start = 2; } // LE BOM + } + let body = bytes.subarray(start); + if (big) { + body = Buffer.from(body); // copy so swap16 doesn't mutate the caller's bytes + if (body.length % 2 === 0) body.swap16(); + } + return body.toString("utf16le"); +} + /** * Parses the JSON envelope returned by the native `run_script` FFI call into an * {@link ExecutionResult}. @@ -82,7 +151,7 @@ export function makeResult( if (!this.success || this.result === null) return null; if (this.binary) return this.result; const bytes = Buffer.from(this.result, "base64"); - return bytes.toString((this.charset as BufferEncoding) ?? "utf-8"); + return decodeBytes(bytes, this.charset); }, }; } diff --git a/native-lib/node/tests/integration/edge-cases.test.ts b/native-lib/node/tests/integration/edge-cases.test.ts new file mode 100644 index 0000000..cfb385e --- /dev/null +++ b/native-lib/node/tests/integration/edge-cases.test.ts @@ -0,0 +1,202 @@ +// Integration edge cases beyond the happy-path suite in dataweave.test.ts: +// concurrency / FFI lifecycle, and the error & edge-input matrix. These run +// against the real native library (dwlib). (Closes gaps F3 and F5 in +// native-lib/node/TESTING_ASSESSMENT.md.) +import { describe, it, expect, afterAll } from "vitest"; +import { DataWeave, run, runStreaming, runTransform, cleanup } from "../../src/index"; +import type { StreamingResult } from "../../src/types"; + +afterAll(() => { + cleanup(); +}); + +/** Drains a streaming/transform generator, returning its chunks and terminal metadata. */ +async function drain(gen: AsyncGenerator) { + const chunks: Buffer[] = []; + let r = await gen.next(); + while (!r.done) { + chunks.push(r.value); + r = await gen.next(); + } + return { text: Buffer.concat(chunks).toString("utf-8"), chunks, metadata: r.value }; +} + +// --- Concurrency / FFI edge cases ---------------------------------------- + +describe("runTransform with async-iterable input", () => { + it("consumes an async generator (pre-buffered read path)", async () => { + async function* asyncChunks(): AsyncGenerator { + for (const s of ["[1, 2, ", "3, 4", ", 5]"]) yield Buffer.from(s); + } + const { text, metadata } = await drain( + runTransform("output application/json\n---\npayload map ($ * 10)", asyncChunks(), { + mimeType: "application/json", + }) + ); + expect(metadata.success).toBe(true); + expect(text).toContain("10"); + expect(text).toContain("50"); + }); + + it("handles an empty async input", async () => { + async function* empty(): AsyncGenerator { /* yields nothing */ } + const { metadata } = await drain( + runTransform("output application/json\n---\n{ ok: true }", empty(), { mimeType: "application/json" }) + ); + // Script ignores payload; empty input must not hang or crash. + expect(metadata.success).toBe(true); + }); + + it("treats a throwing async input as end-of-input for what was produced", async () => { + async function* boom(): AsyncGenerator { + yield Buffer.from("[1, 2, 3]"); + throw new Error("source failed"); + } + const { text, metadata } = await drain( + runTransform("output application/json\n---\nsizeOf(payload)", boom(), { mimeType: "application/json" }) + ); + expect(metadata.success).toBe(true); + expect(text.trim()).toBe("3"); + }); +}); + +describe("multi-instance lifecycle", () => { + it("runs two independent instances and cleans them up independently", () => { + const a = new DataWeave(); + const b = new DataWeave(); + a.initialize(); + b.initialize(); + try { + expect(a.run("1 + 1").getString()).toBe("2"); + expect(b.run("2 + 3").getString()).toBe("5"); + } finally { + a.cleanup(); + b.cleanup(); + } + // After cleanup, a fresh instance still works (runtime not permanently torn down). + const c = new DataWeave(); + c.initialize(); + try { + expect(c.run("6 * 7").getString()).toBe("42"); + } finally { + c.cleanup(); + } + }); + + it("initialize is idempotent and re-initialization after cleanup works", () => { + const dw = new DataWeave(); + dw.initialize(); + dw.initialize(); // no-op, must not throw + expect(dw.run("1").getString()).toBe("1"); + dw.cleanup(); + dw.cleanup(); // double cleanup, must not throw + dw.initialize(); // re-init + try { + expect(dw.run("2").getString()).toBe("2"); + } finally { + dw.cleanup(); + } + }); + + it("run before initialize throws a DataWeaveError", () => { + const dw = new DataWeave(); + expect(() => dw.run("1 + 1")).toThrow(/not initialized/i); + }); + + // NOTE: initialize() with a bad library path throwing a DataWeaveError is only + // reliably observable as the FIRST initialization in a process — the native + // runtime is loaded process-globally (ref-counted, see g_ref_count in + // addon.c), so once any good init has run, a later bad-path init is tolerated + // rather than re-dlopen'd. Because this shared suite loads dwlib in earlier + // tests, that path can't be asserted here without a dedicated isolated process + // (a follow-up: run it via a separate vitest pool/child so no prior init has + // happened). Verified manually: in a fresh process it throws DataWeaveError. +}); + +describe("concurrent execution", () => { + it("runs many concurrent scripts on the shared singleton", async () => { + const results = await Promise.all( + Array.from({ length: 25 }, (_, i) => Promise.resolve().then(() => run(`${i} * 2`))) + ); + results.forEach((r, i) => { + expect(r.success).toBe(true); + expect(r.getString()).toBe(String(i * 2)); + }); + }); + + it("interleaves multiple concurrent streaming generators", async () => { + const gens = Array.from({ length: 4 }, (_, i) => + runStreaming(`output application/json --- (1 to 200) map ($ + ${i * 1000})`) + ); + const drained = await Promise.all(gens.map(drain)); + drained.forEach(({ metadata, text }, i) => { + expect(metadata.success).toBe(true); + expect(text).toContain(String(i * 1000 + 1)); + }); + }); +}); + +// --- Error & edge-input matrix ------------------------------------------- + +describe("binary output", () => { + it("marks binary output and round-trips through getBytes", () => { + const r = run("output application/octet-stream\n---\npayload", { + payload: { content: Buffer.from([0, 1, 2, 255]), mimeType: "application/octet-stream" }, + }); + expect(r.success).toBe(true); + expect(r.binary).toBe(true); + expect(r.getBytes()!.equals(Buffer.from([0, 1, 2, 255]))).toBe(true); + }); +}); + +describe("output charsets (buffered path)", () => { + // Regression guard for the getString() charset bug: the runtime reports IANA + // charset names that Node's Buffer.toString rejects; decodeBytes normalizes them. + for (const enc of ["UTF-8", "UTF-16", "UTF-16LE", "UTF-16BE", "ISO-8859-1"]) { + it(`decodes ${enc} output without throwing`, () => { + const r = run(`output text/plain encoding="${enc}"\n---\n"café"`); + expect(r.success).toBe(true); + expect(() => r.getString()).not.toThrow(); + expect(r.getString()).toBe("café"); + }); + } + + it("substitutes unrepresentable characters for US-ASCII output", () => { + const r = run(`output text/plain encoding="US-ASCII"\n---\n"café"`); + expect(r.success).toBe(true); + // 'é' is not representable in ASCII; the runtime substitutes it on encode. + expect(r.getString()).toBe("caf?"); + }); +}); + +describe("malformed / unsupported input", () => { + it("fails on an unknown input mime type", () => { + const r = run("payload", { payload: { content: "x", mimeType: "application/nonsense-xyz" } }); + expect(r.success).toBe(false); + expect(r.error).toBeTruthy(); + }); + + it("fails on an unknown output mime type", () => { + const r = run("output application/nonsense-xyz\n---\n{ a: 1 }"); + expect(r.success).toBe(false); + expect(r.error).toBeTruthy(); + }); + + it("fails when referencing a missing input", () => { + const r = run("missingInput + 1"); + expect(r.success).toBe(false); + expect(r.error).toBeTruthy(); + }); + + it("fails on a script with a compilation/syntax error", () => { + const r = run("output application/json\n---\n{ unclosed:"); + expect(r.success).toBe(false); + expect(r.error).toBeTruthy(); + }); + + it("surfaces malformed input content as a failed result", () => { + const r = run("payload.value", { payload: { content: "{ not valid json", mimeType: "application/json" } }); + expect(r.success).toBe(false); + expect(r.error).toBeTruthy(); + }); +}); \ No newline at end of file diff --git a/native-lib/node/tests/unit/result.test.ts b/native-lib/node/tests/unit/result.test.ts index 4d9d6d6..3ceb423 100644 --- a/native-lib/node/tests/unit/result.test.ts +++ b/native-lib/node/tests/unit/result.test.ts @@ -1,5 +1,5 @@ import { describe, it, expect } from "vitest"; -import { parseNativeResponse, makeResult, parseStreamingResult } from "../../src/result"; +import { parseNativeResponse, makeResult, parseStreamingResult, decodeBytes } from "../../src/result"; const b64 = (s: string, enc: BufferEncoding = "utf-8") => Buffer.from(s, enc).toString("base64"); @@ -98,6 +98,61 @@ describe("makeResult", () => { }); }); +describe("decodeBytes", () => { + it("defaults to UTF-8 when charset is null", () => { + expect(decodeBytes(Buffer.from("café", "utf-8"), null)).toBe("café"); + }); + + it("decodes UTF-8 by name", () => { + expect(decodeBytes(Buffer.from("café", "utf-8"), "UTF-8")).toBe("café"); + }); + + it("decodes little-endian UTF-16", () => { + const le = Buffer.from("café", "utf16le"); + expect(decodeBytes(le, "UTF-16LE")).toBe("café"); + }); + + it("decodes big-endian UTF-16 by label (byte-swapping)", () => { + const be = Buffer.from("café", "utf16le"); + be.swap16(); + expect(decodeBytes(be, "UTF-16BE")).toBe("café"); + }); + + it("honors a big-endian BOM regardless of label", () => { + const body = Buffer.from("café", "utf16le"); + body.swap16(); + const beWithBom = Buffer.concat([Buffer.from([0xfe, 0xff]), body]); + expect(decodeBytes(beWithBom, "UTF-16")).toBe("café"); + }); + + it("honors a little-endian BOM and strips it", () => { + const leWithBom = Buffer.concat([Buffer.from([0xff, 0xfe]), Buffer.from("café", "utf16le")]); + expect(decodeBytes(leWithBom, "UTF-16")).toBe("café"); + }); + + it("decodes ISO-8859-1 / latin1", () => { + const latin = Buffer.from("café", "latin1"); + expect(decodeBytes(latin, "ISO-8859-1")).toBe("café"); + }); + + it("decodes US-ASCII", () => { + expect(decodeBytes(Buffer.from("hello", "ascii"), "US-ASCII")).toBe("hello"); + }); + + it("falls back to UTF-8 for an unrecognized charset instead of throwing", () => { + expect(() => decodeBytes(Buffer.from("hi", "utf-8"), "x-made-up-charset")).not.toThrow(); + expect(decodeBytes(Buffer.from("hi", "utf-8"), "x-made-up-charset")).toBe("hi"); + }); + + it("does not mutate the caller's buffer when byte-swapping", () => { + const be = Buffer.from("AB", "utf16le"); + be.swap16(); + const snapshot = Buffer.from(be); + decodeBytes(be, "UTF-16BE"); + expect(be.equals(snapshot)).toBe(true); + }); +}); + describe("parseStreamingResult", () => { it("treats an empty string as a failure", () => { const r = parseStreamingResult("");