W-23516608: Add unit tests for extracted DataWeave Node.js binding modules

native-lib/node/tests/unit/errors.test.ts

@@ -0,0 +1,40 @@
+import { describe, it, expect } from "vitest";
+import { DataWeaveError, DataWeaveScriptError } from "../../src/errors";
+import { makeResult } from "../../src/result";
+
+describe("DataWeaveError", () => {
+  it("is an Error with the correct name and message", () => {
+    const e = new DataWeaveError("boom");
+    expect(e).toBeInstanceOf(Error);
+    expect(e).toBeInstanceOf(DataWeaveError);
+    expect(e.name).toBe("DataWeaveError");
+    expect(e.message).toBe("boom");
+  });
+});
+
+describe("DataWeaveScriptError", () => {
+  it("extends DataWeaveError and carries the name", () => {
+    const e = new DataWeaveScriptError(makeResult(false, null, "bad script", false, null, null));
+    expect(e).toBeInstanceOf(Error);
+    expect(e).toBeInstanceOf(DataWeaveError);
+    expect(e).toBeInstanceOf(DataWeaveScriptError);
+    expect(e.name).toBe("DataWeaveScriptError");
+  });
+
+  it("uses the result's error as the message", () => {
+    const result = makeResult(false, null, "unexpected token", false, null, null);
+    const e = new DataWeaveScriptError(result);
+    expect(e.message).toBe("unexpected token");
+  });
+
+  it("falls back to a default message when the result has no error", () => {
+    const e = new DataWeaveScriptError(makeResult(false, null, null, false, null, null));
+    expect(e.message).toBe("Script execution failed");
+  });
+
+  it("attaches the originating result", () => {
+    const result = makeResult(false, null, "err", false, null, null);
+    const e = new DataWeaveScriptError(result);
+    expect(e.result).toBe(result);
+  });
+});

native-lib/node/tests/unit/reader.test.ts

@@ -0,0 +1,116 @@
+import { describe, it, expect } from "vitest";
+import { createChunkReader } from "../../src/reader";
+
+/** Drains a reader with a fixed bufSize, returning the ordered chunks it yielded. */
+function drain(readCb: (n: number) => Buffer | null, bufSize: number): Buffer[] {
+  const out: Buffer[] = [];
+  let chunk: Buffer | null;
+  while ((chunk = readCb(bufSize)) !== null) {
+    out.push(chunk);
+  }
+  return out;
+}
+
+/** Drains a reader and concatenates everything into a single Buffer. */
+function drainAll(readCb: (n: number) => Buffer | null, bufSize: number): Buffer {
+  return Buffer.concat(drain(readCb, bufSize));
+}
+
+async function* asyncOf(...chunks: Array<Buffer | Uint8Array>): AsyncGenerator<Buffer | Uint8Array> {
+  for (const c of chunks) yield c;
+}
+
+describe("createChunkReader (sync input)", () => {
+  it("reassembles the full input across a small bufSize", async () => {
+    const read = await createChunkReader([Buffer.from("hello"), Buffer.from(" world")]);
+    expect(drainAll(read, 3).toString()).toBe("hello world");
+  });
+
+  it("returns null immediately for empty input", async () => {
+    const read = await createChunkReader([]);
+    expect(read(16)).toBeNull();
+  });
+
+  it("skips empty buffers between data", async () => {
+    const read = await createChunkReader([Buffer.from("ab"), Buffer.from(""), Buffer.from("cd")]);
+    expect(drainAll(read, 16).toString()).toBe("abcd");
+  });
+
+  it("bounds each chunk by bufSize", async () => {
+    const read = await createChunkReader([Buffer.from("abcdef")]);
+    const chunks = drain(read, 2);
+    expect(chunks.map((c) => c.toString())).toEqual(["ab", "cd", "ef"]);
+  });
+
+  it("returns a whole buffer when bufSize exceeds its length", async () => {
+    const read = await createChunkReader([Buffer.from("abc"), Buffer.from("de")]);
+    const chunks = drain(read, 100);
+    expect(chunks.map((c) => c.toString())).toEqual(["abc", "de"]);
+  });
+
+  it("accepts Uint8Array chunks", async () => {
+    const read = await createChunkReader([new Uint8Array([104, 105])]);
+    expect(drainAll(read, 16).toString()).toBe("hi");
+  });
+
+  it("keeps returning null after exhaustion", async () => {
+    const read = await createChunkReader([Buffer.from("x")]);
+    drain(read, 16);
+    expect(read(16)).toBeNull();
+    expect(read(16)).toBeNull();
+  });
+
+  it("does not share memory with the source buffer (returns copies)", async () => {
+    const source = Buffer.from("abcd");
+    const read = await createChunkReader([source]);
+    const chunk = read(4)!;
+    source[0] = 0x7a; // mutate 'a' -> 'z' after reading
+    expect(chunk.toString()).toBe("abcd");
+  });
+});
+
+describe("createChunkReader (async input)", () => {
+  it("reassembles the full input across a small bufSize", async () => {
+    const read = await createChunkReader(asyncOf(Buffer.from("foo"), Buffer.from("bar")));
+    expect(drainAll(read, 2).toString()).toBe("foobar");
+  });
+
+  it("returns null immediately for empty async input", async () => {
+    const read = await createChunkReader(asyncOf());
+    expect(read(16)).toBeNull();
+  });
+
+  it("pre-buffers the whole async source before any read", async () => {
+    let produced = 0;
+    async function* counting(): AsyncGenerator<Buffer> {
+      for (const s of ["a", "b", "c"]) {
+        produced++;
+        yield Buffer.from(s);
+      }
+    }
+    const read = await createChunkReader(counting());
+    // createChunkReader resolving means the async source is fully consumed.
+    expect(produced).toBe(3);
+    expect(drainAll(read, 16).toString()).toBe("abc");
+  });
+
+  it("bounds each chunk by bufSize", async () => {
+    const read = await createChunkReader(asyncOf(Buffer.from("abcde")));
+    const chunks = drain(read, 2);
+    expect(chunks.map((c) => c.toString())).toEqual(["ab", "cd", "e"]);
+  });
+
+  it("treats an input error as end-of-input, keeping what was buffered", async () => {
+    async function* boom(): AsyncGenerator<Buffer> {
+      yield Buffer.from("ok");
+      throw new Error("source failed");
+    }
+    const read = await createChunkReader(boom());
+    expect(drainAll(read, 16).toString()).toBe("ok");
+  });
+
+  it("accepts Uint8Array chunks", async () => {
+    const read = await createChunkReader(asyncOf(new Uint8Array([120, 121])));
+    expect(drainAll(read, 16).toString()).toBe("xy");
+  });
+});

native-lib/node/tests/unit/stream.test.ts

@@ -0,0 +1,98 @@
+import { describe, it, expect } from "vitest";
+import { streamFromNative } from "../../src/stream";
+import type { StreamingResult } from "../../src/types";
+
+const okMeta = (extra: Record<string, unknown> = {}) =>
+  JSON.stringify({ success: true, mimeType: "application/json", charset: "utf-8", binary: false, ...extra });
+
+/** Fully consumes a generator, returning its yielded chunks and terminal return value. */
+async function collect(
+  gen: AsyncGenerator<Buffer, StreamingResult, undefined>
+): Promise<{ chunks: Buffer[]; result: StreamingResult }> {
+  const chunks: Buffer[] = [];
+  let r = await gen.next();
+  while (!r.done) {
+    chunks.push(r.value);
+    r = await gen.next();
+  }
+  return { chunks, result: r.value };
+}
+
+function deferred<T>() {
+  let resolve!: (v: T) => void;
+  const promise = new Promise<T>((res) => { resolve = res; });
+  return { promise, resolve };
+}
+
+describe("streamFromNative", () => {
+  it("yields chunks pushed before completion, in order", async () => {
+    const { chunks, result } = await collect(
+      streamFromNative((cb) => {
+        cb(Buffer.from("a"));
+        cb(Buffer.from("b"));
+        cb(Buffer.from("c"));
+        return Promise.resolve(okMeta());
+      })
+    );
+    expect(chunks.map((c) => c.toString())).toEqual(["a", "b", "c"]);
+    expect(result.success).toBe(true);
+    expect(result.mimeType).toBe("application/json");
+  });
+
+  it("returns success metadata with no chunks", async () => {
+    const { chunks, result } = await collect(streamFromNative(() => Promise.resolve(okMeta())));
+    expect(chunks).toEqual([]);
+    expect(result.success).toBe(true);
+  });
+
+  it("parks the consumer until a chunk arrives, then wakes it (backpressure)", async () => {
+    const meta = deferred<string>();
+    let push!: (chunk: Buffer) => void;
+    const gen = streamFromNative((cb) => {
+      push = cb;
+      return meta.promise;
+    });
+
+    // First pull starts the generator and parks — no chunk is ready yet.
+    const pending = gen.next();
+    // Producing a chunk should wake the parked consumer.
+    push(Buffer.from("late"));
+    const first = await pending;
+    expect(first.done).toBe(false);
+    expect(first.value!.toString()).toBe("late");
+
+    // Completing the stream ends the generator with the parsed metadata.
+    meta.resolve(okMeta({ mimeType: "text/plain" }));
+    const last = await gen.next();
+    expect(last.done).toBe(true);
+    expect((last.value as StreamingResult).mimeType).toBe("text/plain");
+  });
+
+  it("drains chunks that arrive together with completion", async () => {
+    const { chunks, result } = await collect(
+      streamFromNative((cb) => {
+        // Chunks buffered but not yet consumed when the native call resolves.
+        cb(Buffer.from("x"));
+        cb(Buffer.from("y"));
+        return Promise.resolve(okMeta());
+      })
+    );
+    expect(chunks.map((c) => c.toString())).toEqual(["x", "y"]);
+    expect(result.success).toBe(true);
+  });
+
+  it("propagates a failure envelope as the terminal result", async () => {
+    const { chunks, result } = await collect(
+      streamFromNative(() => Promise.resolve(JSON.stringify({ success: false, error: "stream boom" })))
+    );
+    expect(chunks).toEqual([]);
+    expect(result.success).toBe(false);
+    expect(result.error).toBe("stream boom");
+  });
+
+  it("treats empty terminal metadata as a failure", async () => {
+    const { result } = await collect(streamFromNative(() => Promise.resolve("")));
+    expect(result.success).toBe(false);
+    expect(result.error).toBe("Empty response");
+  });
+});