Add tests for ErrorStrategies

diff --git a/src/errorStrategies.js b/src/errorStrategies.js
index 7550a03afc4a03128fb42295eac007da6644c6a0..00968adfc260f146622ba6975999f8cae62f6f38 100644 (file)
--- a/src/errorStrategies.js
+++ b/src/errorStrategies.js
@@ -2,6 +2,19 @@ import { setTimeout } from 'node:timers/promises';
 
 /** @import {TransformationErrorFunction} from './core.js'; */
 
+/**
+ * A wrapper around `node:timers/promises`'s async `setTimeout`. This wrapper is required to mock for testing purposes.
+ * 
+ * @param {number} ms The duration to wait before resolving.
+ */
+/* node:coverage ignore next 3 */
+async function sleep(ms) {
+  await setTimeout(ms);
+}
+export const utils = Object.create(null, {
+  sleep: { value: sleep, writable: true, enumerable: true, configurable: true }
+});
+
 /**
  * Create a `TransformationErrorFunction` that handles errors by filling the output row with a given placeholder value.
  * 
@@ -66,7 +79,7 @@ function backoff(iterations, maxExponent, value) {
       const exponent = n > maxExponent ? maxExponent : n;
       const duration = (2 ** exponent) * 1000; // 1s, 2s, 4s, 8s, 16s, ...
       const jitter = Math.random() * 1000;
-      await setTimeout(duration + jitter);
+      await utils.sleep(duration + jitter);
       try { return await fn(row); }
       catch { }
     }

diff --git a/test/errorStrategies.test.js b/test/errorStrategies.test.js
new file mode 100644 (file)
index 0000000..57dd6a3
--- /dev/null
+++ b/test/errorStrategies.test.js
@@ -0,0 +1,248 @@
+import { suite, test } from 'node:test';
+import { rejects } from 'node:assert/strict';
+
+import { csvStreamEqualWritable, createCSVMockStream, assertSleep } from './utils.js';
+import { CSVTransformer } from '../src/core.js';
+import { ErrorStrategies } from '../src/errorStrategies.js';
+
+const { placeholder, retry, backoff } = ErrorStrategies;
+
+/**
+ * Generate a function that will throw errors until it is called **exactly** `count` times.
+ * 
+ * @param {number} count The number of times this function must be called before it returns successfully.
+ * @returns `count` wrapped in an array for confirmation purposes.
+ */
+function retryFactory(count) {
+  let i = 0;
+  return () => {
+    if (++i === count) {
+      return [i];
+    }
+    throw new Error();
+  }
+}
+
+/**
+ * Calculate bounds for the expected sleep duration of an exponential backoff.
+ * 
+ * @param {number} iterations The number of times the exponential backoff will run before truncating.
+ * @returns {{min: number, max: number}} The minimum and maximum bounds of the total sleep time.
+ */
+function getExponentialSleepBounds(iterations) {
+  let min = 0;
+  for (let n = 0; n < iterations; n++) {
+    min += 1000 * (2 ** n);
+  }
+  const max = min + (1000 * iterations);
+  return { min, max };
+}
+
+suite('ErrorStrategies.placeholder', { concurrency: true }, () => {
+  test('sets placeholders on all fields in a row', async () => {
+    await createCSVMockStream([
+      ['header', 'header', 'header'],
+      ['1', '1', '1']
+    ])
+      .pipeThrough(new CSVTransformer(r => { throw new Error(); }, { onError: placeholder('n/a') }))
+      .pipeTo(csvStreamEqualWritable([
+        ['header', 'header', 'header'],
+        ['n/a', 'n/a', 'n/a']
+      ]));
+  });
+  test('sets placeholders on all fields in multiple rows', async () => {
+    await createCSVMockStream([
+      ['header', 'header', 'header'],
+      ['1', '1', '1'],
+      ['1', '1', '1']
+    ])
+      .pipeThrough(new CSVTransformer(r => { throw new Error(); }, { onError: placeholder('n/a') }))
+      .pipeTo(csvStreamEqualWritable([
+        ['header', 'header', 'header'],
+        ['n/a', 'n/a', 'n/a'],
+        ['n/a', 'n/a', 'n/a']
+      ]));
+  });
+  test('sets placeholders on row with 1 field', async () => {
+    await createCSVMockStream([
+      ['header', 'header', 'header'],
+      ['1']
+    ])
+      .pipeThrough(new CSVTransformer(r => { throw new Error(); }, { onError: placeholder('n/a') }))
+      .pipeTo(csvStreamEqualWritable([
+        ['header', 'header', 'header'],
+        ['n/a']
+      ]));
+  });
+  test('sets placeholders on row with 0 fields (empty row)', async () => {
+    await createCSVMockStream([
+      ['header', 'header', 'header'],
+      []
+    ])
+      .pipeThrough(new CSVTransformer(r => { throw new Error(); }, { onError: placeholder('n/a') }))
+      .pipeTo(csvStreamEqualWritable([
+        ['header', 'header', 'header'],
+        ['n/a']
+      ]));
+  });
+});
+
+suite('ErrorStrategies.retry', { concurrency: true }, () => {
+  test('retries until successful (1 retry, 2 total calls)', async () => {
+    const fn = retryFactory(2); // 1st try + 1 retry = 2 total calls
+    await createCSVMockStream([
+      ['header'],
+      ['0']
+    ])
+      .pipeThrough(new CSVTransformer(fn, { onError: retry(1) }))
+      .pipeTo(csvStreamEqualWritable([
+        ['header'],
+        [2]
+      ]));
+  });
+  test('retries until successful (9 retries, 10 total calls)', async () => {
+    const fn = retryFactory(10); // 1st try + 9 retries = pass on 10th and final retry
+    await createCSVMockStream([
+      ['header'],
+      ['0']
+    ])
+      .pipeThrough(new CSVTransformer(fn, { onError: retry(9) }))
+      .pipeTo(csvStreamEqualWritable([
+        ['header'],
+        [10]
+      ]));
+  });
+  test('re-throws error if failed after last iteration', async () => {
+    const fn = retryFactory(10); // 1st try + 8 retries = re-throw after failing on 9th try
+    await rejects(createCSVMockStream([
+      ['header'],
+      ['1']
+    ])
+      .pipeThrough(new CSVTransformer(fn, { onError: retry(8) }))
+      .pipeTo(new WritableStream()));
+  });
+  test('retries until successful (excess retries)', async () => {
+    const fn = retryFactory(10); // 1st try + 10 retries = pass on 10th with extra retries available
+    await createCSVMockStream([
+      ['header'],
+      ['0']
+    ])
+      .pipeThrough(new CSVTransformer(fn, { onError: retry(10) }))
+      .pipeTo(csvStreamEqualWritable([
+        ['header'],
+        [10]
+      ]));
+  });
+  test('sets placeholder if failed after last iteration and placeholder is given', async () => {
+    const fn = retryFactory(100);
+    await createCSVMockStream([
+      ['header'],
+      ['0']
+    ])
+      .pipeThrough(new CSVTransformer(fn, { onError: retry(1, 'n/a') }))
+      .pipeTo(csvStreamEqualWritable([
+        ['header'],
+        ['n/a']
+      ]));
+  });
+});
+
+suite('ErrorStrategies.backoff', { concurrency: 1 }, () => { // TODO: These tests can't run concurrently, fix
+  test('sets 1s backoff for 1 retry', async (t) => {
+    const { min, max } = getExponentialSleepBounds(1);
+    assertSleep(t, { min, max });
+    const fn = retryFactory(2); // 1st try + 1 retry = 2 total calls
+    await createCSVMockStream([
+      ['header'],
+      ['0']
+    ])
+      .pipeThrough(new CSVTransformer(fn, { onError: backoff(1) }))
+      .pipeTo(csvStreamEqualWritable([
+        ['header'],
+        [2]
+      ]));
+  });
+  test('sets exponential backoff for 3 retries', async (t) => {
+    const { min, max } = getExponentialSleepBounds(3);
+    assertSleep(t, { min, max });
+    const fn = retryFactory(4); // 1st try + 3 retries
+    await createCSVMockStream([
+      ['header'],
+      ['0']
+    ])
+      .pipeThrough(new CSVTransformer(fn, { onError: backoff(3) }))
+      .pipeTo(csvStreamEqualWritable([
+        ['header'],
+        [4]
+      ]));
+  });
+  test('sets exponential backoff for 10 retries', async (t) => {
+    const { min, max } = getExponentialSleepBounds(10);
+    assertSleep(t, { min, max });
+    const fn = retryFactory(11); // 1st try + 10 retries
+    await createCSVMockStream([
+      ['header'],
+      ['0']
+    ])
+      .pipeThrough(new CSVTransformer(fn, { onError: backoff(10) }))
+      .pipeTo(csvStreamEqualWritable([
+        ['header'],
+        [11]
+      ]));
+  });
+  test('truncates exponential backoff at 1s', async (t) => {
+    const min = 1000 * 10; // 10x retries at truncated 1s 
+    const max = min + 1000 * 10; // Max jitter is 1s each retry
+    assertSleep(t, { min, max });
+    const fn = retryFactory(11); // 1st try + 10 retries
+    await createCSVMockStream([
+      ['header'],
+      ['0']
+    ])
+      .pipeThrough(new CSVTransformer(fn, { onError: backoff(10, 0) }))
+      .pipeTo(csvStreamEqualWritable([
+        ['header'],
+        [11]
+      ]));
+  });
+  test('truncates exponential backoff at 16s', async (t) => {
+    const min = 1000 * (1 + 2 + 4 + 8 + 16 + 16 + 16); // Exponential until truncating at 16s for 3x retries
+    const max = min + 1000 * 7;
+    assertSleep(t, { min, max });
+    const fn = retryFactory(8); // 1st try + 7 retries
+    await createCSVMockStream([
+      ['header'],
+      ['0']
+    ])
+      .pipeThrough(new CSVTransformer(fn, { onError: backoff(7, 4) }))
+      .pipeTo(csvStreamEqualWritable([
+        ['header'],
+        [8]
+      ]));
+  });
+  test('sets placeholder if failed after last iteration and placeholder is given', async (t) => {
+    const { min, max } = getExponentialSleepBounds(3);
+    assertSleep(t, { min, max });
+    const fn = retryFactory(100);
+    await createCSVMockStream([
+      ['header'],
+      ['0']
+    ])
+      .pipeThrough(new CSVTransformer(fn, { onError: backoff(3, Infinity, 'n/a') }))
+      .pipeTo(csvStreamEqualWritable([
+        ['header'],
+        ['n/a']
+      ]));
+  });
+  test('re-throws error if failed after last iteration', async (t) => {
+    const { min, max } = getExponentialSleepBounds(3);
+    assertSleep(t, { min, max });
+    const fn = retryFactory(100);
+    await rejects(createCSVMockStream([
+      ['header'],
+      ['0']
+    ])
+      .pipeThrough(new CSVTransformer(fn, { onError: backoff(3) }))
+      .pipeTo(new WritableStream()));
+  });
+});

diff --git a/test/utils.js b/test/utils.js
index 7fdb41fe69d189d58135f283867b30deb5af0d90..1d804dc79ea140901f368ecab2ab0970a2523ca6 100644 (file)
--- a/test/utils.js
+++ b/test/utils.js
@@ -1,9 +1,11 @@
-import { deepStrictEqual, notStrictEqual } from 'node:assert/strict';
+import { deepStrictEqual, notStrictEqual, ok } from 'node:assert/strict';
 import { tmpdir } from 'node:os';
 import { mkdtemp, open } from 'node:fs/promises';
 import { normalize } from 'node:path';
 /** @import { TestContext } from 'node:test'; */
 
+import { utils } from '../src/errorStrategies.js';
+
 /**
  * A `WritableStream` sink that compares each field in a CSV stream with a given array, assuming each chunk of the
  * stream is a serialized `Array<any>` representing a row of a CSV.
@@ -89,6 +91,42 @@ export function assertConsole(context, expected = {}) {
   });
 }
 
+/**
+ * @typedef AssertSleepDurations Object containing the minimum and maximum bounds of the total timeout duration.
+ * @property {number} min The minimum total timeout duration that is expected.
+ * @property {number} max The maximum total timeout duration that is expected.
+ */
+
+/**
+ * Progresses asynchronous (`node:timers/promises`) `setTimeout` calls instantly and asserts the total duration passed
+ * to all `setTimeout` calls are within given expected bounds.
+ * 
+ * **Important**: The built-in `setTimeout` can **NOT** be mocked directly. As a workaround, use the `utils.sleep`
+ * wrapper function from `errorStrategies.js`.
+ * 
+ * BAD:
+ * ```javascript
+ * await setTimeout(1000);
+ * ```
+ * 
+ * GOOD:
+ * ```javascript
+ * await utils.sleep(1000);
+ * ```
+ * 
+ * @param {TestContext} context A Node.js `TestContext` for a test case.
+ * @param {AssertSleepDurations} expected Object containing the minimum and maximum bounds of the total sleep
+ * duration.
+ */
+export function assertSleep(context, expected) {
+  let actual = 0;
+  context.mock.method(utils, 'sleep', duration => actual += duration);
+  context.after(() => {
+    ok(actual >= expected.min, `expected minimum ${expected.min} total sleep duration, got ${actual}`);
+    ok(actual <= expected.max, `expected maximum ${expected.max} total sleep duration, got ${actual}`);
+  });
+}
+
 /**
  * Simulate a stream of CSV data for test mocking purposes.
  *