Something went wrong. Try again.
A lexicon-driven AppView for ATProto.
Something went wrong. Try again.
Rust
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826mod common;
use axum::body::Body;use axum::http::{Request, StatusCode};use happyview::db::{adapt_sql, now_rfc3339};use http_body_util::BodyExt;use serde_json::{Value, json};use serial_test::serial;use tower::ServiceExt;
use common::app::TestApp;use common::fixtures;
// ---------------------------------------------------------------------------// Helpers// ---------------------------------------------------------------------------
async fn json_body(resp: axum::response::Response) -> Value { let body = resp.into_body().collect().await.unwrap().to_bytes(); serde_json::from_slice(&body).unwrap()}
fn admin_get( uri: &str, cookie: (axum::http::HeaderName, axum::http::HeaderValue),) -> Request<Body> { Request::builder() .uri(uri) .header(cookie.0, cookie.1) .body(Body::empty()) .unwrap()}
fn admin_post( uri: &str, cookie: (axum::http::HeaderName, axum::http::HeaderValue), body: &Value,) -> Request<Body> { Request::builder() .method("POST") .uri(uri) .header(cookie.0, cookie.1) .header("content-type", "application/json") .body(Body::from(serde_json::to_vec(body).unwrap())) .unwrap()}
fn admin_delete( uri: &str, cookie: (axum::http::HeaderName, axum::http::HeaderValue),) -> Request<Body> { Request::builder() .method("DELETE") .uri(uri) .header(cookie.0, cookie.1) .body(Body::empty()) .unwrap()}
// ---------------------------------------------------------------------------// Auth tests// ---------------------------------------------------------------------------
#[tokio::test]#[serial]async fn admin_no_auth_returns_401() { common::require_db!(); let app = TestApp::new().await;
let resp = app .router .clone() .oneshot( Request::builder() .uri("/admin/lexicons") .body(Body::empty()) .unwrap(), ) .await .unwrap();
assert_eq!(resp.status(), StatusCode::UNAUTHORIZED);}
#[tokio::test]#[serial]async fn admin_wrong_token_returns_401() { common::require_db!(); let app = TestApp::new().await;
// No valid session cookie — the request will be rejected. let resp = app .router .clone() .oneshot( Request::builder() .uri("/admin/lexicons") .header("cookie", "session=invalid-cookie-value") .body(Body::empty()) .unwrap(), ) .await .unwrap();
assert_eq!(resp.status(), StatusCode::UNAUTHORIZED);}
#[tokio::test]#[serial]async fn admin_valid_token_returns_200() { common::require_db!(); let app = TestApp::new().await;
let resp = app .router .clone() .oneshot(admin_get("/admin/lexicons", app.admin_cookie())) .await .unwrap();
assert_eq!(resp.status(), StatusCode::OK);}
#[tokio::test]#[serial]async fn admin_non_admin_did_returns_403() { common::require_db!(); let app = TestApp::new().await;
// Use a DID that is NOT in the admins table. let resp = app .router .clone() .oneshot(admin_get( "/admin/lexicons", common::auth::admin_cookie_header("did:plc:notadmin", &app.state.cookie_key), )) .await .unwrap();
assert_eq!(resp.status(), StatusCode::FORBIDDEN);}
#[tokio::test]#[serial]async fn admin_auto_bootstrap_first_user() { common::require_db!(); let app = TestApp::new().await; let backend = app.state.db_backend;
// Clear the seeded user so the table is empty. sqlx::query("DELETE FROM users") .execute(&app.state.db) .await .unwrap();
// Use a new DID via cookie auth. let bootstrap_did = "did:plc:bootstrap";
let resp = app .router .clone() .oneshot(admin_get( "/admin/lexicons", common::auth::admin_cookie_header(bootstrap_did, &app.state.cookie_key), )) .await .unwrap();
// The first user should be auto-bootstrapped as admin. assert_eq!(resp.status(), StatusCode::OK);
// Verify the DID was inserted. let sql = adapt_sql("SELECT COUNT(*) FROM users WHERE did = ?", backend); let count: (i64,) = sqlx::query_as(&sql) .bind(bootstrap_did) .fetch_one(&app.state.db) .await .unwrap(); assert_eq!(count.0, 1);}
// ---------------------------------------------------------------------------// Lexicon CRUD// ---------------------------------------------------------------------------
#[tokio::test]#[serial]async fn lexicon_create_returns_201() { common::require_db!(); let app = TestApp::new().await; let body = json!({ "lexicon_json": fixtures::game_record_lexicon(), "backfill": true });
let resp = app .router .clone() .oneshot(admin_post("/admin/lexicons", app.admin_cookie(), &body)) .await .unwrap();
assert_eq!(resp.status(), StatusCode::CREATED); let json = json_body(resp).await; assert_eq!(json["id"], "games.gamesgamesgamesgames.game"); assert_eq!(json["revision"], 1);}
#[tokio::test]#[serial]async fn lexicon_upsert_returns_200_with_incremented_revision() { common::require_db!(); let app = TestApp::new().await; let body = json!({ "lexicon_json": fixtures::game_record_lexicon(), "backfill": true });
// First create let resp = app .router .clone() .clone() .oneshot(admin_post("/admin/lexicons", app.admin_cookie(), &body)) .await .unwrap(); assert_eq!(resp.status(), StatusCode::CREATED);
// Upsert let resp = app .router .clone() .oneshot(admin_post("/admin/lexicons", app.admin_cookie(), &body)) .await .unwrap(); assert_eq!(resp.status(), StatusCode::OK); let json = json_body(resp).await; assert_eq!(json["revision"], 2);}
#[tokio::test]#[serial]async fn lexicon_invalid_version_returns_400() { common::require_db!(); let app = TestApp::new().await; let body = json!({ "lexicon_json": { "lexicon": 99, "id": "test.bad" }, });
let resp = app .router .clone() .oneshot(admin_post("/admin/lexicons", app.admin_cookie(), &body)) .await .unwrap();
assert_eq!(resp.status(), StatusCode::BAD_REQUEST);}
#[tokio::test]#[serial]async fn lexicon_missing_id_returns_400() { common::require_db!(); let app = TestApp::new().await; let body = json!({ "lexicon_json": { "lexicon": 1 }, });
let resp = app .router .clone() .oneshot(admin_post("/admin/lexicons", app.admin_cookie(), &body)) .await .unwrap();
assert_eq!(resp.status(), StatusCode::BAD_REQUEST);}
#[tokio::test]#[serial]async fn lexicon_list_all() { common::require_db!(); let app = TestApp::new().await;
// Seed a lexicon app.router .clone() .clone() .oneshot(admin_post( "/admin/lexicons", app.admin_cookie(), &json!({ "lexicon_json": fixtures::game_record_lexicon() }), )) .await .unwrap();
let resp = app .router .clone() .oneshot(admin_get("/admin/lexicons", app.admin_cookie())) .await .unwrap();
assert_eq!(resp.status(), StatusCode::OK); let json = json_body(resp).await; let arr = json.as_array().unwrap(); assert_eq!(arr.len(), 1); assert_eq!(arr[0]["id"], "games.gamesgamesgamesgames.game");}
#[tokio::test]#[serial]async fn lexicon_get_by_id() { common::require_db!(); let app = TestApp::new().await;
app.router .clone() .clone() .oneshot(admin_post( "/admin/lexicons", app.admin_cookie(), &json!({ "lexicon_json": fixtures::game_record_lexicon() }), )) .await .unwrap();
let resp = app .router .clone() .oneshot(admin_get( "/admin/lexicons/games.gamesgamesgamesgames.game", app.admin_cookie(), )) .await .unwrap();
assert_eq!(resp.status(), StatusCode::OK); let json = json_body(resp).await; assert_eq!(json["id"], "games.gamesgamesgamesgames.game");}
#[tokio::test]#[serial]async fn lexicon_get_not_found() { common::require_db!(); let app = TestApp::new().await;
let resp = app .router .clone() .oneshot(admin_get( "/admin/lexicons/nonexistent.lexicon", app.admin_cookie(), )) .await .unwrap();
assert_eq!(resp.status(), StatusCode::NOT_FOUND);}
#[tokio::test]#[serial]async fn lexicon_delete() { common::require_db!(); let app = TestApp::new().await;
app.router .clone() .clone() .oneshot(admin_post( "/admin/lexicons", app.admin_cookie(), &json!({ "lexicon_json": fixtures::game_record_lexicon() }), )) .await .unwrap();
let resp = app .router .clone() .oneshot(admin_delete( "/admin/lexicons/games.gamesgamesgamesgames.game", app.admin_cookie(), )) .await .unwrap();
assert_eq!(resp.status(), StatusCode::NO_CONTENT);}
#[tokio::test]#[serial]async fn lexicon_delete_not_found() { common::require_db!(); let app = TestApp::new().await;
let resp = app .router .clone() .oneshot(admin_delete( "/admin/lexicons/nonexistent.lexicon", app.admin_cookie(), )) .await .unwrap();
assert_eq!(resp.status(), StatusCode::NOT_FOUND);}
// ---------------------------------------------------------------------------// Stats// ---------------------------------------------------------------------------
#[tokio::test]#[serial]async fn stats_empty_db() { common::require_db!(); let app = TestApp::new().await;
let resp = app .router .clone() .oneshot(admin_get("/admin/stats", app.admin_cookie())) .await .unwrap();
assert_eq!(resp.status(), StatusCode::OK); let json = json_body(resp).await; assert_eq!(json["total_records"], 0); assert!(json["collections"].as_array().unwrap().is_empty());}
#[tokio::test]#[serial]async fn stats_with_seeded_records() { common::require_db!(); let app = TestApp::new().await; let backend = app.state.db_backend;
// Seed a lexicon so the stats query can join against it let lexicon_json_val = serde_json::json!({ "lexicon": 1, "id": "test.collection", "defs": { "main": { "type": "record", "key": "tid", "record": { "type": "object", "properties": {} } } } }); let now = now_rfc3339(); let sql = adapt_sql( "INSERT INTO lexicons (id, lexicon_json, created_at) VALUES (?, ?, ?)", backend, ); sqlx::query(&sql) .bind("test.collection") .bind(serde_json::to_string(&lexicon_json_val).unwrap_or_default()) .bind(&now) .execute(&app.state.db) .await .unwrap();
// Seed records directly let record_val = serde_json::json!({"title": "test"}); let sql = adapt_sql( "INSERT INTO records (uri, did, collection, rkey, record, cid, created_at) VALUES (?, ?, ?, ?, ?, ?, ?)", backend, ); sqlx::query(&sql) .bind("at://did:plc:test/test.collection/1") .bind("did:plc:test") .bind("test.collection") .bind("1") .bind(serde_json::to_string(&record_val).unwrap_or_default()) .bind("bafytest") .bind(&now) .execute(&app.state.db) .await .unwrap();
let resp = app .router .clone() .oneshot(admin_get("/admin/stats", app.admin_cookie())) .await .unwrap();
assert_eq!(resp.status(), StatusCode::OK); let json = json_body(resp).await; assert_eq!(json["total_records"], 1); assert_eq!(json["collections"][0]["collection"], "test.collection"); assert_eq!(json["collections"][0]["count"], 1);}
// ---------------------------------------------------------------------------// Backfill// ---------------------------------------------------------------------------
#[tokio::test]#[serial]async fn backfill_create_job() { common::require_db!(); let app = TestApp::new().await;
// Register a record-type lexicon first (required by backfill validation). let lexicon_body = json!({ "lexicon_json": fixtures::game_record_lexicon(), "backfill": true }); app.router .clone() .clone() .oneshot(admin_post( "/admin/lexicons", app.admin_cookie(), &lexicon_body, )) .await .unwrap();
let body = json!({ "collection": "games.gamesgamesgamesgames.game" });
let resp = app .router .clone() .oneshot(admin_post("/admin/backfill", app.admin_cookie(), &body)) .await .unwrap();
assert_eq!(resp.status(), StatusCode::CREATED); let json = json_body(resp).await; assert_eq!(json["status"], "running"); assert!(json.get("id").is_some());}
#[tokio::test]#[serial]async fn backfill_list_jobs() { common::require_db!(); let app = TestApp::new().await;
// Create a job first app.router .clone() .clone() .oneshot(admin_post( "/admin/backfill", app.admin_cookie(), &json!({}), )) .await .unwrap();
let resp = app .router .clone() .oneshot(admin_get("/admin/backfill/status", app.admin_cookie())) .await .unwrap();
assert_eq!(resp.status(), StatusCode::OK); let json = json_body(resp).await; assert_eq!(json.as_array().unwrap().len(), 1);}
#[tokio::test]#[serial]async fn backfill_cancel_running_job() { common::require_db!(); let app = TestApp::new().await; let backend = app.state.db_backend;
// Insert a running job directly so we don't need a real relay. let job_id = uuid::Uuid::new_v4().to_string(); let now = now_rfc3339(); let sql = adapt_sql( "INSERT INTO backfill_jobs (id, status, stage, started_at, created_at) VALUES (?, 'running', 'discovering_repos', ?, ?)", backend, ); sqlx::query(&sql) .bind(&job_id) .bind(&now) .bind(&now) .execute(&app.state.db) .await .unwrap();
let resp = app .router .clone() .oneshot(admin_post( &format!("/admin/backfill/{job_id}/cancel"), app.admin_cookie(), &json!({}), )) .await .unwrap();
assert_eq!(resp.status(), StatusCode::OK); let json = json_body(resp).await; assert_eq!(json["id"], job_id); assert_eq!(json["status"], "cancelling");}
#[tokio::test]#[serial]async fn backfill_cancel_already_cancelling_is_idempotent() { common::require_db!(); let app = TestApp::new().await; let backend = app.state.db_backend;
let job_id = uuid::Uuid::new_v4().to_string(); let now = now_rfc3339(); let sql = adapt_sql( "INSERT INTO backfill_jobs (id, status, stage, started_at, created_at) VALUES (?, 'cancelling', 'fetching_records', ?, ?)", backend, ); sqlx::query(&sql) .bind(&job_id) .bind(&now) .bind(&now) .execute(&app.state.db) .await .unwrap();
let resp = app .router .clone() .oneshot(admin_post( &format!("/admin/backfill/{job_id}/cancel"), app.admin_cookie(), &json!({}), )) .await .unwrap();
assert_eq!(resp.status(), StatusCode::OK); let json = json_body(resp).await; assert_eq!(json["status"], "cancelling");}
#[tokio::test]#[serial]async fn backfill_cancel_completed_returns_400() { common::require_db!(); let app = TestApp::new().await; let backend = app.state.db_backend;
let job_id = uuid::Uuid::new_v4().to_string(); let now = now_rfc3339(); let sql = adapt_sql( "INSERT INTO backfill_jobs (id, status, stage, completed_at, created_at) VALUES (?, 'completed', 'completed', ?, ?)", backend, ); sqlx::query(&sql) .bind(&job_id) .bind(&now) .bind(&now) .execute(&app.state.db) .await .unwrap();
let resp = app .router .clone() .oneshot(admin_post( &format!("/admin/backfill/{job_id}/cancel"), app.admin_cookie(), &json!({}), )) .await .unwrap();
assert_eq!(resp.status(), StatusCode::BAD_REQUEST);}
#[tokio::test]#[serial]async fn backfill_cancel_not_found_returns_404() { common::require_db!(); let app = TestApp::new().await;
let resp = app .router .clone() .oneshot(admin_post( "/admin/backfill/nonexistent-id/cancel", app.admin_cookie(), &json!({}), )) .await .unwrap();
assert_eq!(resp.status(), StatusCode::NOT_FOUND);}
// ---------------------------------------------------------------------------// Admin management// ---------------------------------------------------------------------------
#[tokio::test]#[serial]async fn admin_create_returns_did() { common::require_db!(); let app = TestApp::new().await; let body = json!({ "did": "did:plc:newadmin" });
let resp = app .router .clone() .oneshot(admin_post("/admin/users", app.admin_cookie(), &body)) .await .unwrap();
assert_eq!(resp.status(), StatusCode::CREATED); let json = json_body(resp).await; assert_eq!(json["did"], "did:plc:newadmin"); assert!(json.get("id").is_some());}
#[tokio::test]#[serial]async fn admin_created_did_authenticates() { common::require_db!(); let app = TestApp::new().await;
let new_did = "did:plc:newadmin2"; let body = json!({ "did": new_did });
// Create admin via the existing admin app.router .clone() .clone() .oneshot(admin_post("/admin/users", app.admin_cookie(), &body)) .await .unwrap();
// Use cookie auth with the new DID let resp = app .router .clone() .oneshot(admin_get( "/admin/lexicons", common::auth::admin_cookie_header(new_did, &app.state.cookie_key), )) .await .unwrap();
assert_eq!(resp.status(), StatusCode::OK);}
#[tokio::test]#[serial]async fn admin_list_returns_dids() { common::require_db!(); let app = TestApp::new().await;
let cookie = app.admin_cookie(); let resp = app .router .clone() .oneshot(admin_get("/admin/users", cookie)) .await .unwrap();
assert_eq!(resp.status(), StatusCode::OK); let json = json_body(resp).await; let admins = json.as_array().unwrap(); assert!(!admins.is_empty()); for admin in admins { assert!(admin.get("did").is_some()); assert!(admin.get("api_key").is_none()); assert!(admin.get("api_key_hash").is_none()); }}
#[tokio::test]#[serial]async fn admin_delete_returns_204() { common::require_db!(); let app = TestApp::new().await;
// Create an admin to delete let resp = app .router .clone() .clone() .oneshot(admin_post( "/admin/users", app.admin_cookie(), &json!({ "did": "did:plc:disposable" }), )) .await .unwrap(); let json = json_body(resp).await; let id = json["id"].as_str().unwrap();
let cookie = app.admin_cookie(); let resp = app .router .clone() .oneshot(admin_delete(&format!("/admin/users/{id}"), cookie)) .await .unwrap();
assert_eq!(resp.status(), StatusCode::NO_CONTENT);}
#[tokio::test]#[serial]async fn admin_delete_not_found() { common::require_db!(); let app = TestApp::new().await;
let cookie = app.admin_cookie(); let resp = app .router .clone() .oneshot(admin_delete( "/admin/users/00000000-0000-0000-0000-000000000000", cookie, )) .await .unwrap();
assert_eq!(resp.status(), StatusCode::NOT_FOUND);}