diff --git a/packages/csv-parse/lib/api/index.js b/packages/csv-parse/lib/api/index.js index ca97838cd..71f15608f 100644 --- a/packages/csv-parse/lib/api/index.js +++ b/packages/csv-parse/lib/api/index.js @@ -777,8 +777,12 @@ const transform = function (original_options = {}) { if (this.__isFloat(field)) { return [undefined, Number(field)]; } else if (this.options.cast_date !== false) { - const info = this.__infoField(); - return [undefined, this.options.cast_date.call(null, field, info)]; + try { + const info = this.__infoField(); + return [undefined, this.options.cast_date.call(null, field, info)]; + } catch (err) { + return [err]; + } } return [undefined, field]; }, diff --git a/packages/csv-parse/lib/stream.js b/packages/csv-parse/lib/stream.js index b74c902e6..d403e909b 100644 --- a/packages/csv-parse/lib/stream.js +++ b/packages/csv-parse/lib/stream.js @@ -19,13 +19,13 @@ const parse = (opts) => { transform(chunk) { const error = api.parse(chunk, false, enqueue, terminate); if (error) { - controller.error(error); + throw error; } }, flush() { const error = api.parse(undefined, true, enqueue, terminate); if (error) { - controller.error(error); + throw error; } }, }, diff --git a/packages/csv-parse/test/api.web_stream.ts b/packages/csv-parse/test/api.web_stream.ts index a39592b5c..ef2cb6e91 100644 --- a/packages/csv-parse/test/api.web_stream.ts +++ b/packages/csv-parse/test/api.web_stream.ts @@ -1,8 +1,62 @@ import "should"; +import assert from "node:assert/strict"; import { parse as parseStream } from "../lib/stream.js"; +import type { Options } from "../lib/stream.js"; import { CsvError } from "../lib/index.js"; describe("API Web Stream", function () { + for (const castOption of ["cast", "cast_date"] as const) { + for (const phase of ["write", "close"]) { + it(`rejects ${phase} with the original ${castOption} error`, async function () { + const error = new Error("conversion failed"); + const options: Options = { cast: true }; + options[castOption] = () => { + throw error; + }; + const stream = parseStream(options); + const writer = stream.writable.getWriter(); + const reader = stream.readable.getReader(); + const readError = reader.read().catch((err) => err); + const closedError = writer.closed.catch((err) => err); + let operation; + if (phase === "write") { + operation = writer.write(Buffer.from("bad,next\n")); + } else { + await writer.write(Buffer.from("bad")); + operation = writer.close(); + } + await assert.rejects(operation, (err) => err === error); + assert.equal(await readError, error); + assert.equal(await closedError, error); + }); + } + } + + for (const phase of ["write", "close"]) { + it(`rejects ${phase} with the original CSV syntax error`, async function () { + const stream = parseStream(); + const writer = stream.writable.getWriter(); + const reader = stream.readable.getReader(); + const readError = reader.read().catch((err) => err); + const closedError = writer.closed.catch((err) => err); + let operation; + if (phase === "write") { + operation = writer.write(Buffer.from('bad"quote,next\n')); + } else { + await writer.write(Buffer.from('"unterminated')); + operation = writer.close(); + } + const error = await readError; + await assert.rejects(operation, (err) => err === error); + assert.ok(error instanceof CsvError); + assert.equal( + error.code, + phase === "write" ? "INVALID_OPENING_QUOTE" : "CSV_QUOTE_NOT_CLOSED", + ); + assert.equal(await closedError, error); + }); + } + describe("stream/web/TransformStream", function () { it("simple parse", async function () { const stream = parseStream(); diff --git a/packages/csv-parse/test/option.cast_date.errors.js b/packages/csv-parse/test/option.cast_date.errors.js new file mode 100644 index 000000000..e981dd302 --- /dev/null +++ b/packages/csv-parse/test/option.cast_date.errors.js @@ -0,0 +1,48 @@ +import assert from "node:assert/strict"; +import { parse } from "../lib/index.js"; +import { parse as parseSync } from "../lib/sync.js"; + +describe("Option `cast_date` errors", function () { + for (const input of ["bad,next\n", "good,bad", "good,date\nbad,next\n"]) { + for (const skip_records_with_error of [false, true]) { + it(`reports an error from ${JSON.stringify(input)} with skip=${skip_records_with_error}`, async function () { + const error = new Error("date conversion failed"); + const callbackErrors = []; + const streamErrors = []; + const parser = parse( + { + cast: true, + skip_records_with_error, + cast_date(value) { + if (value === "bad") throw error; + return value; + }, + }, + (err) => callbackErrors.push(err), + ); + parser.on("error", (err) => streamErrors.push(err)); + const closed = new Promise((resolve) => parser.on("close", resolve)); + assert.doesNotThrow(() => parser.end(input)); + await closed; + assert.deepEqual(callbackErrors, [error]); + assert.deepEqual(streamErrors, [error]); + }); + } + } + + it("continues to throw the original error from the sync API", function () { + const error = new Error("date conversion failed"); + let thrown; + try { + parseSync("bad", { + cast: true, + cast_date() { + throw error; + }, + }); + } catch (err) { + thrown = err; + } + assert.equal(thrown, error); + }); +});