test: add unit tests for playTranscodedPreparedStream and transcoder functionality

tests/streaming/playTranscode.test.ts

@@ -0,0 +1,59 @@
+import { describe, it, expect, vi } from "vitest";
+import { PassThrough } from "node:stream";
+
+vi.mock("node:child_process", async () => {
+  const actual = await vi.importActual("node:child_process");
+  return {
+    ...actual,
+    spawn: (cmd: string, args: string[], opts: any) => {
+      const stdout = new PassThrough();
+      const stderr = new PassThrough();
+      const listeners: Record<string, Function[]> = {};
+      const proc: any = {
+        stdout,
+        stderr,
+        kill: vi.fn(() => {
+          (listeners.exit || []).forEach((fn) => fn(0, "SIGKILL"));
+        }),
+        on: (ev: string, fn: Function) => {
+          listeners[ev] = listeners[ev] || [];
+          listeners[ev].push(fn);
+        },
+        off: (ev: string, fn: Function) => {
+          listeners[ev] = (listeners[ev] || []).filter((f) => f !== fn);
+        },
+        stdoutWrite: (d: Buffer | string) => stdout.write(d),
+      };
+      setTimeout(() => {
+        (listeners.exit || []).forEach((fn) => fn(null, null));
+      }, 10);
+      return proc;
+    },
+  };
+});
+
+import { playTranscodedPreparedStream } from "../../src/streaming/index";
+
+describe("playTranscodedPreparedStream", () => {
+  it("pipes transcoder output to session and broadcasts to web", async () => {
+    // mock global broadcast
+    const broadcasts: Buffer[] = [];
+    (globalThis as any).broadcastVideoToWeb = (chunk: Buffer) => broadcasts.push(Buffer.from(chunk));
+
+    const session = {
+      connection: { channel: { id: "c" } },
+      stream: { playVideo: () => null, playAudio: () => null },
+      play: vi.fn().mockImplementation(async (readable) => {
+        // consume a bit from readable to simulate playback
+        readable.on("data", (d: Buffer) => {});
+        // resolve after a short delay
+        await new Promise((r) => setTimeout(r, 5));
+      }),
+      stop: vi.fn(),
+    } as any;
+
+    await playTranscodedPreparedStream("http://example.test/stream", session, { fps: 30 });
+    expect(session.play).toHaveBeenCalled();
+    expect(broadcasts.length).toBeGreaterThanOrEqual(0);
+  });
+});

tests/streaming/transcoder.test.ts

@@ -0,0 +1,52 @@
+import { describe, it, expect, vi } from "vitest";
+import { PassThrough } from "node:stream";
+
+// Mock spawn to avoid calling real ffmpeg
+vi.mock("node:child_process", async () => {
+  const actual = await vi.importActual("node:child_process");
+  return {
+    ...actual,
+    spawn: (cmd: string, args: string[], opts: any) => {
+      const stdout = new PassThrough();
+      const stderr = new PassThrough();
+      const listeners: Record<string, Function[]> = {};
+      const proc: any = {
+        stdout,
+        stderr,
+        kill: vi.fn(() => {
+          // emit exit when killed
+          (listeners.exit || []).forEach((fn) => fn(0, "SIGKILL"));
+        }),
+        on: (ev: string, fn: Function) => {
+          listeners[ev] = listeners[ev] || [];
+          listeners[ev].push(fn);
+        },
+        off: (ev: string, fn: Function) => {
+          listeners[ev] = (listeners[ev] || []).filter((f) => f !== fn);
+        },
+        stdoutWrite: (d: Buffer | string) => stdout.write(d),
+      };
+      // simulate async start
+      setTimeout(() => {
+        (listeners.exit || []).forEach((fn) => fn(null, null));
+      }, 10);
+      return proc;
+    },
+  };
+});
+
+import { prepareTranscoder } from "../../src/streaming/transcoder";
+
+describe("Transcoder", () => {
+  it("starts ffmpeg and returns output stream and command", () => {
+    const { transcoder, command, output } = prepareTranscoder("http://example.test/video", { fps: 24 });
+    expect(transcoder).toBeTruthy();
+    expect(command).toBeTruthy();
+    expect(output).toBeTruthy();
+    expect(typeof command.kill).toBe("function");
+    // write some data and ensure output is readable
+    const wrote = command.stdoutWrite?.("hello");
+    expect(output.readable).toBe(true);
+    transcoder.stop();
+  });
+});