fake-pds#

A lightweight, in-memory AT Protocol PDS for testing. No database, no crypto verification, no Merkle Search Tree — just the XRPC endpoints that matter for integration tests.

Install#

npm install fake-pds

Quick Start#

import { createFakePds } from "fake-pds";

const pds = await createFakePds({
  accounts: [
    { did: "did:plc:alice", handle: "alice.test" },
    { did: "did:plc:bob", handle: "bob.test" },
  ],
});

// Seed data
pds.putRecord("did:plc:alice", "app.opake.publicKey", "self", {
  opakeVersion: 1,
  publicKey: { $bytes: "..." },
  algo: "x25519",
  createdAt: "2026-01-01T00:00:00Z",
});

// Use in tests — it's a real HTTP server
const res = await fetch(
  `${pds.url}/xrpc/com.atproto.repo.getRecord?repo=did:plc:alice&collection=app.opake.publicKey&rkey=self`,
);
const record = await res.json();

// Reset between tests
pds.reset();

// Cleanup
await pds.close();

Endpoints#

Repository (XRPC)#

Endpoint	Method	Auth	Notes
`com.atproto.repo.getRecord`	GET	No	Returns `{ uri, cid, value }` or 404
`com.atproto.repo.listRecords`	GET	No	Cursor pagination, sorted by rkey
`com.atproto.repo.createRecord`	POST	Yes	Auto-generates TID rkey if omitted
`com.atproto.repo.putRecord`	POST	Yes	Upsert by rkey
`com.atproto.repo.deleteRecord`	POST	Yes	No-ops if missing
`com.atproto.repo.uploadBlob`	POST	Yes	Returns blob ref
`com.atproto.sync.getBlob`	GET	No	Returns stored blob bytes

Identity#

Endpoint	Method	Auth	Notes
`com.atproto.identity.resolveHandle`	GET	No	Returns `{ did }` or 404

Auth (Legacy)#

Endpoint	Method	Notes
`com.atproto.server.createSession`	POST	Returns fake JWT tokens
`com.atproto.server.refreshSession`	POST	Invalidates old tokens

OAuth#

Endpoint	Method	Notes
`/.well-known/oauth-protected-resource`	GET	Static metadata
`/.well-known/oauth-authorization-server`	GET	Server metadata
`/oauth/par`	POST	PAR, returns `request_uri` + DPoP nonce
`/oauth/token`	POST	Code exchange or refresh, returns DPoP tokens

Auth Modes#

// No auth checking — fastest for unit tests
const pds = await createFakePds({ auth: "none" });

// Bearer token auth (legacy JWT flow)
const pds = await createFakePds({ auth: "bearer" });

// OAuth with DPoP nonce tracking
const pds = await createFakePds({ auth: "oauth" });

Legacy Auth#

const pds = await createFakePds({
  accounts: [{ did: "did:plc:alice", handle: "alice.test" }],
  auth: "bearer",
});

// Login
const loginRes = await fetch(`${pds.url}/xrpc/com.atproto.server.createSession`, {
  method: "POST",
  headers: { "Content-Type": "application/json" },
  body: JSON.stringify({ identifier: "alice.test", password: "anything" }),
});
const { accessJwt } = await loginRes.json();

// Authenticated request
await fetch(`${pds.url}/xrpc/com.atproto.repo.putRecord`, {
  method: "POST",
  headers: {
    "Content-Type": "application/json",
    Authorization: `Bearer ${accessJwt}`,
  },
  body: JSON.stringify({ repo: "did:plc:alice", collection: "app.test", rkey: "x", record: {} }),
});

OAuth (Direct Token Issuance)#

Skip the browser redirect flow entirely:

const pds = await createFakePds({
  accounts: [{ did: "did:plc:alice", handle: "alice.test" }],
  auth: "oauth",
});

// Get tokens directly — no browser flow needed
const { accessToken, refreshToken, dpopNonce } = pds.issueOAuthTokens("did:plc:alice");

// Use with DPoP header
await fetch(`${pds.url}/xrpc/com.atproto.repo.putRecord`, {
  method: "POST",
  headers: {
    "Content-Type": "application/json",
    Authorization: `DPoP ${accessToken}`,
  },
  body: JSON.stringify({ repo: "did:plc:alice", collection: "app.test", rkey: "x", record: {} }),
});

API#

`createFakePds(options?): Promise<FakePds>`#

Create and start a fake PDS.

Option	Type	Default	Description
`port`	`number`	`0` (random)	Port to listen on
`accounts`	`Account[]`	`[]`	Pre-registered accounts
`auth`	`"none" \| "bearer" \| "oauth"`	`"none"`	Auth enforcement mode

`FakePds`#

Method	Description
`url`	Base URL of the running server
`putRecord(did, collection, rkey, value)`	Seed a record (bypasses auth)
`listRecords(did, collection)`	List records for test inspection
`issueOAuthTokens(did)`	Get OAuth tokens directly
`reset()`	Clear all records, blobs, and sessions
`close()`	Stop the server

`Account`#

interface Account {
  did: string;
  handle: string;
  password?: string; // If set, createSession validates it
}

What This Doesn't Do#

No Merkle Search Tree or commit history
No record validation against lexicon schemas
No federation, sync, or firehose
No DID document hosting
No PKCE or DPoP signature verification
CIDs are fake (sha256-based, not real IPFS CIDs)

These are intentional tradeoffs. The goal is fast, reliable integration tests — not a spec-complete PDS.

License#

AGPL-3.0-or-later

Clone this repository