testcontainers-atproto 0.3.0

Testcontainers module for AT Protocol PDS integration testing

testcontainers-atproto

Spin up ephemeral PDS instances in your Python test suite. Lexicon-agnostic — works with any application built on AT Protocol, not just Bluesky.

---

Installation

pip install testcontainers-atproto

Requires Python 3.10+ and a running Docker daemon.

Extras

Extra	What it adds
testcontainers-atproto[firehose]	websockets, cbor2 for firehose subscription
testcontainers-atproto[sdk]	atproto (MarshalX SDK) for high-level record ops
testcontainers-atproto[all]	Both of the above

---

Quick start

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.test

A 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")

Record operations

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)

Firehose subscription

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]

Error handling

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 detail

---

Pytest fixtures

After 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:")

---

Development

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 versions

---

Glossary

AT Protocol introduces many domain-specific terms. See docs/glossary.md for definitions of PDS, DID, PLC, XRPC, and other initialisms used in this project.

---

License

Apache-2.0. See LICENSE.