Spin up ephemeral PDS instances in your Python test suite. Lexicon-agnostic — works with any application built on AT Protocol, not just Bluesky.
pip install testcontainers-atprotoRequires Python 3.10+ and a running Docker daemon.
| Extra | What it adds |
|---|---|
testcontainers-atproto[firehose] |
websockets, cbor2 for firehose subscription |
testcontainers-atproto[sync] |
cbor2 for CAR file parsing |
testcontainers-atproto[sdk] |
atproto (MarshalX SDK) for high-level record ops |
testcontainers-atproto[all] |
All of the above |
from testcontainers_atproto import PDSContainer
with PDSContainer() as pds:
account = pds.create_account("alice.test")
print(pds.base_url) # http://localhost:<port>
print(account.did) # did:plc:...
print(account.handle) # alice.testA local PLC directory runs alongside the PDS on a shared Docker network — no public internet required. For Postgres-backed PLC parity with production, pass plc_mode="real":
with PDSContainer(plc_mode="real") as pds:
account = pds.create_account("alice.test")with PDSContainer() as pds:
alice = pds.create_account("alice.test")
# Create
ref = alice.create_record("app.bsky.feed.post", {
"$type": "app.bsky.feed.post",
"text": "hello from testcontainers",
"createdAt": "2026-01-01T00:00:00Z",
})
# Read
record = alice.get_record("app.bsky.feed.post", ref.rkey)
# Update
alice.put_record("app.bsky.feed.post", ref.rkey, {
"$type": "app.bsky.feed.post",
"text": "updated text",
"createdAt": "2026-01-01T00:00:00Z",
})
# List & delete
records = alice.list_records("app.bsky.feed.post")
alice.delete_record("app.bsky.feed.post", ref.rkey)Observe real-time repository events via the AT Protocol firehose:
from testcontainers_atproto import PDSContainer
with PDSContainer() as pds:
alice = pds.create_account("alice.test")
alice.create_record("app.bsky.feed.post", {
"$type": "app.bsky.feed.post",
"text": "hello firehose",
"createdAt": "2026-01-01T00:00:00Z",
})
sub = pds.subscribe()
events = sub.collect(count=10, timeout=5.0)
commits = [e for e in events if e["header"].get("t") == "#commit"]
print(commits[-1]["body"]["ops"]) # [{"action": "create", ...}]Requires the firehose extra: pip install testcontainers-atproto[firehose]
Export repositories and retrieve blobs for relay and indexer testing:
with PDSContainer() as pds:
alice = pds.create_account("alice.test")
alice.create_record("app.bsky.feed.post", {
"$type": "app.bsky.feed.post",
"text": "sync test",
"createdAt": "2026-01-01T00:00:00Z",
})
# Export full repository as CAR bytes
car_bytes = alice.export_repo()
# Parse the CAR to inspect blocks
from testcontainers_atproto import parse_car
car = parse_car(car_bytes)
print(f"{len(car.blocks)} blocks, {len(car.roots)} roots")
# Retrieve a specific blob
blob_ref = alice.upload_blob(b"test data", "image/png")
blob_data = alice.get_blob(blob_ref["ref"]["$link"])
assert blob_data == b"test data"CAR parsing requires the sync extra: pip install testcontainers-atproto[sync]
Describe the world, materialize it in one call — no boilerplate account/record setup:
from testcontainers_atproto import PDSContainer, Seed
with PDSContainer() as pds:
world = (
Seed(pds)
.account("alice.test")
.post("Hello from Alice")
.post("Another post")
.account("bob.test")
.post("Bob's first post")
.follow("alice.test")
.like("alice.test", 0) # like Alice's first post
.apply()
)
alice = world.accounts["alice.test"]
bob = world.accounts["bob.test"]
assert len(world.records["alice.test"]) == 2Placeholders let custom records reference other accounts' DIDs and records — resolved at apply() time:
with PDSContainer() as pds:
world = (
Seed(pds)
.account("alice.test")
.record("com.example.project", {
"$type": "com.example.project",
"name": "My Project",
})
.account("bob.test")
.record("com.example.review", {
"$type": "com.example.review",
"reviewer": Seed.did("bob.test"),
"project": Seed.ref("alice.test", 0),
})
.apply()
)Accounts can be revisited to interleave records (e.g. conversation threads):
world = (
Seed(pds)
.account("alice.test")
.post("alice first")
.account("bob.test")
.post("bob replies")
.account("alice.test")
.post("alice continues")
.apply()
)Also available as a dict-based API for data-driven fixtures:
world = pds.seed({
"accounts": [
{"handle": "alice.test", "posts": ["Hello from Alice"]},
{"handle": "bob.test", "follows": ["alice.test"]},
],
})Test email verification and password reset flows with a local Mailpit SMTP server:
with PDSContainer(email_mode="capture") as pds:
alice = pds.create_account("alice.test")
# Request verification email
alice.request_email_confirmation()
# Retrieve it from Mailpit
message = pds.await_email(alice.email)
# Extract token and confirm (token format is PDS-version-dependent)
token = extract_token(message) # your extraction logic
alice.confirm_email(token)Password reset follows the same pattern:
alice.request_password_reset()
message = pds.await_email(alice.email)
token = extract_token(message)
alice.reset_password(token, "new-password")When email_mode="none" (the default), email verification is bypassed and no Mailpit container is started.
Deactivate, reactivate, and delete accounts to test how your app handles state changes:
with PDSContainer() as pds:
alice = pds.create_account("alice.test")
# Deactivate — account becomes inaccessible
alice.deactivate()
# Re-activate — access restored
alice.activate()
# Check status
status = alice.check_account_status()
assert status["activated"] is TrueAdmin operations let you test moderation flows:
with PDSContainer() as pds:
alice = pds.create_account("alice.test")
# Takedown — blocks access
pds.takedown(alice)
# Restore — unblocks access
pds.restore(alice)
# Query status
status = pds.get_subject_status(alice)Account deletion requires email_mode="capture" to retrieve the deletion token:
with PDSContainer(email_mode="capture") as pds:
alice = pds.create_account("alice.test", password="s3cret")
# ... confirm email first ...
alice.request_account_delete()
message = pds.await_email(alice.email)
token = extract_token(message) # your extraction logic
alice.delete_account("s3cret", token)XRPC failures raise XrpcError with structured fields:
from testcontainers_atproto import PDSContainer, XrpcError
with PDSContainer() as pds:
try:
pds.create_account("alice.invalid")
except XrpcError as e:
print(e.status_code) # 400
print(e.error) # "InvalidHandle"
print(e.message) # human-readable detailAfter installing the package, these fixtures are available automatically via the pytest11 entry point:
| Fixture | Scope | Description |
|---|---|---|
pds |
function | Fresh PDS instance per test |
pds_module |
module | Shared PDS instance within a test module |
pds_pair |
function | Two PDS instances for federation testing |
pds_image |
session | PDS image tag (override via ATP_PDS_IMAGE env var) |
def test_create_account(pds):
account = pds.create_account("bob.test")
assert account.did.startswith("did:plc:")make venv # Create virtual environment
source .testcontainers-atproto-3.12/bin/activate # Activate
make test # Run tests
make test-all # Run across all supported Python versionsAT Protocol introduces many domain-specific terms. See docs/glossary.md for definitions of PDS, DID, PLC, XRPC, and other initialisms used in this project.
Apache-2.0. See LICENSE.
