diff --git a/native-lib/node/tests/unit/errors.test.ts b/native-lib/node/tests/unit/errors.test.ts new file mode 100644 index 0000000..1da0fc7 --- /dev/null +++ b/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); + }); +}); \ No newline at end of file diff --git a/native-lib/node/tests/unit/reader.test.ts b/native-lib/node/tests/unit/reader.test.ts new file mode 100644 index 0000000..a546d99 --- /dev/null +++ b/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): AsyncGenerator { + 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 { + 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 { + 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"); + }); +}); \ No newline at end of file diff --git a/native-lib/node/tests/unit/stream.test.ts b/native-lib/node/tests/unit/stream.test.ts new file mode 100644 index 0000000..46ca0e4 --- /dev/null +++ b/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 = {}) => + 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 +): 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() { + let resolve!: (v: T) => void; + const promise = new Promise((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(); + 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"); + }); +}); \ No newline at end of file