From c0de751382593040541b196df13a65bd1a3b1816 Mon Sep 17 00:00:00 2001 From: Stephen von Takach Date: Thu, 15 Oct 2026 00:54:07 +1100 Subject: [PATCH 1/2] fix(stagehand): refetch a system's modules before dropping an unknown alert The module list was cached on a system's first alert and never refreshed, so a module added to the system afterwards could never raise an alert. On a miss the list is now fetched again, at most every 5 seconds, and the alert list is checked again once that window has passed. Co-Authored-By: Claude Opus 5.5 --- .../src/app/dashboards/dashboards.service.ts | 30 +++++++++-- .../src/tests/dashboards.service.spec.ts | 51 +++++++++++++++++++ 2 files changed, 78 insertions(+), 3 deletions(-) diff --git a/apps/stagehand/src/app/dashboards/dashboards.service.ts b/apps/stagehand/src/app/dashboards/dashboards.service.ts index 043ea4c457b..e4e9c130bc0 100644 --- a/apps/stagehand/src/app/dashboards/dashboards.service.ts +++ b/apps/stagehand/src/app/dashboards/dashboards.service.ts @@ -113,6 +113,9 @@ function matchTopics(topic: string, comp: string): boolean { return true; } +/** Minimum time between refetches of a system's modules on an unknown device */ +const MODULE_LIST_REFRESH_MS = 5000; + interface BrokerAlert { time: number; topic: StateTopic; @@ -217,6 +220,8 @@ export class DashboardsService extends AsyncHandler { private _notified_alert_ids = new Set(); /** Cache of system modules: system_id -> list of valid device names */ private _system_modules_cache = new Map(); + /** When each system's module list was fetched, to bound refetches */ + private _system_modules_fetched_at = new Map(); /** Track in-flight module queries to prevent duplicate requests */ private _pending_module_queries = new Map>(); /** Track current alert subscription scope to avoid unnecessary resets */ @@ -425,9 +430,12 @@ export class DashboardsService extends AsyncHandler { * Get valid module device names for a system. * Caches results to avoid repeated API calls. */ - private async _getSystemModuleNames(system_id: string): Promise { + private async _getSystemModuleNames( + system_id: string, + refresh = false, + ): Promise { // Return cached result if available - if (this._system_modules_cache.has(system_id)) { + if (!refresh && this._system_modules_cache.has(system_id)) { return this._system_modules_cache.get(system_id); } @@ -442,6 +450,7 @@ export class DashboardsService extends AsyncHandler { .then((modules) => { const device_names = getModuleDeviceNames(modules); this._system_modules_cache.set(system_id, device_names); + this._system_modules_fetched_at.set(system_id, Date.now()); this._pending_module_queries.delete(system_id); return device_names; }) @@ -469,7 +478,22 @@ export class DashboardsService extends AsyncHandler { device: string, ): Promise { const module_names = await this._getSystemModuleNames(system_id); - return module_names.includes(device); + if (module_names.includes(device)) return true; + // The cached list predates a module added to the system since; check + // again before dropping the alert, but not more often than this + const fetched_at = this._system_modules_fetched_at.get(system_id) ?? 0; + const wait = fetched_at + MODULE_LIST_REFRESH_MS - Date.now(); + if (wait > 0) { + // Nothing may arrive to prompt another look, so schedule one + this.timeout( + 'module_list_recheck', + () => this._updateDashboardAlertList(), + wait + 50, + ); + return false; + } + const refreshed = await this._getSystemModuleNames(system_id, true); + return refreshed.includes(device); } /** diff --git a/apps/stagehand/src/tests/dashboards.service.spec.ts b/apps/stagehand/src/tests/dashboards.service.spec.ts index ad179c86af2..8a372c7230f 100644 --- a/apps/stagehand/src/tests/dashboards.service.spec.ts +++ b/apps/stagehand/src/tests/dashboards.service.spec.ts @@ -320,4 +320,55 @@ describe('DashboardsService', () => { ); }); }); + + // SH-HOT-02: the module list was cached on a system's first alert and never + // refreshed, so a module added afterwards could never raise an alert + describe('module existence check', () => { + const projector = { + id: 'mod-1', + custom_name: 'Projector', + driver: { class_name: 'Harness' }, + }; + const display = { + id: 'mod-2', + name: 'Display', + driver: { class_name: 'Harness' }, + }; + const exists = (device: string) => + (spectator.service as any)._moduleExistsOnSystem('sys-1', device); + + afterEach(() => vi.useRealTimers()); + + it('refetches a stale module list before dropping an unknown device', async () => { + vi.useFakeTimers({ toFake: ['Date'] }); + vi.mocked(ts_client.queryModules) + .mockResolvedValueOnce({ data: [projector] } as any) + .mockResolvedValueOnce({ data: [projector, display] } as any); + + expect(await exists('Projector_1')).toBe(true); + vi.setSystemTime(Date.now() + 6_000); + + expect(await exists('Display_1')).toBe(true); + expect(ts_client.queryModules).toHaveBeenCalledTimes(2); + }); + + it('refetches at most once per interval, and looks again when it may', async () => { + vi.useFakeTimers({ toFake: ['Date'] }); + vi.mocked(ts_client.queryModules).mockResolvedValue({ + data: [projector], + } as any); + const recheck = vi.spyOn(spectator.service as any, 'timeout'); + + expect(await exists('Projector_1')).toBe(true); + expect(await exists('Display_1')).toBe(false); + expect(await exists('Display_1')).toBe(false); + + expect(ts_client.queryModules).toHaveBeenCalledTimes(1); + expect(recheck).toHaveBeenCalledWith( + 'module_list_recheck', + expect.any(Function), + expect.any(Number), + ); + }); + }); }); From da28437fb279b7783423a9ddf6c6ce8abeb6e1da Mon Sep 17 00:00:00 2001 From: Stephen von Takach Date: Thu, 15 Oct 2026 00:54:07 +1100 Subject: [PATCH 2/2] test(e2e): cover stagehand's MQTT state and hot list against a local stack The stack now runs source and mosquitto (with rest-api auth, as deployed), seeds a Broker for source and a state harness driver from PlaceOS/drivers, and serves stagehand for a new `local` Playwright project. The specs hold the broker to the truth: what a fresh client is handed for a system must equal what its modules, names, zones and state imply. They cover publishing, the hot list, reorder, removal, module and system delete, rename, moving a system between levels and cleared statuses. The lifecycle specs need PlaceOS/source v1.22.1. Co-Authored-By: Claude Opus 5.5 --- E2E_USER_STORIES.md | 43 ++ .../e2e/local/stagehand-hotlist.spec.ts | 196 +++++++++ .../local/stagehand-mqtt-lifecycle.spec.ts | 338 +++++++++++++++ .../e2e/local/stagehand-mqtt.spec.ts | 235 +++++++++++ apps/stagehand/playwright.config.ts | 86 +++- e2e/README.md | 66 ++- e2e/stack/docker-compose.yml | 43 +- e2e/stack/mosquitto.conf | 21 + e2e/stack/up.sh | 8 + e2e/support/drivers/drivers.api.ts | 22 +- e2e/support/drivers/drivers.env.ts | 36 ++ e2e/support/mqtt/broker-view.ts | 277 +++++++++++++ e2e/support/mqtt/mqtt.seed.ts | 123 ++++++ e2e/support/seed.ts | 2 +- e2e/support/stagehand/alerts.page.ts | 70 ++++ e2e/support/stagehand/fixtures.ts | 147 +++++++ e2e/support/stagehand/stagehand.api.ts | 384 ++++++++++++++++++ e2e/support/stagehand/stagehand.env.ts | 43 ++ e2e/support/stagehand/stagehand.seed.ts | 90 ++++ 19 files changed, 2206 insertions(+), 24 deletions(-) create mode 100644 apps/stagehand/e2e/local/stagehand-hotlist.spec.ts create mode 100644 apps/stagehand/e2e/local/stagehand-mqtt-lifecycle.spec.ts create mode 100644 apps/stagehand/e2e/local/stagehand-mqtt.spec.ts create mode 100644 e2e/stack/mosquitto.conf create mode 100644 e2e/support/mqtt/broker-view.ts create mode 100644 e2e/support/mqtt/mqtt.seed.ts create mode 100644 e2e/support/stagehand/alerts.page.ts create mode 100644 e2e/support/stagehand/fixtures.ts create mode 100644 e2e/support/stagehand/stagehand.api.ts create mode 100644 e2e/support/stagehand/stagehand.env.ts create mode 100644 e2e/support/stagehand/stagehand.seed.ts diff --git a/E2E_USER_STORIES.md b/E2E_USER_STORIES.md index a89ec6458d7..38e2b28143b 100644 --- a/E2E_USER_STORIES.md +++ b/E2E_USER_STORIES.md @@ -447,6 +447,49 @@ Config gaps caused several production incidents, and they are invisible to UI sp | CFG-04 | P1 | A missing `org`/parented-`level` zone is caught as `/misconfigured`, not as a blank page. | **partial** — WP-E2E-01 asserts the healthy path; the negative case is unasserted | | CFG-05 | P2 | Deployment-shaped run: the app served by nginx at `/workplace/` behind the `verified` cookie gate. | todo — a second project; `mintToken` already captures the cookie | +## 5. Stagehand — MQTT state and the hot list + +Stagehand's hot list is built from MQTT: drivers write state to redis, `source` publishes +each status retained to mosquitto, and stagehand subscribes through `/api/mqtt/`. The stack +runs `source` and `mosquitto` for this, and the specs drive state with a purpose-built driver, +`drivers/place/e2e/state_harness.cr` (a UDP device whose `connected` and other statuses are +set through `exec`). See "MQTT" in `e2e/README.md`. + +**The broker is asserted directly, not only through the page.** Stagehand re-checks every +MQTT alert over REST before showing it (`DashboardsService._updateDashboardAlertList`), which +hides wrong data on the broker. Removing that re-check — it is what makes stagehand slow — is +only safe once the MQTT rows below are all green. + +| ID | P | Story | Status | +|----|---|-------|--------| +| SH-MQTT-01 | P0 | A started module publishes `connected: true`, retained, at its full topic: org, `_` for an absent region and area, building, level, system, driver, module name, index. | **done** — `local/stagehand-mqtt.spec.ts` | +| SH-MQTT-02 | P0 | `connected` follows the module through false and back to true. | **done** — `local/stagehand-mqtt.spec.ts` | +| SH-MQTT-03 | P1 | Any status is published as the JSON value the driver set (boolean, number, string, object), not a string of it. | **done** — `local/stagehand-mqtt.spec.ts` | +| SH-MQTT-04 | P0 | A client connecting later is handed the current value of every status, and nothing the module does not hold. | **done** — `local/stagehand-mqtt.spec.ts` | +| SH-HOT-01 | P0 | The hot list shows a module while it reports `connected: false`, and not while it is connected; a reload agrees. Held against a disconnected control module. | **done** — `local/stagehand-hotlist.spec.ts` | +| SH-MQTT-10 | P0 | Reordering a system's modules moves each module's state to its new index; nothing is left at the old one. | **done** — `local/stagehand-mqtt-lifecycle.spec.ts`. The field report "alerts on the old index". Root cause: source numbered modules in database order (`Module.find_all`), not the system's, so a reorder changed nothing it published. | +| SH-MQTT-11 | P0 | A module removed from a system (still existing) leaves no retained topics under that system. | **done** — `local/stagehand-mqtt-lifecycle.spec.ts` | +| SH-MQTT-12 | P0 | A deleted module leaves no retained topics. | **done** — `local/stagehand-mqtt-lifecycle.spec.ts`. The field report "results for deleted modules". source worked out what to remove from the module's redis state and the current mappings, either of which can be gone first. | +| SH-MQTT-13 | P1 | A deleted system leaves no retained topics. | **done** — `local/stagehand-mqtt-lifecycle.spec.ts` | +| SH-MQTT-14 | P1 | Renaming a module (`custom_name`) moves its state to the new name, and other modules' indexes with it. | **done** — `local/stagehand-mqtt-lifecycle.spec.ts` | +| SH-MQTT-15 | P1 | Moving a system between zones moves its topics. | **done** — `local/stagehand-mqtt-lifecycle.spec.ts` | +| SH-MQTT-16 | P1 | `source` that starts before the broker is reachable publishes once it is. | **done, not automated** — verified by hand (stop mosquitto, restart source, start mosquitto: all state published). Automating it needs a spec that can stop containers. source spec covers the retry path. | +| SH-MQTT-17 | P1 | A status the driver clears is removed from the broker, not retained as `null`. | **done** — `local/stagehand-mqtt-lifecycle.spec.ts` | +| SH-MQTT-18 | P0 | A module added to a system after it started has its state published. | **done** — every lifecycle spec's setup depends on it. Was a race: the state was written before source mapped the module, and nothing republished it. Found while building this coverage. | +| SH-MQTT-19 | P0 | Two quick edits to one system leave source's mapping matching the latest. | **done** — the lifecycle setup makes them (two modules added back to back). pg-orm and placeos-resource handle each change on its own fiber, so source processed them out of order and kept the older module list. Found while building this coverage. | +| SH-MQTT-20 | P1 | After a broker restart, every status is published again. | **done, not automated** — verified by hand (restart mosquitto: 32 of 32 topics back within 2 s; before the fix, 0 until each value changed). | +| SH-MQTT-21 | P1 | Retained topics nothing maps to (left by a module deleted while source was down, or by an older source) are removed when source starts. | **done, not automated** — verified by hand with a planted topic. source spec `sweeps retained topics that nothing maps to`. | +| SH-HOT-02 | P1 | A module added to a system after stagehand has loaded it can still raise an alert. | **done** — `local/stagehand-hotlist.spec.ts`. Stagehand cached each system's module list forever (`_system_modules_cache`); it now refetches, at most every 5 s, before dropping an alert for an unknown device. | + +**The lifecycle specs need source v1.22.1 or later** (PlaceOS/source#65). Against +`placeos-2.2610.1` SH-MQTT-10 … 19 and SH-HOT-02 are red, which is what they are for. Until a +PlaceOS release carries the fix, run them with `E2E_SOURCE_IMAGE=placeos/source:nightly` (a +nightly built after 2026-10-14) or a local build (see `e2e/README.md`, "MQTT"). The advisory CI +runs only the workplace suite, so it is not affected. + +**Not covered yet**: a status set with a TTL (`set_expire`) that expires in redis publishes +nothing, so the broker keeps its last value. source would need redis keyspace notifications to see it. + --- ## Notes & blockers diff --git a/apps/stagehand/e2e/local/stagehand-hotlist.spec.ts b/apps/stagehand/e2e/local/stagehand-hotlist.spec.ts new file mode 100644 index 00000000000..50eb47acf17 --- /dev/null +++ b/apps/stagehand/e2e/local/stagehand-hotlist.spec.ts @@ -0,0 +1,196 @@ +/** + * Stagehand's hot list, end to end: a module that reports `connected: false` + * is listed, and one that is connected is not. + * + * start a harness (UDP, so it comes up connected) -> not listed + * harness reports connected: false -> listed + * harness reports connected: true -> gone again + * + * The broker is asserted alongside the page at every step. When stagehand + * stops re-checking alerts over REST, this spec is what says the page is still + * right; until then, the broker assertions are what say the data is. + * + * "Not listed" needs a control (see e2e/README.md, Conventions): the page shows + * a spinner, not an empty table, until its first alert. So the system also + * holds a second harness, `Projector_1`, that is disconnected first. + * Once the page lists the control, it has the system's state, and a missing + * `Display_1` row means what it says. + */ +import { + BrokerView, + systemFilter, +} from '../../../../e2e/support/mqtt/broker-view'; +import { + expectHotList, + hotListRows, + openHotList, +} from '../../../../e2e/support/stagehand/alerts.page'; +import { expect, test } from '../../../../e2e/support/stagehand/fixtures'; +import { + addHarness, + createSystem, + deleteSystem, + setConnected, + sweepSystems, + waitForConnected, +} from '../../../../e2e/support/stagehand/stagehand.api'; +import { MQTT_PROPAGATION_MS } from '../../../../e2e/support/stagehand/stagehand.env'; + +test.beforeAll(async ({ adminApi }) => { + await sweepSystems(adminApi); +}); + +test('a module is listed while it reports connected: false, and only then', async ({ + page, + adminApi, + stagehandAdminToken, + harnessDriverId, + zones, +}) => { + const system = await createSystem(adminApi, zones, 'hotlist'); + const broker = await BrokerView.open( + stagehandAdminToken, + systemFilter(zones.org, system.id), + ); + // Stagehand logs why it drops an alert; attached so a failure explains itself + const app_log: string[] = []; + page.on('console', (m) => { + if (/ALERTS/.test(m.text())) app_log.push(m.text()); + }); + try { + // Stagehand open first, so every change below arrives live + await openHotList(page); + await expectHotList(page, system.id, []); + + // Both harnesses start connected, so neither is listed yet + const control = await addHarness(adminApi, system.id, harnessDriverId, { + custom_name: 'Projector', + }); + const display = await addHarness(adminApi, system.id, harnessDriverId); + await waitForConnected(adminApi, control.id, true); + await waitForConnected(adminApi, display.id, true); + await broker.waitForValue( + 'Projector', + 1, + 'connected', + true, + MQTT_PROPAGATION_MS, + ); + await broker.waitForValue( + 'Display', + 1, + 'connected', + true, + MQTT_PROPAGATION_MS, + ); + + // NOTE: both modules exist before the system's first alert. Stagehand + // caches a system's module list on its first alert and never refreshes + // it, so a module added after that cannot alert until a reload + // (SH-HOT-02). This spec is about the working path. + await setConnected(adminApi, control.id, false); + await broker.waitForValue( + 'Projector', + 1, + 'connected', + false, + MQTT_PROPAGATION_MS, + ); + + await test.step('a connected module is not listed', async () => { + await expectHotList(page, system.id, ['Projector_1']); + }); + + await test.step('it is listed once it reports connected: false', async () => { + await setConnected(adminApi, display.id, false); + await broker.waitForValue( + 'Display', + 1, + 'connected', + false, + MQTT_PROPAGATION_MS, + ); + await expectHotList(page, system.id, ['Display_1', 'Projector_1']); + + const [row] = (await hotListRows(page, system.id)).filter( + (r) => r.device === 'Display_1', + ); + expect(row.cells.join(' ')).toContain('Disconnected'); + }); + + await test.step('and leaves it when it reconnects', async () => { + await setConnected(adminApi, display.id, true); + await broker.waitForValue( + 'Display', + 1, + 'connected', + true, + MQTT_PROPAGATION_MS, + ); + await expectHotList(page, system.id, ['Projector_1']); + }); + + await test.step('a stagehand opened now agrees', async () => { + await page.reload(); + await expectHotList(page, system.id, ['Projector_1']); + }); + } finally { + await test + .info() + .attach('stagehand ALERTS log', { body: app_log.join('\n') }); + await test + .info() + .attach('broker messages', { body: broker.describeLog() }); + await broker.close(); + await deleteSystem(adminApi, system.id); + } +}); + +test('SH-HOT-02: a module added after the page loaded the system can still be listed', async ({ + page, + adminApi, + stagehandAdminToken, + harnessDriverId, + zones, +}) => { + const system = await createSystem(adminApi, zones, 'late-module'); + const broker = await BrokerView.open( + stagehandAdminToken, + systemFilter(zones.org, system.id), + ); + const app_log: string[] = []; + page.on('console', (m) => { + if (/ALERTS/.test(m.text())) app_log.push(m.text()); + }); + try { + await openHotList(page); + + // The system's first alert, which is when stagehand learns its modules + const control = await addHarness(adminApi, system.id, harnessDriverId, { + custom_name: 'Projector', + }); + await waitForConnected(adminApi, control.id, true); + await setConnected(adminApi, control.id, false); + await expectHotList(page, system.id, ['Projector_1']); + + // Added afterwards, then disconnected + const display = await addHarness(adminApi, system.id, harnessDriverId); + await waitForConnected(adminApi, display.id, true); + await setConnected(adminApi, display.id, false); + await broker.waitForValue( + 'Display', + 1, + 'connected', + false, + MQTT_PROPAGATION_MS, + ); + + await expectHotList(page, system.id, ['Display_1', 'Projector_1']); + } finally { + await test + .info() + .attach('stagehand ALERTS log', { body: app_log.join('\n') }); + await broker.close(); + await deleteSystem(adminApi, system.id); + } +}); diff --git a/apps/stagehand/e2e/local/stagehand-mqtt-lifecycle.spec.ts b/apps/stagehand/e2e/local/stagehand-mqtt-lifecycle.spec.ts new file mode 100644 index 00000000000..cccbfb48fd4 --- /dev/null +++ b/apps/stagehand/e2e/local/stagehand-mqtt-lifecycle.spec.ts @@ -0,0 +1,338 @@ +/** + * The broker stays true while systems change: reordered, renamed, moved + * between zones, modules removed or deleted, statuses cleared. + * + * Each test changes a system the way an admin would and then requires the + * broker to hold EXACTLY what the source of truth says it should: every topic + * a fresh client is handed for the system is compared with the topics computed + * from the system's module list, names, zones and each module's own state + * (`expectedTopics`). A stale topic, a missing one, or a value at the wrong + * index all fail the same comparison. + * + * These are the cases behind the field reports: alerts for deleted modules, + * and alerts on the old index after modules were reordered. Stagehand hides + * both today by re-checking every alert over REST; these hold the broker to + * the standard that lets that re-check go. + */ +import { APIRequestContext } from '@playwright/test'; +import { + BrokerView, + retainedForSystem, + systemFilter, +} from '../../../../e2e/support/mqtt/broker-view'; +import { expect, test } from '../../../../e2e/support/stagehand/fixtures'; +import { + HierarchyZones, + addHarness, + createSystem, + deleteModule, + deleteSystem, + ensureSecondLevel, + exec, + expectedTopics, + getSystem, + removeFromSystem, + renameModule, + setConnected, + setSystemModules, + setSystemZones, + sweepSystems, + waitForConnected, +} from '../../../../e2e/support/stagehand/stagehand.api'; +import { MQTT_PROPAGATION_MS } from '../../../../e2e/support/stagehand/stagehand.env'; + +test.beforeAll(async ({ adminApi }) => { + await sweepSystems(adminApi); +}); + +/** + * Where a fresh client's view of a system differs from the truth, one line per + * topic: `stale` (retained, should not be), `missing` (should be, is not), or + * `wrong` (both, different values). + */ +async function brokerDiscrepancies( + api: APIRequestContext, + token: string, + zones: HierarchyZones, + system_id: string, + levels: string[], +): Promise { + const truth = await expectedTopics(api, system_id, zones, levels); + const broker = await retainedForSystem(token, zones.org, system_id); + const out: string[] = []; + for (const [topic, value] of Object.entries(broker)) { + if (!(topic in truth)) + out.push(`stale ${topic} = ${JSON.stringify(value)}`); + else if (JSON.stringify(truth[topic]) !== JSON.stringify(value)) { + out.push( + `wrong ${topic} = ${JSON.stringify(value)}, module holds ${JSON.stringify(truth[topic])}`, + ); + } + } + for (const [topic, value] of Object.entries(truth)) { + if (!(topic in broker)) + out.push(`missing ${topic} = ${JSON.stringify(value)}`); + } + return out.sort(); +} + +/** Wait until a fresh client is handed exactly the topics the truth implies. */ +async function expectBrokerMatchesTruth( + api: APIRequestContext, + token: string, + zones: HierarchyZones, + system_id: string, + levels: string[] = [zones.level], +) { + await expect + .poll(() => brokerDiscrepancies(api, token, zones, system_id, levels), { + message: `the broker's retained topics for ${system_id} against its modules`, + timeout: MQTT_PROPAGATION_MS, + intervals: [500, 1_000, 2_000], + }) + .toEqual([]); +} + +/** + * A system with two harnesses of the same name, `Display_1` (a) and + * `Display_2` (b), holding different values so a mix up is visible. + */ +async function twoDisplays( + api: APIRequestContext, + token: string, + zones: HierarchyZones, + driver_id: string, + label: string, +) { + const system = await createSystem(api, zones, label); + const a = await addHarness(api, system.id, driver_id); + const b = await addHarness(api, system.id, driver_id); + await waitForConnected(api, a.id, true); + await waitForConnected(api, b.id, true); + await setConnected(api, a.id, false); + await exec(api, a.id, 'set_status', ['which', 'a']); + await exec(api, b.id, 'set_status', ['which', 'b']); + const view = await BrokerView.open( + token, + systemFilter(zones.org, system.id), + ); + try { + await view.waitForValue( + 'Display', + 1, + 'which', + 'a', + MQTT_PROPAGATION_MS, + ); + await view.waitForValue( + 'Display', + 2, + 'which', + 'b', + MQTT_PROPAGATION_MS, + ); + } finally { + await view.close(); + } + await expectBrokerMatchesTruth(api, token, zones, system.id); + return { system, a, b }; +} + +test('SH-MQTT-10: reordering modules moves their state to the new index', async ({ + adminApi, + stagehandAdminToken, + harnessDriverId, + zones, +}) => { + const { system, a, b } = await twoDisplays( + adminApi, + stagehandAdminToken, + zones, + harnessDriverId, + 'reorder', + ); + try { + await setSystemModules(adminApi, await getSystem(adminApi, system.id), [ + b.id, + a.id, + ]); + // b is Display_1 now and connected; a, still disconnected, is Display_2 + await expectBrokerMatchesTruth( + adminApi, + stagehandAdminToken, + zones, + system.id, + ); + } finally { + await deleteSystem(adminApi, system.id); + } +}); + +test('SH-MQTT-11: a module removed from a system leaves nothing under it', async ({ + adminApi, + stagehandAdminToken, + harnessDriverId, + zones, +}) => { + const { system, a, b } = await twoDisplays( + adminApi, + stagehandAdminToken, + zones, + harnessDriverId, + 'remove', + ); + try { + // a was Display_1; b moves up to take its index + await removeFromSystem(adminApi, system.id, a.id); + await expectBrokerMatchesTruth( + adminApi, + stagehandAdminToken, + zones, + system.id, + ); + } finally { + await deleteModule(adminApi, a.id); + await deleteSystem(adminApi, system.id); + } +}); + +test('SH-MQTT-12: a deleted module leaves nothing behind', async ({ + adminApi, + stagehandAdminToken, + harnessDriverId, + zones, +}) => { + const { system, b } = await twoDisplays( + adminApi, + stagehandAdminToken, + zones, + harnessDriverId, + 'delete-module', + ); + try { + await deleteModule(adminApi, b.id); + await expectBrokerMatchesTruth( + adminApi, + stagehandAdminToken, + zones, + system.id, + ); + } finally { + await deleteSystem(adminApi, system.id); + } +}); + +test('SH-MQTT-13: a deleted system leaves nothing behind', async ({ + adminApi, + stagehandAdminToken, + harnessDriverId, + zones, +}) => { + const { system } = await twoDisplays( + adminApi, + stagehandAdminToken, + zones, + harnessDriverId, + 'delete-system', + ); + await deleteSystem(adminApi, system.id); + await expect + .poll( + () => retainedForSystem(stagehandAdminToken, zones.org, system.id), + { + message: `retained topics for deleted ${system.id}`, + timeout: MQTT_PROPAGATION_MS, + }, + ) + .toEqual({}); +}); + +test('SH-MQTT-14: renaming a module moves its state to the new name', async ({ + adminApi, + stagehandAdminToken, + harnessDriverId, + zones, +}) => { + const { system, a } = await twoDisplays( + adminApi, + stagehandAdminToken, + zones, + harnessDriverId, + 'rename', + ); + try { + // a becomes Projector_1, and b moves up to Display_1 + await renameModule(adminApi, a.id, 'Projector'); + await expectBrokerMatchesTruth( + adminApi, + stagehandAdminToken, + zones, + system.id, + ); + } finally { + await deleteSystem(adminApi, system.id); + } +}); + +test('SH-MQTT-15: moving a system to another level moves its topics', async ({ + adminApi, + stagehandAdminToken, + harnessDriverId, + zones, +}) => { + const level2 = await ensureSecondLevel(adminApi, zones.building); + const { system } = await twoDisplays( + adminApi, + stagehandAdminToken, + zones, + harnessDriverId, + 'move', + ); + try { + await setSystemZones(adminApi, await getSystem(adminApi, system.id), [ + zones.org, + zones.building, + level2, + ]); + await expectBrokerMatchesTruth( + adminApi, + stagehandAdminToken, + zones, + system.id, + [zones.level, level2], + ); + } finally { + await deleteSystem(adminApi, system.id); + } +}); + +test('SH-MQTT-17: a status the driver clears is removed from the broker', async ({ + adminApi, + stagehandAdminToken, + harnessDriverId, + zones, +}) => { + const system = await createSystem(adminApi, zones, 'clear'); + try { + const mod = await addHarness(adminApi, system.id, harnessDriverId); + await waitForConnected(adminApi, mod.id, true); + await exec(adminApi, mod.id, 'set_status', ['power', true]); + await expectBrokerMatchesTruth( + adminApi, + stagehandAdminToken, + zones, + system.id, + ); + + // redis drops the key; a retained `{value: null}` is not the same thing + await exec(adminApi, mod.id, 'clear_status', ['power']); + await expectBrokerMatchesTruth( + adminApi, + stagehandAdminToken, + zones, + system.id, + ); + } finally { + await deleteSystem(adminApi, system.id); + } +}); diff --git a/apps/stagehand/e2e/local/stagehand-mqtt.spec.ts b/apps/stagehand/e2e/local/stagehand-mqtt.spec.ts new file mode 100644 index 00000000000..c30102b140b --- /dev/null +++ b/apps/stagehand/e2e/local/stagehand-mqtt.spec.ts @@ -0,0 +1,235 @@ +/** + * The MQTT contract stagehand relies on: what source publishes for a module, + * where, and what a client connecting later is handed. + * + * No browser. These assert on the broker itself, because stagehand re-checks + * every MQTT alert over REST before showing it (`DashboardsService`), so a UI + * that looks right says nothing about whether the broker is right. Removing + * that re-check is only safe once these hold. + * + * Every value is also read from the module's own state through rest-api, which + * is the truth the broker is meant to mirror. + */ +import { moduleState } from '../../../../e2e/support/drivers/drivers.api'; +import { + BrokerView, + parseStateTopic, + retainedSnapshot, + systemFilter, +} from '../../../../e2e/support/mqtt/broker-view'; +import { expect, test } from '../../../../e2e/support/stagehand/fixtures'; +import { + addHarness, + createSystem, + deleteSystem, + exec, + setConnected, + sweepSystems, + waitForConnected, +} from '../../../../e2e/support/stagehand/stagehand.api'; +import { MQTT_PROPAGATION_MS } from '../../../../e2e/support/stagehand/stagehand.env'; + +test.beforeAll(async ({ adminApi }) => { + await sweepSystems(adminApi); +}); + +test('a started module publishes connected: true at its full topic', async ({ + adminApi, + stagehandAdminToken, + harnessDriverId, + zones, +}) => { + const system = await createSystem(adminApi, zones, 'topic'); + const broker = await BrokerView.open( + stagehandAdminToken, + systemFilter(zones.org, system.id), + ); + try { + const mod = await addHarness(adminApi, system.id, harnessDriverId); + await waitForConnected(adminApi, mod.id, true); + + const found = await broker.waitForValue( + 'Display', + 1, + 'connected', + true, + MQTT_PROPAGATION_MS, + ); + + // Every segment, because stagehand parses topics by position. The + // system has no region or area zone, which source writes as `_`. + expect(found.topic).toBe( + [ + 'placeos', + zones.org, + 'state', + '_', + zones.building, + zones.level, + '_', + system.id, + harnessDriverId, + 'Display', + '1', + 'connected', + ].join('/'), + ); + // Unix seconds, and recent: a stale `time` is how a replayed value looks + expect(Math.abs(found.time - Date.now() / 1000)).toBeLessThan(120); + } finally { + await broker.close(); + await deleteSystem(adminApi, system.id); + } +}); + +test('connected follows the module: false, then true again', async ({ + adminApi, + stagehandAdminToken, + harnessDriverId, + zones, +}) => { + const system = await createSystem(adminApi, zones, 'toggle'); + const broker = await BrokerView.open( + stagehandAdminToken, + systemFilter(zones.org, system.id), + ); + try { + const mod = await addHarness(adminApi, system.id, harnessDriverId); + await waitForConnected(adminApi, mod.id, true); + await broker.waitForValue( + 'Display', + 1, + 'connected', + true, + MQTT_PROPAGATION_MS, + ); + + await setConnected(adminApi, mod.id, false); + await waitForConnected(adminApi, mod.id, false); + await broker.waitForValue( + 'Display', + 1, + 'connected', + false, + MQTT_PROPAGATION_MS, + ); + + await setConnected(adminApi, mod.id, true); + await waitForConnected(adminApi, mod.id, true); + await broker.waitForValue( + 'Display', + 1, + 'connected', + true, + MQTT_PROPAGATION_MS, + ); + } finally { + await broker.close(); + await deleteSystem(adminApi, system.id); + } +}); + +test('any status is published as the JSON value the driver set', async ({ + adminApi, + stagehandAdminToken, + harnessDriverId, + zones, +}) => { + const system = await createSystem(adminApi, zones, 'values'); + const broker = await BrokerView.open( + stagehandAdminToken, + systemFilter(zones.org, system.id), + ); + try { + const mod = await addHarness(adminApi, system.id, harnessDriverId); + await waitForConnected(adminApi, mod.id, true); + + const values: Record = { + power: true, + volume: 42.5, + input: 'HDMI2', + inputs: { hdmi: 1, names: ['a', 'b'] }, + }; + for (const [key, value] of Object.entries(values)) { + await exec(adminApi, mod.id, 'set_status', [key, value]); + } + for (const [key, value] of Object.entries(values)) { + // A value, not a JSON string of one: stagehand compares them + await broker.waitForValue( + 'Display', + 1, + key, + value, + MQTT_PROPAGATION_MS, + ); + } + + const state = await moduleState(adminApi, mod.id); + for (const [key, value] of Object.entries(values)) { + expect(state[key], `module state ${key}`).toEqual(value); + } + } finally { + await broker.close(); + await deleteSystem(adminApi, system.id); + } +}); + +test('a client connecting later is handed the current value of every status', async ({ + adminApi, + stagehandAdminToken, + harnessDriverId, + zones, +}) => { + const system = await createSystem(adminApi, zones, 'retained'); + const live = await BrokerView.open( + stagehandAdminToken, + systemFilter(zones.org, system.id), + ); + try { + const mod = await addHarness(adminApi, system.id, harnessDriverId); + await waitForConnected(adminApi, mod.id, true); + await exec(adminApi, mod.id, 'set_status', ['power', true]); + await setConnected(adminApi, mod.id, false); + await live.waitForValue( + 'Display', + 1, + 'connected', + false, + MQTT_PROPAGATION_MS, + ); + await live.waitForValue( + 'Display', + 1, + 'power', + true, + MQTT_PROPAGATION_MS, + ); + + // What a stagehand opened now would load + const snapshot = await retainedSnapshot( + stagehandAdminToken, + systemFilter(zones.org, system.id), + ); + const retained = Object.fromEntries( + [...snapshot].map(([topic, v]) => { + const t = parseStateTopic(topic)!; + return [`${t.module_name}_${t.index}.${t.status}`, v.value]; + }), + ); + expect(retained['Display_1.connected']).toBe(false); + expect(retained['Display_1.power']).toBe(true); + + // ...and nothing the module does not hold + const state = await moduleState(adminApi, mod.id); + for (const key of Object.keys(retained)) { + const status = key.split('.').slice(1).join('.'); + expect( + state, + `retained ${key} is not in the module's state`, + ).toHaveProperty(status); + } + } finally { + await live.close(); + await deleteSystem(adminApi, system.id); + } +}); diff --git a/apps/stagehand/playwright.config.ts b/apps/stagehand/playwright.config.ts index bcd64ab4c08..6984cd699e7 100644 --- a/apps/stagehand/playwright.config.ts +++ b/apps/stagehand/playwright.config.ts @@ -3,32 +3,98 @@ import { defineConfig, devices } from '@playwright/test'; import { workspaceRoot } from '@nx/devkit'; -const baseURL = process.env['BASE_URL'] || 'http://localhost:4210'; +// `stagehand.env.ts` pulls in `e2e/support/env.ts`, which loads e2e/.env first. +import { BACKEND_URL } from '../../e2e/support/env'; +import { + STAGEHAND_URL, + assertStagehandLocal, +} from '../../e2e/support/stagehand/stagehand.env'; + +// The `local` specs create systems and modules: refuse anything not local +// before a browser launches. +assertStagehandLocal(); + +const CI = !!process.env.CI; +const PORT = new URL(STAGEHAND_URL).port || '4210'; export default defineConfig({ ...nxE2EPreset(__filename, { testDir: './e2e' }), + timeout: 120_000, + expect: { timeout: 20_000 }, + // ONE worker. The specs share one admin identity, and the hot list is org + // wide: each spec owns its own system, but a broken spec in a parallel + // worker would show up in every other spec's page too. + workers: 1, + forbidOnly: CI, + retries: CI ? 2 : 0, reporter: [ ['list'], [ 'html', { outputFolder: '../../reports/e2e/stagehand', open: 'never' }, ], + ...(CI + ? ([ + [ + 'json', + { + outputFile: + '../../reports/e2e/stagehand-results.json', + }, + ], + ['github'], + ] as const) + : []), ], use: { - baseURL, + baseURL: STAGEHAND_URL, + ignoreHTTPSErrors: true, trace: 'on-first-retry', - }, - webServer: { - command: 'bunx nx serve stagehand --port=4210', - url: 'http://localhost:4210', - reuseExistingServer: !process.env.CI, - cwd: workspaceRoot, - timeout: 120000, + screenshot: 'only-on-failure', + video: 'retain-on-failure', + actionTimeout: 20_000, + navigationTimeout: 45_000, }, projects: [ + // --- preflight: prove the stack is up, for `local` only ------------- + { + name: 'preflight', + testDir: '../../e2e/support', + testMatch: /preflight\.setup\.ts$/, + use: { ...devices['Desktop Chrome'] }, + }, + + // --- real local backend: source, mosquitto and a harness driver ----- { - name: 'chromium', + name: 'local', + testDir: './e2e/local', + dependencies: ['preflight'], + use: { ...devices['Desktop Chrome'] }, + }, + + // --- mock mode: no backend ------------------------------------------ + { + name: 'mock', + testDir: './e2e', + testIgnore: ['**/local/**'], use: { ...devices['Desktop Chrome'] }, }, ], + webServer: { + command: `bunx nx serve stagehand --port=${PORT}`, + url: STAGEHAND_URL, + reuseExistingServer: !CI, + cwd: workspaceRoot, + timeout: 180_000, + // MUST be set: without it the dev server proxies /api (and the + // /api/mqtt/ websocket) to the shared dev deployment. See the + // concierge config for the long version. + env: { + PLACE_PROXY_DOMAIN: new URL(BACKEND_URL).host, + PLACE_PROXY_SECURE: String( + new URL(BACKEND_URL).protocol === 'https:', + ), + PLACE_PROXY_VALID_SSL: 'false', + }, + }, }); diff --git a/e2e/README.md b/e2e/README.md index 8434f573b35..608609a572e 100644 --- a/e2e/README.md +++ b/e2e/README.md @@ -52,9 +52,9 @@ e2e/stack/down.sh # stop (--volumes to wipe; driver binaries a ``` It is trimmed to what the e2e path exercises: postgres, elasticsearch, redis, -search-ingest, frontend-loader, auth, rest-api, staff-api, core, nginx, init. -Dropped from PlaceOS/local: edge, triggers, dispatch, source, influx, chronograf, -mosquitto, minio and the loki/grafana profile — roughly half the containers. +search-ingest, frontend-loader, auth, rest-api, staff-api, core, source, +mosquitto, nginx, init. Dropped from PlaceOS/local: edge, triggers, dispatch, +influx, chronograf, minio and the loki/grafana profile. The calendar and directory can be backed by the PlaceOS Microsoft 365 sandbox tenant. Set `E2E_O365_TENANT`, `E2E_O365_CLIENT_ID` and `E2E_O365_CLIENT_SECRET` @@ -90,6 +90,57 @@ Two things a cold start taught us that a long-lived stack hides: `seed.ts` patches `login_url` to an absolute URL. A long-lived stack usually has an absolute value already, which is why this never shows up locally. +## MQTT + +Stagehand reads module state from MQTT, so the stack carries that path: + +``` +driver --> redis --> source --> mosquitto:1883 (source publishes, API key auth) + mosquitto:9001 <-- nginx /api/mqtt/ <-- browser (bearer auth) +``` + +- **`source` publishes to Broker rows, nothing else.** `e2e/support/mqtt/mqtt.seed.ts` + creates one for `mosquitto:1883`, signed in with an API key minted as Support (rest-api + only lets support users publish). `up.sh` runs it. +- **mosquitto asks rest-api about every client and topic**, as a deployment does. The routes + are `/api/engine/v2/mqtt/user` and `/mqtt/access`. PlaceOS/local's config names + `mqtt_user`/`mqtt_access`, which 404, so every client is refused there; `e2e/stack/mosquitto.conf` + has the working names. +- **Restarting mosquitto empties it** (no persistence). source before v1.22.1 (so + `placeos-2.2610.1`) neither republishes after reconnecting nor retries a broker it could not + reach at startup, so restart source after mosquitto. v1.22.1 (PlaceOS/source#65) does both. +- **Test against a newer source than the release**: `E2E_SOURCE_IMAGE=placeos/source:nightly`, + or build it (`docker build -t placeos/source:local .` in the source clone), then + `E2E_SOURCE_IMAGE=placeos/source:local e2e/stack/up.sh`, or recreate just + that service with `LOG_LEVEL=info E2E_SOURCE_IMAGE=placeos/source:local docker compose -p + placeos-e2e -f e2e/stack/docker-compose.yml up -d --no-deps --force-recreate source`. + `LOG_LEVEL=info` shows source's `reconciled retained state` and `swept retained state` lines. +- **Assert the broker against the truth, not a value.** `expectedTopics` (stagehand.api.ts) builds + every topic a system should have from its module order, names, zones and each module's own + state; `retainedForSystem` (broker-view.ts) is what a fresh client is handed. The lifecycle + specs require them to be equal, so a stale, missing or misplaced topic all fail the same way. +- **Do not run a Crystal spec suite in Docker alongside the stack** on an 8 GB Docker VM: the + compiler is OOM-killed, and takes Elasticsearch with it. Stop the stack first. +- **Specs read the broker as well as the page.** `e2e/support/mqtt/broker-view.ts` connects the + way stagehand does: `BrokerView` follows topics live, `retainedSnapshot` is what a client + connecting now is handed. Stagehand re-checks alerts over REST, so a right-looking page does not + mean the broker is right. + +The specs drive state with `drivers/place/e2e/state_harness.cr`, pinned in +`e2e/support/drivers/drivers.env.ts` (`HARNESS_*`). It is a UDP device, so it reports +`connected: true` without a peer; `set_connected`, `set_status` and `clear_status` change what +it reports. Create a harness module with `running: true` rather than creating then starting it: +core drops a `start` that arrives before it has loaded the module (`No protocol manager`), and +the module stays `running` with no process behind it. + +If the build farm cannot build for this machine's architecture (arm64 answered 502 for every +build on 2026-10-14 while amd64 worked), run core as amd64 under emulation: + +```bash +DOCKER_DEFAULT_PLATFORM=linux/amd64 docker compose -p placeos-e2e -f e2e/stack/docker-compose.yml \ + up -d --force-recreate --no-deps core +``` + ## Layout ``` @@ -118,6 +169,15 @@ e2e/ shared engine, all apps meeting-flow.page.ts locators for the `meeting-flow-new` booking flow schedule.page.ts extends visitor/your-bookings.page.ts for event cards room.flows.ts multi-step UI flows (bookRoomViaUI) + mqtt/ the MQTT path stagehand reads + mqtt.seed.ts the Broker row source publishes to + broker-view.ts an MQTT client that sees what stagehand sees + stagehand/ stagehand coverage + stagehand.env.ts app URL and timings + stagehand.seed.ts OAuth app for the stagehand origin + the harness driver + stagehand.api.ts systems and harness modules + alerts.page.ts the hot list + fixtures.ts admin session, bearer, harness driver, zones repro/ standalone reproducers for confirmed backend bugs preflight.ts "is the stack up?" — fails in 1s, not 90 preflight.setup.ts setup project — `local` depends on it, `mock` does not diff --git a/e2e/stack/docker-compose.yml b/e2e/stack/docker-compose.yml index 702655e3d87..da13d649dd2 100644 --- a/e2e/stack/docker-compose.yml +++ b/e2e/stack/docker-compose.yml @@ -9,8 +9,14 @@ # without touching it. # 2. It is what CI needs. This is a trimmed stack — only what the e2e path # actually exercises. Dropped from PlaceOS/local: edge, triggers, dispatch, -# source, influx, chronograf, mosquitto, minio and the loki/promtail/grafana -# profile. That is roughly half the containers and most of the RAM. +# influx, chronograf, minio and the loki/promtail/grafana profile. That is +# roughly half the containers and most of the RAM. +# +# source and mosquitto ARE here: they are the MQTT path stagehand reads. A +# status a driver writes reaches the browser as driver -> redis -> source -> +# mosquitto -> nginx `/api/mqtt/` -> websocket, and the stagehand specs assert +# on both ends of it. source runs without influx, which it allows: only its +# MQTT publisher is under test. See "MQTT" in e2e/README.md. # # frontend-loader IS required, despite the SPA being served by the Angular dev # server. It does not only deploy app builds: on startup it clones @@ -284,6 +290,39 @@ services: # The User-Agent core sends the farm, so its logs can tell this stack apart. CLUSTER_NAME: ${E2E_STACK_PROJECT:-placeos-e2e} + # Publishes module state from redis to every Broker row, retained, one topic + # per status. Overridable so the suite can run against a source build that is + # not released yet: `E2E_SOURCE_IMAGE=placeos/source:local e2e/stack/up.sh`. + source: + image: ${E2E_SOURCE_IMAGE:-placeos/source:${PLACEOS_TAG:-latest}} + hostname: source + restart: always + <<: *std-logging + depends_on: + postgres: { condition: service_healthy } + redis: { condition: service_started } + mosquitto: { condition: service_started } + env_file: + - .secrets/.env.secret_key + environment: + <<: [*deployment-env, *postgresdb-client-env, *redis-client-env] + + # The broker stagehand subscribes to. Authentication is the production + # arrangement: mosquitto-go-auth asks rest-api about every client (a bearer for + # a browser, an API key for source) and every topic. Pinned to what + # PlaceOS/local's `latest` resolves to; it has not moved since 2023. + mosquitto: + image: iegomez/mosquitto-go-auth:${E2E_MOSQUITTO_TAG:-2.1.0-mosquitto_2.0.15} + hostname: mosquitto # load-bearing: nginx proxies /api/mqtt/ to mosquitto:9001 + restart: always + <<: *std-logging + depends_on: + - rest-api + volumes: + - ./mosquitto.conf:/etc/mosquitto/mosquitto.conf:ro + environment: + TZ: ${TZ:-UTC} + staff-api: image: placeos/staff-api:${PLACEOS_TAG:-latest} hostname: staff # load-bearing: nginx upstream is `staff` diff --git a/e2e/stack/mosquitto.conf b/e2e/stack/mosquitto.conf new file mode 100644 index 00000000000..ef3479c4fba --- /dev/null +++ b/e2e/stack/mosquitto.conf @@ -0,0 +1,21 @@ +# The e2e stack's MQTT broker. PlaceOS/local's config/mosquitto, with one fix: +# the auth routes. rest-api serves them at `mqtt/user` and `mqtt/access` +# (controllers/mqtt.cr); PlaceOS/local names `mqtt/mqtt_user` and +# `mqtt/mqtt_access`, which answer 404, so every client is refused. +# +# 1883 is where source publishes, 9001 is the websocket listener nginx proxies +# `/api/mqtt/` to. Every client and topic is checked with rest-api, so the specs +# exercise the same access control a deployment does. +listener 1883 +listener 9001 +protocol websockets + +auth_plugin /mosquitto/go-auth.so +auth_opt_backends jwt +auth_opt_jwt_mode remote +auth_opt_jwt_host api +auth_opt_jwt_port 3000 +auth_opt_jwt_getuser_uri /api/engine/v2/mqtt/user +auth_opt_jwt_aclcheck_uri /api/engine/v2/mqtt/access +auth_opt_jwt_response_mode status +auth_opt_jwt_params_mode form diff --git a/e2e/stack/up.sh b/e2e/stack/up.sh index b1582984c56..4aa2883bcce 100755 --- a/e2e/stack/up.sh +++ b/e2e/stack/up.sh @@ -151,6 +151,12 @@ dc run --rm init start step "starting core" dc up -d core +# source subscribes to the module tables at startup, so it comes up after init +# for the same reason core does. It publishes nothing until the seed below +# gives it a Broker row. +step "starting source and mosquitto" +dc up -d mosquitto source + # The last step of the seed waits for core to hold the suite's driver binaries. # A binary already in the drivers volume is used at once. Otherwise a commit the # build farm has already built for this CPU architecture is a download; one it @@ -159,6 +165,8 @@ step "seeding e2e fixtures" cd ../.. E2E_BACKEND_URL="https://localhost:${HTTPS_PORT}" bunx tsx e2e/support/seed.ts E2E_BACKEND_URL="https://localhost:${HTTPS_PORT}" bunx tsx e2e/support/room/room.seed.ts +E2E_BACKEND_URL="https://localhost:${HTTPS_PORT}" bunx tsx e2e/support/mqtt/mqtt.seed.ts +E2E_BACKEND_URL="https://localhost:${HTTPS_PORT}" bunx tsx e2e/support/stagehand/stagehand.seed.ts printf '\n\033[32mstack ready\033[0m backend=https://localhost:%s project=%s\n' "$HTTPS_PORT" "$PROJECT" printf 'run the suite with:\n E2E_BACKEND_URL=https://localhost:%s bunx playwright test --config apps/workplace/playwright.config.ts\n' "$HTTPS_PORT" diff --git a/e2e/support/drivers/drivers.api.ts b/e2e/support/drivers/drivers.api.ts index 20639c65fe6..d83902723e6 100644 --- a/e2e/support/drivers/drivers.api.ts +++ b/e2e/support/drivers/drivers.api.ts @@ -7,6 +7,9 @@ import { APIRequestContext } from '@playwright/test'; import { ENGINE_API, alreadyExists, asList, getJson } from '../api'; import { DRIVERS_COMMIT, DRIVERS_REPOSITORY, DriverSpec } from './drivers.env'; +/** A repository row the suite loads drivers from. */ +export type RepositorySpec = typeof DRIVERS_REPOSITORY; + interface RepositoryRow { id: string; folder_name: string; @@ -38,25 +41,26 @@ const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms)); */ export async function ensureDriversRepository( api: APIRequestContext, + repository: RepositorySpec = DRIVERS_REPOSITORY, ): Promise<{ id: string; created: boolean }> { const find = async () => asList( await getJson(api, `${ENGINE_API}/repositories`, { limit: '500' }), ).find( (r) => - r.folder_name === DRIVERS_REPOSITORY.folder_name && + r.folder_name === repository.folder_name && `${r.repo_type}`.toLowerCase() === 'driver', ); const found = await find(); if (found) { - await syncRepository(api, found); + await syncRepository(api, found, repository); return { id: found.id, created: false }; } const res = await api.post(`${ENGINE_API}/repositories`, { data: { - ...DRIVERS_REPOSITORY, + ...repository, repo_type: 'driver', commit_hash: 'HEAD', description: 'Drivers loaded by the e2e suite. Safe to delete.', @@ -74,13 +78,13 @@ export async function ensureDriversRepository( for (let i = 0; i < 15; i++) { const raced = await find(); if (raced) { - await syncRepository(api, raced); + await syncRepository(api, raced, repository); return { id: raced.id, created: false }; } await sleep(2000); } throw new Error( - `repository ${DRIVERS_REPOSITORY.folder_name} exists but is not listed`, + `repository ${repository.folder_name} exists but is not listed`, ); } @@ -88,8 +92,9 @@ export async function ensureDriversRepository( async function syncRepository( api: APIRequestContext, row: RepositoryRow, + repository: RepositorySpec, ): Promise { - const { uri, branch } = DRIVERS_REPOSITORY; + const { uri, branch } = repository; if (row.uri === uri && row.branch === branch) return; const res = await api.patch(`${ENGINE_API}/repositories/${row.id}`, { data: { uri, branch }, @@ -117,6 +122,7 @@ export async function ensureDriver( repository_id: string, spec: DriverSpec, repositoryIsNew: boolean, + commit: string = DRIVERS_COMMIT, ): Promise<{ id: string; created: boolean }> { const find = async () => asList( @@ -124,7 +130,7 @@ export async function ensureDriver( ).find( (d) => d.file_name === spec.file_name && - d.commit === DRIVERS_COMMIT && + d.commit === commit && d.repository_id === repository_id, ); @@ -142,7 +148,7 @@ export async function ensureDriver( role: spec.role, module_name: spec.module_name, file_name: spec.file_name, - commit: DRIVERS_COMMIT, + commit, repository_id, }, }); diff --git a/e2e/support/drivers/drivers.env.ts b/e2e/support/drivers/drivers.env.ts index a0f004c1866..5bede54d144 100644 --- a/e2e/support/drivers/drivers.env.ts +++ b/e2e/support/drivers/drivers.env.ts @@ -63,3 +63,39 @@ export const ROOM_DRIVERS = { } satisfies Record; export type RoomDriver = keyof typeof ROOM_DRIVERS; + +/** + * Where the state harness comes from: `drivers/place/e2e/state_harness.cr`, a + * UDP device whose `connected` state and other statuses are set by calling its + * functions. The stagehand specs use it to drive what a module publishes over + * MQTT. + * + * A repository row of its own, because it is pinned apart from the room + * drivers: the harness is on master, and they sit on a branch it is not on. + * When both are on master this can collapse into `DRIVERS_REPOSITORY`. + * + * Override with E2E_HARNESS_URI, E2E_HARNESS_BRANCH and E2E_HARNESS_COMMIT. + */ +export const HARNESS_REPOSITORY = { + name: 'E2E Harness Drivers', + folder_name: 'e2e-harness-drivers', + uri: process.env.E2E_HARNESS_URI ?? 'https://github.com/placeos/drivers', + branch: process.env.E2E_HARNESS_BRANCH ?? 'master', +}; + +export const HARNESS_COMMIT = + process.env.E2E_HARNESS_COMMIT ?? + 'a4b072a81bcfc3f8a48baf6236b5c1830a155f7c'; + +/** + * Shows up in systems as `Display`, so a system with two of them has + * `Display_1` and `Display_2` — which is what the reordering specs need. + * Role 1 is a device: it needs an address and a port, and any will do. + */ +export const HARNESS_DRIVER = { + name: 'E2E State Harness', + file_name: 'drivers/place/e2e/state_harness.cr', + module_name: 'Display', + role: 1, + description: 'Owned by the e2e suite. Safe to delete.', +} satisfies DriverSpec; diff --git a/e2e/support/mqtt/broker-view.ts b/e2e/support/mqtt/broker-view.ts new file mode 100644 index 00000000000..860d933efac --- /dev/null +++ b/e2e/support/mqtt/broker-view.ts @@ -0,0 +1,277 @@ +/** + * What the MQTT broker holds, seen the way stagehand sees it. + * + * Connects through nginx at `/api/mqtt/` with a bearer as username and + * password — stagehand's own arrangement, so rest-api checks this client and + * its subscriptions exactly as it checks the app's. + * + * source publishes every status RETAINED, so the broker holds one message per + * topic: the current value. That makes two questions worth asking, and this + * file answers both: + * + * - `BrokerView`: a live subscription. Follows the broker as it changes, so a + * spec can wait for a value to arrive or a topic to be removed. + * - `retainedSnapshot`: a fresh client's first look, which is exactly what a + * stagehand opened right now would load. Stale topics show up here and + * nowhere else: a client that was already connected never sees them again. + * + * Topic layout (source's `MqttPublisher.generate_status_key?`), with `_` for a + * hierarchy zone the system is not in: + * + * placeos//state///////// + */ +import mqtt, { MqttClient } from 'mqtt'; +import { BACKEND_URL } from '../env'; + +export interface StateTopic { + org: string; + region: string; + building: string; + level: string; + area: string; + system_id: string; + driver_id: string; + module_name: string; + index: string; + status: string; +} + +/** A retained status as source publishes it: `{time, value}`. */ +export interface StateValue { + /** Unix seconds. */ + time: number; + value: unknown; +} + +export function parseStateTopic(topic: string): StateTopic | null { + const p = topic.split('/'); + if (p.length !== 12 || p[0] !== 'placeos' || p[2] !== 'state') return null; + const [ + , + org, + , + region, + building, + level, + area, + system_id, + driver_id, + module_name, + index, + status, + ] = p; + return { + org, + region, + building, + level, + area, + system_id, + driver_id, + module_name, + index, + status, + }; +} + +/** Every status topic for one system, in any zone. */ +export function systemFilter(org_id: string, system_id: string): string { + return `placeos/${org_id}/state/+/+/+/+/${system_id}/#`; +} + +/** `Display_2` for `{module_name: 'Display', index: '2'}`. */ +export function moduleSlug( + t: Pick, +): string { + return `${t.module_name}_${t.index}`; +} + +function brokerUrl(): string { + const url = new URL('/api/mqtt/', BACKEND_URL); + url.protocol = url.protocol === 'https:' ? 'wss:' : 'ws:'; + return url.toString(); +} + +async function connect(token: string): Promise { + const client = mqtt.connect(brokerUrl(), { + username: token, + password: token, + // The stack terminates TLS with a self-signed certificate. + rejectUnauthorized: false, + reconnectPeriod: 0, + connectTimeout: 15_000, + clean: true, + }); + await new Promise((resolve, reject) => { + client.once('connect', () => resolve()); + client.once('error', (error) => + reject( + new Error( + `MQTT connect to ${brokerUrl()} failed: ${error.message}. ` + + `Check: docker compose -p placeos-e2e logs mosquitto`, + ), + ), + ); + }); + return client; +} + +async function subscribe(client: MqttClient, filter: string): Promise { + const granted = await client.subscribeAsync(filter, { qos: 0 }); + // MQTT 3.1.1 answers a refused subscription with 0x80 rather than an error + if (granted.some((g) => g.qos === 128)) { + throw new Error(`MQTT subscription to ${filter} was refused`); + } +} + +function decode(payload: Buffer): StateValue | undefined { + // A zero length retained publish removes the topic from the broker + if (payload.length === 0) return undefined; + return JSON.parse(payload.toString()) as StateValue; +} + +const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms)); + +/** A live view of the topics under one filter. */ +export class BrokerView { + /** Topic -> what the broker currently holds for it, as far as this client has seen. */ + readonly topics = new Map(); + /** Every message received, in order, for diagnosing a failure. */ + readonly log: { topic: string; retained: boolean; payload: string }[] = []; + + private constructor( + private readonly client: MqttClient, + readonly filter: string, + ) { + client.on('message', (topic, payload, packet) => { + this.log.push({ + topic, + retained: packet.retain, + payload: payload.toString(), + }); + const value = decode(payload); + if (value === undefined) this.topics.delete(topic); + else this.topics.set(topic, value); + }); + } + + static async open(token: string, filter: string): Promise { + const client = await connect(token); + const view = new BrokerView(client, filter); + await subscribe(client, filter); + return view; + } + + /** The current value of a module's status, matched on name, index and status. */ + find(module_name: string, index: number | string, status: string) { + for (const [topic, value] of this.topics) { + const t = parseStateTopic(topic); + if ( + t && + t.module_name === module_name && + t.index === `${index}` && + t.status === status + ) { + return { topic, ...t, ...value }; + } + } + return undefined; + } + + /** Wait until `module_name`_`index`.`status` holds `expected`. */ + async waitForValue( + module_name: string, + index: number | string, + status: string, + expected: unknown, + timeoutMs: number, + ) { + const deadline = Date.now() + timeoutMs; + for (;;) { + const found = this.find(module_name, index, status); + if ( + found && + JSON.stringify(found.value) === JSON.stringify(expected) + ) + return found; + if (Date.now() > deadline) { + throw new Error( + `${module_name}_${index}.${status} never became ${JSON.stringify(expected)} ` + + `on the broker within ${Math.round(timeoutMs / 1000)}s ` + + `(holds ${found ? JSON.stringify(found.value) : 'nothing'}).\n` + + `Messages on ${this.filter}:\n${this.describeLog()}`, + ); + } + await sleep(100); + } + } + + describeLog(): string { + return this.log.length + ? this.log + .map( + (m) => + ` ${m.retained ? 'R' : ' '} ${m.topic} ${m.payload || ''}`, + ) + .join('\n') + : ' (none)'; + } + + async close(): Promise { + await this.client.endAsync(true); + } +} + +/** + * What a client connecting now is handed: every retained topic under `filter`. + * + * The broker sends retained messages straight after it acknowledges the + * subscription, so this waits until `quietMs` passes without one. Only retained + * messages count; a live update arriving meanwhile is not part of the snapshot. + */ +export async function retainedSnapshot( + token: string, + filter: string, + quietMs = 1_500, +): Promise> { + const client = await connect(token); + const retained = new Map(); + let last = Date.now(); + client.on('message', (topic, payload, packet) => { + if (!packet.retain) return; + last = Date.now(); + const value = decode(payload); + if (value !== undefined) retained.set(topic, value); + }); + try { + await subscribe(client, filter); + last = Date.now(); + while (Date.now() - last < quietMs) await sleep(100); + return retained; + } finally { + await client.endAsync(true); + } +} + +/** + * A fresh client's view of one system: topic after `placeos//state/` -> + * value. Comparable with `expectedTopics` in `stagehand.api.ts`. + */ +export async function retainedForSystem( + token: string, + org_id: string, + system_id: string, +): Promise> { + const snapshot = await retainedSnapshot( + token, + systemFilter(org_id, system_id), + 1_000, + ); + const prefix = `placeos/${org_id}/state/`; + return Object.fromEntries( + [...snapshot].map(([topic, v]) => [ + topic.slice(prefix.length), + v.value, + ]), + ); +} diff --git a/e2e/support/mqtt/mqtt.seed.ts b/e2e/support/mqtt/mqtt.seed.ts new file mode 100644 index 00000000000..1e58c3ad9c5 --- /dev/null +++ b/e2e/support/mqtt/mqtt.seed.ts @@ -0,0 +1,123 @@ +/** + * Connecting `source` to the stack's MQTT broker. + * + * source publishes to every Broker row in the database and to nothing else, so + * without one the stack has a broker that never receives a message. This adds + * the row, with the credentials a deployment uses: an API key, sent as both + * username and password. mosquitto-go-auth hands it to rest-api + * (`/api/engine/v2/mqtt/mqtt_user`), which accepts an API key wherever it + * accepts a bearer and, unlike a bearer, it does not expire mid-run. + * + * Publishing needs support permissions (`MQTT.mqtt_acl_status` in rest-api), so + * the key is minted as Support. A browser only subscribes, which any signed in + * user may do. + * + * Run by `e2e/stack/up.sh`. Idempotent: a Broker row that exists is left alone, + * because the API key's secret is only returned when it is created and cannot + * be read back to compare. + */ +import { APIRequestContext } from '@playwright/test'; +import { ENGINE_API, asList, currentUser, getJson } from '../api'; +import { BACKEND_URL, assertLocalOnly } from '../env'; +import { adminApi } from '../seed'; + +export const BROKER_NAME = 'E2E Mosquitto'; +const API_KEY_NAME = 'e2e-source-mqtt'; + +/** `UserJWT::Permissions::Support` */ +const SUPPORT_PERMISSIONS = 1; + +interface BrokerRow { + id: string; + name: string; + host: string; + port: number; +} + +interface ApiKeyRow { + id: string; + name: string; + x_api_key?: string | null; +} + +export async function ensureBroker( + api: APIRequestContext, +): Promise<{ id: string; created: boolean }> { + const brokers = asList( + await getJson(api, `${ENGINE_API}/brokers`, { limit: '100' }), + ); + const found = brokers.find((b) => b.name === BROKER_NAME); + if (found) return { id: found.id, created: false }; + + const api_key = await mintPublisherKey(api); + const res = await api.post(`${ENGINE_API}/brokers`, { + data: { + name: BROKER_NAME, + description: + 'The e2e stack broker that source publishes to. Safe to delete.', + host: 'mosquitto', + port: 1883, + tls: false, + auth_type: 'UserPassword', + username: api_key, + password: api_key, + }, + }); + if (!res.ok()) { + throw new Error( + `create broker failed: HTTP ${res.status()} ${await res.text()}`, + ); + } + return { id: (await res.json()).id, created: true }; +} + +/** + * A fresh API key for source. One left by an earlier seed whose broker row has + * since gone is replaced: its secret is unrecoverable, and the name is unique. + */ +async function mintPublisherKey(api: APIRequestContext): Promise { + const keys = asList( + await getJson(api, `${ENGINE_API}/api_keys`, { limit: '500' }), + ); + for (const stale of keys.filter((k) => k.name === API_KEY_NAME)) { + await api.delete(`${ENGINE_API}/api_keys/${stale.id}`); + } + const user = await currentUser(api); + const res = await api.post(`${ENGINE_API}/api_keys`, { + data: { + name: API_KEY_NAME, + description: 'source publishing to the e2e broker. Safe to delete.', + user_id: user.id, + scopes: ['public'], + permissions: SUPPORT_PERMISSIONS, + }, + }); + if (!res.ok()) { + throw new Error( + `create api key failed: HTTP ${res.status()} ${await res.text()}`, + ); + } + const key: ApiKeyRow = await res.json(); + if (!key.x_api_key) { + throw new Error('create api key returned no x_api_key'); + } + return key.x_api_key; +} + +if (require.main === module) { + (async () => { + assertLocalOnly(BACKEND_URL); + const api = await adminApi(); + try { + const broker = await ensureBroker(api); + console.log( + ` broker ${broker.created ? 'created' : 'present'} ${BROKER_NAME} -> mosquitto:1883`, + ); + } finally { + await api.dispose(); + } + })().catch((e) => { + console.error(e.message); + process.exit(1); + }); +} diff --git a/e2e/support/seed.ts b/e2e/support/seed.ts index 604116294c3..301000251cb 100644 --- a/e2e/support/seed.ts +++ b/e2e/support/seed.ts @@ -79,7 +79,7 @@ async function authority(api: APIRequestContext, timeoutMs = 90_000) { * `nx serve` the app origin is the dev server, NOT the stack, so this is a * different application from the deployed one. */ -async function ensureOAuthApp(api: APIRequestContext, appUrl: string, name: string) { +export async function ensureOAuthApp(api: APIRequestContext, appUrl: string, name: string) { const redirect_uri = redirectUriFor(appUrl); const uid = clientId(redirect_uri); diff --git a/e2e/support/stagehand/alerts.page.ts b/e2e/support/stagehand/alerts.page.ts new file mode 100644 index 00000000000..2852fce8ece --- /dev/null +++ b/e2e/support/stagehand/alerts.page.ts @@ -0,0 +1,70 @@ +/** + * Stagehand's hot list: the alerts page with the built in "Disconnected + * Devices" view, which lists every module whose `connected` is false. + * + * The table is `simple-table`, so rows are read with the concierge page object + * that already knows how to regroup its flat grid. A hot list row carries the + * device as `_` and the location as the system's name with + * its id underneath, so a row is attributed to a system by the id. + * + * Until the first alert arrives the page shows a spinner instead of the table, + * for up to 15 seconds. "No row for my module" is therefore only meaningful + * once some row is showing; the specs hold a disconnected control module for + * exactly that. + */ +import { Page, expect } from '@playwright/test'; +import { readTable } from '../concierge/simple-table.page'; + +export interface HotListRow { + device: string; + cells: string[]; +} + +export async function openHotList(page: Page): Promise { + await page.goto('/#/alerts'); + await expect( + page.getByRole('heading', { name: 'AV Systems Alerts' }), + ).toBeVisible({ + timeout: 60_000, + }); +} + +/** The hot list rows that belong to `system_id`, read once. */ +export async function hotListRows( + page: Page, + system_id: string, +): Promise { + if (!(await page.locator('simple-table [role=table]').count())) return []; + const table = await readTable(page); + const device_column = table.headers.findIndex((h) => /device/i.test(h)); + return table.rows + .filter((cells) => cells.some((c) => c.includes(system_id))) + .map((cells) => ({ + // `Display_1~~Control System`: the slug comes first + device: (cells[device_column] ?? '').split('~')[0].trim(), + cells, + })); +} + +/** The devices in the hot list for `system_id`, sorted, for an exact assertion. */ +export async function hotListDevices( + page: Page, + system_id: string, +): Promise { + return (await hotListRows(page, system_id)).map((r) => r.device).sort(); +} + +/** Wait until the hot list for `system_id` is exactly `devices`. */ +export async function expectHotList( + page: Page, + system_id: string, + devices: string[], + timeout = 30_000, +): Promise { + await expect + .poll(() => hotListDevices(page, system_id), { + message: `stagehand's hot list for ${system_id}`, + timeout, + }) + .toEqual([...devices].sort()); +} diff --git a/e2e/support/stagehand/fixtures.ts b/e2e/support/stagehand/fixtures.ts new file mode 100644 index 00000000000..c110c56e184 --- /dev/null +++ b/e2e/support/stagehand/fixtures.ts @@ -0,0 +1,147 @@ +/** + * Fixtures for the STAGEHAND app: an admin session bound to the stagehand + * origin, an admin API context, the admin's bearer for MQTT clients, and the + * harness driver. + * + * Same shape as `concierge/fixtures.ts`, for the same reason: a storage state + * belongs to one origin, so stagehand cannot borrow workplace's. + */ +import { + APIRequestContext, + test as base, + request as pwRequest, +} from '@playwright/test'; +import * as fs from 'fs'; +import * as path from 'path'; +import { ENGINE_API } from '../api'; +import { buildStorageState, mintToken } from '../auth'; +import { AUTH_DIR, BACKEND_URL, roleFor } from '../env'; +import { adminApi as bootstrapApi } from '../seed'; +import { HierarchyZones, hierarchyZones } from './stagehand.api'; +import { STAGEHAND_URL, assertStagehandLocal } from './stagehand.env'; +import { ensureHarnessDriver, ensureStagehandOAuthApp } from './stagehand.seed'; + +interface StagehandWorkerFixtures { + /** storageState path for the admin, bound to the stagehand origin. */ + stagehandAdminState: string; + /** A bearer for the admin: what the specs' own MQTT clients sign in with. */ + stagehandAdminToken: string; + /** The state harness driver's id. Loaded by `up.sh`; checked here. */ + harnessDriverId: string; + /** The org, building and level the specs' systems go in. */ + zones: HierarchyZones; +} + +interface StagehandTestFixtures { + adminApi: APIRequestContext; +} + +async function apiWithToken(token: string): Promise { + return pwRequest.newContext({ + baseURL: BACKEND_URL, + ignoreHTTPSErrors: true, + extraHTTPHeaders: { Authorization: `Bearer ${token}` }, + }); +} + +async function mintForStagehand(workerIndex: number) { + assertStagehandLocal(); + const setup = await bootstrapApi(); + try { + await ensureStagehandOAuthApp(setup); + } finally { + await setup.dispose(); + } + const admin = roleFor('admin', workerIndex); + const mint = await mintToken( + BACKEND_URL, + STAGEHAND_URL, + admin.email, + admin.password, + ); + fs.mkdirSync(AUTH_DIR, { recursive: true }); + const state_path = path.join( + AUTH_DIR, + `stagehand-admin-${workerIndex}.json`, + ); + fs.writeFileSync( + state_path, + JSON.stringify(buildStorageState(mint, STAGEHAND_URL), null, 2), + ); + return { state_path, token: mint.accessToken }; +} + +export const test = base.extend( + { + storageState: ({ stagehandAdminState }, use) => + use(stagehandAdminState), + baseURL: ({}, use) => use(STAGEHAND_URL), + + stagehandAdminState: [ + async ({}, use) => { + const { state_path } = await mintForStagehand( + test.info().parallelIndex, + ); + await use(state_path); + }, + { scope: 'worker' }, + ], + + stagehandAdminToken: [ + async ({}, use) => { + const { token } = await mintForStagehand( + test.info().parallelIndex, + ); + // Same check the other suites make: a token that parses but is + // refused downstream fails in confusing ways much later. + const api = await apiWithToken(token); + try { + const res = await api.get(`${ENGINE_API}/users/current`); + if (!res.ok()) { + throw new Error( + `minted admin token was rejected: HTTP ${res.status()}`, + ); + } + } finally { + await api.dispose(); + } + await use(token); + }, + { scope: 'worker' }, + ], + + harnessDriverId: [ + async ({ stagehandAdminToken }, use) => { + const api = await apiWithToken(stagehandAdminToken); + try { + // A short budget: `up.sh` already waited for the build farm, so + // a binary that is not there now is a broken bring-up. + await use(await ensureHarnessDriver(api, () => {}, 60_000)); + } finally { + await api.dispose(); + } + }, + { scope: 'worker', timeout: 120_000 }, + ], + + zones: [ + async ({ stagehandAdminToken }, use) => { + const api = await apiWithToken(stagehandAdminToken); + try { + await use(await hierarchyZones(api)); + } finally { + await api.dispose(); + } + }, + { scope: 'worker' }, + ], + + adminApi: async ({ stagehandAdminToken }, use) => { + const api = await apiWithToken(stagehandAdminToken); + await use(api); + await api.dispose(); + }, + }, +); + +export { expect } from '@playwright/test'; diff --git a/e2e/support/stagehand/stagehand.api.ts b/e2e/support/stagehand/stagehand.api.ts new file mode 100644 index 00000000000..e460b31854f --- /dev/null +++ b/e2e/support/stagehand/stagehand.api.ts @@ -0,0 +1,384 @@ +/** + * Systems and harness modules for the stagehand specs, through the engine API. + * + * Every spec builds its own system, so nothing it asserts can be disturbed by + * another spec or by a leftover: stagehand's hot list is org wide, and the only + * part of it a spec can own is the rows for its own system. + * + * Needs ADMIN. + */ +import { APIRequestContext } from '@playwright/test'; +import { ENGINE_API, asList, getJson, uniqueTitle, zonesWithTag } from '../api'; +import { moduleState } from '../drivers/drivers.api'; +import { STAGEHAND_SYSTEM_PREFIX } from './stagehand.env'; + +export interface HierarchyZones { + org: string; + building: string; + level: string; +} + +/** The org, building and level `init` seeded. */ +export async function hierarchyZones( + api: APIRequestContext, +): Promise { + const [[org], [building], [level]] = await Promise.all( + ['org', 'building', 'level'].map((tag) => zonesWithTag(api, tag)), + ); + if (!org || !building || !level) { + throw new Error( + `the stack has no org/building/level zones (got ${!!org}/${!!building}/${!!level}); ` + + `\`init start\` creates them`, + ); + } + return { org: org.id, building: building.id, level: level.id }; +} + +export interface SystemRow { + id: string; + name: string; + zones: string[]; + modules: string[]; +} + +/** + * A system in the org, building and level. + * + * The org zone is what makes source publish its modules at all: a status topic + * begins with the org, and source drops events for a system it cannot scope. + */ +export async function createSystem( + api: APIRequestContext, + zones: HierarchyZones, + label: string, +): Promise { + const res = await api.post(`${ENGINE_API}/systems`, { + data: { + name: uniqueTitle(`${STAGEHAND_SYSTEM_PREFIX} ${label}`), + description: 'Created by the stagehand e2e specs. Safe to delete.', + zones: [zones.org, zones.building, zones.level], + }, + }); + if (!res.ok()) { + throw new Error( + `create system failed: HTTP ${res.status()} ${await res.text()}`, + ); + } + return res.json(); +} + +export async function getSystem( + api: APIRequestContext, + id: string, +): Promise { + return getJson(api, `${ENGINE_API}/systems/${id}`); +} + +/** Replace a system's module list, which is also its module ORDER. */ +export async function setSystemModules( + api: APIRequestContext, + system: SystemRow, + modules: string[], +): Promise { + const current = await getSystem(api, system.id); + const res = await api.patch(`${ENGINE_API}/systems/${system.id}`, { + params: { + version: `${(current as unknown as { version: number }).version}`, + }, + data: { modules }, + }); + if (!res.ok()) { + throw new Error( + `update modules of ${system.id} failed: HTTP ${res.status()} ${await res.text()}`, + ); + } + return res.json(); +} + +/** Delete a system and every module in it. */ +export async function deleteSystem( + api: APIRequestContext, + system_id: string, +): Promise { + const system = await getSystem(api, system_id).catch(() => null); + if (!system) return; + for (const module_id of system.modules ?? []) { + await api.delete(`${ENGINE_API}/modules/${module_id}`); + } + await api.delete(`${ENGINE_API}/systems/${system_id}`); +} + +/** + * Delete systems a crashed run left behind. + * + * Only ones older than `olderThanMs`, so a sweep in one worker never takes a + * system another worker is using right now. + */ +export async function sweepSystems( + api: APIRequestContext, + olderThanMs = 30 * 60 * 1000, +): Promise { + const res = await api.get(`${ENGINE_API}/systems`, { + params: { q: STAGEHAND_SYSTEM_PREFIX, limit: '500' }, + }); + if (!res.ok()) return; + const cutoff = Date.now() / 1000 - olderThanMs / 1000; + for (const system of asList( + await res.json(), + )) { + if ( + system.name.startsWith(STAGEHAND_SYSTEM_PREFIX) && + system.created_at < cutoff + ) { + await deleteSystem(api, system.id); + } + } +} + +export interface HarnessModule { + id: string; + custom_name?: string; +} + +/** + * Add a running state harness to the end of the system's module list. + * + * A device module needs an address and a port; the harness talks to nothing, + * so loopback and the discard port will do. Unlike a logic module, a device + * may not name a system when it is created (422, "should not be associated for + * Device modules"): it joins one through the system's module list. + * + * Created already running, rather than created and then started: core loads a + * module asynchronously on the create event, and a `start` that lands before + * the load has finished is dropped ("No protocol manager for mod-…") and never + * retried, leaving a module that is `running` with no process behind it. + */ +export async function addHarness( + api: APIRequestContext, + system_id: string, + driver_id: string, + options: { custom_name?: string } = {}, +): Promise { + const res = await api.post(`${ENGINE_API}/modules`, { + data: { + driver_id, + ip: '127.0.0.1', + port: 9, + udp: true, + running: true, + custom_name: options.custom_name, + notes: 'Created by the stagehand e2e specs. Safe to delete.', + }, + }); + if (!res.ok()) { + throw new Error( + `create harness module failed: HTTP ${res.status()} ${await res.text()}`, + ); + } + const mod: { id: string } = await res.json(); + const added = await api.put( + `${ENGINE_API}/systems/${system_id}/module/${mod.id}`, + ); + if (!added.ok()) { + throw new Error( + `add module ${mod.id} to ${system_id} failed: HTTP ${added.status()} ${await added.text()}`, + ); + } + return { id: mod.id, custom_name: options.custom_name }; +} + +/** Call a function on a module. */ +export async function exec( + api: APIRequestContext, + module_id: string, + method: string, + args: unknown[] = [], +): Promise { + const res = await api.post( + `${ENGINE_API}/modules/${module_id}/exec/${method}`, + { + data: args, + }, + ); + if (!res.ok()) { + throw new Error( + `exec ${method} on ${module_id} failed: HTTP ${res.status()} ${await res.text()}`, + ); + } + return res.json().catch(() => null); +} + +/** Report the module as connected or not, as a device that stopped answering would. */ +export async function setConnected( + api: APIRequestContext, + module_id: string, + connected: boolean, +): Promise { + await exec(api, module_id, 'set_connected', [connected]); +} + +/** + * Wait until the module's own state (redis, read through rest-api) holds + * `connected: expected`. The source of truth the broker is compared against. + */ +export async function waitForConnected( + api: APIRequestContext, + module_id: string, + expected: boolean, + timeoutMs = 60_000, +): Promise { + const deadline = Date.now() + timeoutMs; + for (;;) { + const state = await moduleState(api, module_id).catch( + () => ({}) as Record, + ); + if (state['connected'] === expected) return; + if (Date.now() > deadline) { + throw new Error( + `module ${module_id} never reported connected=${expected} within ` + + `${Math.round(timeoutMs / 1000)}s (state: ${JSON.stringify(state)}). ` + + `A module with no state did not start: docker compose -p placeos-e2e logs core`, + ); + } + await new Promise((r) => setTimeout(r, 500)); + } +} + +/** Take a module out of a system without deleting it. */ +export async function removeFromSystem( + api: APIRequestContext, + system_id: string, + module_id: string, +): Promise { + const res = await api.delete( + `${ENGINE_API}/systems/${system_id}/module/${module_id}`, + ); + if (!res.ok()) { + throw new Error( + `remove ${module_id} from ${system_id} failed: HTTP ${res.status()} ${await res.text()}`, + ); + } +} + +export async function deleteModule( + api: APIRequestContext, + module_id: string, +): Promise { + const res = await api.delete(`${ENGINE_API}/modules/${module_id}`); + if (!res.ok() && res.status() !== 404) { + throw new Error( + `delete module ${module_id} failed: HTTP ${res.status()} ${await res.text()}`, + ); + } +} + +/** Set a module's `custom_name`, which is the name systems address it by. */ +export async function renameModule( + api: APIRequestContext, + module_id: string, + custom_name: string, +): Promise { + const res = await api.patch(`${ENGINE_API}/modules/${module_id}`, { + data: { custom_name }, + }); + if (!res.ok()) { + throw new Error( + `rename ${module_id} failed: HTTP ${res.status()} ${await res.text()}`, + ); + } +} + +/** Replace a system's zones. */ +export async function setSystemZones( + api: APIRequestContext, + system: SystemRow, + zones: string[], +): Promise { + const current = await getSystem(api, system.id); + const res = await api.patch(`${ENGINE_API}/systems/${system.id}`, { + params: { + version: `${(current as unknown as { version: number }).version}`, + }, + data: { zones }, + }); + if (!res.ok()) { + throw new Error( + `update zones of ${system.id} failed: HTTP ${res.status()} ${await res.text()}`, + ); + } + return res.json(); +} + +/** A level zone of the suite's own under `building`, for moving systems between. */ +export async function ensureSecondLevel( + api: APIRequestContext, + building: string, +): Promise { + const name = 'E2E Stagehand Level 2'; + const found = (await zonesWithTag(api, 'level')).find( + (z) => z.name === name, + ); + if (found) return found.id; + const res = await api.post(`${ENGINE_API}/zones`, { + data: { + name, + tags: ['level'], + parent_id: building, + description: 'Safe to delete.', + }, + }); + if (!res.ok()) { + throw new Error( + `create level zone failed: HTTP ${res.status()} ${await res.text()}`, + ); + } + return (await res.json()).id; +} + +interface ModuleRow { + id: string; + name: string; + custom_name?: string | null; + driver_id: string; +} + +/** + * Every status topic the broker SHOULD retain for a system, from the source of + * truth: the system's module list (order gives the index), each module's name, + * the system's zones, and each module's own state in redis. + * + * Keys are the topic after `placeos//state/`; values are the status + * values. `levels` lists the level zones a system may be in, so the topic's + * level segment can be filled from whichever the system holds. + */ +export async function expectedTopics( + api: APIRequestContext, + system_id: string, + zones: HierarchyZones, + levels: string[] = [zones.level], +): Promise> { + const system = await getSystem(api, system_id).catch(() => null); + if (!system) return {}; + if (!system.zones.includes(zones.org)) return {}; + const building = system.zones.includes(zones.building) + ? zones.building + : '_'; + const level = levels.find((l) => system.zones.includes(l)) ?? '_'; + const counts: Record = {}; + const out: Record = {}; + for (const module_id of system.modules ?? []) { + const mod: ModuleRow = await getJson( + api, + `${ENGINE_API}/modules/${module_id}`, + ); + const name = mod.custom_name || mod.name; + const index = (counts[name] = (counts[name] ?? 0) + 1); + const state = await moduleState(api, module_id); + for (const [status, value] of Object.entries(state)) { + out[ + `_/${building}/${level}/_/${system_id}/${mod.driver_id}/${name}/${index}/${status}` + ] = value; + } + } + return out; +} diff --git a/e2e/support/stagehand/stagehand.env.ts b/e2e/support/stagehand/stagehand.env.ts new file mode 100644 index 00000000000..8f62cfcb3b3 --- /dev/null +++ b/e2e/support/stagehand/stagehand.env.ts @@ -0,0 +1,43 @@ +/** + * Config for the STAGEHAND app's e2e specs. + * + * Stagehand is an AV support console. Its hot list is built from MQTT: `source` + * publishes every module status to the broker, retained, and stagehand + * subscribes to `.../connected` (and to whatever a dashboard's custom alerts + * name). A module whose `connected` is false is a row in the hot list. + * + * So these specs own the whole path, and assert on both ends of it: + * + * harness module --exec--> redis --> source --> mosquitto --> stagehand + * | + * an MQTT client in the spec (the broker's view) + * + * The broker's view matters on its own. Stagehand double checks every MQTT + * alert against the REST API before showing it, which hides wrong data on the + * broker. Asserting on the broker directly is what proves that data is right, + * and is the precondition for removing the double check. + * + * Signed in as ADMIN: stagehand is a support tool, and the systems and modules + * the specs create need admin rights anyway. + */ +import { APP_URL, BACKEND_URL, assertLocalOnly } from '../env'; + +/** Where the stagehand app is served. Its Playwright config serves it here. */ +export const STAGEHAND_URL = + process.env.E2E_STAGEHAND_URL ?? 'http://localhost:4210'; + +/** Refuse to run against anything that is not local. */ +export function assertStagehandLocal(): void { + assertLocalOnly(BACKEND_URL, APP_URL, STAGEHAND_URL); +} + +/** Prefix of every system these specs create, so a sweep can find leftovers. */ +export const STAGEHAND_SYSTEM_PREFIX = 'E2E Stagehand'; + +/** + * How long a status may take to travel from a module to the broker. + * + * source batches redis events every 40 ms and the broker is local, so this is + * normally well under a second. The budget is for a cold stack. + */ +export const MQTT_PROPAGATION_MS = 20_000; diff --git a/e2e/support/stagehand/stagehand.seed.ts b/e2e/support/stagehand/stagehand.seed.ts new file mode 100644 index 00000000000..0b3ea1802df --- /dev/null +++ b/e2e/support/stagehand/stagehand.seed.ts @@ -0,0 +1,90 @@ +/** + * Seeding for the stagehand specs: an OAuth application for the stagehand + * origin, and the state harness driver. + * + * Owned by the stagehand specs rather than added to `e2e/support/seed.ts`, for + * the reason `concierge.seed.ts` gives: the shared seed is what every workplace + * suite needs to boot, and a harness the build farm cannot build must not stop + * them. `up.sh` runs this after the shared seed; the fixtures run it again, + * which costs a couple of list requests. + */ +import { APIRequestContext } from '@playwright/test'; +import { + ensureDriver, + ensureDriversRepository, + waitForDriverCompiled, +} from '../drivers/drivers.api'; +import { + HARNESS_COMMIT, + HARNESS_DRIVER, + HARNESS_REPOSITORY, +} from '../drivers/drivers.env'; +import { DRIVER_LOAD_TIMEOUT_MS } from '../drivers/drivers.seed'; +import { adminApi, ensureOAuthApp } from '../seed'; +import { STAGEHAND_URL, assertStagehandLocal } from './stagehand.env'; + +export async function ensureStagehandOAuthApp(api: APIRequestContext) { + return ensureOAuthApp(api, STAGEHAND_URL, 'stagehand (e2e dev server)'); +} + +/** + * The state harness driver row, once core holds its binary. Returns its id. + * + * A first run on an architecture waits for the build farm to compile it, which + * takes a few minutes; `up.sh` pays that, not a spec. + */ +export async function ensureHarnessDriver( + api: APIRequestContext, + log: (message: string) => void = () => {}, + timeoutMs = DRIVER_LOAD_TIMEOUT_MS, +): Promise { + const repository = await ensureDriversRepository(api, HARNESS_REPOSITORY); + const driver = await ensureDriver( + api, + repository.id, + HARNESS_DRIVER, + repository.created, + HARNESS_COMMIT, + ); + await waitForDriverCompiled( + api, + driver.id, + `${HARNESS_DRIVER.name} (${HARNESS_DRIVER.file_name} @ ${HARNESS_COMMIT.slice(0, 7)})`, + timeoutMs, + log, + ); + return driver.id; +} + +if (require.main === module) { + (async () => { + assertStagehandLocal(); + const api = await adminApi(); + try { + const app = await ensureStagehandOAuthApp(api); + console.log( + ` oauth app ${app.created ? 'created' : 'present'} ${app.uid} ${app.redirect_uri}`, + ); + // Like the room drivers: a farm that cannot build the harness + // fails the stagehand specs, not the bring-up. + try { + const id = await ensureHarnessDriver(api, (message) => + console.log(` harness ${message}`), + ); + console.log( + ` harness ${id} ${HARNESS_REPOSITORY.uri}@${HARNESS_REPOSITORY.branch} ${HARNESS_COMMIT.slice(0, 7)}`, + ); + } catch (error) { + console.warn( + ` harness NOT LOADED; the stagehand specs will fail.\n` + + `${error instanceof Error ? error.message : error}`, + ); + } + } finally { + await api.dispose(); + } + })().catch((e) => { + console.error(e.message); + process.exit(1); + }); +}