chadtmiller.com/knotless
Git hosted in your atproto account, on any stock PDS, fast enough for agents.
git clone knotless://chadtmiller.com/knotlessneeds the knotless git helper. set it up
//! Forks, pulls, issues and reviews, with an appview that follows the
//! reference PDS's firehose.
#[path = "../../knotless/tests/common/mod.rs"]
mod common;
use common::*;
use futures::{SinkExt, StreamExt};
use knotless_serve::gateway::Gateway;
use knotless_serve::index::Index;
use knotless_serve::{App, router};
use serde_json::Value;
use std::sync::Arc;
use std::time::Duration;
struct Appview {
base: String,
did: String,
app: Arc<App>,
rt: tokio::runtime::Runtime,
_proxy: TlsProxy,
_sandbox: Sandbox,
}
/// An appview that is a gateway too, following the PDS.
fn appview() -> Appview {
let mut s = Sandbox::new();
s.set("KNOTLESS_PASSWORD", "");
s.set("KNOTLESS_DID_WEB_SCHEME", "http");
s.set("KNOTLESS_ADMIN_PASSWORD", "hunter2");
let rt = tokio::runtime::Runtime::new().unwrap();
let listener = rt
.block_on(tokio::net::TcpListener::bind("127.0.0.1:0"))
.unwrap();
let port = listener.local_addr().unwrap().port();
let base = format!("http://localhost:{port}");
let mut gw = Gateway::new(&base).unwrap();
// Authorization servers reach a gateway, an OAuth client, over https.
let proxy = tls_proxy(port);
gw.oauth_url = proxy.url.clone();
let did = gw.did.clone();
let index = Index::open(&s.path("index.db")).unwrap();
let app = App::build(ctx(&s), Duration::ZERO, Some(gw), Some(index), false);
let router = router(app.clone());
rt.spawn(async move { axum::serve(listener, router).await.unwrap() });
let ws = net().pds.replace("http://", "ws://");
rt.spawn(knotless_serve::firehose::watch(app.clone(), ws));
Appview {
base,
did,
app,
rt,
_proxy: proxy,
_sandbox: s,
}
}
fn person(account: Account, appview: &Appview) -> Sandbox {
let mut s = Sandbox::with(account);
s.set("KNOTLESS_PASSWORD", "");
s.set("KNOTLESS_DID_WEB_SCHEME", "http");
s.set("KNOTLESS_APPVIEW", &appview.base);
s.set("KNOTLESS_BROWSER", &browser(&s.account.clone(), ""));
let password = s.account.password.clone();
let handle = s.account.handle.clone();
assert_eq!(
s.cli(&["login", &handle, "--password-stdin"], password.as_bytes())
.0,
0
);
s
}
fn ok(s: &Sandbox, args: &[&str]) -> String {
let (code, out) = s.cli(args, b"");
assert_eq!(code, 0, "{args:?}: {out}");
out
}
fn fails(s: &Sandbox, args: &[&str]) {
let (code, out) = s.cli(args, b"");
assert_eq!(code, 1, "{args:?} should fail: {out}");
}
/// Waits until `list` shows `line`.
fn listed(s: &Sandbox, url: &str, line: &str) {
wait_until(|| s.cli(&["list", url], b"").1.contains(line));
}
#[test]
fn forks_pulls_reviews_and_merges() {
let av = appview();
let owner = person(account(), &av);
let contrib = person(account(), &av);
let url = owner.url("forkable");
let work = owner.work("w", 1);
owner.git(&work, &["push", "-q", &url, "main"]);
// A fork in the contributor's account, with a branch to offer.
let fork = ok(&contrib, &["fork", &url]);
assert_eq!(fork, format!("knotless://{}/forkable", contrib.account.did));
let clone = contrib.path("clone");
contrib.git(contrib.dir.path(), &["clone", "-q", &fork, "clone"]);
contrib.git(&clone, &["checkout", "-q", "-b", "feature"]);
let feature = contrib.commit_file(&clone, "b", "from the fork\n");
contrib.git(&clone, &["push", "-q", "origin", "feature"]);
fails(
&contrib,
&["pull", &url, "main", &fork, "no-such-branch", "nope"],
);
let pull = ok(
&contrib,
&[
"pull", &url, "main", &fork, "feature", "Add b", "-m", "please",
],
);
assert!(pull.starts_with(&format!("at://{}/dev.knotless.pull/", contrib.account.did)));
listed(&owner, &url, &format!("pull\topen\t{pull}\tAdd b"));
ok(
&contrib,
&["comment", &url, &pull, "looks right to me", "--approve"],
);
ok(&owner, &["comment", &url, &pull, "thanks"]);
fails(&owner, &["comment", &url, &pull, "?", "--bogus"]);
wait_until(|| owner.cli(&["show", &url, &pull], b"").1.contains("thanks"));
let shown = ok(&owner, &["show", &url, &pull]);
assert!(shown.starts_with("open pull by"), "{shown}");
assert!(
shown.contains("please") && shown.contains("[approve]: looks right"),
"{shown}"
);
// The base has not moved: a fast-forward.
let merged = ok(&owner, &["merge", &url, &pull]);
assert_eq!(merged, feature);
let (_, refs) = owner.cli(&["refs", &url], b"");
assert!(
refs.contains(&format!("{feature}\trefs/heads/main")),
"{refs}"
);
listed(&owner, &url, &format!("pull\tmerged\t{pull}"));
fails(&owner, &["merge", &url, &pull]);
// The base moved: a merge commit with both parents.
owner.git(&work, &["pull", "-q", &url, "main"]);
owner.commit_file(&work, "c", "owner's\n");
owner.git(&work, &["push", "-q", &url, "main"]);
contrib.git(&clone, &["checkout", "-q", "-b", "two", &feature]);
let two = contrib.commit_file(&clone, "d", "more\n");
contrib.git(&clone, &["push", "-q", "origin", "two"]);
let second = ok(&contrib, &["pull", &url, "main", &fork, "two", "Add d"]);
listed(&owner, &url, &second);
let merge = ok(&owner, &["merge", &url, &second]);
let check = owner.path("check");
owner.git(owner.dir.path(), &["clone", "-q", &url, "check"]);
assert_eq!(owner.git(&check, &["rev-parse", "HEAD"]), merge);
assert_eq!(owner.git(&check, &["rev-parse", "HEAD^2"]), two);
assert!(check.join("c").exists() && check.join("d").exists());
// Both sides changed one file: a conflict, and nothing moves.
owner.git(&work, &["pull", "-q", &url, "main"]);
owner.commit_file(&work, "f", "owner's f\n");
owner.git(&work, &["push", "-q", &url, "main"]);
contrib.git(&clone, &["checkout", "-q", "-b", "clash", &two]);
contrib.commit_file(&clone, "f", "contributor's f\n");
contrib.git(&clone, &["push", "-q", "origin", "clash"]);
let clash = ok(
&contrib,
&["pull", &url, "main", &fork, "clash", "Change f"],
);
listed(&owner, &url, &clash);
let (merged, _, err) = owner.knotless(&["merge", &url, &clash], b"");
assert!(!merged && err.contains("conflicts in f"), "{err}");
// Branches can go between opening a pull and merging it.
contrib.git(&clone, &["push", "-q", "origin", ":clash"]);
let (merged, _, err) = owner.knotless(&["merge", &url, &clash], b"");
assert!(!merged && err.ends_with("has no branch clash"), "{err}");
fails(
&contrib,
&["pull", &url, "release", &fork, "two", "Into release"],
);
owner.git(&work, &["push", "-q", &url, "main:release"]);
let release = ok(
&contrib,
&["pull", &url, "release", &fork, "two", "Into release"],
);
owner.git(&work, &["push", "-q", &url, ":release"]);
listed(&owner, &url, &release);
let (merged, _, err) = owner.knotless(&["merge", &url, &release], b"");
assert!(!merged && err.ends_with("has no branch release"), "{err}");
// A pull outlives its branches by pointing at others: the target by
// anyone who may close it, the head by its author alone.
owner.git(&work, &["push", "-q", &url, "main:stable"]);
for bad in [
vec!["retarget", &url, &release],
vec!["retarget", &url, &release, "--sideways", "x"],
vec!["retarget", &url, &release, "--base", "nope"],
vec!["retarget", &url, &release, "--head", "two"],
vec!["retarget", &url, &pull, "--base", "stable"],
] {
fails(&owner, &bad);
}
ok(&owner, &["retarget", &url, &release, "--base", "stable"]);
fails(&contrib, &["retarget", &url, &clash, "--head", "nope"]);
ok(&contrib, &["retarget", &url, &clash, "--head", "two"]);
let issue = ok(&contrib, &["issue", &url, "Not a pull"]);
fails(&owner, &["retarget", &url, &issue, "--base", "main"]);
// Once the index has the new target, the pull merges into it.
wait_until(|| owner.knotless(&["merge", &url, &release], b"").0);
let check = owner.path("stable");
owner.git(
owner.dir.path(),
&["clone", "-q", "-b", "stable", &url, "stable"],
);
owner.git(&check, &["merge-base", "--is-ancestor", &two, "stable"]);
// A collaborator merges, through the gateway.
ok(&owner, &["delegate", &owner.account.handle, &av.did]);
ok(&owner, &["share", &url, &contrib.account.handle, "--write"]);
contrib.git(&clone, &["fetch", "-q", &url, "main"]);
contrib.git(&clone, &["checkout", "-q", "-b", "three", &merge]);
contrib.commit_file(&clone, "e", "collaborating\n");
contrib.git(&clone, &["push", "-q", "origin", "three"]);
let third = ok(&contrib, &["pull", &url, "main", &fork, "three", "Add e"]);
listed(&contrib, &url, &third);
let sha = ok(&contrib, &["merge", &url, &third]);
let (_, refs) = owner.cli(&["refs", &url], b"");
assert!(refs.contains(&format!("{sha}\trefs/heads/main")), "{refs}");
// And makes and deletes branches the same way, as the owner does.
assert_eq!(ok(&contrib, &["branch", &url, "topic"]), sha);
assert_eq!(
ok(&contrib, &["branch", &url, "topic", "--delete"]),
"deleted topic"
);
assert_eq!(ok(&owner, &["branch", &url, "pinned", &two]), two);
ok(&owner, &["branch", &url, "pinned", "--delete"]);
// Someone else's commit, elsewhere on the firehose, changes nothing.
av.app.observe(knotless_serve::firehose::Event::default());
assert_eq!(av.app.events_for("did:plc:nobody"), 0);
// A record gone before the index fetched it is an error to skip.
let gone = knotless_serve::firehose::Op {
action: "create".into(),
collection: "dev.knotless.issue".into(),
rkey: "3zzzzzzzzzzzz".into(),
record: None,
};
let index = av.app.index.as_ref().unwrap();
assert!(
av.rt
.block_on(knotless_serve::index_op(
&av.app.ctx,
index,
&owner.account.did,
&gone
))
.is_err()
);
// The index knows the fork, and finds repositories by name.
let found: Value = reqwest::blocking::get(format!(
"{}/xrpc/dev.knotless.listRepos?q=forkable",
av.base
))
.unwrap()
.json()
.unwrap();
assert_eq!(found["repos"].as_array().unwrap().len(), 2);
assert!(found.get("cursor").is_none());
// A page at a time, the same two.
let list = |cursor: Option<&str>| -> Value {
let mut query = vec![("q", "forkable"), ("limit", "1")];
query.extend(cursor.map(|c| ("cursor", c)));
reqwest::blocking::Client::new()
.get(format!("{}/xrpc/dev.knotless.listRepos", av.base))
.query(&query)
.send()
.unwrap()
.json()
.unwrap()
};
let first = list(None);
let second = list(first["cursor"].as_str());
assert_eq!(
[&first["repos"][0], &second["repos"][0]],
[&found["repos"][0], &found["repos"][1]]
);
assert!(
list(second["cursor"].as_str())["repos"]
.as_array()
.unwrap()
.is_empty()
);
let uri = format!("at://{}/dev.knotless.repo/forkable", owner.account.did);
let indexed = av.app.index.as_ref().unwrap();
let key = indexed.repo(&uri).unwrap().unwrap()["key"]
.as_str()
.unwrap()
.to_string();
let forks = indexed.forks(&owner.account.did, &key);
assert_eq!(forks.unwrap().len(), 1);
// A second appview, backfilled, sees the same.
let s = Sandbox::new();
let index = Index::open(&s.path("i.db")).unwrap();
let n = av
.rt
.block_on(knotless_serve::backfill(&ctx(&s), &index, &net().pds))
.unwrap();
assert!(n >= 10, "{n}");
let report = av.rt.block_on(knotless_serve::backfilled(
&ctx(&s),
&index,
"http://127.0.0.1:9",
));
assert!(
report.starts_with("backfilling from http://127.0.0.1:9: "),
"{report}"
);
let report = av
.rt
.block_on(knotless_serve::backfilled(&ctx(&s), &index, &net().pds));
assert!(report.starts_with("backfilled "), "{report}");
let did = owner.account.did.as_str();
let backfilled = index.forge(did, &key).unwrap().len();
wait_until(|| indexed.forge(did, &key).unwrap().len() == backfilled);
// A deletion takes the record out of the index.
let pull_rkey = third.rsplit('/').next().unwrap();
let deleted = knotless_serve::firehose::Op {
action: "delete".into(),
collection: "dev.knotless.pull".into(),
rkey: pull_rkey.into(),
record: None,
};
av.rt
.block_on(knotless_serve::index_op(
&ctx(&s),
&index,
&contrib.account.did,
&deleted,
))
.unwrap();
assert_eq!(index.forge(did, &key).unwrap().len(), backfilled - 1);
}
#[test]
fn pulls_squash_into_a_linear_branch() {
let av = appview();
let owner = person(account(), &av);
let contrib = person(account(), &av);
let url = owner.url("squashed");
let work = owner.work("w", 1);
owner.git(&work, &["push", "-q", &url, "main"]);
let base = owner.git(&work, &["rev-parse", "HEAD"]);
let fork = ok(&contrib, &["fork", &url]);
let clone = contrib.path("clone");
contrib.git(contrib.dir.path(), &["clone", "-q", &fork, "clone"]);
contrib.git(&clone, &["checkout", "-q", "-b", "two"]);
contrib.commit_file(&clone, "a", "one\n");
contrib.commit_file(&clone, "b", "two\n");
contrib.git(&clone, &["push", "-q", "origin", "two"]);
let pull = ok(
&contrib,
&[
"pull",
&url,
"main",
&fork,
"two",
"Add a and b",
"-m",
"both",
],
);
listed(&owner, &url, &pull);
// A fast-forward would bring both commits; protected, it is one.
ok(&owner, &["protect", &url, "main"]);
let sha = ok(&owner, &["merge", &url, &pull]);
let check = owner.path("check");
owner.git(owner.dir.path(), &["clone", "-q", &url, "check"]);
assert_eq!(owner.git(&check, &["rev-parse", "HEAD"]), sha);
assert_eq!(owner.git(&check, &["rev-parse", "HEAD^"]), base);
assert_eq!(owner.git(&check, &["rev-list", "--count", "HEAD"]), "2");
assert!(check.join("a").exists() && check.join("b").exists());
let message = owner.git(&check, &["log", "-1", "--format=%an|%cn|%B"]);
// The head's author wrote it, and the owner, who merged it, committed it.
let committer = &owner.account.handle;
assert_eq!(
message,
format!("Test|{committer}|Add a and b\n\nboth\n\n{pull}")
);
}
#[test]
fn issues_and_who_may_close_them() {
let av = appview();
let owner = person(account(), &av);
let author = person(account(), &av);
let stranger = person(account(), &av);
let url = owner.url("tracked");
ok(&owner, &["create", &url]);
let issue = ok(
&author,
&["issue", &url, "It breaks", "-m", "when I do this"],
);
let quiet = ok(&author, &["issue", &url, "Also this"]);
listed(&owner, &url, &issue);
listed(&owner, &url, &quiet);
assert_eq!(ok(&author, &["close", &url, &issue]), "closed");
listed(&owner, &url, &format!("issue\tclosed\t{issue}"));
// A stranger's reopening counts for nothing; the owner's does.
ok(&stranger, &["reopen", &url, &issue]);
std::thread::sleep(Duration::from_millis(500));
let listing = ok(&owner, &["list", &url, "--issues"]);
assert!(
listing.contains(&format!("issue\tclosed\t{issue}")),
"{listing}"
);
ok(&owner, &["reopen", &url, &issue]);
listed(&owner, &url, &format!("issue\topen\t{issue}"));
assert_eq!(ok(&owner, &["list", &url, "--pulls"]), "");
assert!(ok(&owner, &["show", &url, &quiet]).ends_with(": Also this"));
fails(&owner, &["list", &url, "--bogus"]);
fails(&owner, &["merge", &url, &issue]);
fails(
&owner,
&["show", &url, "at://did:plc:nobody/dev.knotless.issue/x"],
);
// A comment is about a pull or an issue on this repository.
fails(
&owner,
&[
"comment",
&url,
"at://did:plc:x/dev.knotless.state/3abc",
"?",
],
);
fails(
&owner,
&[
"comment",
&url,
&format!("{}x", issue.trim_end_matches(|c: char| c.is_alphanumeric())),
"?",
],
);
ok(&owner, &["create", &owner.url("elsewhere")]);
fails(&owner, &["comment", &owner.url("elsewhere"), &issue, "?"]);
fails(&owner, &["issue", &url, "t", "-x"]);
fails(&owner, &["issue", &url, "t", "-m"]);
fails(&owner, &["fork", &owner.url("nothing")]);
// Without an appview there is nowhere to find them.
let mut lost = person(owner.account.clone(), &av);
lost.set("KNOTLESS_APPVIEW", "");
let (code, out) = lost.cli(&["list", &url], b"");
assert_eq!(code, 1, "{out}");
// A server without an index says so.
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 base = format!("http://{}", listener.local_addr().unwrap());
let bare = App::new(ctx(&s), Duration::ZERO);
let router = router(bare.clone());
rt.spawn(async move { axum::serve(listener, router).await.unwrap() });
for path in [
"dev.knotless.getForge?repo=at://did:plc:a/dev.knotless.repo/r&key=k",
"dev.knotless.listRepos",
"com.atproto.label.queryLabels?uriPatterns=*",
] {
let res = reqwest::blocking::get(format!("{base}/xrpc/{path}")).unwrap();
assert_eq!(res.status().as_u16(), 404);
}
// Nor can it follow a labeler, and an account is not one: each try
// fails, and it tries again.
for (app, did) in [
(&bare, av.did.clone()),
(&av.app, owner.account.did.clone()),
] {
let following = knotless_serve::moderation::follow(app.clone(), did);
let _ =
rt.block_on(async { tokio::time::timeout(Duration::from_secs(1), following).await });
}
// A repository is named by its record's URI.
let res = reqwest::blocking::get(format!(
"{}/xrpc/dev.knotless.getForge?repo=x&key=k",
av.base
))
.unwrap();
assert_eq!(res.status().as_u16(), 400);
// Every record comes back, a page at a time.
let key = owner.record("dev.knotless.repo", "tracked")["key"].clone();
let repo = format!("at://{}/dev.knotless.repo/tracked", owner.account.did);
let forge = |cursor: &str| -> Value {
reqwest::blocking::get(format!(
"{}/xrpc/dev.knotless.getForge?repo={repo}&key={}&limit=1&cursor={cursor}",
av.base,
key.as_str().unwrap()
))
.unwrap()
.json()
.unwrap()
};
let (mut paged, mut cursor) = (0, String::new());
loop {
let page = forge(&cursor);
assert!(page["entries"].as_array().unwrap().len() <= 1);
paged += page["entries"].as_array().unwrap().len();
match page["cursor"].as_str() {
Some(next) => cursor = next.to_string(),
None => break,
}
}
assert!(paged > 1, "{paged}");
}
#[test]
fn private_repositories_keep_their_issues_in_the_space() {
if !net().spaces {
eprintln!("skipped: the PDS is not on the spaces alpha");
return;
}
let av = appview();
let owner = person(account(), &av);
let reader = person(account(), &av);
let url = owner.url("inner");
ok(&owner, &["create", &url, "--private"]);
let work = owner.work("w", 1);
owner.git(&work, &["push", "-q", &url, "main"]);
ok(&owner, &["share", &url, &reader.account.handle]);
let issue = ok(&reader, &["issue", &url, "Private trouble"]);
owner.git(&work, &["checkout", "-q", "-b", "fix"]);
let fix = owner.commit_file(&work, "fix", "fixed\n");
owner.git(&work, &["push", "-q", &url, "fix"]);
let pull = ok(&owner, &["pull", &url, "main", &url, "fix", "Fix it"]);
ok(&reader, &["comment", &url, &pull, "ship it", "--approve"]);
// No appview involved: the members' repos in the space are read directly.
let listing = ok(&owner, &["list", &url]);
assert!(
listing.contains(&format!("issue\topen\t{issue}")),
"{listing}"
);
assert!(
listing.contains(&format!("pull\topen\t{pull}")),
"{listing}"
);
// Its pulls and issues are in the space, which cannot move.
fails(&owner, &["transfer", &url, &reader.account.handle]);
// The owner's gate and blocks hold in the space, with no appview.
ok(&owner, &["hide", &url, &issue]);
assert!(!ok(&reader, &["list", &url]).contains(&issue));
ok(&owner, &["unhide", &url, &issue]);
ok(&owner, &["block", &reader.account.handle]);
assert!(!ok(&owner, &["list", &url]).contains(&issue));
ok(&owner, &["unblock", &reader.account.handle]);
assert!(ok(&owner, &["list", &url]).contains(&issue));
// A block from this repository alone is in the space with it.
ok(&owner, &["block", &reader.account.handle, &url]);
assert!(!ok(&owner, &["list", &url]).contains(&issue));
ok(&owner, &["unblock", &reader.account.handle, &url]);
assert!(ok(&owner, &["list", &url]).contains(&issue));
assert_eq!(ok(&owner, &["merge", &url, &pull]), fix);
let listing = ok(&reader, &["list", &url]);
assert!(
listing.contains(&format!("pull\tmerged\t{pull}")),
"{listing}"
);
// The appview never saw any of it.
let uri = format!("at://{}/dev.knotless.repo/inner", owner.account.did);
assert_eq!(av.app.index.as_ref().unwrap().repo(&uri).unwrap(), None);
// A private repository forks into a private one.
let fork = ok(&reader, &["fork", &url, "inner-fork"]);
let (_, refs) = reader.cli(&["refs", &fork], b"");
assert!(refs.contains(&format!("{fix}\trefs/heads/main")), "{refs}");
let anonymous = Sandbox::new();
assert_eq!(anonymous.cli(&["refs", &fork], b"").0, 1);
// Its source's owner reads it, to take a pull from it.
let (code, refs) = owner.cli(&["refs", &fork], b"");
assert_eq!(code, 0, "{refs}");
let readers = ok(&reader, &["members", &fork]);
assert!(readers.contains(owner.account.did.as_str()), "{readers}");
// A fork of one's own is shared with nobody more.
let own = ok(&owner, &["fork", &url, "inner-copy"]);
let readers = ok(&owner, &["members", &own]);
assert_eq!(readers.lines().count(), 1, "{readers}");
}
#[test]
fn moving_to_another_account() {
let av = appview();
let owner = person(account(), &av);
let heir = person(account(), &av);
let stranger = person(account(), &av);
let url = owner.url("moving");
let work = owner.work("w", 1);
owner.git(&work, &["push", "-q", &url, "main"]);
let before = ok(&owner, &["refs", &url]);
let closed = ok(&stranger, &["issue", &url, "Closed before the move"]);
let open = ok(&stranger, &["issue", &url, "Still open"]);
ok(&owner, &["close", &url, &closed]);
listed(&owner, &url, &format!("issue\tclosed\t{closed}"));
// Nothing moves until it is offered, and only to whom it is offered.
fails(&heir, &["accept", &url]);
fails(&owner, &["transfer", &url, &owner.account.handle]);
ok(&owner, &["transfer", &url, &heir.account.handle]);
fails(&stranger, &["accept", &url]);
ok(&owner, &["transfer", &url, "--cancel"]);
fails(&heir, &["accept", &url]);
let elsewhere = format!("knotless://{}/moving", heir.account.handle);
ok(&owner, &["transfer", &url, &elsewhere]);
ok(&owner, &["transfer", &url, "--cancel"]);
fails(
&owner,
&["transfer", &owner.url("nothing"), &heir.account.handle],
);
let offered = ok(&owner, &["transfer", &url, &heir.account.handle]);
assert!(offered.contains("knotless accept"), "{offered}");
let moved = ok(&heir, &["accept", &url]);
let new = format!("knotless://{}/moving", heir.account.did);
assert_eq!(moved, format!("moved to {new}"));
// The old URL leads to the new place, for reads and pushes alike.
assert_eq!(ok(&heir, &["refs", &url]), before);
heir.git(heir.dir.path(), &["clone", "-q", &url, "clone"]);
let clone = heir.path("clone");
heir.commit_file(&clone, "g", "after the move\n");
let (pushed, _, err) = heir.git_raw(&clone, &["push", "-q", "origin", "main"]);
assert!(
pushed && err.contains(&format!("has moved to {new}")),
"{err}"
);
assert_ne!(ok(&heir, &["refs", &new]), before);
// Its issues came with it, and what the old owner did before the move
// still counts; nothing they do after it does.
listed(&heir, &new, &format!("issue\tclosed\t{closed}"));
ok(&owner, &["close", &url, &open]);
std::thread::sleep(Duration::from_secs(1));
listed(&heir, &new, &format!("issue\topen\t{open}"));
ok(&heir, &["comment", &new, &open, "moved along"]);
ok(&heir, &["close", &new, &open]);
listed(&heir, &new, &format!("issue\tclosed\t{open}"));
// The old copy can go, leaving a redirect in its place; its old names'
// redirects lead to the new place too, and others' stay as they were.
let key = owner.record("dev.knotless.repo", "moving")["key"].clone();
let at = format!("at://{}/dev.knotless.repo/moving", owner.account.did);
owner.put_record(
"dev.knotless.redirect",
"once-moving",
serde_json::json!({"$type": "dev.knotless.redirect", "to": {"uri": at, "key": key},
"createdAt": "2026-10-21T00:00:00.000Z"}),
);
ok(&owner, &["create", &owner.url("spare")]);
ok(&owner, &["rename", &owner.url("spare"), "spare2"]);
ok(&owner, &["delete", &url, "--yes"]);
assert!(
owner.record("dev.knotless.redirect", "once-moving")["to"]["uri"]
.as_str()
.unwrap()
.starts_with(&format!("at://{}/", heir.account.did))
);
assert_eq!(
owner.record("dev.knotless.redirect", "spare")["to"]["uri"],
format!("at://{}/dev.knotless.repo/spare2", owner.account.did)
);
assert!(
owner.record("dev.knotless.redirect", "moving")["to"]["uri"]
.as_str()
.unwrap()
.starts_with(&format!("at://{}/", heir.account.did))
);
assert_eq!(
ok(&stranger, &["refs", &url]),
ok(&stranger, &["refs", &new])
);
listed(&stranger, &url, &format!("issue\tclosed\t{closed}"));
let direct = reqwest::blocking::Client::builder()
.redirect(reqwest::redirect::Policy::none())
.build()
.unwrap();
let old_page = || {
direct
.get(format!(
"{}/_data/{}/moving/issues",
av.base, owner.account.handle
))
.send()
.unwrap()
};
wait_until(|| old_page().status().as_u16() == 307);
assert_eq!(
old_page().headers()["location"],
format!("/_data/{}/moving/issues", heir.account.did)
);
// Saying a repository came from another proves nothing without its
// owner's word: none of its issues come along.
let other = owner.url("other");
owner.git(&work, &["push", "-q", &other, "main"]);
let theirs = ok(&stranger, &["issue", &other, "Not yours"]);
listed(&owner, &other, &theirs);
let key = owner.record("dev.knotless.repo", "other")["key"].clone();
let from = serde_json::json!({"uri": format!("at://{}/dev.knotless.repo/other", owner.account.did), "key": key});
stranger.put_record(
"dev.knotless.repo",
"claimed",
serde_json::json!({"$type": "dev.knotless.repo", "key": "3kabcdefghijk", "name": "claimed",
"transferredFrom": from, "createdAt": "2026-10-21T00:00:00.000Z"}),
);
let claimed = stranger.url("claimed");
std::thread::sleep(Duration::from_secs(1));
assert!(!ok(&stranger, &["list", &claimed]).contains(&theirs));
}
/// Waits until `list` no longer shows `line`.
fn unlisted(s: &Sandbox, url: &str, line: &str) {
wait_until(|| {
let (code, out) = s.cli(&["list", url], b"");
code == 0 && !out.contains(line)
});
}
#[test]
fn moderating() {
let av = appview();
let owner = person(account(), &av);
let troll = person(account(), &av);
let reporter = person(account(), &av);
let url = owner.url("moderated");
let work = owner.work("w", 1);
owner.git(&work, &["push", "-q", &url, "main"]);
let trolling = ok(&troll, &["issue", &url, "Trolling"]);
let trolling = trolling.trim();
let fair = ok(&reporter, &["issue", &url, "A fair point"]);
let fair = fair.trim();
let reply = ok(&troll, &["comment", &url, fair, "nonsense"]);
let reply = reply.trim();
listed(&owner, &url, trolling);
// The owner hides an issue, and what was said about it goes with it.
assert_eq!(ok(&owner, &["hide", &url, fair]), "hidden");
assert_eq!(ok(&owner, &["hide", &url, fair]), "hidden");
unlisted(&reporter, &url, fair);
assert_eq!(reporter.cli(&["show", &url, fair], b"").0, 1);
ok(&owner, &["hide", &url, reply]);
assert_eq!(ok(&owner, &["unhide", &url, fair]), "shown");
listed(&reporter, &url, fair);
assert!(!ok(&reporter, &["show", &url, fair]).contains("nonsense"));
ok(&owner, &["unhide", &url, reply]);
wait_until(|| ok(&reporter, &["show", &url, fair]).contains("nonsense"));
// Only the owner, and only what is said about a repository.
fails(&troll, &["hide", &url, fair]);
let repo_uri = format!("at://{}/dev.knotless.repo/moderated", owner.account.did);
fails(&owner, &["hide", &url, &repo_uri]);
// A block keeps out everything the account writes.
let blocked = ok(&owner, &["block", &troll.account.handle]);
assert_eq!(blocked, format!("blocked {}", troll.account.did));
ok(&owner, &["block", &troll.account.handle]);
unlisted(&reporter, &url, trolling);
wait_until(|| !ok(&reporter, &["show", &url, fair]).contains("nonsense"));
fails(&owner, &["block", &owner.account.handle]);
ok(&owner, &["unblock", &troll.account.handle]);
listed(&reporter, &url, trolling);
// Or keeps them out of one repository only, and only its owner can.
ok(&owner, &["block", &troll.account.handle, &url]);
ok(&owner, &["block", &troll.account.handle, &url]);
unlisted(&reporter, &url, trolling);
// Nor does it show by its URI, however new.
assert_eq!(reporter.cli(&["show", &url, trolling], b"").0, 1);
fails(&reporter, &["block", &troll.account.handle, &url]);
ok(&owner, &["unblock", &troll.account.handle, &url]);
listed(&reporter, &url, trolling);
// Reports go to the appview's moderation service, from an OAuth login
// too, whose scope names the call.
let handle = reporter.account.handle.clone();
assert_eq!(reporter.cli(&["login", &handle], b"").0, 0);
let sent = ok(
&reporter,
&["report", trolling, "spam", "-m", "the same thing again"],
);
assert_eq!(sent, "report 1 sent");
ok(&reporter, &["report", &troll.account.handle, "rude"]);
ok(&reporter, &["report", &url, "other"]);
fails(&reporter, &["report", &url, "boring"]);
fails(&reporter, &["report", &format!("{trolling}x"), "spam"]);
let mut admin = person(account(), &av);
fails(&admin, &["admin", "reports"]);
admin.set("KNOTLESS_ADMIN_PASSWORD", "wrong");
fails(&admin, &["admin", "reports"]);
admin.set("KNOTLESS_ADMIN_PASSWORD", "hunter2");
let reports = ok(&admin, &["admin", "reports"]);
let lines: Vec<&str> = reports.lines().collect();
assert_eq!(lines.len(), 3, "{reports}");
assert!(
lines[0].starts_with("3\t") && lines[0].contains(&repo_uri),
"{reports}"
);
assert!(
lines[1].contains(&format!("rude\t{}", troll.account.did)),
"{reports}"
);
assert_eq!(
lines[2],
format!(
"1\t{}\tspam\t{trolling}\tthe same thing again",
reporter.account.did
)
);
assert_eq!(ok(&admin, &["admin", "reports", "1"]).lines().count(), 1);
fails(&admin, &["admin", "reports", "x"]);
fails(&admin, &["admin", "nothing"]);
// Without an appview, reports go where KNOTLESS_MODERATOR says.
let mut direct = person(reporter.account.clone(), &av);
direct.set("KNOTLESS_APPVIEW", "");
fails(&direct, &["report", trolling, "spam"]);
direct.set("KNOTLESS_MODERATOR", &owner.account.did);
fails(&direct, &["report", trolling, "spam"]);
fails(&reporter, &["report", &owner.url("nothing"), "spam"]);
direct.set("KNOTLESS_MODERATOR", &av.did);
assert_eq!(ok(&direct, &["report", trolling, "spam"]), "report 4 sent");
// A second appview follows this one's labels.
let other = appview();
other.rt.spawn(knotless_serve::moderation::follow(
other.app.clone(),
av.did.clone(),
));
let down_there = |subject: &str| other.app.index.as_ref().unwrap().down(&[subject]).unwrap();
// The operator takes things down.
assert_eq!(ok(&admin, &["admin", "status", trolling]), "up");
assert_eq!(ok(&admin, &["admin", "takedown", trolling]), "taken down");
assert_eq!(ok(&admin, &["admin", "status", trolling]), "taken down");
unlisted(&reporter, &url, trolling);
wait_until(|| down_there(trolling));
fails(&admin, &["admin", "takedown", trolling, "--later"]);
assert_eq!(
ok(&admin, &["admin", "takedown", trolling, "--undo"]),
"restored"
);
listed(&reporter, &url, trolling);
wait_until(|| !down_there(trolling));
// An account taken down takes its repositories with it.
let page = || {
reqwest::blocking::get(format!(
"{}/_data/{}/moderated",
av.base, owner.account.handle
))
.unwrap()
.status()
.as_u16()
};
assert_eq!(page(), 200);
ok(&admin, &["admin", "takedown", &owner.account.handle]);
assert_eq!(
ok(&admin, &["admin", "status", &owner.account.handle]),
"taken down"
);
assert_eq!(page(), 404);
let listing = reqwest::blocking::get(format!("{}/xrpc/dev.knotless.listRepos", av.base))
.unwrap()
.text()
.unwrap();
assert!(!listing.contains(&owner.account.did), "{listing}");
let labels: Value = reqwest::blocking::get(format!(
"{}/xrpc/com.atproto.label.queryLabels?uriPatterns=did:*&uriPatterns={}",
av.base, trolling
))
.unwrap()
.json()
.unwrap();
assert_eq!(labels["labels"][0]["uri"], owner.account.did.as_str());
assert_eq!(labels["labels"][0]["val"], "!takedown");
assert_eq!(labels["labels"].as_array().unwrap().len(), 1);
wait_until(|| down_there(&owner.account.did));
ok(&admin, &["admin", "takedown", &url]);
assert_eq!(ok(&admin, &["admin", "status", &url]), "taken down");
ok(
&admin,
&["admin", "takedown", &owner.account.handle, "--undo"],
);
assert_eq!(page(), 404);
ok(&admin, &["admin", "takedown", &url, "--undo"]);
assert_eq!(page(), 200);
// The calls themselves, by hand.
let client = reqwest::blocking::Client::new();
let xrpc = |nsid: &str| format!("{}/xrpc/{nsid}", av.base);
let post = |nsid: &str, auth: Option<&str>, body: Value| {
let mut req = client.post(xrpc(nsid)).json(&body);
if let Some(auth) = auth {
req = req.header("authorization", auth);
}
req.send().unwrap().status().as_u16()
};
let basic = "Basic YWRtaW46aHVudGVyMg==";
let status = "com.atproto.admin.updateSubjectStatus";
assert_eq!(post(status, None, serde_json::json!({})), 401);
assert_eq!(
post(status, Some(basic), serde_json::json!({"subject": {}})),
400
);
let account =
serde_json::json!({"$type": "com.atproto.admin.defs#repoRef", "did": "did:plc:x"});
assert_eq!(
post(status, Some(basic), serde_json::json!({"subject": account})),
200
);
let get_status = |q: &str| {
client
.get(xrpc(&format!("com.atproto.admin.getSubjectStatus{q}")))
.header("authorization", basic)
.send()
.unwrap()
.status()
.as_u16()
};
assert_eq!(get_status(""), 400);
assert_eq!(get_status("?did=did:plc:x"), 200);
let report = "com.atproto.moderation.createReport";
assert_eq!(post(report, None, serde_json::json!({})), 401);
let token = |ctx: &knotless::env::Ctx| {
av.rt.block_on(async {
let me = knotless::session::me(ctx).await.unwrap();
knotless::auth::service_token(ctx, &me, &av.did, report)
.await
.unwrap()
})
};
let bearer = format!("Bearer {}", token(&ctx(&reporter)));
assert_eq!(
post(
report,
Some(&bearer),
serde_json::json!({"subject": account})
),
400
);
let bearer = format!("Bearer {}", token(&ctx(&reporter)));
assert_eq!(
post(report, Some(&bearer), serde_json::json!({"subject": {}})),
400
);
let page2 = client
.get(xrpc("dev.knotless.listReports?limit=2"))
.header("authorization", basic)
.send()
.unwrap()
.json::<Value>()
.unwrap();
assert_eq!(page2["cursor"], "3");
// The label stream, from the start: everything given out so far, then
// what comes; from past the end, only what comes after it.
let stream = |cursor: Option<i64>| {
let url = format!(
"{}/xrpc/com.atproto.label.subscribeLabels{}",
av.base.replace("http", "ws"),
cursor.map(|c| format!("?cursor={c}")).unwrap_or_default()
);
av.rt
.block_on(tokio_tungstenite::connect_async(url))
.unwrap()
.0
};
let (mut from_start, mut from_end) = (stream(Some(0)), stream(Some(1000)));
let next = |ws: &mut tokio_tungstenite::WebSocketStream<_>| {
let msg = av.rt.block_on(ws.next()).unwrap().unwrap();
knotless_serve::moderation::parse(&msg.into_data()).unwrap()
};
assert_eq!(next(&mut from_start).0, 1);
for _ in 2..=6 {
next(&mut from_start);
}
ok(&admin, &["admin", "takedown", trolling]);
let (seq, labels) = next(&mut from_start);
assert_eq!((seq, labels[0]["uri"].as_str()), (7, Some(trolling)));
let ping = tokio_tungstenite::tungstenite::Message::Ping(vec![].into());
av.rt.block_on(from_start.send(ping)).unwrap();
av.rt.block_on(from_start.close(None)).unwrap();
av.rt.block_on(from_end.close(None)).unwrap();
av.rt.block_on(stream(None).close(None)).unwrap();
ok(&admin, &["admin", "takedown", trolling, "--undo"]);
// Labels follow from where they were read to.
let labels = other.app.index.as_ref().unwrap();
assert!(labels.labeler_seq(&av.did).unwrap().is_some());
// One taken down as soon as it is opened does not show by its URI,
// indexed yet or not.
let fresh = ok(&troll, &["issue", &url, "Again"]);
let fresh = fresh.trim();
ok(&admin, &["admin", "takedown", fresh]);
assert_eq!(reporter.cli(&["show", &url, fresh], b"").0, 1);
}