johnandersen777.bsky.social/knotless
git that lives in your Atmosphere account
Forked from chadtmiller.com/knotless
0
0
//! knotless-serve against a reference PDS: real git clients over smart
//! HTTP, and the API.
#[path = "../../knotless/tests/common/mod.rs"]
mod common;
use common::*;
use knotless::commit::{Change, Spec};
use knotless_serve::{App, router};
use std::io::Write;
use std::sync::Arc;
use std::sync::atomic::Ordering;
use std::time::Duration;
struct Server {
base: String,
app: Arc<App>,
rt: tokio::runtime::Runtime,
}
fn serve(s: &Sandbox, ttl: Duration) -> Server {
let rt = tokio::runtime::Runtime::new().unwrap();
let app = App::new(ctx(s), ttl);
let listener = rt
.block_on(tokio::net::TcpListener::bind("127.0.0.1:0"))
.unwrap();
let base = format!("http://{}", listener.local_addr().unwrap());
let router = router(app.clone());
rt.spawn(async move { axum::serve(listener, router).await.unwrap() });
Server { base, app, rt }
}
fn put(path: &str, data: &str) -> Change {
Change::Put {
path: path.into(),
data: data.as_bytes().to_vec(),
executable: false,
}
}
fn commit(s: &Sandbox, name: &str, changes: Vec<Change>, message: &str) -> String {
let rt = tokio::runtime::Runtime::new().unwrap();
let url = knotless::url::RepoUrl::parse(&s.url(name)).unwrap();
let spec = Spec {
branch: "main".into(),
message: message.into(),
changes,
author: None,
retries: 3,
..Default::default()
};
rt.block_on(knotless::commit::commit(&ctx(s), &url, &spec))
.unwrap()
.sha
}
fn get(url: &str) -> (u16, String, reqwest::header::HeaderMap) {
let rt = tokio::runtime::Runtime::new().unwrap();
rt.block_on(async {
let res = reqwest::get(url).await.unwrap();
let status = res.status().as_u16();
let headers = res.headers().clone();
(status, res.text().await.unwrap(), headers)
})
}
#[test]
fn git_clients_over_smart_http() {
let s = Sandbox::new();
commit(
&s,
"web",
vec![put("a.txt", "one\n"), put("dir/b.txt", "two\n")],
"first",
);
let tip = commit(&s, "web", vec![put("a.txt", "three\n")], "second");
let server = serve(&s, Duration::ZERO);
let url = format!("{}/{}/web.git", server.base, s.account.handle);
let cases: [(&str, &[&str], &[&str]); 4] = [
("v2", &[], &[]),
("v0", &["-c", "protocol.version=0"], &[]),
("partial", &[], &["--filter=blob:none"]),
("shallow", &[], &["--depth", "1"]),
];
for (dir, config, flags) in cases {
let path = s.path(dir);
let mut args: Vec<&str> = config.to_vec();
args.extend(["clone", "-q"]);
args.extend(flags);
args.extend([url.as_str(), path.to_str().unwrap()]);
s.git(s.dir.path(), &args);
assert_eq!(s.git(&path, &["rev-parse", "HEAD"]), tip, "{dir}");
assert_eq!(
std::fs::read_to_string(path.join("a.txt")).unwrap(),
"three\n",
"{dir}"
);
}
assert_eq!(
s.git(&s.path("shallow"), &["rev-list", "--count", "HEAD"]),
"1"
);
// Pushing goes to the PDS, never here.
let (status, _, _) = get(&format!("{url}/info/refs?service=git-receive-pack"));
assert_eq!(status, 403);
let (status, body, _) = get(&format!("{}/healthz", server.base));
assert_eq!((status, body.as_str()), (200, "ok"));
drop(server.rt);
}
#[test]
fn gzip_requests() {
let s = Sandbox::new();
commit(&s, "gz", vec![put("a", "a")], "c");
let server = serve(&s, Duration::from_secs(60));
let url = format!("{}/{}/gz/git-upload-pack", server.base, s.account.handle);
let mut gz = flate2::write::GzEncoder::new(Vec::new(), flate2::Compression::default());
gz.write_all(b"0000").unwrap();
let body = gz.finish().unwrap();
server.rt.block_on(async {
let client = reqwest::Client::new();
let ok = client
.post(&url)
.header("content-encoding", "gzip")
.body(body)
.send()
.await
.unwrap();
assert_eq!(ok.status(), 200);
let bad = client
.post(&url)
.header("content-encoding", "gzip")
.body("nope")
.send()
.await
.unwrap();
assert_eq!(bad.status(), 400);
});
}
#[test]
fn api_reads_without_a_clone() {
let s = Sandbox::new();
let first = commit(
&s,
"api",
vec![put("a.txt", "one\n"), put("dir/b.txt", "two\n")],
"first",
);
let second = commit(&s, "api", vec![put("a.txt", "three\n")], "second\n\nbody");
let server = serve(&s, Duration::from_secs(60));
let api = format!("{}/api/{}/api", server.base, s.account.handle);
let (status, body, _) = get(&format!("{api}/refs"));
assert_eq!(status, 200);
let v: serde_json::Value = serde_json::from_str(&body).unwrap();
assert_eq!(v["refs"][0]["target"], second);
assert_eq!(v["head"], "refs/heads/main");
let (_, body, headers) = get(&format!("{api}/tree?rev={second}"));
let v: serde_json::Value = serde_json::from_str(&body).unwrap();
let names: Vec<&str> = v["entries"]
.as_array()
.unwrap()
.iter()
.map(|e| e["name"].as_str().unwrap())
.collect();
assert_eq!(names, vec!["a.txt", "dir"]);
assert!(
headers["cache-control"]
.to_str()
.unwrap()
.contains("immutable")
);
let (_, body, _) = get(&format!("{api}/tree?path=/dir/"));
assert!(body.contains("b.txt"));
let (status, body, headers) = get(&format!("{api}/blob?path=a.txt"));
assert_eq!((status, body.as_str()), (200, "three\n"));
assert!(headers.contains_key("etag"));
let (_, body, _) = get(&format!("{api}/blob?path=a.txt&rev={first}"));
assert_eq!(body, "one\n");
let (_, body, _) = get(&format!("{api}/log?limit=10"));
let v: serde_json::Value = serde_json::from_str(&body).unwrap();
assert_eq!(v["commits"].as_array().unwrap().len(), 2);
assert_eq!(v["commits"][0]["message"], "second\n\nbody");
assert_eq!(v["commits"][0]["parents"][0], first);
let (_, body, _) = get(&format!("{api}/log?path=dir"));
let v: serde_json::Value = serde_json::from_str(&body).unwrap();
assert_eq!(v["commits"].as_array().unwrap().len(), 1);
let (_, body, _) = get(&format!("{api}/diff?base={first}&head={second}"));
assert!(body.contains("-one") && body.contains("+three"), "{body}");
let (status, body, headers) = get(&format!("{api}/object/{second}"));
assert_eq!(status, 200);
assert!(body.starts_with("tree "));
assert_eq!(headers["x-git-type"], "commit");
for (path, want) in [
("/blob?rev=--output=x", 400),
("/blob?path=nope", 404),
("/object/xyz", 400),
(&format!("/object/{}", "0".repeat(40)), 404),
] {
assert_eq!(get(&format!("{api}{path}")).0, want, "{path}");
}
let (status, _, _) = get(&format!(
"{}/api/{}/missing/refs",
server.base, s.account.handle
));
assert_eq!(status, 404);
let (status, _, _) = get(&format!(
"{}/api/{}/bad%20name/refs",
server.base, s.account.handle
));
assert_eq!(status, 400);
let (status, _, _) = get(&format!("{}/api/nobody.invalid/x/refs", server.base));
assert_eq!(status, 404);
}
#[test]
fn ttl_and_firehose_freshness() {
let s = Sandbox::new();
commit(&s, "fresh", vec![put("a", "1")], "one");
let server = serve(&s, Duration::from_secs(3600));
let blob = format!("{}/api/{}/fresh/blob?path=a", server.base, s.account.handle);
assert_eq!(get(&blob).1, "1");
// Within the TTL and unwatched, the mirror is trusted as is.
commit(&s, "fresh", vec![put("a", "2")], "two");
assert_eq!(get(&blob).1, "1");
// The firehose marks it stale and refreshes it.
let ws = format!(
"{}/xrpc/com.atproto.sync.subscribeRepos",
net().pds.replace("http", "ws")
);
server
.rt
.spawn(knotless_serve::firehose::watch(server.app.clone(), ws));
wait_until(|| server.app.watching.load(Ordering::Relaxed));
commit(&s, "fresh", vec![put("a", "3")], "three");
wait_until(|| get(&blob).1 == "3");
// Watched and quiet: served without asking the PDS.
assert_eq!(get(&blob).1, "3");
// A firehose that cannot connect leaves the TTL in charge.
let lost = App::new(ctx(&s), Duration::ZERO);
lost.watching.store(true, Ordering::Relaxed);
server.rt.spawn(knotless_serve::firehose::watch(
lost.clone(),
"ws://127.0.0.1:9".into(),
));
wait_until(|| !lost.watching.load(Ordering::Relaxed));
}
#[test]
fn binary_lifecycle() {
let s = Sandbox::new();
let exe = env!("CARGO_BIN_EXE_knotless-serve");
let out = s.command(exe, s.dir.path()).arg("--nope").output().unwrap();
assert_eq!(out.status.code(), Some(2));
let out = s
.command(exe, s.dir.path())
.args(["--listen", "not-an-address"])
.output()
.unwrap();
assert_eq!(out.status.code(), Some(1));
let out = s
.command(exe, s.dir.path())
.args(["--public-url", "ftp://nope"])
.output()
.unwrap();
assert_eq!(out.status.code(), Some(1));
// `--private` is gone, and says what replaced it.
let out = s
.command(exe, s.dir.path())
.args(["--private", "--listen", "127.0.0.1:0"])
.output()
.unwrap();
assert_eq!(out.status.code(), Some(2));
let said = String::from_utf8_lossy(&out.stderr);
assert!(said.contains(knotless_serve::PRIVATE_GONE), "{said}");
// Bots' logins are where bots reach it.
let out = s
.command(exe, s.dir.path())
.args(["--bots", "bots"])
.output()
.unwrap();
assert_eq!(out.status.code(), Some(1));
assert!(String::from_utf8_lossy(&out.stderr).contains("--public-url"));
// So are a site's owners sent from its root.
let out = s
.command(exe, s.dir.path())
.args(["--pages-domain", "knotless.page"])
.output()
.unwrap();
assert_eq!(out.status.code(), Some(1));
assert!(String::from_utf8_lossy(&out.stderr).contains("--pages-domain takes --public-url"));
// Apple's push key takes its id and team, and has to be one.
let apns = |args: &[&str]| {
let out = s.command(exe, s.dir.path()).args(args).output().unwrap();
assert_eq!(out.status.code(), Some(1));
String::from_utf8_lossy(&out.stderr).to_string()
};
let key = s.path("apns.p8");
let key = key.to_string_lossy();
assert!(apns(&["--apns-key", &key]).contains("--apns-key-id and --apns-team"));
let named = [
"--apns-key",
&*key,
"--apns-key-id",
"K",
"--apns-team",
"T",
];
assert!(apns(&named).contains("reading"));
std::fs::write(s.path("apns.p8"), "not a key").unwrap();
assert!(apns(&named).contains("apns.p8"));
let signing = p256::ecdsa::SigningKey::from_bytes((&[7u8; 32]).into()).unwrap();
let pem = p256::pkcs8::EncodePrivateKey::to_pkcs8_pem(&signing, Default::default()).unwrap();
std::fs::write(s.path("apns.p8"), pem.as_bytes()).unwrap();
std::fs::create_dir_all(s.path("releases")).unwrap();
std::fs::write(s.path("releases/latest"), "0.1.0\n").unwrap();
let mut child = s
.command(exe, s.dir.path())
.args([
"--listen",
"127.0.0.1:0",
"--ttl",
"1",
"--firehose",
"ws://127.0.0.1:9",
"--jetstream",
"ws://127.0.0.1:9",
"--public-url",
"http://localhost:1",
"--oauth-url",
"https://gateway.example/",
"--index",
&s.path("index.db").to_string_lossy(),
"--backfill",
"http://127.0.0.1:9",
"--relay",
"http://127.0.0.1:9",
"--labeler",
"did:web:127.0.0.1%3A9",
"--web",
"http://127.0.0.1:9",
"--releases",
&s.path("releases").to_string_lossy(),
"--metrics",
"127.0.0.1:0",
"--usage",
&s.path("usage.db").to_string_lossy(),
"--logs",
&s.path("logs.db").to_string_lossy(),
"--logs-keep",
"100000",
"--code-index",
&s.path("code").to_string_lossy(),
"--pages-domain",
"knotless.page",
"--apns-key",
&key,
"--apns-key-id",
"K",
"--apns-team",
"T",
"--client-minimum",
"0.1.0",
"--client-minimum-reason",
"Compaction lost objects.",
"--ios-minimum-build",
"7",
"--ios-minimum-reason",
"Older builds lose drafts.",
])
.stdout(std::process::Stdio::piped())
.stderr(std::process::Stdio::piped())
.spawn()
.unwrap();
// The sweep at startup, told of on stderr, which is read to its end.
let (swept, sweeping) = std::sync::mpsc::channel();
let stderr = std::io::BufReader::new(child.stderr.take().unwrap());
std::thread::spawn(move || {
for line in std::io::BufRead::lines(stderr).map_while(Result::ok) {
if line.contains("index sweep ended") {
let _ = swept.send(());
}
}
});
let mut out = std::io::BufReader::new(child.stdout.take().unwrap());
let mut said = |prefix: &str| {
let mut line = String::new();
std::io::BufRead::read_line(&mut out, &mut line).unwrap();
line.trim().strip_prefix(prefix).unwrap().to_string()
};
let base = said("listening on ");
let metrics = said("metrics on ");
assert_eq!(get(&format!("{base}/healthz")).1, "ok");
// The oldest client it asks for, and why.
let described = get(&format!("{base}/xrpc/dev.knotless.describeServer")).1;
let described: serde_json::Value = serde_json::from_str(&described).unwrap();
assert_eq!(
described["client"],
serde_json::json!({"minimum": "0.1.0", "reason": "Compaction lost objects."})
);
assert_eq!(
described["ios"],
serde_json::json!({"minimumBuild": 7, "reason": "Older builds lose drafts."})
);
// Each request is counted, by its route, and given its trace.
assert_eq!(get(&format!("{base}/healthz")).2["x-trace-id"].len(), 32);
// Or the trace a traceparent names, as the pages hand theirs on.
let id = "4bf92f3577b34da6a3ce929d0e0e4736";
let asked = reqwest::blocking::Client::new()
.get(format!("{base}/healthz"))
.header("traceparent", format!("00-{id}-00f067aa0ba902b7-01"))
.send()
.unwrap();
assert_eq!(asked.headers()["x-trace-id"], id);
let counted = get(&format!("{metrics}/metrics")).1;
assert!(counted.contains("http_route=\"/healthz\""), "{counted}");
let http = reqwest::blocking::Client::new();
let set = |level: &str| {
let res = http.put(format!("{metrics}/log")).body(level.to_string());
res.send().unwrap().status().as_u16()
};
assert_eq!(set("info,knotless::outgoing=debug"), 200);
assert_eq!(set("[[["), 400);
assert!(
get(&format!("{base}/.well-known/did.json"))
.1
.contains("did:web:localhost%3A1")
);
assert!(
get(&format!("{base}/install.sh"))
.1
.contains("http://localhost:1/releases")
);
// The script and the releases are answered here, even what is not
// there, rather than handed on to the pages.
assert_eq!(get(&format!("{base}/releases/latest")).1, "0.1.0\n");
assert_eq!(get(&format!("{base}/releases/nope")).0, 404);
assert!(
get(&format!("{base}/.well-known/oauth-protected-resource/mcp"))
.1
.contains("http://localhost:1/mcp")
);
// The pages are handed on, here to nothing.
assert_eq!(get(&format!("{base}/someone.test")).0, 502);
// Stopped only once the sweep has finished, which it may not have
// yet on a slow machine.
sweeping.recv_timeout(Duration::from_secs(30)).unwrap();
std::process::Command::new("kill")
.args(["-TERM", &child.id().to_string()])
.status()
.unwrap();
assert!(child.wait().unwrap().success());
// Each line it logged was kept as it went to stderr, the last as it
// stopped; the requests with their traces.
let db = rusqlite::Connection::open(s.path("logs.db")).unwrap();
let kept = |sql: &str| db.query_row(sql, [], |r| r.get::<_, i64>(0)).unwrap();
assert_eq!(
kept("SELECT count(*) FROM line WHERE message = 'stopped'"),
1
);
let traced = "SELECT count(*) FROM line WHERE message = 'request' AND length(trace) = 32";
assert!(kept(traced) >= 1);
// Each request was counted, under the route it matched, and written as
// serve stopped; the health check was not.
let db = rusqlite::Connection::open(s.path("usage.db")).unwrap();
let mut stmt = db
.prepare(
"SELECT surface, route, class, sum(n) FROM request GROUP BY 1, 2, 3 ORDER BY 1, 2, 3",
)
.unwrap();
let counted: Vec<(String, String, i64, i64)> = stmt
.query_map([], |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?, r.get(3)?)))
.unwrap()
.map(Result::unwrap)
.collect();
let row = |surface: &str, route: &str, class, n| (surface.into(), route.into(), class, n);
assert_eq!(
counted,
[
row("install", "/install.sh", 2, 1),
row("install", "/releases/*", 2, 1),
row("install", "/releases/*", 4, 1),
row("mcp", "/.well-known/oauth-protected-resource/mcp", 2, 1),
row("other", "/.well-known/did.json", 2, 1),
row("pages", "*", 5, 1),
row("xrpc", "/xrpc/dev.knotless.describeServer", 2, 1),
]
);
}
#[test]
fn an_index_is_checked_before_a_rolling_deploy() {
let dir = tempfile::tempdir().unwrap();
let exe = env!("CARGO_BIN_EXE_knotless-serve");
let path = dir.path().join("index.db");
let check = || {
std::process::Command::new(exe)
.arg("--check-index")
.arg(&path)
.output()
.unwrap()
};
// None yet: nothing to rebuild, and checking makes none.
assert!(check().status.success());
assert!(!path.exists());
knotless_serve::index::Index::open(&path).unwrap();
let out = check();
assert!(out.status.success());
assert!(String::from_utf8_lossy(&out.stdout).contains("keeps it"));
// Of other tables, starting would rebuild it, and checking leaves it.
rusqlite::Connection::open(&path)
.unwrap()
.execute_batch("PRAGMA user_version = 1")
.unwrap();
let out = check();
assert_eq!(out.status.code(), Some(1));
assert!(String::from_utf8_lossy(&out.stderr).contains("would rebuild it"));
assert_eq!(knotless_serve::index::version_at(&path).unwrap(), Some(1));
}
#[test]
fn arguments() {
use knotless_serve::parse_args;
let args = |a: &[&str]| {
parse_args(
&std::iter::once("x")
.chain(a.iter().copied())
.map(String::from)
.collect::<Vec<_>>(),
)
};
let o = args(&[
"--listen",
"0.0.0.0:1",
"--ttl",
"9",
"--firehose",
"wss://r",
"--jetstream",
"wss://jetstream.us-east.bsky.network",
])
.unwrap();
assert_eq!(o.listen, "0.0.0.0:1");
assert_eq!(o.ttl, Duration::from_secs(9));
assert_eq!(o.firehose.as_deref(), Some("wss://r"));
assert_eq!(
o.jetstream.as_deref(),
Some("wss://jetstream.us-east.bsky.network")
);
assert_eq!(o.public_url, None);
// Empty, a stream or backfill is off.
let o = args(&["--firehose", "", "--jetstream", "", "--backfill", ""]).unwrap();
assert_eq!((o.firehose, o.jetstream, o.backfill), (None, None, None));
let o = args(&["--backfill", "https://r"]).unwrap();
assert_eq!(o.backfill.as_deref(), Some("https://r"));
// Directories in turn, and a mirror; empty, none.
let o = args(&[
"--directory",
"https://a",
"--directory",
"",
"--directory",
"https://b",
"--mirror",
"https://m",
])
.unwrap();
assert_eq!(o.sources.directories, ["https://a", "https://b"]);
assert_eq!(o.sources.mirror.as_deref(), Some("https://m"));
assert_eq!(args(&["--mirror", ""]).unwrap().sources.mirror, None);
let gone = args(&["--private"]).err();
assert_eq!(gone.as_deref(), Some(knotless_serve::PRIVATE_GONE));
let o = args(&["--public-url", "https://knot.example"]).unwrap();
assert_eq!(o.public_url.as_deref(), Some("https://knot.example"));
let o = args(&["--labeler", "did:web:a", "--labeler", "did:web:b"]).unwrap();
assert_eq!(o.labelers, ["did:web:a", "did:web:b"]);
let o = args(&["--default-runner", "did:plc:r"]).unwrap();
assert_eq!(o.default_runner.as_deref(), Some("did:plc:r"));
assert_eq!(
args(&["--default-runner", ""]).unwrap().default_runner,
None
);
assert_eq!(
args(&[
"--default-runner",
"did:plc:r",
"--default-runner",
"did:plc:s"
])
.unwrap_err(),
"--default-runner names one runner"
);
assert_eq!(
args(&["--default-runner", "runner.example"]).unwrap_err(),
"--default-runner takes a DID, not runner.example"
);
assert_eq!(o.web, None);
// The oldest client asked for is a version; empty, none is.
let o = args(&["--client-minimum", "0.3.0", "--client-minimum-reason", ""]).unwrap();
assert_eq!(o.client_minimum.as_deref(), Some("0.3.0"));
assert_eq!(o.client_minimum_reason, None);
let o = args(&["--client-minimum", "", "--client-minimum-reason", "Why"]).unwrap();
assert_eq!(o.client_minimum, None);
assert_eq!(o.client_minimum_reason.as_deref(), Some("Why"));
assert!(args(&["--client-minimum", "soon"]).is_err());
let o = args(&["--ios-minimum-build", "5", "--ios-minimum-reason", ""]).unwrap();
assert_eq!(o.ios_minimum_build, Some(5));
assert_eq!(o.ios_minimum_reason, None);
let o = args(&["--ios-minimum-build", "", "--ios-minimum-reason", "Why"]).unwrap();
assert_eq!(o.ios_minimum_build, None);
assert_eq!(o.ios_minimum_reason.as_deref(), Some("Why"));
let refused = args(&["--ios-minimum-build", "soon"]).err();
assert_eq!(
refused.as_deref(),
Some("--ios-minimum-build takes a build number")
);
let o = args(&["--web", "http://127.0.0.1:2591"]).unwrap();
assert_eq!(o.web.as_deref(), Some("http://127.0.0.1:2591"));
assert_eq!(o.releases, None);
let o = args(&["--releases", "/var/lib/knotless/releases"]).unwrap();
let dir = knotless_serve::releases::Source::Dir("/var/lib/knotless/releases".into());
assert_eq!(o.releases, Some(dir));
// Or a repository's releases.
let o = args(&["--releases", "knotless://alice.test/knotless"]).unwrap();
let url = knotless::url::RepoUrl::parse("knotless://alice.test/knotless").unwrap();
assert_eq!(
o.releases,
Some(knotless_serve::releases::Source::Repo(url))
);
assert!(args(&["--releases", "knotless://alice.test"]).is_err());
assert_eq!(o.stale, None);
let o = args(&["--stale", "300"]).unwrap();
assert_eq!(o.stale, Some(Duration::from_secs(300)));
assert!(args(&["--stale", "x"]).is_err());
assert_eq!(
(o.logs.as_deref(), o.logs_keep),
(None, knotless_serve::logs::KEEP)
);
let o2 = args(&["--logs", "/data/logs.db", "--logs-keep", "9"]).unwrap();
assert_eq!(
(o2.logs.as_deref(), o2.logs_keep),
(Some(std::path::Path::new("/data/logs.db")), 9)
);
assert!(args(&["--logs-keep", "x"]).is_err());
assert_eq!(o.usage, None);
let o = args(&["--usage", "/var/lib/knotless/usage.db"]).unwrap();
assert_eq!(
o.usage.as_deref(),
Some(std::path::Path::new("/var/lib/knotless/usage.db"))
);
assert!(args(&["--ttl", "x"]).is_err());
assert!(args(&["--ttl"]).is_err());
assert_eq!(args(&[]).unwrap().metrics, None);
let o = args(&["--metrics", "127.0.0.1:9464"]).unwrap();
assert_eq!(o.metrics.as_deref(), Some("127.0.0.1:9464"));
assert_eq!(o.bots, None);
let o = args(&["--bots", "/var/lib/knotless/bots"]).unwrap();
assert_eq!(
o.bots.as_deref(),
Some(std::path::Path::new("/var/lib/knotless/bots"))
);
assert_eq!(o.sign_up_pds, "https://selfhosted.social");
let o = args(&["--sign-up-pds", "https://pds.example"]).unwrap();
assert_eq!(o.sign_up_pds, "https://pds.example");
assert!(args(&["--sign-up-pds"]).is_err());
assert_eq!(o.oauth_url, None);
assert_eq!(o.pages_domain, None);
let o = args(&["--pages-domain", "knotless.page"]).unwrap();
assert_eq!(o.pages_domain.as_deref(), Some("knotless.page"));
assert_eq!(o.apns_key, None);
let o = args(&[
"--apns-key",
"/data/apns.p8",
"--apns-key-id",
"K",
"--apns-team",
"T",
])
.unwrap();
assert_eq!(
o.apns_key.as_deref(),
Some(std::path::Path::new("/data/apns.p8"))
);
assert_eq!(o.apns_key_id.as_deref(), Some("K"));
assert_eq!(o.apns_team.as_deref(), Some("T"));
let o = args(&["--apns-key", "", "--apns-key-id", "", "--apns-team", ""]).unwrap();
assert_eq!((o.apns_key, o.apns_key_id, o.apns_team), (None, None, None));
let o = args(&["--oauth-url", "https://gateway.example"]).unwrap();
assert_eq!(o.oauth_url.as_deref(), Some("https://gateway.example"));
}
#[test]
fn firehose_frames_of_every_kind() {
use futures::SinkExt;
use tokio_tungstenite::tungstenite::Message;
let s = Sandbox::new();
let rt = tokio::runtime::Runtime::new().unwrap();
let listener = rt
.block_on(tokio::net::TcpListener::bind("127.0.0.1:0"))
.unwrap();
let addr = listener.local_addr().unwrap();
rt.spawn(async move {
let (tcp, _) = listener.accept().await.unwrap();
let mut ws = tokio_tungstenite::accept_async(tcp).await.unwrap();
let value = |pairs: Vec<(&str, ciborium::Value)>| {
ciborium::Value::Map(
pairs
.into_iter()
.map(|(k, v)| (ciborium::Value::Text(k.into()), v))
.collect(),
)
};
let mut frame = Vec::new();
ciborium::into_writer(
&value(vec![("t", ciborium::Value::Text("#identity".into()))]),
&mut frame,
)
.unwrap();
ciborium::into_writer(
&value(vec![
("did", ciborium::Value::Text("did:plc:watched".into())),
("seq", ciborium::Value::Integer(5.into())),
]),
&mut frame,
)
.unwrap();
ws.send(Message::Text("hello".into())).await.unwrap();
ws.send(Message::Binary(vec![0xff].into())).await.unwrap();
ws.send(Message::Binary(frame.into())).await.unwrap();
ws.close(None).await.unwrap();
});
let app = App::new(ctx(&s), Duration::ZERO);
rt.spawn(knotless_serve::firehose::watch(
app.clone(),
format!("ws://{addr}"),
));
wait_until(|| app.events_for("did:plc:watched") == 1);
wait_until(|| !app.watching.load(Ordering::Relaxed));
}
#[test]
fn a_cache_it_cannot_write_is_a_server_error() {
let mut s = Sandbox::new();
commit(&s, "nocache", vec![put("a", "a")], "c");
std::fs::write(s.path("blocked"), b"").unwrap();
let blocked = s.path("blocked").join("cache");
s.set("KNOTLESS_CACHE_DIR", blocked.to_str().unwrap());
let server = serve(&s, Duration::ZERO);
let (status, _, _) = get(&format!(
"{}/api/{}/nocache/refs",
server.base, s.account.handle
));
assert_eq!(status, 500);
}