{
pkgs,
self,
...
}:
pkgs.testers.nixosTest {
name = "tranquil-pds";
containers.server = {
config,
pkgs,
...
}: {
imports = [self.nixosModules.default];
services.tranquil-pds = {
enable = true;
database.createLocally = true;
settings = {
server.hostname = "pds.test";
server.host = "0.0.0.0";
server.disable_rate_limiting = true;
server.invite_code_required = false;
server.enable_pds_hosted_did_web = true;
secrets.jwt_secret = "test-jwt-secret-must-be-32-chars-long";
secrets.dpop_secret = "test-dpop-secret-must-be-32-chars-long";
secrets.master_key = "test-master-key-must-be-32-chars-long";
secrets.allow_insecure = true;
};
};
services.nginx = let
vhost = {
locations."/" = {
proxyPass = "http://127.0.0.1:3000";
proxyWebsockets = true;
extraConfig = ''
proxy_set_header Host $host;
proxy_set_header X-Forwarded-Proto $scheme;
proxy_read_timeout 120s;
'';
};
};
in {
enable = true;
recommendedProxySettings = true;
virtualHosts."pds.test" = vhost;
virtualHosts."*.pds.test" = vhost;
};
};
testScript = ''
import json
server.wait_for_unit("postgresql.service")
server.wait_for_unit("tranquil-pds.service")
server.wait_for_unit("nginx.service")
server.wait_for_open_port(3000)
server.wait_for_open_port(80)
def xrpc(method, endpoint, *, headers=None, data=None, raw_body=None, via="nginx"):
host_header = "-H 'Host: pds.test'" if via == "nginx" else ""
base = "http://localhost" if via == "nginx" else "http://localhost:3000"
url = f"{base}/xrpc/{endpoint}"
parts = ["curl", "-s", "-w", r"'\n%{http_code}'", "-X", method, host_header]
if headers:
parts.extend(f"-H '{k}: {v}'" for k, v in headers.items())
if data is not None:
parts.append("-H 'Content-Type: application/json'")
parts.append(f"-d '{json.dumps(data)}'")
if raw_body:
parts.append(f"--data-binary @{raw_body}")
parts.append(f"'{url}'")
result = server.succeed(" ".join(parts))
lines = result.rsplit("\n", 1)
body = lines[0] if len(lines) > 1 else result
status = lines[1].strip() if len(lines) > 1 else "000"
assert status.startswith("2"), f"xrpc {endpoint} returned HTTP {status}: {body}"
return body
def xrpc_json(method, endpoint, **kwargs):
return json.loads(xrpc(method, endpoint, **kwargs))
def xrpc_status(endpoint, *, headers=None, via="nginx"):
host_header = "-H 'Host: pds.test'" if via == "nginx" else ""
base = "http://localhost" if via == "nginx" else "http://localhost:3000"
url = f"{base}/xrpc/{endpoint}"
parts = ["curl", "-s", "-o", "/dev/null", "-w", "'%{http_code}'", host_header]
if headers:
parts.extend(f"-H '{k}: {v}'" for k, v in headers.items())
parts.append(f"'{url}'")
return server.succeed(" ".join(parts)).strip()
def http_status(path, *, host="pds.test", via="nginx"):
base = "http://localhost" if via == "nginx" else "http://localhost:3000"
return server.succeed(
f"curl -s -o /dev/null -w '%{{http_code}}' -H 'Host: {host}' '{base}{path}'"
).strip()
def http_get(path, *, host="pds.test"):
return server.succeed(
f"curl -sf -H 'Host: {host}' 'http://localhost{path}'"
)
def http_header(path, header, *, host="pds.test"):
return server.succeed(
f"curl -sI -H 'Host: {host}' 'http://localhost{path}'"
f" | grep -i '^{header}:'"
).strip()
# --- testing that stuff is up in general ---
with subtest("service is running"):
status = server.succeed("systemctl is-active tranquil-pds")
assert "active" in status
with subtest("data directories exist"):
server.succeed("test -d /var/lib/tranquil-pds/blobs")
with subtest("postgres database created"):
server.succeed("runuser -u tranquil-pds -- psql -d tranquil-pds -c 'SELECT 1'")
with subtest("healthcheck via backend"):
xrpc("GET", "_health", via="backend")
with subtest("healthcheck via nginx"):
xrpc("GET", "_health")
with subtest("describeServer"):
desc = xrpc_json("GET", "com.atproto.server.describeServer")
assert "availableUserDomains" in desc
assert "did" in desc
assert desc.get("inviteCodeRequired") == False
with subtest("nginx serves frontend"):
result = server.succeed("curl -sf -H 'Host: pds.test' http://localhost/")
assert "<html" in result.lower() or "<!" in result
with subtest("well-known proxied"):
code = http_status("/.well-known/atproto-did")
assert code != "502" and code != "504", f"well-known proxy broken: {code}"
with subtest("health endpoint proxied"):
code = http_status("/health")
assert code != "404" and code != "502", f"/health not proxied: {code}"
with subtest("robots.txt proxied"):
code = http_status("/robots.txt")
assert code != "404" and code != "502", f"/robots.txt not proxied: {code}"
with subtest("metrics endpoint proxied"):
code = http_status("/metrics")
assert code != "502", f"/metrics not proxied: {code}"
with subtest("oauth path proxied"):
code = http_status("/oauth/.well-known/openid-configuration")
assert code != "502" and code != "504", f"oauth proxy broken: {code}"
with subtest("subdomain routing works"):
code = http_status("/xrpc/_health", host="alice.pds.test")
assert code == "200", f"subdomain routing failed: {code}"
with subtest("oauth-client-metadata.json served with host substitution"):
meta_raw = http_get("/oauth-client-metadata.json")
meta = json.loads(meta_raw)
assert "client_id" in meta, f"no client_id in oauth-client-metadata: {meta}"
assert "pds.test" in meta_raw, "host substitution did not apply"
with subtest("static assets location exists"):
code = http_status("/assets/nonexistent.js")
assert code == "404", f"expected 404 for missing asset, got {code}"
with subtest("spa fallback works"):
code = http_status("/app/some/deep/route")
assert code == "200", f"SPA fallback broken: {code}"
with subtest("firewall ports open"):
server.succeed("ss -tlnp | grep ':80 '")
server.succeed("ss -tlnp | grep ':3000 '")
# --- test little bit of an account lifecycle ---
with subtest("create account"):
account = xrpc_json("POST", "com.atproto.server.createAccount", data={
"handle": "alice",
"password": "NixOS-Test-Pass-99!",
"email": "alice@pds.test",
"didType": "web",
})
assert "accessJwt" in account, f"no accessJwt: {account}"
assert "did" in account, f"no did: {account}"
access_token = account["accessJwt"]
did = account["did"]
assert did.startswith("did:web:"), f"expected did:web, got {did}"
with subtest("mark account verified"):
server.succeed(
f"runuser -u tranquil-pds -- psql -d tranquil-pds "
f"-c \"UPDATE users SET email_verified = true WHERE did = '{did}'\""
)
auth = {"Authorization": f"Bearer {access_token}"}
with subtest("get session"):
session = xrpc_json("GET", "com.atproto.server.getSession", headers=auth)
assert session["did"] == did
assert session["handle"] == "alice.pds.test", f"unexpected handle: {session['handle']}"
with subtest("create record"):
created = xrpc_json("POST", "com.atproto.repo.createRecord", headers=auth, data={
"repo": did,
"collection": "app.bsky.feed.post",
"record": {
"$type": "app.bsky.feed.post",
"text": "hello from lewis silly nix integration test",
"createdAt": "2025-01-01T00:00:00.000Z",
},
})
assert "uri" in created, f"no uri: {created}"
assert "cid" in created, f"no cid: {created}"
record_uri = created["uri"]
record_cid = created["cid"]
rkey = record_uri.split("/")[-1]
with subtest("read record back"):
fetched = xrpc_json(
"GET",
f"com.atproto.repo.getRecord?repo={did}&collection=app.bsky.feed.post&rkey={rkey}",
)
assert fetched["uri"] == record_uri
assert fetched["cid"] == record_cid
assert fetched["value"]["text"] == "hello from lewis silly nix integration test"
with subtest("upload blob"):
server.succeed("dd if=/dev/urandom bs=1024 count=4 of=/tmp/testblob.bin 2>/dev/null")
blob_resp = xrpc_json(
"POST",
"com.atproto.repo.uploadBlob",
headers={**auth, "Content-Type": "application/octet-stream"},
raw_body="/tmp/testblob.bin",
)
assert "blob" in blob_resp, f"no blob: {blob_resp}"
blob_ref = blob_resp["blob"]
assert blob_ref["size"] == 4096
with subtest("export repo as car"):
server.succeed(
f"curl -sf -H 'Host: pds.test' "
f"-o /tmp/repo.car "
f"'http://localhost/xrpc/com.atproto.sync.getRepo?did={did}'"
)
size = int(server.succeed("stat -c%s /tmp/repo.car").strip())
assert size > 0, "exported car is empty"
with subtest("delete record"):
xrpc_json("POST", "com.atproto.repo.deleteRecord", headers=auth, data={
"repo": did,
"collection": "app.bsky.feed.post",
"rkey": rkey,
})
with subtest("deleted record gone"):
code = xrpc_status(
f"com.atproto.repo.getRecord?repo={did}&collection=app.bsky.feed.post&rkey={rkey}",
)
assert code != "200", f"expected non-200 for deleted record, got {code}"
with subtest("service still healthy after lifecycle"):
xrpc("GET", "_health")
'';
}