# print a failure message, kill any running hydrant pids, and exit with code 1.
export def fail [msg: string, ...pids: int] {
print $" FAILED: ($msg)"
for pid in $pids {
try { kill $pid }
}
exit 1
}
# pipe a full http response (obtained via -f -e flags) through this to assert an expected
# status code. on mismatch, prints the status, response body, kills any supplied pids, and exits.
# on success, returns the response record so callers can inspect fields further.
export def assert-status [expected: int, label: string, ...pids: int] {
let resp = $in
if $resp.status != $expected {
print $" ($label): expected status ($expected), got ($resp.status)"
try { print $" body: ($resp.body)" }
for pid in $pids { try { kill $pid } }
exit 1
}
$resp
}
# resolve the api port this test instance should use.
export def resolve-test-port [default: int] {
$env | get --optional HYDRANT_API_PORT | default ($default | into string) | into int
}
# resolve the debug port this test instance should use.
export def resolve-test-debug-port [default: int] {
$env | get --optional HYDRANT_DEBUG_PORT | default ($default | into string) | into int
}
# resolve the mock relay port for tests that need one.
export def resolve-test-mock-port [default: int] {
$env | get --optional HYDRANT_TEST_MOCK_PORT | default ($default | into string) | into int
}
export def resolve-binary [default: string] {
$env | get --optional HYDRANT_BINARY | default $default
}
export def load-env-file [] {
if (".env" | path exists) {
let content = (open .env)
$content | lines
| where { |x| ($x | str trim | is-empty) == false and ($x | str trim | str starts-with "#") == false }
| each { |x|
let parts = ($x | split row "=" -n 2)
{ key: ($parts.0 | str trim), value: ($parts.1 | str trim | str trim -c '"' | str trim -c "'") }
}
| reduce -f {} { |it, acc| $acc | insert $it.key $it.value }
} else {
{}
}
}
# plc directories to use, best first. the mirror is unthrottled but has gone down for days,
# so a dead mirror falls back to plc.directory instead of failing every test that resolves.
# set HYDRANT_TEST_PLC_URL (comma-separated) to pick your own.
export def plc-urls [] {
let candidates = (
$env | get --optional HYDRANT_TEST_PLC_URL
| default "https://plc.klbr.net,https://plc.directory"
| split row ","
| each { str trim | str trim -r -c "/" }
| where { is-not-empty }
)
for c in ($candidates | enumerate) {
if (try { http get --max-time 5sec $"($c.item)/_health"; true } catch { false }) {
return ($candidates | skip $c.index)
}
}
# nothing answers, so keep them all and let the real request say why
$candidates
}
export def resolve-pds [did: string] {
mut errors = []
for plc in (plc-urls) {
let attempt = try {
let doc = (http get --raw $"($plc)/($did)" | from json)
{pds: ($doc.service | where type == "AtprotoPersonalDataServer" | first).serviceEndpoint}
} catch { |e|
{error: $"($plc): ($e.msg)"}
}
if "pds" in $attempt {
return $attempt.pds
}
$errors = ($errors | append $attempt.error)
}
error make {msg: $"could not resolve ($did):\n($errors | str join "\n")"}
}
export def authenticate [pds_url: string, identifier: string, password: string] {
http post -t application/json $"($pds_url)/xrpc/com.atproto.server.createSession" {
identifier: $identifier,
password: $password
}
}
export def create-record [pds_url: string, jwt: string, repo: string, collection: string, record: any] {
http post -t application/json -H ["Authorization" $"Bearer ($jwt)"] $"($pds_url)/xrpc/com.atproto.repo.createRecord" {
repo: $repo,
collection: $collection,
record: $record
}
}
export def delete-record [pds_url: string, jwt: string, repo: string, collection: string, rkey: string] {
http post -t application/json -H ["Authorization" $"Bearer ($jwt)"] $"($pds_url)/xrpc/com.atproto.repo.deleteRecord" {
repo: $repo,
collection: $collection,
rkey: $rkey
}
}
export def deactivate-account [pds_url: string, jwt: string] {
http post -t application/json -H ["Authorization" $"Bearer ($jwt)"] $"($pds_url)/xrpc/com.atproto.server.deactivateAccount" {}
}
export def activate-account [pds_url: string, jwt: string] {
mut last_message = ""
for attempt in 1..5 {
let result = (curl --fail-with-body --silent --show-error --max-time 20 -X POST -H "Content-Type: application/json" -H $"Authorization: Bearer ($jwt)" $"($pds_url)/xrpc/com.atproto.server.activateAccount" | complete)
if $result.exit_code == 0 {
return $result.stdout
}
$last_message = $"($result.stdout)\n($result.stderr)" | str trim
if $attempt < 5 {
sleep 1sec
}
}
error make {msg: $"activateAccount failed: ($last_message)"}
}
# extract the hydrant executable path from cargo's json build output
def parse-hydrant-executable [output: string] {
let items = ($output
| lines
| each { |line| try { $line | from json } catch { null } }
| compact
| where { |r| $r.reason? == "compiler-artifact" and $r.executable? != null and ($r.target?.name? == "hydrant") })
if ($items | is-empty) {
error make {msg: "cargo succeeded without reporting a hydrant executable"}
}
$items | last | get executable
}
def require-build-success [out: record] {
if $out.exit_code != 0 {
let diagnostics = ([$out.stdout $out.stderr] | str join "\n" | str trim)
error make {msg: $"cargo build failed:\n($diagnostics)"}
}
parse-hydrant-executable $out.stdout
}
# build the hydrant binary
export def build-hydrant [] {
if ($env | get --optional HYDRANT_BINARY | is-not-empty) {
return $env.HYDRANT_BINARY
}
print "building hydrant..."
let out = (^cargo build --message-format json | complete)
require-build-success $out
}
# build the hydrant binary with extra cargo features (space-separated string)
export def build-hydrant-features [features: string] {
if ($env | get --optional HYDRANT_BINARY | is-not-empty) {
return $env.HYDRANT_BINARY
}
print $"building hydrant with features: ($features)..."
let out = (^cargo build --features $features --message-format json | complete)
require-build-success $out
}
# build the hydrant binary for relay only (no default features)
export def build-hydrant-relay [] {
if ($env | get --optional HYDRANT_BINARY | is-not-empty) {
return $env.HYDRANT_BINARY
}
print "building hydrant for relay only..."
let out = (^cargo build --no-default-features --features relay --message-format json | complete)
require-build-success $out
}
# build the hydrant binary for relay with jetstream support (no default features)
export def build-hydrant-relay-jetstream [] {
if ($env | get --optional HYDRANT_BINARY | is-not-empty) {
return $env.HYDRANT_BINARY
}
print "building hydrant for relay with jetstream..."
let out = (^cargo build --no-default-features --features relay,jetstream --message-format json | complete)
require-build-success $out
}
# start a websocket client in the background that writes each message as a line of out_file,
# and return its pid. stderr goes to out_file.err so it never reads as a message.
# websocat's async stdio fails with EINVAL on darwin when stdin is /dev/null, which it is
# for a background job, hence --no-async-stdio.
export def start-ws-listener [url: string, out_file: string, --ping] {
# with --ping the client gives up once a pong is 4s late, so a dead keepalive ends it
let flags = if $ping { "--ping-interval 1 --ping-timeout 4" } else { "" }
bash -c $"websocat --no-async-stdio -n ($flags) '($url)' > '($out_file)' 2> '($out_file).err' & echo $!"
| str trim
| into int
}
# start hydrant in the background
export def start-hydrant [binary: string, db_path: string, port: int] {
let occupied = (^bash -c $"echo > /dev/tcp/127.0.0.1/($port)" | complete)
if $occupied.exit_code == 0 {
error make {msg: $"cannot start test hydrant: port ($port) is already in use"}
}
let log_file = $"($db_path)/hydrant.log"
print $"starting hydrant - logs at ($log_file)..."
let hydrant_vars = ($env | transpose k v | where k =~ "HYDRANT_" | reduce -f {} { |it, acc| $acc | upsert $it.k $it.v })
let env_vars = {
HYDRANT_DATABASE_PATH: ($db_path),
HYDRANT_FULL_NETWORK: "false",
HYDRANT_API_BIND: $"127.0.0.1:($port)",
# tests drive the management api over tcp, but it's unix-only by default
HYDRANT_API_TCP_MANAGEMENT: "true",
HYDRANT_ENABLE_DEBUG: "true",
HYDRANT_DEBUG_PORT: (resolve-test-debug-port ($port + 1) | into string),
HYDRANT_PLC_URL: (plc-urls | str join ","),
RUST_LOG: "debug,hyper=error,tokio=error,h2=error,tower=error,rustls=error"
} | merge $hydrant_vars
let pid = (with-env $env_vars {
sh -c $"($binary) >($log_file) 2>&1 & echo $!" | str trim | into int
})
print $"hydrant started with pid: ($pid)"
{ pid: $pid, log: $log_file }
}
# wait for the api to become responsive
export def wait-for-api [url: string] {
print "waiting for api to be ready..."
for i in 1..30 {
try {
http get $"($url)/stats"
return true
} catch {
sleep 1sec
}
}
false
}
# poll stats until backfill is complete or fails
export def wait-for-backfill [url: string] {
print "waiting for backfill to complete..."
let attempts = 240
for i in 1..$attempts {
let stats = (http get $"($url)/stats").counts
let pending = ($stats.pending | into int)
let records = ($stats.records | into int)
let repos = ($stats.repos | into int)
let resync = ($stats.resync | into int)
print $"[($i)/($attempts)] pending: ($pending), records: ($records), repos: ($repos), resync: ($resync)"
if $resync > 0 {
print "resync state detected (failure or gone)!"
print ($stats | table)
return false
}
if ($pending == 0) and ($repos > 0) and ($records > 0) {
print "backfill complete."
return true
}
sleep 1sec
}
false
}