c86b75ac52
WAI toolchain decision 11: joule-moq-relay builds wai-relay, joule-moq-room wai-room, and joule-moq-ingest wai-ingest. wai dev finds them by those names (WAI_RELAY_BIN, WAI_ROOM_BIN), and the quickstart archive, the release workflow, the examples' workflow, the browser tests, the interop scripts, the example host script, the Scoop manifest and the crates' READMEs and example configurations name them so. Wire and record identifiers that carry a crate's name (joule-moq-relay/session-summary/2, joule-moq-ingest/session/1, the relay's MOQT implementation string) and the log variables (JOULE_MOQ_*_LOG) are unchanged. Co-Authored-By: Claude Opus 5.5 <[email protected]>
300 lines
18 KiB
JavaScript
300 lines
18 KiB
JavaScript
// End-to-end tests of the messaging view (web/room/messages.html) and the call flow
|
|
// (web/room/call.html) against a room service (wai-room) on loopback, in the host's
|
|
// installed browser driven by the automation package of e2e/package.json, with a fake camera
|
|
// and microphone (WAI live-messaging).
|
|
//
|
|
// node e2e/messaging.e2e.mjs --room <wai-room> --relay <wai-relay> --web <web dir>
|
|
// [--browser <executable>] [--only messaging,call] [--out <dir>] [--runs N]
|
|
//
|
|
// Scenarios:
|
|
// messaging a conversation kept by the service with a history store: three pages join, each
|
|
// making its own key, key through the service's log (KeyPackages, the clock's
|
|
// proposals, a GroupInfo, commits, Welcomes), and message one another, each
|
|
// opening what the others seal; one page goes offline (its network cut), the
|
|
// others start a thread, reply and react, the service is stopped and started again
|
|
// from its store, the root is edited and a message sent and retracted; the page
|
|
// offline comes back and catches up from the store, opening every message of
|
|
// both epochs, the thread, the reaction, the edit and the retraction shown as the
|
|
// others show them; each page acknowledges what it read; the owner removes a
|
|
// member, whose page is told it is no longer a member, and opens nothing written
|
|
// after; the store holds no plaintext.
|
|
// call a call between two pages: the caller places it naming the callee's key, the
|
|
// callee's inbox rings after checking the invitation and its call, the callee
|
|
// accepts, and both pages' rooms publish camera and microphone and decode the
|
|
// other's video and audio; then a call to a third page that never answers is
|
|
// missed at its deadline, signed by the room's clock, and shown as missed.
|
|
//
|
|
// It prints one line per check and `messaging e2e: <passed> of <total> runs passed`, and exits
|
|
// non-zero when any check fails.
|
|
|
|
import fs from "node:fs";
|
|
import path from "node:path";
|
|
import { createRequire } from "node:module";
|
|
import { spawn } from "node:child_process";
|
|
import net from "node:net";
|
|
import { serveSameOrigin, startRelay, roomKeys, startRoomService, browserArgs, memguardedBrowser, killStrays, tmpdir, until, sleep } from "./lib.mjs";
|
|
|
|
const args = Object.fromEntries(process.argv.slice(2).reduce((a, x, i, all) => (x.startsWith("--") ? [...a, [x.slice(2), all[i + 1] && !all[i + 1].startsWith("--") ? all[i + 1] : "1"]] : a), []));
|
|
const need = (k) => { if (!args[k]) { console.error(`messaging e2e: --${k} is required`); process.exit(2); } return path.resolve(args[k]); };
|
|
const ROOM = need("room");
|
|
const RELAY = need("relay");
|
|
const WEB = need("web");
|
|
const ONLY = (args.only || "messaging,call").split(",");
|
|
const RUNS = Number(args.runs || 1);
|
|
const OUT = args.out ? path.resolve(args.out) : tmpdir("msg-out");
|
|
fs.mkdirSync(OUT, { recursive: true });
|
|
const require = createRequire(path.join(process.env.PLAYWRIGHT_HOST || path.join(path.dirname(new URL(import.meta.url).pathname), "node_modules"), "x"));
|
|
const { chromium } = require("playwright");
|
|
|
|
const STEP = 60000;
|
|
const results = [];
|
|
let failed = 0;
|
|
function check(name, ok, detail = "") {
|
|
results.push({ name, ok, detail });
|
|
if (!ok) failed++;
|
|
console.log(`${ok ? "ok " : "FAIL"} ${name}${detail ? ` — ${detail}` : ""}`);
|
|
}
|
|
|
|
process.on("unhandledRejection", (e) => { console.error("unhandled:", e && e.stack || e); killStrays(); process.exit(1); });
|
|
|
|
async function launch(dir) {
|
|
const bin = args.browser ? path.resolve(args.browser) : null;
|
|
const opts = { headless: true, args: browserArgs() };
|
|
if (bin) opts.executablePath = memguardedBrowser(bin, dir).executable;
|
|
const b = await chromium.launch(opts);
|
|
console.log(`browser: ${b.version()}`);
|
|
return b;
|
|
}
|
|
|
|
async function openPage(browser, url, name, dir) {
|
|
const ctx = await browser.newContext({ permissions: ["camera", "microphone"], colorScheme: "dark" });
|
|
const p = await ctx.newPage();
|
|
const logs = [];
|
|
p.on("console", (m) => logs.push(`[${m.type()}] ${m.text()}`));
|
|
p.on("pageerror", (e) => logs.push(`[pageerror] ${e.stack || e}`));
|
|
await p.goto(url, { waitUntil: "domcontentloaded", timeout: STEP });
|
|
p.ctx = ctx;
|
|
p.name = name;
|
|
p.dump = () => fs.writeFileSync(path.join(dir, `${name}.console.log`), logs.join("\n"));
|
|
return p;
|
|
}
|
|
|
|
function freePort() {
|
|
return new Promise((ok) => { const s = net.createServer(); s.listen(0, "127.0.0.1", () => { const p = s.address().port; s.close(() => ok(p)); }); });
|
|
}
|
|
|
|
// The room service keeping one conversation, its store in `store`; started again on the same
|
|
// port and store by `restart()`.
|
|
async function startConversation(dir, keys, store) {
|
|
const port = await freePort();
|
|
const toml = path.join(dir, "conversation.toml");
|
|
fs.writeFileSync(toml, `listen = "127.0.0.1:${port}"\nkeys = ${JSON.stringify(keys)}\nrelay_url = "ws://127.0.0.1:1/moq"\n# The path the front serves the service under: a proof names the path the page addressed.\npublic_prefix = "/room-api"\n\n[messaging]\nstore = ${JSON.stringify(store)}\nmax_members = 8\n\n[rate]\nclient_burst = 64\n`);
|
|
const url = `http://127.0.0.1:${port}`;
|
|
let proc = null, exited = null, n = 0;
|
|
const start = async () => {
|
|
const log = path.join(dir, `conversation-${n++}.log`);
|
|
const out = fs.openSync(log, "w");
|
|
proc = spawn(ROOM, ["--config", toml], { stdio: ["ignore", out, out], env: { ...process.env, NO_COLOR: "1" } });
|
|
exited = new Promise((r) => proc.on("exit", (code, signal) => r({ code, signal })));
|
|
await until("the conversation's service", 30000, async () => {
|
|
if (proc.exitCode != null) throw new Error(`the service exited: ${fs.readFileSync(log, "utf8").slice(-400)}`);
|
|
try { return (await fetch(`${url}/msg`)).ok; } catch { return false; }
|
|
}, 200);
|
|
};
|
|
const stop = async () => {
|
|
if (proc.exitCode == null) proc.kill("SIGTERM");
|
|
const e = await Promise.race([exited, sleep(10000).then(() => null)]);
|
|
if (!e) { proc.kill("SIGKILL"); await exited; }
|
|
};
|
|
await start();
|
|
return { url, stop, restart: async () => { await stop(); await start(); } };
|
|
}
|
|
|
|
const st = (p) => p.evaluate(() => window.msgTest.state);
|
|
const view = (p) => p.evaluate(() => window.msgTest.view());
|
|
const texts = (v) => v.messages.filter((m) => m.outcome === "accepted" && (m.kind === "text" || m.kind === "reply")).map((m) => m.text);
|
|
|
|
async function scenarioMessaging(run) {
|
|
const dir = path.join(OUT, `run${run}-messaging`);
|
|
fs.mkdirSync(dir, { recursive: true });
|
|
const k = roomKeys(ROOM, dir);
|
|
const store = path.join(dir, "store");
|
|
const svc = await startConversation(dir, k.keys, store);
|
|
const srv = await serveSameOrigin(WEB, { service: svc.url, relayPort: 1 });
|
|
const browser = await launch(dir);
|
|
const pages = [];
|
|
try {
|
|
const url = `${srv.url}/room/messages.html?service=/room-api`;
|
|
for (const name of ["alice", "bob", "carol"]) {
|
|
const p = await openPage(browser, url, name, dir);
|
|
pages.push(p);
|
|
await until(`${name}'s page ready`, STEP, () => p.evaluate(() => !!(window.msgTest && window.msgTest.join)), 200);
|
|
const phase = await p.evaluate(() => window.msgTest.join());
|
|
if (phase !== "member") throw new Error(`${name} did not join: ${phase}`);
|
|
}
|
|
const [A, B, C] = pages;
|
|
const who = {};
|
|
for (const p of pages) who[p.name] = (await st(p)).me;
|
|
await until("every page keyed into one epoch of three members", STEP, async () => {
|
|
const s = await Promise.all(pages.map(st));
|
|
return s.every((x) => x.members.length === 3 && x.epoch && x.epoch.epoch === s[0].epoch.epoch) && s;
|
|
}, 300);
|
|
check("messaging: three pages join, each with its own key, and key through the service's log into one group", true, `epoch ${(await st(A)).epoch.epoch}`);
|
|
|
|
// Three members message.
|
|
await A.evaluate(() => window.msgTest.send("hello from alice"));
|
|
await B.evaluate(() => window.msgTest.send("hello from bob"));
|
|
await C.evaluate(() => window.msgTest.send("hello from carol"));
|
|
for (const p of pages) {
|
|
await until(`${p.name} opens the three`, STEP, async () => texts(await view(p)).length === 3, 300);
|
|
}
|
|
const all3 = await Promise.all(pages.map(async (p) => texts(await view(p)).sort().join("|")));
|
|
check("messaging: each page opens what the others seal", all3.every((x) => x === "hello from alice|hello from bob|hello from carol"), all3[0]);
|
|
|
|
// Carol goes offline; the others start a thread, reply and react.
|
|
await C.ctx.setOffline(true);
|
|
const root = await A.evaluate(() => window.msgTest.send("lunch at noon?"));
|
|
await until("bob sees the root", STEP, async () => (await view(B)).messages.some((m) => m.link === root), 300);
|
|
const reply = await B.evaluate((r) => window.msgTest.reply(r, "yes"), root);
|
|
await until("alice sees the reply", STEP, async () => (await view(A)).messages.some((m) => m.link === reply), 300);
|
|
await A.evaluate((r) => window.msgTest.react(r, "+1"), reply);
|
|
|
|
// The service stops and starts again from its store.
|
|
await svc.restart();
|
|
check("messaging: the service started again from its history store", true);
|
|
await A.evaluate((r) => window.msgTest.edit(r, "lunch at one?"), root);
|
|
const oops = await B.evaluate(() => window.msgTest.send("wrong conversation"));
|
|
await B.evaluate((r) => window.msgTest.retract(r), oops);
|
|
|
|
// Carol comes back and catches up from the store.
|
|
await C.ctx.setOffline(false);
|
|
const vc = await until("carol catches up", STEP * 2, async () => {
|
|
const v = await view(C);
|
|
const r = v.messages.find((m) => m.link === root);
|
|
const x = v.messages.find((m) => m.link === oops);
|
|
return r && r.status === "edited" && x && x.status === "retracted" && r.replies.length === 1 && v;
|
|
}, 500);
|
|
const r = vc.messages.find((m) => m.link === root);
|
|
const rp = vc.messages.find((m) => m.link === reply);
|
|
const x = vc.messages.find((m) => m.link === oops);
|
|
check("messaging: the page offline catches up from the store after the service's restart: the thread, the reaction, the edit", r.text === "lunch at one?" && r.edits === 1 && r.replies[0] === reply && rp.reactions.length === 1 && rp.reactions[0][1] === "+1", JSON.stringify({ text: r.text, edits: r.edits, replies: r.replies.length, reactions: rp.reactions }));
|
|
check("messaging: a retracted message stays in verification, stated retracted, its body shown nowhere", x.outcome === "accepted" && x.status === "retracted" && x.text === null && !(await C.evaluate(() => document.getElementById("messages").textContent)).includes("wrong conversation"), JSON.stringify(x));
|
|
const sc = await st(C);
|
|
check("messaging: every message the page received opened and verified", sc.unopened === 0 && vc.messages.every((m) => m.outcome === "accepted") && sc.opened === vc.messages.length, `${sc.opened} opened of ${vc.messages.length}`);
|
|
// Acknowledgements: each page acknowledges, as read, what it showed.
|
|
const ack = await until("alice holds carol's read acknowledgement", STEP, async () => {
|
|
const v = await view(A);
|
|
const a = v.acknowledged[who.alice];
|
|
return a && a[who.carol] && a[who.carol].read != null && a;
|
|
}, 500);
|
|
check("messaging: the pages acknowledge what they read, and each acknowledgement verifies", ack[who.carol].read >= 1, JSON.stringify(ack));
|
|
|
|
// The owner removes Bob: his page is told, and opens nothing written after.
|
|
const rm = await A.evaluate((p) => window.msgTest.remove(p), who.bob);
|
|
check("messaging: the owner removes a member", rm.status === 200, String(rm.status));
|
|
await until("bob's page is told it is no longer a member", STEP, async () => (await st(B)).phase === "removed", 300);
|
|
const ep = (await st(A)).epoch.epoch;
|
|
await until("alice and carol in the epoch without bob", STEP, async () => { const [a, c] = await Promise.all([st(A), st(C)]); return a.members.length === 2 && c.members.length === 2 && !a.members.includes(who.bob); }, 300);
|
|
await A.evaluate(() => window.msgTest.send("bob cannot read this"));
|
|
await until("carol reads it", STEP, async () => texts(await view(C)).includes("bob cannot read this"), 300);
|
|
await sleep(1500);
|
|
const vb = await view(B);
|
|
const mb = await B.evaluate(() => window.msgTest.memberState());
|
|
check("messaging: the removed member opens nothing written after its removal, and the store serves it nothing", !texts(vb).includes("bob cannot read this") && mb.removed != null, `removed at epoch ${mb.removed}, the conversation at ${ep + 0}`);
|
|
|
|
// The store holds no plaintext.
|
|
const journal = fs.readFileSync(path.join(store, "journal"));
|
|
const leaks = ["hello from", "lunch at", "wrong conversation", "bob cannot read"].filter((t) => journal.includes(Buffer.from(t)));
|
|
check("messaging: the history store holds no plaintext", leaks.length === 0, leaks.join(", ") || `${journal.length} bytes of ciphertext, receipts and signed records`);
|
|
const errs = (await Promise.all(pages.map(st))).filter((s) => s.error).map((s) => s.error);
|
|
check("messaging: no page met an error", errs.length === 0, errs.join("; "));
|
|
} finally {
|
|
for (const p of pages) p.dump();
|
|
await browser.close().catch(() => {});
|
|
await srv.close();
|
|
await svc.stop();
|
|
killStrays();
|
|
}
|
|
}
|
|
|
|
async function scenarioCall(run) {
|
|
const dir = path.join(OUT, `run${run}-call`);
|
|
fs.mkdirSync(dir, { recursive: true });
|
|
const k = roomKeys(ROOM, dir);
|
|
const relay = await startRelay(RELAY, { no_quic: true, auth: true, extra: k.lines, summary_out: path.join(dir, "summaries.jsonl") }, dir, "relay");
|
|
const svc = await startRoomService(ROOM, { keys: k.keys, relayWs: relay.wsUrl, room: { max_publishers: 4 }, top: { public_prefix: '"/room-api"' }, extra: "[call]\nring_ms = 8000" }, dir);
|
|
const srv = await serveSameOrigin(WEB, { service: svc.url, relayPort: relay.ws });
|
|
const browser = await launch(dir);
|
|
const pages = [];
|
|
try {
|
|
const url = `${srv.url}/room/call.html?service=/room-api`;
|
|
for (const name of ["alice", "bob", "carol"]) {
|
|
const p = await openPage(browser, url, name, dir);
|
|
pages.push(p);
|
|
await until(`${name}'s call page ready`, STEP, () => p.evaluate(() => window.callTest && window.callTest.state.phase === "ready"), 200);
|
|
}
|
|
const [A, B, C] = pages;
|
|
const key = (p) => p.evaluate(() => window.callTest.state.me);
|
|
const bob = await key(B);
|
|
await A.evaluate((k) => { window.callTest.place([k]).catch((e) => { window.callTest.state.error = String(e); }); }, bob);
|
|
const ringing = await until("bob's inbox rings", STEP, async () => { const s = await B.evaluate(() => window.callTest.state); return s.inbox.find((x) => x.state === "ringing") && s; }, 300);
|
|
check("call: the callee's inbox holds the invitation, checked with its call before it rings", ringing.inbox[0].check.ok && ringing.inbox[0].check.caller === (await key(A)) && ringing.refused.length === 0, JSON.stringify(ringing.inbox[0].check).slice(0, 160));
|
|
await B.evaluate(() => { window.callTest.accept().catch((e) => { window.callTest.state.error = String(e); }); });
|
|
const roomFrame = (p) => p.frames().find((f) => f.url().includes("/room/index.html"));
|
|
const snap = (p) => roomFrame(p).evaluate(() => window.roomTest.snapshot());
|
|
const principalOf = (p) => roomFrame(p).evaluate(() => window.roomTest.snapshot().me.principal);
|
|
for (const p of [A, B]) {
|
|
await until(`${p.name}'s room frame publishes`, STEP * 2, async () => { const f = roomFrame(p); return f && (await f.evaluate(() => window.roomTest && window.roomTest.state.phase)) === "publishing"; }, 500);
|
|
}
|
|
const [pa, pb] = [await principalOf(A), await principalOf(B)];
|
|
for (const [p, other] of [[A, pb], [B, pa]]) {
|
|
const s = await until(`${p.name} decodes the other's video and audio`, STEP * 2, async () => {
|
|
const x = await snap(p);
|
|
const t = x.subs[other];
|
|
return t && t.video.decoded >= 20 && t.audio.decoded >= 30 && x;
|
|
}, 500);
|
|
check(`call: ${p.name}'s room carries the other's audio and video, decoded`, true, `${s.subs[other].video.decoded} video frames, ${s.subs[other].audio.decoded} audio frames`);
|
|
}
|
|
check("call: the caller and the callee publish under the grants their call came with", pa === (await key(A)) && pb === bob, `${pa.slice(0, 12)} / ${pb.slice(0, 12)}`);
|
|
// The call ends as a room does; the next is missed.
|
|
for (const p of [A, B]) await roomFrame(p).evaluate(() => window.roomTest.leave());
|
|
const carol = await key(C);
|
|
await until("a new call placed", STEP, async () => {
|
|
const r = await A.evaluate(async (k) => { try { await window.callTest.place([k]); return true; } catch (e) { return String(e); } }, carol);
|
|
return r === true;
|
|
}, 1000);
|
|
await until("carol's inbox rings", STEP, async () => (await C.evaluate(() => window.callTest.state.inbox)).some((x) => x.state === "ringing"), 300);
|
|
const missed = await until("carol's call is missed at its deadline", STEP, async () => { const s = await C.evaluate(() => window.callTest.state); return s.inbox.find((x) => x.state === "missed") && s; }, 500);
|
|
const shown = await C.evaluate(() => document.getElementById("inbox").textContent);
|
|
check("call: an invitation unanswered is missed, signed by the room's clock, and shown as missed", missed.inbox.some((x) => x.state === "missed") && shown.includes("Missed call"), shown);
|
|
const errs = [];
|
|
for (const p of pages) { const e = await p.evaluate(() => window.callTest.state.error); if (e) errs.push(`${p.name}: ${e}`); }
|
|
check("call: no page met an error", errs.length === 0, errs.join("; "));
|
|
} finally {
|
|
for (const p of pages) p.dump();
|
|
await browser.close().catch(() => {});
|
|
await srv.close();
|
|
await svc.stop();
|
|
await relay.stop();
|
|
killStrays();
|
|
}
|
|
}
|
|
|
|
let passedRuns = 0;
|
|
for (let run = 1; run <= RUNS; run++) {
|
|
const before = failed;
|
|
console.log(`== run ${run} of ${RUNS}`);
|
|
for (const [name, f] of [["messaging", scenarioMessaging], ["call", scenarioCall]]) {
|
|
if (!ONLY.includes(name)) continue;
|
|
try {
|
|
await f(run);
|
|
} catch (e) {
|
|
check(`${name}: the scenario ran to its end`, false, (e && e.stack || String(e)).split("\n").slice(0, 3).join(" | "));
|
|
}
|
|
}
|
|
if (failed === before) passedRuns++;
|
|
}
|
|
fs.writeFileSync(path.join(OUT, "messaging-results.json"), JSON.stringify({ checks: results }, null, 1));
|
|
console.log(`messaging e2e: ${passedRuns} of ${RUNS} runs passed; ${results.filter((r) => r.ok).length} of ${results.length} checks ok; artifacts in ${OUT}`);
|
|
killStrays();
|
|
process.exit(failed ? 1 : 0);
|