Nora
209 lines
8.0 KiB
TypeScript
209 lines
8.0 KiB
TypeScript
// @vitest-environment node
|
|
// `streamTail` pipes a `fetch()` response body through a `TextDecoderStream`
|
|
// and a manual NDJSON-line splitter. We mock `fetch` with `vi.stubGlobal`
|
|
// and drive a fake `ReadableStream` from the test, so no DOM/happy-dom
|
|
// is needed — Node 22's globals (Response, ReadableStream, TextDecoderStream)
|
|
// are sufficient.
|
|
|
|
import { afterEach, describe, expect, it, vi } from "vitest";
|
|
import { streamTail, type TailEvent } from "./tail-stream";
|
|
|
|
/**
|
|
* Build a controllable ReadableStream<Uint8Array> and a tiny driver. The
|
|
* stream's controller is captured at construction so the test can decide
|
|
* exactly when each NDJSON line is enqueued and when to close. The body
|
|
* is wrapped in a `Response` (matching what `fetch()` would return).
|
|
*/
|
|
function makeStream() {
|
|
const encoder = new TextEncoder();
|
|
let controllerRef: ReadableStreamDefaultController<Uint8Array> | undefined;
|
|
const stream = new ReadableStream<Uint8Array>({
|
|
start(c) {
|
|
controllerRef = c;
|
|
},
|
|
});
|
|
return {
|
|
response: new Response(stream, {
|
|
status: 200,
|
|
headers: { "content-type": "application/x-ndjson" },
|
|
}),
|
|
push: (line: string) => {
|
|
controllerRef?.enqueue(encoder.encode(line + "\n"));
|
|
},
|
|
close: () => controllerRef?.close(),
|
|
};
|
|
}
|
|
|
|
/**
|
|
* Drain an async iterable into an array. Useful for asserting the full
|
|
* emission list without manually awaiting each `.next()`.
|
|
*/
|
|
async function collect<T>(iter: AsyncIterableIterator<T>): Promise<T[]> {
|
|
const out: T[] = [];
|
|
for await (const v of iter) out.push(v);
|
|
return out;
|
|
}
|
|
|
|
describe("streamTail", () => {
|
|
afterEach(() => {
|
|
vi.unstubAllGlobals();
|
|
});
|
|
|
|
it("test_parses_well_formed_ndjson_into_typed_events", async () => {
|
|
const driver = makeStream();
|
|
const fetchMock = vi.fn().mockResolvedValue(driver.response);
|
|
vi.stubGlobal("fetch", fetchMock);
|
|
|
|
const gen = streamTail("claims");
|
|
driver.push('{"type":"item","data":{"id":"CLM-1"}}');
|
|
driver.push('{"type":"item","data":{"id":"CLM-2"}}');
|
|
driver.push('{"type":"item","data":{"id":"CLM-3"}}');
|
|
driver.close();
|
|
|
|
const first = await gen.next();
|
|
const second = await gen.next();
|
|
const third = await gen.next();
|
|
const tail = await gen.next();
|
|
|
|
expect(first.done).toBe(false);
|
|
expect(second.done).toBe(false);
|
|
expect(third.done).toBe(false);
|
|
expect(tail.done).toBe(true);
|
|
expect([first.value, second.value, third.value]).toEqual<TailEvent[]>([
|
|
{ type: "item", data: { id: "CLM-1" } },
|
|
{ type: "item", data: { id: "CLM-2" } },
|
|
{ type: "item", data: { id: "CLM-3" } },
|
|
]);
|
|
|
|
// The fetch call itself must target /api/claims/stream with the
|
|
// NDJSON Accept header.
|
|
expect(fetchMock).toHaveBeenCalledTimes(1);
|
|
const [calledUrl, calledInit] = fetchMock.mock.calls[0] as [string, RequestInit];
|
|
expect(calledUrl).toBe("/api/claims/stream");
|
|
expect(calledInit.headers).toMatchObject({ Accept: "application/x-ndjson" });
|
|
});
|
|
|
|
it("test_yields_typed_error_on_error_line", async () => {
|
|
const driver = makeStream();
|
|
vi.stubGlobal("fetch", vi.fn().mockResolvedValue(driver.response));
|
|
const gen = streamTail("remittances");
|
|
driver.push('{"type":"error","data":{"message":"x"}}');
|
|
driver.close();
|
|
const events = await collect(gen);
|
|
expect(events).toEqual<TailEvent[]>([
|
|
{ type: "error", data: { message: "x" } },
|
|
]);
|
|
});
|
|
|
|
it("test_targets_acks_stream_endpoint", async () => {
|
|
// SP25: the `acks` resource is part of TailResource; the URL must
|
|
// be /api/acks/stream so useTailStream("acks") can open it.
|
|
const driver = makeStream();
|
|
const fetchMock = vi.fn().mockResolvedValue(driver.response);
|
|
vi.stubGlobal("fetch", fetchMock);
|
|
const gen = streamTail("acks");
|
|
// Drive the generator one step so the fetch() call is reached
|
|
// before we close the stream.
|
|
await gen.next();
|
|
driver.close();
|
|
await gen.return?.(undefined);
|
|
expect(fetchMock).toHaveBeenCalledTimes(1);
|
|
const [calledUrl] = fetchMock.mock.calls[0] as [string];
|
|
expect(calledUrl).toBe("/api/acks/stream");
|
|
});
|
|
|
|
it("test_targets_ta1_acks_stream_endpoint", async () => {
|
|
// SP25: the `ta1_acks` resource is part of TailResource; the URL
|
|
// must be /api/ta1-acks/stream.
|
|
const driver = makeStream();
|
|
const fetchMock = vi.fn().mockResolvedValue(driver.response);
|
|
vi.stubGlobal("fetch", fetchMock);
|
|
const gen = streamTail("ta1_acks");
|
|
await gen.next();
|
|
driver.close();
|
|
await gen.return?.(undefined);
|
|
expect(fetchMock).toHaveBeenCalledTimes(1);
|
|
const [calledUrl] = fetchMock.mock.calls[0] as [string];
|
|
expect(calledUrl).toBe("/api/ta1-acks/stream");
|
|
});
|
|
|
|
it("test_targets_claim_acks_stream_endpoint", async () => {
|
|
// SP28: the `claim-acks` resource is part of TailResource; the URL
|
|
// must be /api/claim-acks/stream. The claim↔ack link stream is a
|
|
// shared resource (not per-claim or per-ack) — both the ClaimDrawer
|
|
// and AckDrawer subscribe to the same endpoint and filter
|
|
// server-side via the page-specific filterFn passed to
|
|
// `useMergedTail`.
|
|
const driver = makeStream();
|
|
const fetchMock = vi.fn().mockResolvedValue(driver.response);
|
|
vi.stubGlobal("fetch", fetchMock);
|
|
// Drive an event so the iterator's first read resolves — same
|
|
// pattern as `test_parses_well_formed_ndjson_into_typed_events`
|
|
// (the `test_targets_acks_stream_endpoint` shape hangs in CI
|
|
// because it awaits `gen.next()` before any chunk arrives).
|
|
const gen = streamTail("claim-acks");
|
|
driver.push('{"type":"snapshot_end","data":{"count":0}}');
|
|
driver.close();
|
|
await gen.next();
|
|
await gen.return?.(undefined);
|
|
expect(fetchMock).toHaveBeenCalledTimes(1);
|
|
const [calledUrl] = fetchMock.mock.calls[0] as [string];
|
|
expect(calledUrl).toBe("/api/claim-acks/stream");
|
|
});
|
|
|
|
it("test_handles_heartbeat_silently", async () => {
|
|
// "Silently" here means: still yields the correctly-typed heartbeat
|
|
// (no special filtering) — the consumer (useTailStream) decides what
|
|
// to do with it. The lib just has to forward it.
|
|
const driver = makeStream();
|
|
vi.stubGlobal("fetch", vi.fn().mockResolvedValue(driver.response));
|
|
const gen = streamTail("activity");
|
|
driver.push('{"type":"heartbeat","data":{"ts":"2026-06-20T12:00:00Z"}}');
|
|
driver.close();
|
|
const events = await collect(gen);
|
|
expect(events).toEqual<TailEvent[]>([
|
|
{ type: "heartbeat", data: { ts: "2026-06-20T12:00:00Z" } },
|
|
]);
|
|
});
|
|
|
|
it("test_abort_signal_cancels_mid_stream", async () => {
|
|
const driver = makeStream();
|
|
vi.stubGlobal("fetch", vi.fn().mockResolvedValue(driver.response));
|
|
const ac = new AbortController();
|
|
const gen = streamTail("claims", { signal: ac.signal });
|
|
|
|
// First event arrives normally.
|
|
driver.push('{"type":"item","data":{"id":"CLM-1"}}');
|
|
const first = await gen.next();
|
|
expect(first.done).toBe(false);
|
|
|
|
// Abort before any more events arrive, then close the upstream so
|
|
// any pending read completes. The generator must exit cleanly
|
|
// (no throw) and not yield further.
|
|
ac.abort();
|
|
driver.close();
|
|
const second = await gen.next();
|
|
expect(second.done).toBe(true);
|
|
});
|
|
|
|
it("test_malformed_line_skipped_next_valid_still_emitted", async () => {
|
|
const warnSpy = vi.spyOn(console, "warn").mockImplementation(() => {});
|
|
const driver = makeStream();
|
|
vi.stubGlobal("fetch", vi.fn().mockResolvedValue(driver.response));
|
|
const gen = streamTail("claims");
|
|
driver.push('{"type":"item","data":{"id":"CLM-1"}}');
|
|
driver.push("this is not json");
|
|
driver.push('{"type":"item","data":{"id":"CLM-2"}}');
|
|
driver.close();
|
|
const events = await collect(gen);
|
|
expect(events).toEqual<TailEvent[]>([
|
|
{ type: "item", data: { id: "CLM-1" } },
|
|
{ type: "item", data: { id: "CLM-2" } },
|
|
]);
|
|
// The malformed line must produce a console.warn so operators can spot
|
|
// a buggy backend without crashing the whole tail.
|
|
expect(warnSpy).toHaveBeenCalled();
|
|
warnSpy.mockRestore();
|
|
});
|
|
});
|