From 4414d4e0f64acd8a889d6e3f9a63b6cc9bbdb006 Mon Sep 17 00:00:00 2001 From: Lewis Date: Sun, 10 May 2026 20:36:05 +0000 Subject: [PATCH] test(xrpc): bulk endpoint & search-filter integration Lewis: May this revision serve well! --- crates/xrpc/tests/bulk.rs | 347 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ crates/xrpc/tests/search.rs | 146 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++- 2 file(s) changed, 492 insertion(s)(+), 1 deletion(s)(-) diff --git a/crates/xrpc/tests/bulk.rs b/crates/xrpc/tests/bulk.rs new file mode 100644 --- /dev/null +++ b/crates/xrpc/tests/bulk.rs @@ -0,0 +1,347 @@ +use std::sync::Arc; + +use axum::body::{Body, to_bytes}; +use bobbin_edge_index::{CoverageWatch, EdgeStore}; +use bobbin_knot_proxy::{KnotHttpConfig, KnotProxy, KnotProxyConfig}; +use bobbin_record_lru::{CacheCapacity, LruRecordStore}; +use bobbin_runtime::{RuntimeHasher, SystemClock}; +use bobbin_search::{DEFAULT_WRITER_HEAP_BYTES, SearchIndex, SearchReader}; +use bobbin_slingshot_client::SlingshotClient; +use bobbin_xrpc::{AppState, router}; +use http::{Request, StatusCode}; +use serde_json::{Value, json}; +use tower::ServiceExt; +use url::Url; +use url::form_urlencoded::byte_serialize; +use wiremock::matchers::{method, path, query_param}; +use wiremock::{Mock, MockServer, ResponseTemplate}; + +const CID: &str = "bafyreieqygohnz2zqyvtvktbjpvhutphobcmbsnt4q5lc36ri7vpcmoz4i"; + +struct Harness { + server: MockServer, + state: AppState, +} + +impl Harness { + async fn new() -> Self { + let server = MockServer::start().await; + let coverage = Arc::new(CoverageWatch::new()); + let state = AppState::new( + Arc::new(LruRecordStore::new(CacheCapacity::from_bytes(64 * 1024))), + SlingshotClient::with_default_http(Url::parse(&server.uri()).unwrap()).unwrap(), + Arc::new(EdgeStore::new(RuntimeHasher::default())), + coverage, + Arc::new( + KnotProxy::new( + KnotProxyConfig::default(), + KnotHttpConfig::default(), + Arc::new(SystemClock::new()), + RuntimeHasher::default(), + ) + .unwrap(), + ), + Arc::new( + SearchIndex::new(DEFAULT_WRITER_HEAP_BYTES, Arc::new(SystemClock::new())).unwrap(), + ) as Arc, + ); + Self { server, state } + } + + async fn mount(&self, did: &str, collection: &str, rkey: &str, value: Value) { + let uri = format!("at://{did}/{collection}/{rkey}"); + Mock::given(method("GET")) + .and(path("/xrpc/com.atproto.repo.getRecord")) + .and(query_param("repo", did)) + .and(query_param("collection", collection)) + .and(query_param("rkey", rkey)) + .respond_with(ResponseTemplate::new(200).set_body_json(json!({ + "uri": uri, + "cid": CID, + "value": value, + }))) + .mount(&self.server) + .await; + } + + async fn mount_404(&self, did: &str, collection: &str, rkey: &str) { + Mock::given(method("GET")) + .and(path("/xrpc/com.atproto.repo.getRecord")) + .and(query_param("repo", did)) + .and(query_param("collection", collection)) + .and(query_param("rkey", rkey)) + .respond_with(ResponseTemplate::new(404).set_body_json( + json!({"error": "RecordNotFound", "message": "missing"}), + )) + .mount(&self.server) + .await; + } +} + +fn enc(s: &str) -> String { + byte_serialize(s.as_bytes()).collect() +} + +fn bulk_request(endpoint: &str, key: &str, values: &[&str]) -> Request { + let qs = values + .iter() + .map(|v| format!("{key}={}", enc(v))) + .collect::>() + .join("&"); + Request::builder() + .uri(format!("/xrpc/{endpoint}?{qs}")) + .body(Body::empty()) + .unwrap() +} + +async fn json_response(resp: axum::response::Response) -> (StatusCode, Value) { + let status = resp.status(); + let bytes = to_bytes(resp.into_body(), 1 << 20).await.unwrap(); + let parsed: Value = serde_json::from_slice(&bytes).expect("JSON body"); + (status, parsed) +} + +fn issue_body(repo_did: &str, title: &str) -> Value { + json!({ + "$type": "sh.tangled.repo.issue", + "repoDid": repo_did, + "title": title, + "createdAt": "2026-05-01T00:00:00Z" + }) +} + +fn pull_body(target_repo: &str, title: &str) -> Value { + json!({ + "$type": "sh.tangled.repo.pull", + "title": title, + "createdAt": "2026-05-01T00:00:00Z", + "rounds": [], + "target": { + "branch": "main", + "repo": target_repo + } + }) +} + +fn repo_body(name: &str) -> Value { + json!({ + "$type": "sh.tangled.repo", + "name": name, + "knot": "oyster.cafe", + "createdAt": "2026-05-01T00:00:00Z" + }) +} + +fn profile_body(handle: &str) -> Value { + json!({ + "$type": "sh.tangled.actor.profile", + "bluesky": false, + "preferredHandle": handle + }) +} + +#[tokio::test] +async fn get_repos_returns_all_resolved_records() { + let h = Harness::new().await; + h.mount("did:plc:nel", "sh.tangled.repo", "abalone", repo_body("abalone")) + .await; + h.mount("did:plc:teq", "sh.tangled.repo", "limpet", repo_body("limpet")) + .await; + let app = router(h.state.clone()); + let (status, body) = json_response( + app.oneshot(bulk_request( + "sh.tangled.repo.getRepos", + "repos", + &[ + "at://did:plc:nel/sh.tangled.repo/abalone", + "at://did:plc:teq/sh.tangled.repo/limpet", + ], + )) + .await + .unwrap(), + ) + .await; + assert_eq!(status, StatusCode::OK); + let items = body["items"].as_array().unwrap(); + assert_eq!(items.len(), 2); + let names: Vec<&str> = items + .iter() + .map(|v| v["value"]["name"].as_str().unwrap()) + .collect(); + assert!(names.contains(&"abalone")); + assert!(names.contains(&"limpet")); + assert!(body["coverage"]["ready"].is_boolean()); +} + +#[tokio::test] +async fn get_profiles_returns_all_resolved_profiles() { + let h = Harness::new().await; + h.mount( + "did:plc:nel", + "sh.tangled.actor.profile", + "self", + profile_body("witchcraft.systems"), + ) + .await; + h.mount( + "did:plc:teq", + "sh.tangled.actor.profile", + "self", + profile_body("olaren.dev"), + ) + .await; + let app = router(h.state.clone()); + let (status, body) = json_response( + app.oneshot(bulk_request( + "sh.tangled.actor.getProfiles", + "actors", + &[ + "at://did:plc:nel/sh.tangled.actor.profile/self", + "at://did:plc:teq/sh.tangled.actor.profile/self", + ], + )) + .await + .unwrap(), + ) + .await; + assert_eq!(status, StatusCode::OK); + let items = body["items"].as_array().unwrap(); + assert_eq!(items.len(), 2); +} + +#[tokio::test] +async fn get_issues_returns_all_resolved_issues() { + let h = Harness::new().await; + let repo = "did:plc:abalone"; + h.mount( + "did:plc:nel", + "sh.tangled.repo.issue", + "i1", + issue_body(repo, "first"), + ) + .await; + h.mount( + "did:plc:olaren", + "sh.tangled.repo.issue", + "i2", + issue_body(repo, "second"), + ) + .await; + let app = router(h.state.clone()); + let (status, body) = json_response( + app.oneshot(bulk_request( + "sh.tangled.repo.getIssues", + "issues", + &[ + "at://did:plc:nel/sh.tangled.repo.issue/i1", + "at://did:plc:olaren/sh.tangled.repo.issue/i2", + ], + )) + .await + .unwrap(), + ) + .await; + assert_eq!(status, StatusCode::OK); + let items = body["items"].as_array().unwrap(); + assert_eq!(items.len(), 2); +} + +#[tokio::test] +async fn get_pulls_returns_all_resolved_pulls() { + let h = Harness::new().await; + let target = "at://did:plc:abalone/sh.tangled.repo/abalone"; + h.mount( + "did:plc:nel", + "sh.tangled.repo.pull", + "p1", + pull_body(target, "patch one"), + ) + .await; + let app = router(h.state.clone()); + let (status, body) = json_response( + app.oneshot(bulk_request( + "sh.tangled.repo.getPulls", + "pulls", + &["at://did:plc:nel/sh.tangled.repo.pull/p1"], + )) + .await + .unwrap(), + ) + .await; + assert_eq!(status, StatusCode::OK); + let items = body["items"].as_array().unwrap(); + assert_eq!(items.len(), 1); + assert_eq!(items[0]["value"]["title"], json!("patch one")); +} + +#[tokio::test] +async fn missing_records_are_dropped_silently() { + let h = Harness::new().await; + h.mount("did:plc:nel", "sh.tangled.repo", "abalone", repo_body("abalone")) + .await; + h.mount_404("did:plc:teq", "sh.tangled.repo", "ghost").await; + let app = router(h.state.clone()); + let (status, body) = json_response( + app.oneshot(bulk_request( + "sh.tangled.repo.getRepos", + "repos", + &[ + "at://did:plc:nel/sh.tangled.repo/abalone", + "at://did:plc:teq/sh.tangled.repo/ghost", + ], + )) + .await + .unwrap(), + ) + .await; + assert_eq!(status, StatusCode::OK); + let items = body["items"].as_array().unwrap(); + assert_eq!(items.len(), 1, "missing records must be dropped not fail the bulk call"); + assert_eq!(items[0]["value"]["name"], json!("abalone")); +} + +#[tokio::test] +async fn empty_uri_list_is_rejected() { + let h = Harness::new().await; + let app = router(h.state.clone()); + let resp = app + .oneshot( + Request::builder() + .uri("/xrpc/sh.tangled.repo.getRepos") + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::BAD_REQUEST); +} + +#[tokio::test] +async fn over_limit_uri_list_is_rejected() { + let h = Harness::new().await; + let uris: Vec = (0..51) + .map(|i| format!("at://did:plc:nel/sh.tangled.repo/r{i}")) + .collect(); + let refs: Vec<&str> = uris.iter().map(|s| s.as_str()).collect(); + let app = router(h.state.clone()); + let resp = app + .oneshot(bulk_request("sh.tangled.repo.getRepos", "repos", &refs)) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::BAD_REQUEST); +} + +#[tokio::test] +async fn malformed_uri_in_list_returns_400() { + let h = Harness::new().await; + let app = router(h.state.clone()); + let resp = app + .oneshot(bulk_request( + "sh.tangled.repo.getRepos", + "repos", + &["not-a-uri"], + )) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::BAD_REQUEST); +} diff --git a/crates/xrpc/tests/search.rs b/crates/xrpc/tests/search.rs --- a/crates/xrpc/tests/search.rs +++ b/crates/xrpc/tests/search.rs @@ -12,7 +12,7 @@ use http::{Request, StatusCode}; use jacquard_common::DefaultStr; use jacquard_common::types::nsid::Nsid; -use jacquard_common::types::string::AtUri; +use jacquard_common::types::string::{AtUri, Did}; use serde_json::{Value, json}; use tower::ServiceExt; use url::Url; @@ -73,6 +73,18 @@ } async fn index_issue(&self, did: &str, rkey: &str, title: &str, body: &str) { + self.index_issue_at(did, rkey, title, body, None, None).await; + } + + async fn index_issue_at( + &self, + did: &str, + rkey: &str, + title: &str, + body: &str, + created_at: Option, + repo: Option<&str>, + ) { let uri = format!("at://{did}/sh.tangled.repo.issue/{rkey}"); self.search .upsert(SearchDoc { @@ -80,6 +92,9 @@ nsid: nsid("sh.tangled.repo.issue"), title: title.to_owned(), body: body.to_owned(), + author: Some(Did::::new_owned(did).unwrap()), + created_at, + repo: repo.map(|d| Did::::new_owned(d).unwrap()), }) .await; self.search.flush().await; @@ -94,6 +109,9 @@ nsid: nsid("sh.tangled.string"), title: filename.to_owned(), body: contents.to_owned(), + author: None, + created_at: None, + repo: None, }) .await; self.search.flush().await; @@ -395,6 +413,9 @@ nsid: nsid("sh.tangled.repo.issue"), title: "kelp".to_owned(), body: "ocean".to_owned(), + author: None, + created_at: None, + repo: None, }) .await; h.search.flush().await; @@ -421,6 +442,9 @@ nsid: nsid("sh.tangled.repo.issue"), title: "abalone".to_owned(), body: "shell".to_owned(), + author: None, + created_at: None, + repo: None, }) .await; h.search.flush().await; @@ -451,6 +475,9 @@ nsid: nsid("sh.tangled.repo.issue"), title: "kelp".to_owned(), body: "ocean".to_owned(), + author: None, + created_at: None, + repo: None, }) .await; h.search.flush().await; @@ -481,3 +508,120 @@ .unwrap(); let _ = app.oneshot(search_request(&[("q", "kelp")])).await.unwrap(); } + +#[tokio::test] +async fn author_filter_narrows_to_matching_did() { + let h = Harness::new().await; + h.index_issue("did:plc:nel", "i1", "kelp tide", "") + .await; + h.index_issue("did:plc:teq", "i2", "kelp wave", "") + .await; + let app = router(h.state.clone()); + let resp = app + .oneshot(search_request(&[("q", "kelp"), ("author", "did:plc:nel")])) + .await + .unwrap(); + let (status, body) = json_response(resp).await; + assert_eq!(status, StatusCode::OK); + let hits = body["hits"].as_array().unwrap(); + assert_eq!(hits.len(), 1); + assert!( + hits[0]["uri"] + .as_str() + .unwrap() + .starts_with("at://did:plc:nel/") + ); +} + +#[tokio::test] +async fn since_until_window_filters_by_created_at() { + let h = Harness::new().await; + let early = 1_700_000_000; + let mid = 1_750_000_000; + let late = 1_800_000_000; + h.index_issue_at("did:plc:nel", "i1", "kelp early", "", Some(early), None) + .await; + h.index_issue_at("did:plc:nel", "i2", "kelp mid", "", Some(mid), None) + .await; + h.index_issue_at("did:plc:nel", "i3", "kelp late", "", Some(late), None) + .await; + let app = router(h.state.clone()); + let resp = app + .oneshot(search_request(&[ + ("q", "kelp"), + ("since", "2025-01-01T00:00:00Z"), + ("until", "2027-01-01T00:00:00Z"), + ])) + .await + .unwrap(); + let (status, body) = json_response(resp).await; + assert_eq!(status, StatusCode::OK); + let hits = body["hits"].as_array().unwrap(); + assert_eq!(hits.len(), 1, "only the mid record falls in [2025, 2027)"); + assert_eq!( + hits[0]["uri"].as_str().unwrap(), + "at://did:plc:nel/sh.tangled.repo.issue/i2" + ); +} + +#[tokio::test] +async fn repo_filter_scopes_to_owning_repo() { + let h = Harness::new().await; + let abalone = "did:plc:abalone"; + let limpet = "did:plc:limpet"; + h.index_issue_at("did:plc:nel", "i1", "kelp one", "", None, Some(abalone)) + .await; + h.index_issue_at("did:plc:teq", "i2", "kelp two", "", None, Some(limpet)) + .await; + let app = router(h.state.clone()); + let resp = app + .oneshot(search_request(&[("q", "kelp"), ("repo", abalone)])) + .await + .unwrap(); + let (status, body) = json_response(resp).await; + assert_eq!(status, StatusCode::OK); + let hits = body["hits"].as_array().unwrap(); + assert_eq!(hits.len(), 1); + assert_eq!( + hits[0]["uri"].as_str().unwrap(), + "at://did:plc:nel/sh.tangled.repo.issue/i1" + ); +} + +#[tokio::test] +async fn invalid_author_did_returns_400() { + let h = Harness::new().await; + let app = router(h.state.clone()); + let resp = app + .oneshot(search_request(&[("q", "kelp"), ("author", "not-a-did")])) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::BAD_REQUEST); +} + +#[tokio::test] +async fn invalid_since_returns_400() { + let h = Harness::new().await; + let app = router(h.state.clone()); + let resp = app + .oneshot(search_request(&[("q", "kelp"), ("since", "yesterday")])) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::BAD_REQUEST); +} + +#[tokio::test] +async fn since_after_until_returns_400() { + let h = Harness::new().await; + let app = router(h.state.clone()); + let resp = app + .oneshot(search_request(&[ + ("q", "kelp"), + ("since", "2027-01-01T00:00:00Z"), + ("until", "2025-01-01T00:00:00Z"), + ])) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::BAD_REQUEST); +} + -- tangled.sh