// node --test rust/host/cli.test.mjs (needs `cargo wasm` and `cargo host`; no network) // The atlibs CLI end to end against the local fixture registry (fixture-registry.mjs) publishing // compose and vibescrobble as alice.test. add -> lockfile + vendored copies; stage // --target spa -> client metadata at the configured paths; run and serve the added lib. import assert from "node:assert/strict"; import { execFileSync, spawn } from "node:child_process"; import { createHash } from "node:crypto"; import { cp, mkdir, mkdtemp, readFile, symlink, writeFile } from "node:fs/promises"; import { createServer as createHttpsServer } from "node:https"; import { tmpdir } from "node:os"; import { join } from "node:path"; import { after, before, test } from "node:test"; import { modulePin } from "../../javascript/src/run-module/index.js"; import { contextOptions, createNodeContext } from "../../javascript/src/server/index.js"; import { atlibs, hostBin } from "./conformance.mjs"; import { DID, fixtureRegistry } from "./fixture-registry.mjs"; const ROOT = new URL("../../", import.meta.url); let registry, records, blobs, put, publish, auth, seen, base, env, dir; before(async () => { registry = await fixtureRegistry(); ({ records, blobs, put, publish, auth, seen, base } = registry); await publish("compose"); await publish("vibescrobble"); dir = await mkdtemp(join(tmpdir(), "atlibs-cli-")); env = { ...registry.env, ATLIBS_SERVER_STATE: join(dir, "state") }; }); after(() => registry?.close()); test("add / -> stage --target spa -> client metadata at the configured paths; run the added lib", async () => { await atlibs(["add", "alice.test/compose", "alice.test/vibescrobble"], { cwd: dir, env }); const lock = JSON.parse(await readFile(join(dir, "atlibs.lock.json"), "utf8")); assert.equal(lock.modules.vibescrobble.uri, `at://${DID}/com.fedfork.exec.wasm.module/vibescrobble`); assert.equal(lock.modules.vibescrobble.ref, "alice.test/vibescrobble"); assert.equal(lock.modules.vibescrobble.session, "vibescrobble"); assert.equal(lock.sessions.vibescrobble.record.name, JSON.parse(await readFile(new URL("records/vibescrobble.json", ROOT), "utf8")).name); // the lockfile's pin covers the wasm and the record's grants: the JS modulePin of the vendored pair const vendored = { ...JSON.parse(await readFile(join(dir, "modules/compose.json"), "utf8")), module: await readFile(join(dir, "modules/compose.wasm")) }; assert.equal(await modulePin(vendored), lock.modules.compose.pin); const out = join(dir, "site"); const { stdout } = await atlibs(["stage", "--target", "spa", "--out", out, "--origin", "https://site.example/app/", "--oauth-path", "auth/clients"], { cwd: dir, env }); const staged = JSON.parse(stdout); const doc = JSON.parse(await readFile(join(out, "auth/clients/vibescrobble/client-metadata.json"), "utf8")); assert.equal(doc.client_id, "https://site.example/app/auth/clients/vibescrobble/client-metadata.json"); assert.deepEqual(doc.redirect_uris, ["https://site.example/app/"]); assert.equal(staged.clientId, doc.client_id, "one session: it is the page's client"); assert.equal(JSON.parse(await readFile(join(out, "modules/vibescrobble.json"), "utf8")).session, "vibescrobble"); assert.deepEqual(await readFile(join(out, "modules/compose.wasm")), await readFile(join(dir, "modules/compose.wasm"))); // the lockfile name runs the vendored copy; so does the / ref, straight from the registry const input = JSON.stringify({ call: { input: { n: 4 } } }); for (const ref of ["compose", "alice.test/compose"]) { const ran = JSON.parse((await atlibs(["run", ref, "even", input], { cwd: dir, env })).stdout); assert.deepEqual(ran.result, { allow: true, violations: [] }, ref); } await assert.rejects(atlibs(["run", "nope", "even"], { cwd: dir, env }), /no module nope/); await assert.rejects(atlibs(["add", "bob.test/compose"], { cwd: dir, env }), /resolving bob.test/); await assert.rejects(atlibs(["add", "alice.test/missing"], { cwd: dir, env }), /publishes no module or composition named missing/); }); test("add refuses a blob that does not match its CID, and a pin that does not match", async () => { const work = await mkdtemp(join(tmpdir(), "atlibs-cli-bad-")); await assert.rejects(atlibs(["add", "alice.test/compose@00000000"], { cwd: work, env }), /pin 00000000 is too short/); await assert.rejects(atlibs(["add", `alice.test/compose@${"0".repeat(32)}`], { cwd: work, env }), /pin 0{32} not published/); const { blob } = await publish("compose"); const good = blobs.get(blob); blobs.set(blob, Buffer.from("tampered")); try { await assert.rejects(atlibs(["add", "alice.test/compose"], { cwd: work, env }), /blob does not match its CID/); } finally { blobs.set(blob, good); } }); test("serve: the index, a run over HTTP, and client metadata on the request's origin", async () => { const child = spawn(hostBin(), ["serve", "compose", "--port", "0"], { cwd: dir, env: { PATH: process.env.PATH, HOME: process.env.HOME, ...env } }); try { const url = await new Promise((resolve, reject) => { let err = ""; child.stderr.on("data", (chunk) => { err += chunk; const found = err.match(/atlibs serve: (http:\/\/\S+\/)/); if (found) resolve(found[1]); }); child.on("exit", (code) => reject(new Error(`serve exited ${code}: ${err}`))); }); const index = await (await fetch(url)).json(); assert.ok(index.run.includes("/run/even"), JSON.stringify(index)); const ran = await (await fetch(new URL("run/even", url), { method: "POST", headers: { "content-type": "application/json" }, body: JSON.stringify({ call: { input: { n: 3 } } }) })).json(); assert.equal(ran.result.allow, false); assert.equal((await fetch(new URL("run/even", url))).status, 405); // an entry the module does not export: 404, with the ones it does const nope = await fetch(new URL("run/nope", url), { method: "POST", headers: { "content-type": "application/json" }, body: "{}" }); assert.deepEqual([nope.status, (await nope.json()).message], [404, "no entry nope (caller, even, lookup, spy)"]); assert.match(await (await fetch(url, { headers: { accept: "text/html" } })).text(), / { let err = ""; vibescrobble.stderr.on("data", (chunk) => (err += chunk).match(/atlibs serve: (http:\/\/\S+\/)/) && resolve(err.match(/atlibs serve: (http:\/\/\S+\/)/)[1])); vibescrobble.on("exit", (code) => reject(new Error(`serve exited ${code}: ${err}`))); }); const doc = await (await fetch(new URL("oauth/vibescrobble/client-metadata.json", url))).json(); assert.equal(doc.client_id, new URL("oauth/vibescrobble/client-metadata.json", url).href); } finally { vibescrobble.kill(); } }); // One module, many runs at once: concurrent POST /run on one atlibs serve (one thread, runs interleave) // and concurrent atlibs run processes sharing a state dir. Each answers its own input. test("concurrent runs of one module: POST /run on one atlibs serve, and parallel atlibs run processes", async () => { const child = spawn(hostBin(), ["serve", "compose", "--port", "0"], { cwd: dir, env: { PATH: process.env.PATH, HOME: process.env.HOME, ...env } }); const even = (n) => JSON.stringify({ call: { input: { n } } }); const expected = Array.from({ length: 8 }, (_, n) => n % 2 === 0); try { const url = await new Promise((resolve, reject) => { let err = ""; child.stderr.on("data", (chunk) => (err += chunk).match(/atlibs serve: (http:\/\/\S+\/)/) && resolve(err.match(/atlibs serve: (http:\/\/\S+\/)/)[1])); child.on("exit", (code) => reject(new Error(`serve exited ${code}: ${err}`))); }); const ran = await Promise.all(expected.map((_, n) => fetch(new URL("run/even", url), { method: "POST", headers: { "content-type": "application/json" }, body: even(n) }).then((r) => r.json()))); assert.deepEqual(ran.map((r) => r.result?.allow), expected, JSON.stringify(ran)); } finally { child.kill(); } const runs = await Promise.all(expected.map((_, n) => atlibs(["run", "compose", "even", even(n)], { cwd: dir, env }))); assert.deepEqual(runs.map((r) => JSON.parse(r.stdout).result.allow), expected); }); // ATLIBS_WASM_CONCURRENCY=1: the runs queue for the one slot (queued time is not wall time) and all answer; // a value that is not a positive integer stops atlibs at startup. test("ATLIBS_WASM_CONCURRENCY: 4 overlapping POST /run with one slot all answer; a bad value is a startup error", async () => { const child = spawn(hostBin(), ["serve", "compose", "--port", "0"], { cwd: dir, env: { PATH: process.env.PATH, HOME: process.env.HOME, ...env, ATLIBS_WASM_CONCURRENCY: "1" } }); const expected = [true, false, true, false]; try { const url = await new Promise((resolve, reject) => { let err = ""; child.stderr.on("data", (chunk) => (err += chunk).match(/atlibs serve: (http:\/\/\S+\/)/) && resolve(err.match(/atlibs serve: (http:\/\/\S+\/)/)[1])); child.on("exit", (code) => reject(new Error(`serve exited ${code}: ${err}`))); }); const ran = await Promise.all(expected.map((_, n) => fetch(new URL("run/even", url), { method: "POST", headers: { "content-type": "application/json" }, body: JSON.stringify({ call: { input: { n } } }) }).then((r) => r.json()))); assert.deepEqual(ran.map((r) => r.result?.allow), expected, JSON.stringify(ran)); } finally { child.kill(); } for (const bad of ["0", "-1", "two"]) { const stderr = await atlibs(["serve", "compose", "--port", "0"], { cwd: dir, env: { ...env, ATLIBS_WASM_CONCURRENCY: bad } }).then(() => assert.fail(`${bad} started`), (e) => e.stderr); assert.equal(stderr, `ATLIBS_WASM_CONCURRENCY must be a positive integer, not "${bad}"\n`); } }); test("run --handler: a composition's declared handler from a wasm module (with a callback), an http endpoint, or an opt-in command", async () => { const work = await mkdtemp(join(tmpdir(), "atlibs-cli-handlers-")); const comp = join(work, "confirm.json"); await writeFile(comp, JSON.stringify({ $type: "com.fedfork.exec.composition", name: "confirm", instances: [{ name: "ui", module: "compose" }], handlers: [{ name: "confirm", description: "someone approves" }], edges: [{ from: "ui", nsid: "com.example.compose.confirm", handler: "confirm", reason: "ui asks before acting" }], })); const NSID = "com.example.compose.confirm"; const run = (...extra) => atlibs(["run", comp, "ui", "caller", JSON.stringify({ nsid: NSID, input: { n: 1 } }), ...extra], { env }); await assert.rejects(run(), /handler confirm was not supplied/); // wasm: the compose module's lookup entry answers, calling the progress callback ui passed it const wasm = JSON.parse((await run("--handler", "confirm=wasm:compose#lookup")).stdout); assert.equal(wasm.error, undefined, wasm.error); assert.deepEqual(wasm.result, { q: { ok: true, output: { doubled: 2, ack: { seen: { step: 1 } }, caller: { instance: "ui", nsid: NSID, reason: "ui asks before acting" } } }, progress: [{ step: 1 }] }); assert.deepEqual(wasm.graph.map((g) => [g.from, g.nsid, g.handler ?? g.cap ?? null, g.ok]).at(-1), ["ui", NSID, "confirm", true]); // http: the call as JSON, no credentials const http = JSON.parse((await run("--handler", `confirm=${base}/handler`)).stdout).result.q.output; assert.equal(http.approved, true); assert.deepEqual([http.got.handler, http.got.caller, http.got.nsid, http.got.reason, http.got.input.n, http.auth], ["confirm", "ui", NSID, "ui asks before acting", 1, null]); assert.match(http.got.input.progress.$cap, /^cap\.[0-9a-f]{32}\.\d+$/, "a cap stays a token"); // exec: only with --allow-exec; the command gets the call on stdin and prints the answer await assert.rejects(run("--handler", "confirm=exec:cat"), /--allow-exec/); const exec = JSON.parse((await run("--handler", "confirm=exec:cat", "--allow-exec")).stdout).result.q.output; assert.deepEqual([exec.handler, exec.caller, exec.input.n], ["confirm", "ui", 1]); const failed = JSON.parse((await run("--handler", "confirm=exec:echo no >&2; exit 2", "--allow-exec")).stdout).result.q; assert.deepEqual(failed, { ok: false, error: "handler confirm: echo no >&2; exit 2 exited 2: no" }); // names are checked against the manifest; a plain module takes no handlers await assert.rejects(run("--handler", "nope=exec:cat", "--allow-exec"), /declares no handler nope \(it declares: confirm\)/); await assert.rejects(atlibs(["run", "compose", "even", "{}", "--handler", "confirm=exec:cat", "--allow-exec"], { env }), /one module/); await assert.rejects(run("--handler", "confirm=http://example.com/x"), /only to localhost/); }); // One saved session, an expired access token, and several Rust and JS processes sharing the state dir // at once: exactly one refreshSession (under the state dir's "session" lease), no createSession. test("concurrent atlibs and cli.mjs processes sharing a state dir refresh the saved session once", async () => { const state = join(await mkdtemp(join(tmpdir(), "atlibs-cli-session-")), "state"); const seed = createNodeContext(contextOptions(state)); await seed.settings.set("session", { identifier: "alice.test", did: DID, handle: "alice.test", accessJwt: "a1", refreshJwt: "r1", didDoc: { service: [{ id: "#atproto_pds", serviceEndpoint: base }] } }); seed.close(); Object.assign(auth, { refreshes: 0, creates: 0 }); // the account is explicit (flags; the password from the variable --password-env names), never ATPROTO_* env const signedIn = { PATH: process.env.PATH, HOME: process.env.HOME, ATLIBS_SERVER_STATE: state, TEST_PW: "pw", TEST_PDS: base }; const account = ["--app-password", "alice.test", "--password-env", "TEST_PW", "--pds", base]; const node = `import { contextOptions, createNodeContext, login, sessionStore } from ${JSON.stringify(new URL("../../javascript/src/server/index.js", import.meta.url).href)}; const ctx = createNodeContext(contextOptions(process.env.ATLIBS_SERVER_STATE)); const session = await login({ identifier: "alice.test", password: process.env.TEST_PW, service: process.env.TEST_PDS, ...sessionStore(ctx) }); await session.xrpc("com.atproto.repo.deleteRecord", { input: { repo: session.did, collection: "app.test", rkey: "x" } });`; const exit = (child) => { let err = ""; child.stderr.on("data", (chunk) => (err += chunk)); return new Promise((resolve) => child.on("exit", (code) => resolve(code === 0 ? 0 : err))); }; const runs = await Promise.all([ ...[1, 2, 3].map(() => exit(spawn(hostBin(), ["publish", "delete", "app.test/x", ...account], { env: signedIn }))), ...[1, 2, 3].map(() => exit(spawn(process.execPath, ["--input-type=module", "-e", node], { env: signedIn }))), ]); assert.deepEqual(runs, Array(6).fill(0)); assert.deepEqual(auth, { refreshes: 1, creates: 0 }); }); // 3a L1 / 3c M3: the input as `-` (stdin) or `@file` keeps a secret out of argv; a parse error does // not echo the input; a run that errors exits 1 after printing its result. test("atlibs run reads the input from - or @file, never echoes it, and exits 1 on a run error", async () => { const dir = await mkdtemp(join(tmpdir(), "atlibs-cli-input-")); const env = { ATLIBS_SERVER_STATE: join(dir, "state") }; const run = (args, stdin) => new Promise((resolve) => { const child = spawn(hostBin(), ["run", "examples/rust/compose", ...args], { cwd: ROOT, env: { PATH: process.env.PATH, HOME: process.env.HOME, ...env } }); let stdout = "", stderr = ""; child.stdout.on("data", (d) => (stdout += d)); child.stderr.on("data", (d) => (stderr += d)); child.on("close", (code) => resolve({ code, stdout, stderr })); child.stdin.end(stdin ?? ""); }); const input = JSON.stringify({ call: { input: { n: 4 } }, stdin: "tok" }); const piped = await run(["even", "-"], input); assert.equal(piped.code, 0, piped.stderr); assert.deepEqual(JSON.parse(piped.stdout).result, { allow: true, violations: [] }); await writeFile(join(dir, "in.json"), input); assert.deepEqual(JSON.parse((await run(["even", `@${join(dir, "in.json")}`])).stdout).result, { allow: true, violations: [] }); const bad = await run(["even", "-"], '{"stdin":"secret-token"'); assert.equal(bad.code, 1); assert.match(bad.stderr, /input is not JSON/); assert.doesNotMatch(bad.stderr, /secret-token/); const failed = await run(["lookup", "{}"]); assert.equal(failed.code, 1, "a run error is the exit status"); assert.equal(JSON.parse(failed.stdout).error, "n is required", "and the result is still printed"); // an entry it does not export is refused before any run (review 4c): nothing printed, its entries named const nope = await run(["no_such_entry", "{}"]); assert.deepEqual([nope.code, nope.stdout, nope.stderr], [1, "", "no entry no_such_entry (caller, even, lookup, spy)\n"]); }); // Each command's own usage on a bad invocation (not the whole help); `run ` with no entry lists the entries. test("a bad invocation prints that command's usage; run with no entry lists the module's entries", async () => { const env = { ATLIBS_SERVER_STATE: join(await mkdtemp(join(tmpdir(), "atlibs-cli-usage-")), "state") }; const fails = (args) => atlibs(args, { env }).then(() => assert.fail(`${args.join(" ")} succeeded`), (e) => e.stderr); const listed = await fails(["run", "examples/rust/compose"]); assert.match(listed, /^usage: atlibs run /); assert.match(listed, /compose entries: caller, lookup, even/); assert.match(await fails(["run", "examples/rust/compose", "nope"]), /^no entry nope \(caller, even, lookup, spy\)$/m); for (const [args, line] of [[["stage", "--orign", "x"], /unknown flag --orign\nusage: atlibs stage /], [["serve"], /^usage: atlibs serve /], [["add"], /^usage: atlibs add /], [["verify"], /^usage: atlibs verify /], [["publish", "--nope=1"], /unknown flag --nope\nusage: atlibs publish \[ \.\.\.\][^\n]*\n {7}atlibs publish module [^\n]*\n {7}atlibs publish records/]]) { const stderr = await fails(args); assert.match(stderr, line, args.join(" ")); assert.doesNotMatch(stderr, /usage: atlibs /, args.join(" ")); } const { stdout } = await atlibs(["help"], { env }); for (const piece of ["publish delete", "--check-live", "--password-file ", "--allow-dirty", "--out records/wasm", "@pin|@cid", "--dir .", "--record ", "--oauth-path oauth/] [--handler", "--allow-exec", "'' | - | @file", "[--target spa|server]", "a name under examples/rust", "`bridge`: internal"]) assert.ok(stdout.includes(piece), piece); assert.ok(!stdout.includes("webrtc-peer"), "webrtc-peer is the atlibs-webrtc repo's own binary"); }); // Regression (verify never matched on macOS: a docker look-alike's `run` returned before the build and exited 0, and // a host build is not the bytes a Linux rebuild makes, cargo hashing the host triple into every crate): a clone of // this repo publishes echo to the fixture registry (built in the container) and `atlibs verify` rebuilds the same // bytes. Needs a container engine (CONTAINER_ENGINE, or podman / docker / Apple's container answering `info`) and // the network (the rust image, rustup, crates); skipped without an engine. Offline (stand-in engine): registry.rs. test("atlibs publish module builds in the container and atlibs verify reproduces it", async (t) => { const registry = await fixtureRegistry(); // its own: the publish test counts the shared one's writes t.after(() => registry.close()); const tmp = await mkdtemp(join(tmpdir(), "atlibs-cli-verify-")); const ws = join(tmp, "ws"); execFileSync("git", ["clone", "-q", new URL(".", ROOT).pathname, ws]); await writeFile(join(tmp, "password"), "pw\n", { mode: 0o600 }); const signedIn = { ...registry.env, ...(process.env.CONTAINER_ENGINE && { CONTAINER_ENGINE: process.env.CONTAINER_ENGINE }), XDG_CACHE_HOME: join(tmp, "cache"), ATLIBS_SERVER_STATE: join(tmp, "state") }; const as = ["--app-password", "alice.test", "--password-file", join(tmp, "password"), "--pds", registry.base]; await atlibs(["publish", "records", "records/echo.json", ...as], { cwd: ws, env: signedIn }); try { await atlibs(["publish", "module", "echo", "--out", join(tmp, "out"), ...as], { cwd: ws, env: signedIn }); } catch (e) { if (/no container engine answers/.test(e.stderr)) return t.skip(e.stderr.trim()); throw e; } const published = JSON.parse(await readFile(join(tmp, "out", "echo.manifest.json"), "utf8")).output.sha256; const { stdout } = await atlibs(["verify", "alice.test/echo"], { env: signedIn }); assert.deepEqual(JSON.parse(stdout), { built: published, manifest: published, published, reproducible: true }); }); // Another repo publishes the same way: its own cargo workspace (modules/ and a path dependency under crates/), // nothing of this repo's layout. The sources and the build come from that workspace; the dry run builds in the // container and its pin is what publish then writes; verify reproduces it. Needs a container engine, as above. test("atlibs publish from another repo's workspace builds in the container and atlibs verify reproduces it", async (t) => { const registry = await fixtureRegistry(); t.after(() => registry.close()); const tmp = await mkdtemp(join(tmpdir(), "atlibs-cli-other-")); const ws = join(tmp, "other"); const git = (...args) => execFileSync("git", ["-c", "user.name=t", "-c", "user.email=t@t", "-C", ws, ...args]); await cp(new URL("rust/src/abi", ROOT), join(ws, "crates/abi"), { recursive: true }); await cp(new URL("examples/rust/hello", ROOT), join(ws, "modules/hello"), { recursive: true }); const root = await readFile(new URL("Cargo.toml", ROOT), "utf8"); await writeFile(join(ws, "Cargo.toml"), root.replace(/^members = .*$/m, 'members = ["modules/*"]').replace('path = "rust/src/abi"', 'path = "crates/abi"')); await cp(new URL("Cargo.lock", ROOT), join(ws, "Cargo.lock")); execFileSync("cargo", ["metadata", "--offline", "--format-version", "1", "--manifest-path", join(ws, "Cargo.toml")], { stdio: "ignore" }); // prunes the lock to this workspace await writeFile(join(ws, ".gitignore"), "target\n"); git("init", "-q"); git("remote", "add", "origin", "https://example.test/other"); git("add", "."); git("commit", "-qm", "one"); await writeFile(join(tmp, "password"), "pw\n", { mode: 0o600 }); const signedIn = { ...registry.env, ...(process.env.CONTAINER_ENGINE && { CONTAINER_ENGINE: process.env.CONTAINER_ENGINE }), XDG_CACHE_HOME: join(tmp, "cache"), ATLIBS_SERVER_STATE: join(tmp, "state") }; const publish = async (...args) => { const { stdout } = await atlibs(["publish", "--dir", "modules", "--app-password", "alice.test", "--password-file", join(tmp, "password"), "--pds", registry.base, ...args], { cwd: ws, env: signedIn }); return Object.fromEntries(stdout.trim().split("\n").slice(1).map((line) => line.split(/\s+/)).map(([status, name, pin]) => [name, { status, pin }])); }; let dry; try { dry = await publish("--dry-run"); } catch (e) { if (/no container engine answers/.test(e.stderr)) return t.skip(e.stderr.trim()); throw e; } assert.equal(dry.hello.status, "would-publish"); const done = await publish(); assert.deepEqual(done.hello, { status: "published", pin: dry.hello.pin }); const manifest = JSON.parse(await readFile(join(ws, "records/wasm/hello.manifest.json"), "utf8")); assert.deepEqual(manifest.files.map((f) => f.path).sort(), ["Cargo.lock", "Cargo.toml", "crates/abi/Cargo.toml", ...["aturi", "ids", "lib", "out", "records", "repo"].map((m) => `crates/abi/src/${m}.rs`), "modules/hello/Cargo.toml", "modules/hello/src/lib.rs"].sort()); assert.equal(manifest.origin.repository, "https://example.test/other"); const { stdout } = await atlibs(["verify", "alice.test/hello"], { env: signedIn }); assert.equal(JSON.parse(stdout).reproducible, true, stdout); }); // Regression (`atlibs help | head` exited 101 with println!'s broken-pipe panic): a reader that closed before // the output is written ends the process quietly, exit 0. test("a closed stdout (atlibs help | head) exits 0 without a panic", async () => { const child = spawn(hostBin(), ["help"], { stdio: ["ignore", "pipe", "pipe"] }); child.stdout.destroy(); let stderr = ""; child.stderr.on("data", (chunk) => (stderr += chunk)); const code = await new Promise((resolve) => child.on("close", resolve)); assert.doesNotMatch(stderr, /panicked/); assert.equal(code, 0, stderr); }); // `atlibs publish`: a workspace (a git repo with two modules and a session record) as alice.test. The // first run publishes all, the next none (the cache, no requests), a committed change only that module; // --dry-run publishes nothing; dirty sources are skipped; --check-live reports drift; the cache is by DID; // and what was published is what `atlibs add /@` fetches, signed out. test("atlibs publish: only what changed, by a cache per DID; dry run, dirty sources, drift", async () => { const tmp = await mkdtemp(join(tmpdir(), "atlibs-cli-publish-")); const ws = join(tmp, "ws"); const git = (...args) => execFileSync("git", ["-c", "user.name=t", "-c", "user.email=t@t", "-C", ws, ...args]); const release = "target/wasm32-unknown-unknown/release"; const vibescrobble = JSON.parse(await readFile(new URL("examples/rust/vibescrobble/module.json", ROOT), "utf8")); await mkdir(join(ws, release), { recursive: true }); await mkdir(join(ws, "records")); await cp(new URL(`${release}/compose.wasm`, ROOT), join(ws, release, "pa.wasm")); await cp(new URL(`${release}/vibescrobble.wasm`, ROOT), join(ws, release, "pb.wasm")); await cp(new URL("examples/rust/compose", ROOT), join(ws, "examples/rust/pa"), { recursive: true }); await cp(new URL("examples/rust/vibescrobble", ROOT), join(ws, "examples/rust/pb"), { recursive: true }); await writeFile(join(ws, "examples/rust/pb/module.json"), JSON.stringify({ ...vibescrobble, session: "pb" })); await cp(new URL("records/vibescrobble.json", ROOT), join(ws, "records/pb.json")); await symlink(new URL("lexicons", ROOT).pathname, join(ws, "lexicons")); await writeFile(join(ws, ".gitignore"), "target\nlexicons\n"); git("init", "-q"); git("remote", "add", "origin", "https://example.test/ws"); git("add", "."); git("commit", "-qm", "one"); const cache = join(tmp, "cache"); await writeFile(join(tmp, "password"), "pw\n", { mode: 0o600 }); const signedIn = { XDG_CACHE_HOME: cache, ATLIBS_SERVER_STATE: join(tmp, "state") }; const publish = async (...args) => { const { stdout } = await atlibs(["publish", "--no-build", "--app-password", "alice.test", "--password-file", join(tmp, "password"), "--pds", base, ...args], { cwd: ws, env: signedIn }); return Object.fromEntries(stdout.trim().split("\n").slice(1).map((line) => line.split(/\s+/)).map(([status, name, pin, cid, ref]) => [name, { status, pin, cid, ref }])); }; const statuses = (rows) => Object.fromEntries(Object.entries(rows).map(([name, row]) => [name, row.status])); const writes = () => seen.filter((p) => /(put|create)Record|uploadBlob/.test(p)).length; assert.deepEqual(statuses(await publish("--dry-run")), { "records/pb": "would-publish", pa: "would-publish", pb: "would-publish" }); assert.equal(writes(), 0, "a dry run publishes nothing"); const first = await publish(); assert.deepEqual(statuses(first), { "records/pb": "published", pa: "published", pb: "published" }); assert.equal(first.pa.ref, "alice.test/pa", "by the --app-password handle, as atlibs add takes it"); assert.equal(records.get("com.fedfork.exec.wasm.module/pb").value.session.cid, first["records/pb"].cid, "the module names its session record"); const cached = JSON.parse(await readFile(join(cache, "atlibs", DID, "pa.json"), "utf8")); assert.deepEqual([cached.pin, cached.cid, cached.commit], [first.pa.pin, first.pa.cid, git("rev-parse", "HEAD").toString().trim()], "the cache is keyed by DID"); const requests = seen.length; assert.deepEqual(statuses(await publish()), { "records/pb": "unchanged", pa: "unchanged", pb: "unchanged" }); assert.equal(seen.length, requests, "unchanged by the cache: no requests"); const pa = JSON.parse(await readFile(join(ws, "examples/rust/pa/module.json"), "utf8")); await writeFile(join(ws, "examples/rust/pa/module.json"), JSON.stringify({ ...pa, calls: [...(pa.calls ?? []), "com.atproto.repo.getRecord"] })); git("commit", "-qam", "two"); const second = await publish(); assert.deepEqual(statuses(second), { "records/pb": "unchanged", pa: "published", pb: "unchanged" }); assert.notEqual(second.pa.pin, first.pa.pin, "a changed grant is a new pin"); // signed out: add /@ fetches exactly what was published const out = await mkdtemp(join(tmpdir(), "atlibs-cli-publish-add-")); await atlibs(["add", `alice.test/pa@${second.pa.pin}`], { cwd: out, env }); assert.equal(JSON.parse(await readFile(join(out, "atlibs.lock.json"), "utf8")).modules.pa.pin, second.pa.pin); // ... and the first pin still the first version: a republish leaves it in place await atlibs(["add", `alice.test/pa@${first.pa.pin}`], { cwd: out, env }); assert.equal(JSON.parse(await readFile(join(out, "atlibs.lock.json"), "utf8")).modules.pa.pin, first.pa.pin); await assert.rejects(atlibs(["add", `alice.test/pa@${"0".repeat(32)}`], { cwd: out, env }), /pin 0{32} not published/); // dirty sources: skipped, nothing written, exit 0 await writeFile(join(ws, "examples/rust/pb/module.json"), JSON.stringify({ ...vibescrobble, session: "pb", description: "edited" })); const before = writes(); assert.deepEqual(statuses(await publish("pb")), { "records/pb": "unchanged", pb: "skipped:dirty" }); assert.equal(writes(), before); git("checkout", "--", "examples/rust/pb/module.json"); // drift: the live record is not the one this cache published const live = records.get("com.fedfork.exec.wasm.module/pa").value; const moved = put("com.fedfork.exec.wasm.module", "pa", { ...live, createdAt: "2020-01-01T00:00:00.000Z" }); const drift = await publish("--check-live"); assert.deepEqual(statuses(drift), { "records/pb": "unchanged", pa: "drift", pb: "unchanged" }); assert.equal(drift.pa.cid, moved.cid, "the cache now holds the live record"); assert.equal(writes(), before, "same content: reported, not republished"); }); // Regression (atproto-crates linked a second rustls provider; every wss subscribe then panicked until main's // atlibs_host::tls()): the built binary runs echo's `watch` over wss:// against a local TLS server (minimal // WebSocket upgrade, one text frame), trusting the fixture cert through the test-only ATLIBS_EXTRA_CA. test("atlibs run subscribes over wss:// without a rustls CryptoProvider panic", async () => { const fixtures = new URL("tests/fixtures/", import.meta.url); const tls = createHttpsServer({ cert: await readFile(new URL("localhost.pem", fixtures)), key: await readFile(new URL("localhost.key.pem", fixtures)) }); tls.on("upgrade", (req, socket) => { const accept = createHash("sha1").update(`${req.headers["sec-websocket-key"]}258EAFA5-E914-47DA-95CA-C5AB0DC85B11`).digest("base64"); socket.write(`HTTP/1.1 101 Switching Protocols\r\nUpgrade: websocket\r\nConnection: Upgrade\r\nSec-WebSocket-Accept: ${accept}\r\n\r\n`); const payload = Buffer.from(JSON.stringify({ $type: "com.atproto.sync.subscribeRepos#commit", seq: 1 })); socket.write(Buffer.concat([Buffer.from([0x81, payload.length]), payload])); // FIN text frame, < 126 bytes socket.on("data", () => {}).on("error", () => {}); // the client's close (the module cancels at its limit) }); await new Promise((resolve) => tls.listen(0, "127.0.0.1", resolve)); try { const service = `https://127.0.0.1:${tls.address().port}`; const { stdout, stderr } = await atlibs(["run", "examples/rust/echo", "watch", JSON.stringify({ did: DID, service, limit: 1, seconds: 10 })], { env: { ATLIBS_EXTRA_CA: new URL("localhost.pem", fixtures).pathname } }); assert.doesNotMatch(stderr, /panicked|CryptoProvider/); const { result } = JSON.parse(stdout); assert.deepEqual([result.count, result.last, result.stopped], [1, { $type: "com.atproto.sync.subscribeRepos#commit", seq: 1 }, "limit"], stderr); } finally { tls.close(); tls.closeAllConnections?.(); } });