stream: expand stream/iter test coverage

Author: jasnell

test/parallel/test-stream-iter-broadcast-coverage.js

@@ -0,0 +1,115 @@
+// Flags: --experimental-stream-iter
+'use strict';
+
+// Coverage tests for broadcast.js: signal abort on pending write,
+// sync iterable from, ringbuffer grow.
+
+const common = require('../common');
+const assert = require('assert');
+const {
+  broadcast,
+  Broadcast,
+  text,
+} = require('stream/iter');
+
+// Signal abort on pending write (covers wireBroadcastWriteSignal + removeAt)
+async function testBroadcastWriteAbort() {
+  const { writer, broadcast: bc } = broadcast({
+    highWaterMark: 1,
+    backpressure: 'block',
+  });
+  const consumer = bc.push();
+
+  // Fill the buffer to capacity
+  writer.writeSync(new Uint8Array([1]));
+
+  // Next write will block — pass a signal
+  const ac = new AbortController();
+  const writePromise = writer.write(new Uint8Array([2]),
+                                    { signal: ac.signal });
+
+  // Abort the signal
+  ac.abort();
+
+  await assert.rejects(writePromise, { name: 'AbortError' });
+
+  // Clean up
+  writer.endSync();
+  // Drain the consumer
+  const result = [];
+  for await (const batch of consumer) {
+    result.push(...batch);
+  }
+  assert.ok(result.length >= 1);
+}
+
+// Broadcast.from with sync iterable (generator)
+async function testBroadcastFromSyncIterable() {
+  function* source() {
+    yield [new Uint8Array([10, 20])];
+    yield [new Uint8Array([30, 40])];
+  }
+
+  const { broadcast: bc } = Broadcast.from(source());
+  const consumer = bc.push();
+  // Just verify it completes without error and produces data
+  let count = 0;
+  for await (const batch of consumer) {
+    count += batch.length;
+  }
+  assert.ok(count > 0);
+}
+
+// Broadcast.from with sync iterable — string chunks
+async function testBroadcastFromSyncIterableStrings() {
+  function* source() {
+    yield 'hello';
+    yield ' world';
+  }
+  const { broadcast: bc } = Broadcast.from(source());
+  const consumer = bc.push();
+  const result = await text(consumer);
+  assert.strictEqual(result, 'hello world');
+}
+
+// Ringbuffer grow — push > 16 items without consumer draining
+async function testRingbufferGrow() {
+  const { writer, broadcast: bc } = broadcast({ highWaterMark: 32 });
+  const consumer = bc.push();
+
+  // Push 20 items (exceeds default ringbuffer capacity of 16)
+  for (let i = 0; i < 20; i++) {
+    writer.writeSync(new Uint8Array([i]));
+  }
+  writer.endSync();
+
+  // Read all items back and verify order
+  const items = [];
+  for await (const batch of consumer) {
+    for (const chunk of batch) {
+      items.push(chunk[0]);
+    }
+  }
+  assert.strictEqual(items.length, 20);
+  for (let i = 0; i < 20; i++) {
+    assert.strictEqual(items[i], i);
+  }
+}
+
+// Broadcast drainableProtocol after close returns null
+async function testDrainableAfterClose() {
+  const { drainableProtocol } = require('stream/iter');
+  const { writer } = broadcast();
+  writer.endSync();
+  const result = writer[drainableProtocol]();
+  // After close, desired should be null
+  assert.strictEqual(result, null);
+}
+
+Promise.all([
+  testBroadcastWriteAbort(),
+  testBroadcastFromSyncIterable(),
+  testBroadcastFromSyncIterableStrings(),
+  testRingbufferGrow(),
+  testDrainableAfterClose(),
+]).then(common.mustCall());

test/parallel/test-stream-iter-from-coverage.js

@@ -0,0 +1,144 @@
+// Flags: --experimental-stream-iter
+'use strict';
+
+// Coverage tests for from.js: sub-batching >128, DataView in generator,
+// non-Uint8Array TypedArray normalization.
+
+const common = require('../common');
+const assert = require('assert');
+const {
+  from,
+  fromSync,
+  bytes,
+  bytesSync,
+} = require('stream/iter');
+
+// fromSync: Uint8Array[] with > 128 elements triggers sub-batching
+async function testFromSyncSubBatching() {
+  const bigBatch = Array.from({ length: 200 },
+                              (_, i) => new Uint8Array([i & 0xFF]));
+  const batches = [];
+  for (const batch of fromSync(bigBatch)) {
+    batches.push(batch);
+  }
+  // Should be split into sub-batches: 128 + 72
+  assert.strictEqual(batches.length, 2);
+  assert.strictEqual(batches[0].length, 128);
+  assert.strictEqual(batches[1].length, 72);
+  // Verify no data loss
+  let totalChunks = 0;
+  for (const batch of batches) totalChunks += batch.length;
+  assert.strictEqual(totalChunks, 200);
+}
+
+// from: Uint8Array[] with > 128 elements triggers sub-batching (async)
+async function testFromAsyncSubBatching() {
+  const bigBatch = Array.from({ length: 200 },
+                              (_, i) => new Uint8Array([i & 0xFF]));
+  const batches = [];
+  for await (const batch of from(bigBatch)) {
+    batches.push(batch);
+  }
+  assert.strictEqual(batches.length, 2);
+  assert.strictEqual(batches[0].length, 128);
+  assert.strictEqual(batches[1].length, 72);
+}
+
+// Exact boundary: 128 elements → single batch (no split)
+async function testFromSubBatchingBoundary() {
+  const exactBatch = Array.from({ length: 128 },
+                                (_, i) => new Uint8Array([i]));
+  const batches = [];
+  for (const batch of fromSync(exactBatch)) {
+    batches.push(batch);
+  }
+  assert.strictEqual(batches.length, 1);
+  assert.strictEqual(batches[0].length, 128);
+}
+
+// 129 elements → 2 batches (128 + 1)
+async function testFromSubBatchingBoundaryPlus1() {
+  const batch129 = Array.from({ length: 129 },
+                              (_, i) => new Uint8Array([i & 0xFF]));
+  const batches = [];
+  for await (const batch of from(batch129)) {
+    batches.push(batch);
+  }
+  assert.strictEqual(batches.length, 2);
+  assert.strictEqual(batches[0].length, 128);
+  assert.strictEqual(batches[1].length, 1);
+}
+
+// DataView yielded from a sync generator → normalizeSyncValue path
+async function testFromSyncDataViewInGenerator() {
+  function* gen() {
+    const buf = new ArrayBuffer(3);
+    const dv = new DataView(buf);
+    dv.setUint8(0, 65);
+    dv.setUint8(1, 66);
+    dv.setUint8(2, 67);
+    yield dv;
+  }
+  const data = bytesSync(fromSync(gen()));
+  assert.deepStrictEqual(data, new Uint8Array([65, 66, 67]));
+}
+
+// DataView yielded from an async generator → normalizeAsyncValue path
+async function testFromAsyncDataViewInGenerator() {
+  async function* gen() {
+    const buf = new ArrayBuffer(3);
+    const dv = new DataView(buf);
+    dv.setUint8(0, 68);
+    dv.setUint8(1, 69);
+    dv.setUint8(2, 70);
+    yield dv;
+  }
+  const data = await bytes(from(gen()));
+  assert.deepStrictEqual(data, new Uint8Array([68, 69, 70]));
+}
+
+// Int16Array yielded from generator → primitiveToUint8Array fallback
+async function testFromSyncInt16ArrayInGenerator() {
+  function* gen() {
+    yield new Int16Array([0x0102, 0x0304]);
+  }
+  const data = bytesSync(fromSync(gen()));
+  assert.strictEqual(data.byteLength, 4); // 2 int16 = 4 bytes
+}
+
+// Float64Array as top-level input to from()
+async function testFromFloat64Array() {
+  const f64 = new Float64Array([1.0]);
+  const batches = [];
+  for await (const batch of from(f64)) {
+    batches.push(batch);
+  }
+  assert.strictEqual(batches.length, 1);
+  assert.strictEqual(batches[0][0].byteLength, 8); // 1 float64 = 8 bytes
+}
+
+// Sync generator yielding invalid type → ERR_INVALID_ARG_TYPE
+async function testFromSyncInvalidYield() {
+  function* gen() {
+    yield 42; // Not a valid stream value
+  }
+  assert.throws(
+    () => {
+      // eslint-disable-next-line no-unused-vars
+      for (const batch of fromSync(gen())) { /* consume */ }
+    },
+    { code: 'ERR_INVALID_ARG_TYPE' },
+  );
+}
+
+Promise.all([
+  testFromSyncSubBatching(),
+  testFromAsyncSubBatching(),
+  testFromSubBatchingBoundary(),
+  testFromSubBatchingBoundaryPlus1(),
+  testFromSyncDataViewInGenerator(),
+  testFromAsyncDataViewInGenerator(),
+  testFromSyncInt16ArrayInGenerator(),
+  testFromFloat64Array(),
+  testFromSyncInvalidYield(),
+]).then(common.mustCall());

test/parallel/test-stream-iter-pipeto-edge.js

@@ -0,0 +1,68 @@
+// Flags: --experimental-stream-iter
+'use strict';
+
+// Edge case tests for pipeToSync: endSync fallback, preventFail.
+
+const common = require('../common');
+const assert = require('assert');
+const { pipeToSync, fromSync } = require('stream/iter');
+
+// pipeToSync endSync returns negative → falls back to end()
+async function testPipeToSyncEndSyncFallback() {
+  let endCalled = false;
+  const writer = {
+    writeSync() { return true; },
+    endSync() { return -1; }, // Negative → triggers end() fallback
+    end() { endCalled = true; },
+  };
+  pipeToSync(fromSync('data'), writer);
+  assert.strictEqual(endCalled, true);
+}
+
+// pipeToSync endSync missing → falls back to end()
+async function testPipeToSyncNoEndSync() {
+  let endCalled = false;
+  const writer = {
+    writeSync() { return true; },
+    end() { endCalled = true; },
+  };
+  pipeToSync(fromSync('data'), writer);
+  assert.strictEqual(endCalled, true);
+}
+
+// pipeToSync with preventFail: true — source error does NOT call fail()
+async function testPipeToSyncPreventFail() {
+  let failCalled = false;
+  const writer = {
+    writeSync() { return true; },
+    endSync() { return 0; },
+    fail() { failCalled = true; },
+  };
+  function* badSource() {
+    yield [new Uint8Array([1])];
+    throw new Error('source error');
+  }
+  assert.throws(
+    () => pipeToSync(badSource(), writer, { preventFail: true }),
+    { message: 'source error' },
+  );
+  assert.strictEqual(failCalled, false);
+}
+
+// pipeToSync with preventClose: true — end/endSync not called
+async function testPipeToSyncPreventClose() {
+  let endCalled = false;
+  const writer = {
+    writeSync() { return true; },
+    endSync() { endCalled = true; return 0; },
+  };
+  pipeToSync(fromSync('data'), writer, { preventClose: true });
+  assert.strictEqual(endCalled, false);
+}
+
+Promise.all([
+  testPipeToSyncEndSyncFallback(),
+  testPipeToSyncNoEndSync(),
+  testPipeToSyncPreventFail(),
+  testPipeToSyncPreventClose(),
+]).then(common.mustCall());