test(stream): ensure it writes head & pipe

src/stream.ts

@@ -36,12 +36,12 @@ export type HeaderStream = NodeJS.WritableStream & WriteHeaders
 export class Stream extends Transform {
   readonly #uid: string
 
-  constructor(uid: string, request: IncomingMessage) {
+  constructor(uid: string, request?: IncomingMessage) {
     super({ objectMode: true })
 
     this.#uid = uid
 
-    if (request.socket) {
+    if (request?.socket) {
       request.socket.setKeepAlive(true)
       request.socket.setNoDelay(true)
       request.socket.setTimeout(0)

test-helpers/sink.ts

@@ -0,0 +1,26 @@
+import { Writable } from 'node:stream'
+import type { HeaderStream } from '../src/stream.js'
+
+export class Sink extends Writable implements HeaderStream {
+  #chunks: Buffer[] = []
+
+  constructor() {
+    super({ objectMode: true })
+  }
+
+  assertWriteHead(assertion: (statusCode: number, headers: any) => void) {
+    // @ts-expect-error - Mocking the writeHead method
+    this.writeHead = (statusCode, headers) => {
+      assertion(statusCode, headers)
+    }
+  }
+
+  get content() {
+    return this.#chunks.join('')
+  }
+
+  _write(chunk: Buffer, _encoding: BufferEncoding, callback: (error?: Error | null) => void): void {
+    this.#chunks.push(chunk)
+    callback()
+  }
+}

tests/stream.spec.ts

@@ -0,0 +1,45 @@
+import { randomUUID } from 'node:crypto'
+import { test } from '@japa/runner'
+import { Stream } from '../src/stream.js'
+import { Sink } from '../test-helpers/sink.js'
+
+test.group('Stream', () => {
+  test('should write multiple chunks to the stream', async ({ assert }) => {
+    const stream = new Stream(randomUUID())
+    const sink = new Sink()
+    stream.pipe(sink)
+
+    stream.writeMessage({ data: { channel: 'foo', payload: 'bar' } })
+    stream.writeMessage({ data: { channel: 'baz', payload: 'qux' } })
+
+    assert.equal(
+      sink.content,
+      [
+        `:ok\n\n`,
+        `data: {"channel":"foo","payload":"bar"}\n\n`,
+        `data: {"channel":"baz","payload":"qux"}\n\n`,
+      ].join('')
+    )
+  })
+
+  test('sets headers on the response', async ({ assert }) => {
+    assert.plan(2)
+
+    const stream = new Stream(randomUUID())
+    const sink = new Sink()
+
+    sink.assertWriteHead((statusCode, headers) => {
+      assert.equal(statusCode, 200)
+      assert.deepEqual(headers, {
+        'Cache-Control': 'private, no-cache, no-store, must-revalidate, max-age=0, no-transform',
+        'Connection': 'keep-alive',
+        'Content-Type': 'text/event-stream',
+        'Expire': '0',
+        'Pragma': 'no-cache',
+        'X-Accel-Buffering': 'no',
+      })
+    })
+
+    stream.pipe(sink)
+  })
+})