From dfabf5564407ae0b13f51bea475e3c25cff5771a Mon Sep 17 00:00:00 2001 From: Theo Zourzouvillys Date: Tue, 24 Mar 2026 04:10:56 -0700 Subject: [PATCH] Add full boat and cloud server API coverage Add history(), liveness(), readiness() to Client. Add health/liveness/readiness to CloudClient. Add bus field to Filter/Frame/Device/SendParams, decoded field to Frame, device_removed SSE event type, SubscribeOptions with decode param, HistoryParams, expanded HealthStatus with broker/replication sub-objects, and full ReplicationStatus counters. Update CLI to handle device_removed events. --- CLAUDE.md | 12 +- packages/lplex-cli/src/main.ts | 4 + packages/lplex/src/client.ts | 110 ++++++++++++++--- packages/lplex/src/cloud.ts | 40 ++++++ packages/lplex/src/index.ts | 5 + packages/lplex/src/sse.ts | 14 ++- packages/lplex/src/types.ts | 80 +++++++++++- packages/lplex/test/client.test.ts | 192 +++++++++++++++++++++++++++++ packages/lplex/test/cloud.test.ts | 60 +++++++++ packages/lplex/test/sse.test.ts | 15 +++ 10 files changed, 506 insertions(+), 26 deletions(-) diff --git a/CLAUDE.md b/CLAUDE.md index 2553b25..63a6397 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -31,15 +31,17 @@ npx tsx src/main.ts --server http://inuc1.local:8089 | File | Owns | |---|---| -| `src/types.ts` | `Frame`, `Device`, `Event` (discriminated union), `Filter`, `PGNValue`, `DeviceValues`, `SessionConfig`, `SessionInfo`, `SendParams` | +| `src/types.ts` | `Frame`, `Device`, `DeviceRemoved`, `Event` (discriminated union), `Filter`, `SubscribeOptions`, `PGNValue`, `DeviceValues`, `DecodedPGNValue`, `DecodedDeviceValues`, `QueryParams`, `HistoryParams`, `BrokerHealth`, `ReplicationHealth`, `HealthStatus`, `ReplicationStatus`, `SessionConfig`, `SessionInfo`, `SendParams`, `InstanceSummary`, `InstanceStatus`, `SeqRange`, `ReplicationEvent`, `ReplicationEventType` | | `src/errors.ts` | `LplexError`, `HttpError` | -| `src/sse.ts` | `parseSSE` async generator: reads `data:` lines from `ReadableStream`, yields `Event` objects | -| `src/client.ts` | `Client` class: `devices()`, `values()`, `subscribe()`, `send()`, `createSession()`. Injectable `fetch`. | +| `src/sse.ts` | `parseSSE` async generator: reads `data:` lines from `ReadableStream`, yields `Event` objects (frame, device, device_removed) | +| `src/client.ts` | `Client` class: `devices()`, `values()`, `decodedValues()`, `subscribe()`, `send()`, `query()`, `history()`, `health()`, `liveness()`, `readiness()`, `replicationStatus()`, `createSession()`. Injectable `fetch`. | +| `src/cloud.ts` | `CloudClient` class: `client()`, `instances()`, `status()`, `replicationEvents()`, `health()`, `liveness()`, `readiness()`. | | `src/session.ts` | `Session` class: `subscribe()`, `ack()`, `info`, `lastAckedSeq` | | `src/index.ts` | Barrel exports | | `README.md` | npm package page README | -| `test/sse.test.ts` | SSE parser unit tests (8 tests) | -| `test/client.test.ts` | Client + Session tests with injected fetch (14 tests) | +| `test/sse.test.ts` | SSE parser unit tests (9 tests) | +| `test/client.test.ts` | Client + Session tests with injected fetch (40 tests) | +| `test/cloud.test.ts` | CloudClient tests with injected fetch (12 tests) | ### packages/lplex-cli/ File Map diff --git a/packages/lplex-cli/src/main.ts b/packages/lplex-cli/src/main.ts index 65a73b6..9e8d43e 100644 --- a/packages/lplex-cli/src/main.ts +++ b/packages/lplex-cli/src/main.ts @@ -197,6 +197,8 @@ async function runEphemeral(client: Client, devices: DeviceMap): Promise { if (aborted) break; if (event.type === "device") { handleDevice(event.device, devices); + } else if (event.type === "device_removed") { + devices.delete(event.deviceRemoved.src); } else { writeFrame(event, devices); } @@ -242,6 +244,8 @@ async function runBuffered(client: Client, devices: DeviceMap): Promise { if (aborted) break; if (event.type === "device") { handleDevice(event.device, devices); + } else if (event.type === "device_removed") { + devices.delete(event.deviceRemoved.src); } else { lastSeq = event.frame.seq; writeFrame(event, devices); diff --git a/packages/lplex/src/client.ts b/packages/lplex/src/client.ts index 1e4d181..dce6bba 100644 --- a/packages/lplex/src/client.ts +++ b/packages/lplex/src/client.ts @@ -9,11 +9,13 @@ import type { Filter, Frame, HealthStatus, + HistoryParams, QueryParams, ReplicationStatus, SendParams, SessionConfig, SessionInfo, + SubscribeOptions, } from "./types.js"; type FetchFn = typeof globalThis.fetch; @@ -62,18 +64,33 @@ export class Client { /** * Open an ephemeral SSE stream with optional filtering. * No session, no replay, no ACK. + * + * Accepts either a Filter (for backwards compatibility) or + * SubscribeOptions for additional control (decode, signal). */ async subscribe( - filter?: Filter, + filterOrOptions?: Filter | SubscribeOptions, signal?: AbortSignal, ): Promise> { + let filter: Filter | undefined; + let decode: boolean | undefined; + let sig: AbortSignal | undefined = signal; + + if (filterOrOptions && isSubscribeOptions(filterOrOptions)) { + filter = filterOrOptions.filter; + decode = filterOrOptions.decode; + sig = filterOrOptions.signal ?? sig; + } else { + filter = filterOrOptions as Filter | undefined; + } + let url = `${this.#baseURL}/events`; - const qs = filterToQueryString(filter); + const qs = filterToQueryString(filter, decode); if (qs) url += `?${qs}`; const resp = await this.#fetch(url, { headers: { Accept: "text/event-stream" }, - signal, + signal: sig, }); if (!resp.ok) { @@ -124,7 +141,7 @@ export class Client { /** * Send an ISO Request (PGN 59904) and wait for the response frame. - * Returns the response frame, or throws HttpError with status 504 on timeout. + * Returns the response frame, or throws HttpError on timeout (408). */ async query(params: QueryParams, signal?: AbortSignal): Promise { const url = `${this.#baseURL}/query`; @@ -143,7 +160,32 @@ export class Client { return resp.json() as Promise; } - /** Check server health. */ + /** + * Query historical frames (requires journaling on the server). + * Returns an array of frames matching the query parameters. + */ + async history(params: HistoryParams, signal?: AbortSignal): Promise { + const qs = new URLSearchParams(); + qs.set("from", params.from); + if (params.to) qs.set("to", params.to); + for (const p of params.pgn ?? []) qs.append("pgn", p.toString()); + for (const s of params.src ?? []) qs.append("src", s.toString()); + if (params.limit !== undefined) qs.set("limit", params.limit.toString()); + if (params.interval) qs.set("interval", params.interval); + if (params.decode) qs.set("decode", "true"); + + const url = `${this.#baseURL}/history?${qs.toString()}`; + const resp = await this.#fetch(url, { signal }); + + if (!resp.ok) { + const body = await resp.text(); + throw new HttpError("GET", url, resp.status, body); + } + + return resp.json() as Promise; + } + + /** Check server health (GET /healthz). */ async health(signal?: AbortSignal): Promise { const url = `${this.#baseURL}/healthz`; const resp = await this.#fetch(url, { signal }); @@ -156,6 +198,32 @@ export class Client { return resp.json() as Promise; } + /** Liveness probe (GET /livez). */ + async liveness(signal?: AbortSignal): Promise { + const url = `${this.#baseURL}/livez`; + const resp = await this.#fetch(url, { signal }); + + if (!resp.ok) { + const body = await resp.text(); + throw new HttpError("GET", url, resp.status, body); + } + + return resp.json() as Promise; + } + + /** Readiness probe (GET /readyz). */ + async readiness(signal?: AbortSignal): Promise { + const url = `${this.#baseURL}/readyz`; + const resp = await this.#fetch(url, { signal }); + + if (!resp.ok) { + const body = await resp.text(); + throw new HttpError("GET", url, resp.status, body); + } + + return resp.json() as Promise; + } + /** Fetch boat-side replication status (only available when replication is configured). */ async replicationStatus(signal?: AbortSignal): Promise { const url = `${this.#baseURL}/replication/status`; @@ -200,6 +268,12 @@ export class Client { } } +function isSubscribeOptions( + obj: Filter | SubscribeOptions, +): obj is SubscribeOptions { + return "decode" in obj || "signal" in obj || "filter" in obj; +} + function filterIsEmpty(f: Filter): boolean { return ( !f.pgn?.length && @@ -207,21 +281,26 @@ function filterIsEmpty(f: Filter): boolean { !f.manufacturer?.length && !f.instance?.length && !f.name?.length && - !f.exclude_name?.length + !f.exclude_name?.length && + !f.bus?.length ); } -function filterToQueryString(f?: Filter): string { - if (!f || filterIsEmpty(f)) return ""; +function filterToQueryString(f?: Filter, decode?: boolean): string { + if ((!f || filterIsEmpty(f)) && !decode) return ""; const params = new URLSearchParams(); - for (const p of f.pgn ?? []) params.append("pgn", p.toString()); - for (const p of f.exclude_pgn ?? []) - params.append("exclude_pgn", p.toString()); - for (const m of f.manufacturer ?? []) params.append("manufacturer", m); - for (const i of f.instance ?? []) params.append("instance", i.toString()); - for (const n of f.name ?? []) params.append("name", n); - for (const n of f.exclude_name ?? []) params.append("exclude_name", n); + if (f) { + for (const p of f.pgn ?? []) params.append("pgn", p.toString()); + for (const p of f.exclude_pgn ?? []) + params.append("exclude_pgn", p.toString()); + for (const m of f.manufacturer ?? []) params.append("manufacturer", m); + for (const i of f.instance ?? []) params.append("instance", i.toString()); + for (const n of f.name ?? []) params.append("name", n); + for (const n of f.exclude_name ?? []) params.append("exclude_name", n); + for (const b of f.bus ?? []) params.append("bus", b); + } + if (decode) params.set("decode", "true"); return params.toString(); } @@ -233,5 +312,6 @@ function filterToJSON(f: Filter): Record { if (f.instance?.length) m.instance = f.instance; if (f.name?.length) m.name = f.name; if (f.exclude_name?.length) m.exclude_name = f.exclude_name; + if (f.bus?.length) m.bus = f.bus; return m; } diff --git a/packages/lplex/src/cloud.ts b/packages/lplex/src/cloud.ts index 532bf17..049fed8 100644 --- a/packages/lplex/src/cloud.ts +++ b/packages/lplex/src/cloud.ts @@ -1,6 +1,7 @@ import { Client, type ClientOptions } from "./client.js"; import { HttpError } from "./errors.js"; import type { + HealthStatus, InstanceStatus, InstanceSummary, ReplicationEvent, @@ -88,4 +89,43 @@ export class CloudClient { return resp.json() as Promise; } + + /** Check cloud server health (GET /healthz). */ + async health(signal?: AbortSignal): Promise { + const url = `${this.#baseURL}/healthz`; + const resp = await this.#fetch(url, { signal }); + + if (!resp.ok) { + const body = await resp.text(); + throw new HttpError("GET", url, resp.status, body); + } + + return resp.json() as Promise; + } + + /** Liveness probe (GET /livez). */ + async liveness(signal?: AbortSignal): Promise { + const url = `${this.#baseURL}/livez`; + const resp = await this.#fetch(url, { signal }); + + if (!resp.ok) { + const body = await resp.text(); + throw new HttpError("GET", url, resp.status, body); + } + + return resp.json() as Promise; + } + + /** Readiness probe (GET /readyz). */ + async readiness(signal?: AbortSignal): Promise { + const url = `${this.#baseURL}/readyz`; + const resp = await this.#fetch(url, { signal }); + + if (!resp.ok) { + const body = await resp.text(); + throw new HttpError("GET", url, resp.status, body); + } + + return resp.json() as Promise; + } } diff --git a/packages/lplex/src/index.ts b/packages/lplex/src/index.ts index 4b56838..bb25b67 100644 --- a/packages/lplex/src/index.ts +++ b/packages/lplex/src/index.ts @@ -7,13 +7,18 @@ export { LplexError, HttpError } from "./errors.js"; export type { Frame, Device, + DeviceRemoved, Event, Filter, + SubscribeOptions, PGNValue, DeviceValues, DecodedPGNValue, DecodedDeviceValues, QueryParams, + HistoryParams, + BrokerHealth, + ReplicationHealth, HealthStatus, ReplicationStatus, SessionConfig, diff --git a/packages/lplex/src/sse.ts b/packages/lplex/src/sse.ts index 1fdb43c..1cedfbb 100644 --- a/packages/lplex/src/sse.ts +++ b/packages/lplex/src/sse.ts @@ -1,4 +1,4 @@ -import type { Device, Event, Frame } from "./types.js"; +import type { Device, DeviceRemoved, Event, Frame } from "./types.js"; /** * Parse an SSE stream from a fetch Response into an async iterable of Events. @@ -67,8 +67,16 @@ export async function* parseSSE( } function classify(obj: Record): Event | null { - if ("type" in obj && obj.type === "device") { - return { type: "device", device: obj as unknown as Device }; + if ("type" in obj) { + if (obj.type === "device") { + return { type: "device", device: obj as unknown as Device }; + } + if (obj.type === "device_removed") { + return { + type: "device_removed", + deviceRemoved: obj as unknown as DeviceRemoved, + }; + } } if ("seq" in obj) { return { type: "frame", frame: obj as unknown as Frame }; diff --git a/packages/lplex/src/types.ts b/packages/lplex/src/types.ts index 654a484..ac40fb0 100644 --- a/packages/lplex/src/types.ts +++ b/packages/lplex/src/types.ts @@ -2,15 +2,18 @@ export interface Frame { seq: number; ts: string; + bus?: string; prio: number; pgn: number; src: number; dst: number; data: string; + decoded?: Record; } /** An NMEA 2000 device discovered on the bus. */ export interface Device { + bus?: string; src: number; name: string; manufacturer: string; @@ -30,10 +33,18 @@ export interface Device { byte_count: number; } +/** A device-removed notification from the bus. */ +export interface DeviceRemoved { + type: "device_removed"; + bus?: string; + src: number; +} + /** Discriminated union for SSE events. */ export type Event = | { type: "frame"; frame: Frame } - | { type: "device"; device: Device }; + | { type: "device"; device: Device } + | { type: "device_removed"; deviceRemoved: DeviceRemoved }; /** * Filter for CAN frames. @@ -46,6 +57,15 @@ export interface Filter { instance?: number[]; name?: string[]; exclude_name?: string[]; + bus?: string[]; +} + +/** Options for ephemeral SSE subscription. */ +export interface SubscribeOptions { + filter?: Filter; + /** When true, frames include decoded field values. */ + decode?: boolean; + signal?: AbortSignal; } /** Configuration for creating a buffered session. */ @@ -70,6 +90,7 @@ export interface SendParams { dst: number; prio: number; data: string; + bus?: string; } // --- Values types --- @@ -112,13 +133,62 @@ export interface DecodedDeviceValues { /** Parameters for an ISO Request query (POST /query). */ export interface QueryParams { pgn: number; - dst: number; + /** Destination address. Defaults to 0xFF (broadcast) on the server. */ + dst?: number; timeout?: string; + bus?: string; +} + +/** Parameters for historical data query (GET /history). */ +export interface HistoryParams { + /** Start timestamp (RFC 3339). */ + from: string; + /** End timestamp (RFC 3339). Defaults to now. */ + to?: string; + /** Filter by PGN(s). */ + pgn?: number[]; + /** Filter by source address(es). */ + src?: number[]; + /** Max frames to return. Defaults to 10000. */ + limit?: number; + /** Downsample interval (e.g. "1s", "PT1M"). */ + interval?: string; + /** Include decoded values in response. */ + decode?: boolean; +} + +// --- Health types --- + +/** Broker health details. */ +export interface BrokerHealth { + status: string; + frames_total: number; + head_seq: number; + last_frame_time: string; + device_count: number; + ring_entries: number; + ring_capacity: number; +} + +/** Replication component health (within health response). */ +export interface ReplicationHealth { + status: string; + connected: boolean; + live_lag: number; + backfill_remaining_seqs: number; + last_ack: string; } -/** Health check response from GET /healthz. */ +/** Health check response from GET /healthz or /readyz. */ export interface HealthStatus { status: string; + broker?: BrokerHealth; + replication?: ReplicationHealth; + components?: Record; + /** Cloud-only: total known instances. */ + instances_total?: number; + /** Cloud-only: currently connected instances. */ + instances_connected?: number; } /** Boat-side replication status from GET /replication/status. */ @@ -131,6 +201,10 @@ export interface ReplicationStatus { live_lag: number; backfill_remaining_seqs: number; last_ack: string; + live_frames_sent: number; + backfill_blocks_sent: number; + backfill_bytes_sent: number; + reconnects: number; } // --- Cloud types --- diff --git a/packages/lplex/test/client.test.ts b/packages/lplex/test/client.test.ts index db6c356..d4c32da 100644 --- a/packages/lplex/test/client.test.ts +++ b/packages/lplex/test/client.test.ts @@ -566,6 +566,198 @@ describe("Client.replicationStatus", () => { }); }); +describe("Client.subscribe with SubscribeOptions", () => { + it("passes decode=true as query param", async () => { + let capturedURL = ""; + const mockFetch = async (url: string | URL | Request) => { + capturedURL = String(url); + return sseResponse(""); + }; + + const client = new Client("http://localhost:8089", { + fetch: mockFetch as typeof fetch, + }); + await client.subscribe({ decode: true }); + const parsed = new URL(capturedURL); + expect(parsed.searchParams.get("decode")).toBe("true"); + }); + + it("passes filter and decode together", async () => { + let capturedURL = ""; + const mockFetch = async (url: string | URL | Request) => { + capturedURL = String(url); + return sseResponse(""); + }; + + const client = new Client("http://localhost:8089", { + fetch: mockFetch as typeof fetch, + }); + await client.subscribe({ + filter: { pgn: [129025] }, + decode: true, + }); + const parsed = new URL(capturedURL); + expect(parsed.searchParams.getAll("pgn")).toEqual(["129025"]); + expect(parsed.searchParams.get("decode")).toBe("true"); + }); +}); + +describe("Client.history", () => { + it("fetches historical frames", async () => { + const frames = [ + { + seq: 1, + ts: "2026-03-01T00:00:00Z", + prio: 6, + pgn: 129025, + src: 1, + dst: 255, + data: "aabb", + }, + ]; + let capturedURL = ""; + const mockFetch = async (url: string | URL | Request) => { + capturedURL = String(url); + return jsonResponse(frames); + }; + + const client = new Client("http://localhost:8089", { + fetch: mockFetch as typeof fetch, + }); + const result = await client.history({ from: "2026-03-01T00:00:00Z" }); + expect(result).toHaveLength(1); + expect(result[0].pgn).toBe(129025); + const parsed = new URL(capturedURL); + expect(parsed.pathname).toBe("/history"); + expect(parsed.searchParams.get("from")).toBe("2026-03-01T00:00:00Z"); + }); + + it("encodes all query params", async () => { + let capturedURL = ""; + const mockFetch = async (url: string | URL | Request) => { + capturedURL = String(url); + return jsonResponse([]); + }; + + const client = new Client("http://localhost:8089", { + fetch: mockFetch as typeof fetch, + }); + await client.history({ + from: "2026-03-01T00:00:00Z", + to: "2026-03-02T00:00:00Z", + pgn: [129025, 129026], + src: [1, 2], + limit: 500, + interval: "1s", + decode: true, + }); + const parsed = new URL(capturedURL); + expect(parsed.searchParams.get("from")).toBe("2026-03-01T00:00:00Z"); + expect(parsed.searchParams.get("to")).toBe("2026-03-02T00:00:00Z"); + expect(parsed.searchParams.getAll("pgn")).toEqual(["129025", "129026"]); + expect(parsed.searchParams.getAll("src")).toEqual(["1", "2"]); + expect(parsed.searchParams.get("limit")).toBe("500"); + expect(parsed.searchParams.get("interval")).toBe("1s"); + expect(parsed.searchParams.get("decode")).toBe("true"); + }); + + it("throws HttpError on 503", async () => { + const mockFetch = async () => errorResponse(503, "journaling not enabled"); + const client = new Client("http://localhost:8089", { + fetch: mockFetch as typeof fetch, + }); + await expect( + client.history({ from: "2026-03-01T00:00:00Z" }), + ).rejects.toThrow(HttpError); + }); +}); + +describe("Client.liveness", () => { + it("returns liveness status", async () => { + const mockFetch = async (url: string | URL | Request) => { + expect(url).toBe("http://localhost:8089/livez"); + return jsonResponse({ status: "ok" }); + }; + + const client = new Client("http://localhost:8089", { + fetch: mockFetch as typeof fetch, + }); + const result = await client.liveness(); + expect(result.status).toBe("ok"); + }); +}); + +describe("Client.readiness", () => { + it("returns readiness status", async () => { + const mockFetch = async (url: string | URL | Request) => { + expect(url).toBe("http://localhost:8089/readyz"); + return jsonResponse({ status: "ok" }); + }; + + const client = new Client("http://localhost:8089", { + fetch: mockFetch as typeof fetch, + }); + const result = await client.readiness(); + expect(result.status).toBe("ok"); + }); + + it("throws HttpError when not ready", async () => { + const mockFetch = async () => errorResponse(503, "not ready"); + const client = new Client("http://localhost:8089", { + fetch: mockFetch as typeof fetch, + }); + await expect(client.readiness()).rejects.toThrow(HttpError); + }); +}); + +describe("Filter with bus", () => { + it("encodes bus filter in query params", async () => { + let capturedURL = ""; + const mockFetch = async (url: string | URL | Request) => { + capturedURL = String(url); + return jsonResponse([]); + }; + + const client = new Client("http://localhost:8089", { + fetch: mockFetch as typeof fetch, + }); + await client.values({ bus: ["can0", "can1"] }); + const parsed = new URL(capturedURL); + expect(parsed.searchParams.getAll("bus")).toEqual(["can0", "can1"]); + }); + + it("includes bus in session filter body", async () => { + let capturedBody = ""; + const mockFetch = async ( + _url: string | URL | Request, + init?: RequestInit, + ) => { + if (init?.method === "PUT") { + capturedBody = init.body as string; + return jsonResponse({ + client_id: "bus-test", + seq: 1, + cursor: 0, + devices: [], + }); + } + return errorResponse(404, "not found"); + }; + + const client = new Client("http://localhost:8089", { + fetch: mockFetch as typeof fetch, + }); + await client.createSession({ + clientId: "bus-test", + bufferTimeout: "PT1M", + filter: { bus: ["can0"] }, + }); + + const body = JSON.parse(capturedBody); + expect(body.filter.bus).toEqual(["can0"]); + }); +}); + describe("Filter exclusions", () => { it("encodes exclude_pgn in query params", async () => { let capturedURL = ""; diff --git a/packages/lplex/test/cloud.test.ts b/packages/lplex/test/cloud.test.ts index 37f60bf..d3a8aaa 100644 --- a/packages/lplex/test/cloud.test.ts +++ b/packages/lplex/test/cloud.test.ts @@ -145,6 +145,66 @@ describe("CloudClient.replicationEvents", () => { }); }); +describe("CloudClient.health", () => { + it("fetches cloud health status", async () => { + const payload = { + status: "ok", + instances_total: 10, + instances_connected: 8, + }; + const mockFetch = async (url: string | URL | Request) => { + expect(url).toBe("https://cloud.example.com/healthz"); + return jsonResponse(payload); + }; + + const client = new CloudClient("https://cloud.example.com", { + fetch: mockFetch as typeof fetch, + }); + const result = await client.health(); + expect(result.status).toBe("ok"); + expect(result.instances_total).toBe(10); + expect(result.instances_connected).toBe(8); + }); +}); + +describe("CloudClient.liveness", () => { + it("fetches liveness", async () => { + const mockFetch = async (url: string | URL | Request) => { + expect(url).toBe("https://cloud.example.com/livez"); + return jsonResponse({ status: "ok" }); + }; + + const client = new CloudClient("https://cloud.example.com", { + fetch: mockFetch as typeof fetch, + }); + const result = await client.liveness(); + expect(result.status).toBe("ok"); + }); +}); + +describe("CloudClient.readiness", () => { + it("fetches readiness", async () => { + const mockFetch = async (url: string | URL | Request) => { + expect(url).toBe("https://cloud.example.com/readyz"); + return jsonResponse({ status: "ok" }); + }; + + const client = new CloudClient("https://cloud.example.com", { + fetch: mockFetch as typeof fetch, + }); + const result = await client.readiness(); + expect(result.status).toBe("ok"); + }); + + it("throws HttpError when not ready", async () => { + const mockFetch = async () => errorResponse(503, "not ready"); + const client = new CloudClient("https://cloud.example.com", { + fetch: mockFetch as typeof fetch, + }); + await expect(client.readiness()).rejects.toThrow(HttpError); + }); +}); + describe("CloudClient.client", () => { it("returns a Client scoped to the instance", async () => { const devices = [{ src: 1, manufacturer: "Garmin" }]; diff --git a/packages/lplex/test/sse.test.ts b/packages/lplex/test/sse.test.ts index d99c8ac..3aed22e 100644 --- a/packages/lplex/test/sse.test.ts +++ b/packages/lplex/test/sse.test.ts @@ -143,6 +143,21 @@ describe("parseSSE", () => { expect(events).toHaveLength(0); }); + it("parses a device_removed event", async () => { + const removed = { type: "device_removed", bus: "can0", src: 5 }; + const stream = makeStream(`data: ${JSON.stringify(removed)}\n\n`); + const events = []; + for await (const event of parseSSE(stream)) { + events.push(event); + } + expect(events).toHaveLength(1); + expect(events[0].type).toBe("device_removed"); + if (events[0].type === "device_removed") { + expect(events[0].deviceRemoved.src).toBe(5); + expect(events[0].deviceRemoved.bus).toBe("can0"); + } + }); + it("skips JSON that is not an object", async () => { const text = `data: 42\ndata: "hello"\ndata: [1,2,3]\ndata: null\ndata: ${JSON.stringify(sampleFrame)}\n\n`; const stream = makeStream(text);