use arrow_array::{Array, StringArray};
use arrow_ipc::reader::StreamReader;
use axum::body::{to_bytes, Body};
use axum::http::{Request, StatusCode};
use axum::Router;
use core_api::{
json_to_value, Explanation, FkSkip, IngestReport, Predicate, PredicateSummary, RuleDef,
RuleStats, SharedDb, Stats, Value,
};
use serde_json::{json, Value as Json};
#[cfg(feature = "embed-ui")]
use server::router_with_embedded_ui;
use server::{router, router_with_auth, router_with_ui, serve};
use std::io::Cursor;
use std::path::PathBuf;
use tower::ServiceExt;
fn tmp(name: &str) -> PathBuf {
let d = std::env::temp_dir().join(format!("graphdb-http-{}-{}", name, std::process::id()));
let _ = std::fs::remove_dir_all(&d);
d
}
fn open(name: &str) -> (Router, SharedDb) {
let db = SharedDb::open(&tmp(name)).unwrap();
(router(db.clone()), db)
}
async fn send(app: Router, req: Request<Body>) -> (StatusCode, Vec<u8>, Option<String>) {
let res = app.oneshot(req).await.unwrap();
let status = res.status();
let ctype = res
.headers()
.get(axum::http::header::CONTENT_TYPE)
.and_then(|v| v.to_str().ok())
.map(str::to_string);
let body = to_bytes(res.into_body(), usize::MAX)
.await
.unwrap()
.to_vec();
(status, body, ctype)
}
fn json_req(method: &str, uri: &str, body: Json) -> Request<Body> {
Request::builder()
.method(method)
.uri(uri)
.header(axum::http::header::CONTENT_TYPE, "application/json")
.body(Body::from(body.to_string()))
.unwrap()
}
fn get(uri: &str) -> Request<Body> {
Request::builder()
.method("GET")
.uri(uri)
.body(Body::empty())
.unwrap()
}
fn seed_person(db: &SharedDb, key: &str) {
db.write()
.insert_node("Person", key, vec![("id".into(), Value::Str(key.into()))])
.unwrap();
}
fn parse_json(bytes: &[u8]) -> Json {
serde_json::from_slice(bytes)
.unwrap_or_else(|e| panic!("json: {e}: {}", String::from_utf8_lossy(bytes)))
}
#[tokio::test]
async fn health_is_unauthenticated() {
let db = SharedDb::open(&tmp("health-unauth")).unwrap();
let app = router_with_auth(db, Some("t".into()));
let (status, body, _) = send(app, get("/health")).await;
assert_eq!(status, StatusCode::OK);
let v = parse_json(&body);
assert_eq!(v["ok"], json!(true));
assert_eq!(v["nodes"], json!(0));
assert_eq!(v["edges"], json!(0));
assert_eq!(v["addr"], json!("127.0.0.1:8080"));
}
#[tokio::test]
async fn health_reports_counts() {
let (app, db) = open("health-counts");
seed_person(&db, "a");
seed_person(&db, "b");
db.write().insert_edge("KNOWS", "a", "b").unwrap();
let (status, body, _) = send(app, get("/health")).await;
assert_eq!(status, StatusCode::OK);
let v = parse_json(&body);
assert_eq!(v["ok"], json!(true));
assert_eq!(v["nodes"], json!(2));
assert_eq!(v["edges"], json!(1));
assert_eq!(v["addr"], json!("127.0.0.1:8080"));
}
#[tokio::test]
async fn query_without_bearer_is_401_when_token_configured() {
let db = SharedDb::open(&tmp("query-no-bearer")).unwrap();
let app = router_with_auth(db, Some("t".into()));
let (status, body, _) = send(
app,
json_req("POST", "/query", json!({"cypher": "MATCH (n) RETURN n"})),
)
.await;
assert_eq!(status, StatusCode::UNAUTHORIZED);
let v = parse_json(&body);
assert!(
v["error"].as_str().is_some_and(|s| !s.is_empty()),
"401 body must be {{\"error\":\"...\"}}, got {v}"
);
}
#[tokio::test]
async fn query_with_bearer_succeeds_when_token_configured() {
let db = SharedDb::open(&tmp("query-bearer")).unwrap();
let app = router_with_auth(db, Some("t".into()));
let req = Request::builder()
.method("POST")
.uri("/query?format=json")
.header(axum::http::header::CONTENT_TYPE, "application/json")
.header(axum::http::header::AUTHORIZATION, "Bearer t")
.body(Body::from(
json!({"cypher": "MATCH (n) RETURN n"}).to_string(),
))
.unwrap();
let (status, _, _) = send(app, req).await;
assert_eq!(status, StatusCode::OK);
}
#[tokio::test]
async fn query_with_query_token_succeeds_when_token_configured() {
let db = SharedDb::open(&tmp("query-qs-token")).unwrap();
let app = router_with_auth(db, Some("t".into()));
let (status, _, _) = send(
app,
json_req(
"POST",
"/query?token=t&format=json",
json!({"cypher": "MATCH (n) RETURN n"}),
),
)
.await;
assert_eq!(status, StatusCode::OK);
}
#[tokio::test]
async fn query_with_wrong_query_token_is_401() {
let db = SharedDb::open(&tmp("query-qs-wrong")).unwrap();
let app = router_with_auth(db, Some("t".into()));
let (status, body, _) = send(
app,
json_req(
"POST",
"/query?token=wrong",
json!({"cypher": "MATCH (n) RETURN n"}),
),
)
.await;
assert_eq!(status, StatusCode::UNAUTHORIZED);
let v = parse_json(&body);
assert!(
v["error"].as_str().is_some_and(|s| !s.is_empty()),
"401 body must be {{\"error\":\"...\"}}, got {v}"
);
}
#[tokio::test]
async fn watch_with_query_token_is_not_401_when_token_configured() {
let db = SharedDb::open(&tmp("watch-qs-token")).unwrap();
let app = router_with_auth(db, Some("t".into()));
let (status, _, _) = send(app, get("/watch?token=t")).await;
assert_ne!(
status,
StatusCode::UNAUTHORIZED,
"GET /watch?token=t must pass auth (upgrade may still fail without WS headers)"
);
}
#[tokio::test]
async fn query_token_percent_decoded_matches_configured_token() {
let db = SharedDb::open(&tmp("query-qs-encoded-slash")).unwrap();
let app = router_with_auth(db, Some("a/b".into()));
let (status, _, _) = send(
app,
json_req(
"POST",
"/query?token=a%2Fb&format=json",
json!({"cypher": "MATCH (n) RETURN n"}),
),
)
.await;
assert_eq!(
status,
StatusCode::OK,
"configured token \"a/b\" must match URL-encoded ?token=a%2Fb"
);
let db = SharedDb::open(&tmp("query-qs-encoded-plus")).unwrap();
let app = router_with_auth(db, Some("a+b".into()));
let (status, _, _) = send(
app,
json_req(
"POST",
"/query?token=a%2Bb&format=json",
json!({"cypher": "MATCH (n) RETURN n"}),
),
)
.await;
assert_eq!(
status,
StatusCode::OK,
"configured token \"a+b\" must match URL-encoded ?token=a%2Bb"
);
}
#[tokio::test]
async fn watch_with_url_encoded_query_token_is_not_401() {
let db = SharedDb::open(&tmp("watch-qs-encoded")).unwrap();
let app = router_with_auth(db, Some("a/b".into()));
let (status, _, _) = send(app, get("/watch?token=a%2Fb")).await;
assert_ne!(
status,
StatusCode::UNAUTHORIZED,
"GET /watch?token=a%2Fb must pass auth for configured token \"a/b\""
);
}
async fn send_headers(
app: Router,
req: Request<Body>,
) -> (StatusCode, Vec<u8>, axum::http::HeaderMap) {
let res = app.oneshot(req).await.unwrap();
let status = res.status();
let headers = res.headers().clone();
let body = to_bytes(res.into_body(), usize::MAX)
.await
.unwrap()
.to_vec();
(status, body, headers)
}
#[tokio::test]
async fn html_query_token_sets_auth_cookie() {
let ui = tmp("ui-cookie");
std::fs::create_dir_all(&ui).unwrap();
std::fs::write(
ui.join("index.html"),
"<!doctype html><title>graph-db</title>",
)
.unwrap();
let db = SharedDb::open(&tmp("ui-cookie-db")).unwrap();
let app = router_with_ui(db, &ui, Some("t".into()));
let (status, _, headers) = send_headers(app, get("/?token=t")).await;
assert_eq!(status, StatusCode::OK);
let cookie = headers
.get(axum::http::header::SET_COOKIE)
.and_then(|v| v.to_str().ok())
.unwrap_or("");
assert!(
cookie.contains("mushroomdb_token=t"),
"Set-Cookie must include mushroomdb_token=, got {cookie:?}"
);
assert!(
cookie.contains("Path=/"),
"Set-Cookie Path=/, got {cookie:?}"
);
assert!(
cookie.contains("SameSite=Lax"),
"Set-Cookie SameSite=Lax, got {cookie:?}"
);
assert!(
cookie.contains("HttpOnly"),
"Set-Cookie HttpOnly, got {cookie:?}"
);
}
#[tokio::test]
async fn missing_asset_with_cookie_is_404_not_401() {
let db = SharedDb::open(&tmp("asset-cookie")).unwrap();
let app = router_with_auth(db, Some("t".into()));
let req = Request::builder()
.method("GET")
.uri("/no-such.js")
.header(axum::http::header::COOKIE, "mushroomdb_token=t")
.body(Body::empty())
.unwrap();
let (status, _, _) = send(app, req).await;
assert_eq!(status, StatusCode::NOT_FOUND);
}
#[tokio::test]
async fn missing_asset_without_auth_is_401() {
let db = SharedDb::open(&tmp("asset-noauth")).unwrap();
let app = router_with_auth(db, Some("t".into()));
let (status, _, _) = send(app, get("/no-such.js")).await;
assert_eq!(status, StatusCode::UNAUTHORIZED);
}
#[tokio::test]
async fn stats_with_cookie_succeeds_when_token_configured() {
let db = SharedDb::open(&tmp("stats-cookie")).unwrap();
let app = router_with_auth(db, Some("t".into()));
let req = Request::builder()
.method("GET")
.uri("/stats")
.header(axum::http::header::COOKIE, "mushroomdb_token=t")
.body(Body::empty())
.unwrap();
let (status, body, _) = send(app, req).await;
assert_eq!(status, StatusCode::OK);
let v = parse_json(&body);
assert!(v.get("nodes_live").is_some(), "/stats JSON, got {v}");
}
#[tokio::test]
async fn query_default_returns_arrow_ipc() {
let (app, db) = open("query-arrow");
seed_person(&db, "p1");
let (status, body, ctype) = send(
app,
json_req(
"POST",
"/query",
json!({"cypher": "MATCH (t:Person {id: $tid}) RETURN t", "params": {"tid": "p1"}}),
),
)
.await;
assert_eq!(status, StatusCode::OK);
assert_eq!(
ctype.as_deref(),
Some("application/vnd.apache.arrow.stream")
);
let mut reader = StreamReader::try_new(Cursor::new(body), None).unwrap();
let batch = reader.next().expect("one batch").unwrap();
assert!(reader.next().is_none(), "single batch stream");
assert_eq!(batch.num_rows(), 1);
assert_eq!(batch.schema().field(0).name(), "t");
let col = batch
.column(0)
.as_any()
.downcast_ref::<StringArray>()
.unwrap();
assert_eq!(col.value(0), "p1");
}
#[tokio::test]
async fn query_format_json_matches_columns_and_rows() {
let (app, db) = open("query-json");
seed_person(&db, "p1");
let (status, body, ctype) = send(
app,
json_req(
"POST",
"/query?format=json",
json!({
"cypher": "MATCH (t:Person {id: $tid}) RETURN t",
"params": {"tid": "p1"}
}),
),
)
.await;
assert_eq!(status, StatusCode::OK);
assert_eq!(ctype.as_deref(), Some("application/json"));
let v = parse_json(&body);
assert_eq!(v["columns"], json!(["t"]));
assert_eq!(v["rows"], json!([["p1"]]));
}
#[tokio::test]
async fn query_count_star_format_json_wire_shape() {
let (app, db) = open("query-count-star");
seed_person(&db, "p1");
seed_person(&db, "p2");
seed_person(&db, "p3");
let (status, body, ctype) = send(
app,
json_req(
"POST",
"/query?format=json",
json!({"cypher": "MATCH (p:Person) RETURN COUNT(*)"}),
),
)
.await;
assert_eq!(status, StatusCode::OK);
assert_eq!(ctype.as_deref(), Some("application/json"));
let v = parse_json(&body);
assert_eq!(
v["columns"],
json!(["COUNT(*)"]),
"column name must be COUNT(*)"
);
assert_eq!(
v["rows"],
json!([[3]]),
"three Person nodes must be counted"
);
}
#[tokio::test]
async fn query_bad_cypher_is_400_with_parse_prefix() {
let (app, _) = open("query-bad");
let (status, body, _) = send(
app,
json_req("POST", "/query", json!({"cypher": "MATCH (n)"})),
)
.await;
assert_eq!(status, StatusCode::BAD_REQUEST);
let v = parse_json(&body);
let err = v["error"].as_str().expect("error string");
assert!(
err.starts_with("parse:"),
"expected parse:-prefixed detail, got {err}"
);
}
#[tokio::test]
async fn stats_round_trips_serialize() {
let (app, db) = open("stats");
seed_person(&db, "p1");
let (status, body, ctype) = send(app, get("/stats")).await;
assert_eq!(status, StatusCode::OK);
assert_eq!(ctype.as_deref(), Some("application/json"));
let v = parse_json(&body);
assert_eq!(v["nodes_live"], json!(1));
assert_eq!(v["nodes_tombstoned"], json!(0));
assert_eq!(v["edges"], json!(0));
assert_eq!(v["rules"], json!([]));
let live = Stats {
nodes_live: 1,
nodes_tombstoned: 2,
edges: 3,
rules: vec![RuleStats {
name: "r".into(),
edges: 4,
tripped: true,
fires: 5,
approximate: false,
}],
};
let encoded = serde_json::to_value(&live).expect("Stats: Serialize");
assert_eq!(encoded["nodes_live"], json!(1));
assert_eq!(encoded["nodes_tombstoned"], json!(2));
assert_eq!(encoded["edges"], json!(3));
assert_eq!(encoded["rules"][0]["name"], json!("r"));
assert_eq!(encoded["rules"][0]["edges"], json!(4));
assert_eq!(encoded["rules"][0]["tripped"], json!(true));
assert_eq!(encoded["rules"][0]["fires"], json!(5));
assert_eq!(encoded["rules"][0]["approximate"], json!(false));
}
#[tokio::test]
async fn ingest_happy_path() {
let (app, db) = open("ingest-ok");
db.write().insert_node("Org", "acme", vec![]).unwrap();
let (status, body, ctype) = send(
app,
json_req(
"POST",
"/ingest",
json!({
"label": "Person",
"rows": [{"id": "p1", "org_id": "acme", "name": "ada"}],
"options": {}
}),
),
)
.await;
assert_eq!(status, StatusCode::OK);
assert_eq!(ctype.as_deref(), Some("application/json"));
let v = parse_json(&body);
assert_eq!(v["inserted"], json!(1));
assert_eq!(v["row_errors"], json!([]));
assert_eq!(v["rules_created"], json!(["auto_fk_person_org_id"]));
assert_eq!(v["skipped_fk_fields"], json!([]));
assert!(db.read().has_node("p1"));
}
#[tokio::test]
async fn ingest_many_rows_is_one_wal_commit() {
let dir = tmp("ingest-one-wal");
let db = SharedDb::open(&dir).unwrap();
let app = router(db.clone());
let (status, _, _) = send(
app,
json_req(
"POST",
"/ingest",
json!({
"label": "Person",
"rows": [{"id": "a"}, {"id": "b"}, {"id": "c"}],
"options": {"auto_fk": "off"}
}),
),
)
.await;
assert_eq!(status, StatusCode::OK);
assert_eq!(core_api::wal_commit_count_at(&dir).unwrap(), 1);
assert_eq!(db.read().node_count(), 3);
}
#[tokio::test]
async fn ingest_shape_error_is_400() {
let (app, _) = open("ingest-shape");
let (status, body, _) = send(
app,
json_req(
"POST",
"/ingest",
json!({"label": "Person", "rows": {"id": "p1"}}),
),
)
.await;
assert_eq!(status, StatusCode::BAD_REQUEST);
let v = parse_json(&body);
let err = v["error"].as_str().expect("error string");
assert!(
err.contains("array of objects"),
"expected ingest shape detail, got {err}"
);
}
#[tokio::test]
async fn explain_happy_path() {
let (app, db) = open("explain-ok");
{
let mut w = db.write();
w.insert_node("Org", "o1", vec![]).unwrap();
w.create_rule(RuleDef {
name: "works_at".into(),
src_label: "Person".into(),
dst_label: "Org".into(),
predicate: Predicate::KeyMatch {
field: "org_id".into(),
},
edge_type: "WORKS_AT".into(),
weight_prop: None,
max_edges: None,
approximate: false,
via_label: None,
via_edge: None,
via_dir: None,
})
.unwrap();
w.insert_node(
"Person",
"p1",
vec![("org_id".into(), Value::Str("o1".into()))],
)
.unwrap();
}
let (status, body, ctype) = send(app, get("/explain?a=p1&b=o1")).await;
assert_eq!(status, StatusCode::OK);
assert_eq!(ctype.as_deref(), Some("application/json"));
let v = parse_json(&body);
assert!(v.is_array());
assert_eq!(v[0]["rule"], json!("works_at"));
assert_eq!(v[0]["edge_type"], json!("WORKS_AT"));
assert_eq!(v[0]["src_key"], json!("p1"));
assert_eq!(v[0]["dst_key"], json!("o1"));
assert_eq!(v[0]["weight"], Json::Null);
assert_eq!(v[0]["predicate"]["kind"], json!("key_match"));
assert_eq!(v[0]["predicate"]["fields"], json!(["org_id"]));
let pred = v[0]["predicate"].as_object().expect("predicate object");
for key in ["min", "tolerance", "km", "parts"] {
assert!(
pred.contains_key(key),
"leaf JSON must present {key} as null, not omit it"
);
assert_eq!(pred[key], Json::Null);
}
}
#[tokio::test]
async fn explain_predicate_all_json_shape() {
let (app, db) = open("explain-all");
{
let mut w = db.write();
w.insert_node(
"Org",
"o1",
vec![
("ind".into(), Value::Str("arch".into())),
("tags".into(), Value::List(vec![Value::Str("x".into())])),
],
)
.unwrap();
w.create_rule(RuleDef {
name: "both".into(),
src_label: "Person".into(),
dst_label: "Org".into(),
predicate: Predicate::All(vec![
Predicate::FieldEqual {
field: "ind".into(),
},
Predicate::Overlap {
field: "tags".into(),
min: 0.5,
},
]),
edge_type: "BOTH".into(),
weight_prop: Some("score".into()),
max_edges: None,
approximate: false,
via_label: None,
via_edge: None,
via_dir: None,
})
.unwrap();
w.insert_node(
"Person",
"p1",
vec![
("ind".into(), Value::Str("arch".into())),
("tags".into(), Value::List(vec![Value::Str("x".into())])),
],
)
.unwrap();
}
let (status, body, _) = send(app, get("/explain?a=p1&b=o1")).await;
assert_eq!(status, StatusCode::OK);
let v = parse_json(&body);
assert_eq!(v[0]["predicate"]["kind"], json!("all"));
assert_eq!(v[0]["predicate"]["fields"], json!(["ind", "tags"]));
assert_eq!(v[0]["predicate"]["parts"][0]["kind"], json!("field_equal"));
assert_eq!(v[0]["predicate"]["parts"][1]["kind"], json!("overlap"));
assert_eq!(v[0]["predicate"]["parts"][1]["min"], json!(0.5));
}
#[tokio::test]
async fn explain_unknown_key_is_400() {
let (app, db) = open("explain-miss");
seed_person(&db, "p1");
let (status, body, _) = send(app, get("/explain?a=p1&b=ghost")).await;
assert_eq!(status, StatusCode::BAD_REQUEST);
let v = parse_json(&body);
let err = v["error"].as_str().expect("error string");
assert!(
err.contains("ghost"),
"expected unknown key in detail, got {err}"
);
}
#[tokio::test]
async fn neighborhood_depth_and_dir() {
let (app, db) = open("nbhd");
{
let mut w = db.write();
w.insert_node("Person", "a", vec![]).unwrap();
w.insert_node("Person", "b", vec![]).unwrap();
w.insert_node("Person", "c", vec![]).unwrap();
w.insert_edge("KNOWS", "a", "b").unwrap();
w.insert_edge("KNOWS", "b", "c").unwrap();
}
let (status, body, ctype) =
send(app.clone(), get("/node/a/neighborhood?depth=1&dir=out")).await;
assert_eq!(status, StatusCode::OK);
assert_eq!(ctype.as_deref(), Some("application/json"));
let hop1 = parse_json(&body);
assert_eq!(hop1["columns"], json!(["key", "label", "depth"]));
assert_eq!(hop1["rows"], json!([["b", "Person", 1]]));
let (_, body, _) = send(app.clone(), get("/node/a/neighborhood?depth=2&dir=out")).await;
let hop2 = parse_json(&body);
assert_eq!(
hop2["rows"],
json!([["b", "Person", 1], ["c", "Person", 2]])
);
let (_, body, _) = send(app, get("/node/b/neighborhood?depth=1&dir=in")).await;
let incoming = parse_json(&body);
assert_eq!(incoming["rows"], json!([["a", "Person", 1]]));
}
#[tokio::test]
async fn neighborhood_edge_types_filter() {
let (app, db) = open("nbhd-etypes");
{
let mut w = db.write();
w.insert_node("Person", "a", vec![]).unwrap();
w.insert_node("Person", "b", vec![]).unwrap();
w.insert_node("Person", "c", vec![]).unwrap();
w.insert_edge("KNOWS", "a", "b").unwrap();
w.insert_edge("LIKES", "a", "c").unwrap();
}
let (status, body, _) = send(app.clone(), get("/node/a/neighborhood?depth=1&dir=out")).await;
assert_eq!(status, StatusCode::OK);
let unfiltered = parse_json(&body);
assert_eq!(
unfiltered["rows"],
json!([["b", "Person", 1], ["c", "Person", 1]])
);
let (_, body, _) = send(
app.clone(),
get("/node/a/neighborhood?depth=1&dir=out&edge_types=KNOWS"),
)
.await;
let knows = parse_json(&body);
assert_eq!(knows["rows"], json!([["b", "Person", 1]]));
let (_, body, _) = send(
app,
get("/node/a/neighborhood?depth=1&dir=out&edge_types=KNOWS,LIKES"),
)
.await;
let both = parse_json(&body);
assert_eq!(
both["rows"],
json!([["b", "Person", 1], ["c", "Person", 1]])
);
}
#[tokio::test]
async fn node_info_json_shape() {
let (app, db) = open("node-info");
{
let mut w = db.write();
w.insert_node(
"Person",
"p1",
vec![
("years".into(), Value::Int(8)),
("name".into(), Value::Str("ada".into())),
("ok".into(), Value::Bool(true)),
("rating".into(), Value::Float(0.5)),
(
"tags".into(),
Value::List(vec![Value::Str("x".into()), Value::Str("y".into())]),
),
],
)
.unwrap();
}
let (status, body, ctype) = send(app, get("/node/p1")).await;
assert_eq!(status, StatusCode::OK);
assert_eq!(ctype.as_deref(), Some("application/json"));
let v = parse_json(&body);
assert_eq!(v["key"], json!("p1"));
assert_eq!(v["label"], json!("Person"));
assert_eq!(v["props"]["name"], json!("ada"));
assert_eq!(v["props"]["ok"], json!(true));
assert_eq!(v["props"]["years"], json!(8));
assert_eq!(v["props"]["rating"], json!(0.5));
assert_eq!(v["props"]["tags"], json!(["x", "y"]));
let keys: Vec<&str> = v["props"]
.as_object()
.expect("props object")
.keys()
.map(String::as_str)
.collect();
assert_eq!(keys, vec!["name", "ok", "rating", "tags", "years"]);
}
#[tokio::test]
async fn node_info_unknown_key_is_404() {
let (app, _) = open("node-info-miss");
let (status, body, _) = send(app, get("/node/ghost")).await;
assert_eq!(status, StatusCode::NOT_FOUND);
let v = parse_json(&body);
assert_eq!(v, json!({"error": "node key not found: ghost"}));
}
#[tokio::test]
async fn node_edges_json_shape_user_and_derived() {
let (app, db) = open("node-edges");
{
let mut w = db.write();
w.insert_node("Org", "acme", vec![]).unwrap();
w.create_rule(core_api::RuleDef {
name: "works_at".into(),
src_label: "Person".into(),
dst_label: "Org".into(),
predicate: core_api::Predicate::KeyMatch {
field: "org_id".into(),
},
edge_type: "WORKS_AT".into(),
weight_prop: None,
max_edges: None,
approximate: false,
via_label: None,
via_edge: None,
via_dir: None,
})
.unwrap();
w.insert_node(
"Person",
"p1",
vec![("org_id".into(), Value::Str("acme".into()))],
)
.unwrap();
w.insert_node("Person", "p2", vec![]).unwrap();
w.insert_edge("KNOWS", "p1", "p2").unwrap();
}
let (status, body, ctype) = send(app, get("/node/p1/edges")).await;
assert_eq!(status, StatusCode::OK);
assert_eq!(ctype.as_deref(), Some("application/json"));
let v = parse_json(&body);
assert_eq!(
v,
json!({
"edges": [
{
"edge_type": "KNOWS",
"src_key": "p1",
"dst_key": "p2",
"derived": false
},
{
"edge_type": "WORKS_AT",
"src_key": "p1",
"dst_key": "acme",
"derived": true
}
]
})
);
}
#[tokio::test]
async fn node_edges_unknown_key_is_404() {
let (app, _) = open("node-edges-miss");
let (status, body, _) = send(app, get("/node/ghost/edges")).await;
assert_eq!(status, StatusCode::NOT_FOUND);
let v = parse_json(&body);
assert_eq!(v, json!({"error": "node key not found: ghost"}));
}
#[tokio::test]
async fn neighborhood_unknown_key_is_400() {
let (app, _) = open("nbhd-miss");
let (status, body, _) = send(app, get("/node/ghost/neighborhood")).await;
assert_eq!(status, StatusCode::BAD_REQUEST);
let v = parse_json(&body);
let err = v["error"].as_str().expect("error string");
assert!(
err.contains("ghost"),
"expected unknown key in detail, got {err}"
);
}
#[tokio::test]
async fn ui_fallback_serves_static_and_stats_stays_json() {
let ui = tmp("ui-dist");
std::fs::create_dir_all(&ui).unwrap();
std::fs::write(
ui.join("index.html"),
"<!doctype html><title>graph-db</title>",
)
.unwrap();
std::fs::write(ui.join("hello.txt"), "hello-static").unwrap();
let db = SharedDb::open(&tmp("ui-api")).unwrap();
let app = router_with_ui(db, &ui, None);
let (st, body, _) = send(app.clone(), get("/hello.txt")).await;
assert_eq!(st, StatusCode::OK);
assert_eq!(body, b"hello-static");
let (st, body, _) = send(app.clone(), get("/")).await;
assert_eq!(st, StatusCode::OK);
let html = String::from_utf8_lossy(&body);
assert!(
html.contains("graph-db"),
"GET / should serve index.html, got {html}"
);
let (st, body, ctype) = send(app, get("/stats")).await;
assert_eq!(st, StatusCode::OK);
let ctype = ctype.expect("stats content-type");
assert!(ctype.contains("json"), "/stats must stay JSON, got {ctype}");
let j = parse_json(&body);
assert!(
j.get("nodes_live").is_some(),
"/stats JSON must include nodes_live, got {j}"
);
}
#[tokio::test]
async fn serve_readiness_returns_local_addr() {
let db = SharedDb::open(&tmp("serve")).unwrap();
let (tx, rx) = tokio::sync::oneshot::channel();
let handle = tokio::spawn(async move {
serve(db, "127.0.0.1:0".parse().unwrap(), tx, None)
.await
.unwrap();
});
let addr = rx.await.expect("readiness");
assert_ne!(addr.port(), 0, "ephemeral port must be resolved");
handle.abort();
}
#[tokio::test]
async fn ingest_edges_inserts_user_edge() {
let (app, db) = open("ingest-edges");
db.write().insert_node("Person", "a", vec![]).unwrap();
db.write().insert_node("Person", "b", vec![]).unwrap();
let (status, body, _) = send(
app,
json_req(
"POST",
"/ingest",
json!({
"label": "Person",
"rows": [],
"edges": [{"edge_type": "KNOWS", "src": "a", "dst": "b"}]
}),
),
)
.await;
assert_eq!(status, StatusCode::OK);
let v = parse_json(&body);
assert_eq!(v["edges_inserted"], json!(1));
let edges = db.read().node_edges("a").unwrap();
assert!(
edges.iter().any(|e| {
e.edge_type == "KNOWS" && e.src_key == "a" && e.dst_key == "b" && !e.derived
}),
"expected user KNOWS a→b, got {edges:?}"
);
}
#[tokio::test]
async fn ingest_edges_unknown_endpoint_is_400() {
let (app, _) = open("ingest-edge-miss");
let (status, body, _) = send(
app,
json_req(
"POST",
"/ingest",
json!({
"label": "Person",
"rows": [],
"edges": [{"edge_type": "KNOWS", "src": "ghost", "dst": "ghost"}]
}),
),
)
.await;
assert_eq!(status, StatusCode::BAD_REQUEST);
let v = parse_json(&body);
let err = v["error"].as_str().unwrap();
assert!(
err.contains("node key not found"),
"expected KeyNotFound register, got {err}"
);
}
#[tokio::test]
async fn ingest_bad_edge_is_atomic() {
let (app, db) = open("ingest-atomic");
let (status, body, _) = send(
app,
json_req(
"POST",
"/ingest",
json!({
"label": "Person",
"rows": [{"id": "newbie"}],
"edges": [{"edge_type": "KNOWS", "src": "newbie", "dst": "ghost"}]
}),
),
)
.await;
assert_eq!(status, StatusCode::BAD_REQUEST);
let v = parse_json(&body);
let err = v["error"].as_str().unwrap();
assert!(
err.contains("node key not found"),
"preview error, got {err}"
);
assert!(
!db.read().has_node("newbie"),
"newbie must not persist after a rejected mixed batch"
);
}
#[tokio::test]
async fn ingest_duplicate_edge_counts_zero() {
let (app, db) = open("ingest-dup-edge");
{
let mut w = db.write();
w.insert_node("Person", "a", vec![]).unwrap();
w.insert_node("Person", "b", vec![]).unwrap();
w.insert_edge("KNOWS", "a", "b").unwrap();
}
let (status, body, _) = send(
app,
json_req(
"POST",
"/ingest",
json!({
"label": "Person",
"rows": [],
"edges": [{"edge_type": "KNOWS", "src": "a", "dst": "b"}]
}),
),
)
.await;
assert_eq!(status, StatusCode::OK);
let v = parse_json(&body);
assert_eq!(v["edges_inserted"], json!(0));
}
#[tokio::test]
async fn create_rule_http_and_validation() {
let (app, db) = open("rules-post");
db.write()
.insert_node("Org", "o1", vec![("founded_year".into(), Value::Int(2010))])
.unwrap();
db.write()
.insert_node("Org", "o2", vec![("founded_year".into(), Value::Int(2011))])
.unwrap();
let (status, body, _) = send(
app.clone(),
json_req(
"POST",
"/rules",
json!({
"name": "founded_within",
"src_label": "Org",
"dst_label": "Org",
"predicate": {"NumericWithin": {"field": "founded_year", "tolerance": 2.0}},
"edge_type": "FOUNDED_WITHIN",
"weight_prop": "score",
"max_edges": null
}),
),
)
.await;
assert_eq!(status, StatusCode::OK);
assert_eq!(
parse_json(&body),
json!({"ok": true, "name": "founded_within"})
);
assert!(db.read().rules().iter().any(|r| r.name == "founded_within"));
assert_eq!(
db.read()
.rules()
.iter()
.find(|r| r.name == "founded_within")
.unwrap()
.max_edges,
Some(32),
"JSON null max_edges fills default scored top-k"
);
let (status, body, _) = send(
app,
json_req(
"POST",
"/rules",
json!({
"name": "",
"src_label": "Org",
"dst_label": "Org",
"predicate": {"NumericWithin": {"field": "founded_year", "tolerance": 2.0}},
"edge_type": "FOUNDED_WITHIN",
"weight_prop": "score",
"max_edges": null
}),
),
)
.await;
assert_eq!(status, StatusCode::BAD_REQUEST);
let v = parse_json(&body);
let err = v["error"].as_str().unwrap();
assert!(
err.contains("invalid rule:"),
"engine message verbatim, got {err}"
);
}
#[tokio::test]
async fn create_rule_http_omitted_max_edges_fills_default() {
let (app, db) = open("rules-omit-max");
db.write().insert_node("Org", "o1", vec![]).unwrap();
db.write()
.insert_node(
"Person",
"p1",
vec![("org_id".into(), Value::Str("o1".into()))],
)
.unwrap();
let (status, _, _) = send(
app.clone(),
json_req(
"POST",
"/rules",
json!({
"name": "works_at",
"src_label": "Person",
"dst_label": "Org",
"predicate": {"KeyMatch": {"field": "org_id"}},
"edge_type": "WORKS_AT",
"weight_prop": null
}),
),
)
.await;
assert_eq!(status, StatusCode::OK);
assert_eq!(
db.read()
.rules()
.iter()
.find(|r| r.name == "works_at")
.unwrap()
.max_edges,
Some(1)
);
}
#[test]
fn wire_types_serialize() {
let stats = Stats {
nodes_live: 0,
nodes_tombstoned: 0,
edges: 0,
rules: vec![],
};
serde_json::to_value(&stats).unwrap();
serde_json::to_value(&RuleStats {
name: "n".into(),
edges: 0,
tripped: false,
fires: 0,
approximate: false,
})
.unwrap();
serde_json::to_value(&IngestReport {
inserted: 0,
row_errors: vec![],
rules_created: vec![],
skipped_fk_fields: vec![FkSkip {
field: "org_id".into(),
reason: "no matching target keys".into(),
}],
edges_inserted: 0,
})
.unwrap();
let expl = serde_json::to_value(&Explanation {
rule: "r".into(),
edge_type: "E".into(),
src_key: "a".into(),
dst_key: "b".into(),
weight: Some(0.5),
predicate: PredicateSummary::from(&Predicate::KeyMatch { field: "fk".into() }),
})
.unwrap();
assert_eq!(expl["predicate"]["kind"], json!("key_match"));
assert_eq!(expl["predicate"]["fields"], json!(["fk"]));
let pred = expl["predicate"].as_object().expect("predicate object");
for key in ["min", "tolerance", "km", "parts"] {
assert!(pred.contains_key(key), "wire summary must include {key}");
assert_eq!(pred[key], Json::Null);
}
assert_eq!(
json_to_value(json!("p1")),
Some(Value::Str("p1".into())),
"params reuse json_to_value"
);
}
#[tokio::test]
async fn cypher_write_over_http_is_durable() {
let dir = tmp("cypher-write-http");
let db = SharedDb::open(&dir).unwrap();
let app = router(db.clone());
let (status, body, ctype) = send(
app.clone(),
json_req(
"POST",
"/query?format=json",
json!({"cypher": "CREATE (n:Person {id: 'alice'})"}),
),
)
.await;
assert_eq!(status, StatusCode::OK, "write must succeed");
assert_eq!(ctype.as_deref(), Some("application/json"));
let v = parse_json(&body);
assert_eq!(
v["columns"],
json!(["created", "properties_set", "deleted"])
);
let rows = v["rows"].as_array().expect("rows array");
assert_eq!(rows.len(), 1);
assert_eq!(rows[0][0], json!(1), "created=1");
assert_eq!(rows[0][1], json!(0), "properties_set=0");
assert_eq!(rows[0][2], json!(0), "deleted=0");
assert!(
db.read().has_node("alice"),
"node must be queryable immediately"
);
drop(app);
drop(db);
let db2 = SharedDb::open(&dir).unwrap();
assert!(
db2.read().has_node("alice"),
"node must survive DB re-open (WAL fsynced before HTTP response)"
);
}
#[cfg(feature = "embed-ui")]
#[tokio::test]
async fn embedded_ui_serves_index_and_stats_wins() {
let db = SharedDb::open(&tmp("embed-ui")).unwrap();
let app = router_with_embedded_ui(db);
let (st, body, ctype) = send(app.clone(), get("/")).await;
assert_eq!(st, StatusCode::OK);
let html = String::from_utf8_lossy(&body);
assert!(
html.contains("mushroomdb") || html.contains("<!doctype") || html.contains("<!DOCTYPE"),
"embedded GET / should be index.html, got {html}"
);
let ctype = ctype.unwrap_or_default();
assert!(
ctype.contains("html"),
"index content-type html, got {ctype}"
);
let (st, body, ctype) = send(app, get("/stats")).await;
assert_eq!(st, StatusCode::OK);
let ctype = ctype.expect("stats content-type");
assert!(ctype.contains("json"), "/stats must stay JSON, got {ctype}");
let j = parse_json(&body);
assert!(j.get("nodes_live").is_some(), "/stats JSON, got {j}");
}
#[tokio::test]
async fn http_params_read_round_trip() {
let (app, db) = open("http-params-read");
{
let mut w = db.write();
w.insert_node("HP", "alice", vec![("age".into(), Value::Int(30))])
.unwrap();
w.insert_node("HP", "bob", vec![("age".into(), Value::Int(25))])
.unwrap();
}
let (status, body, _) = send(
app,
json_req(
"POST",
"/query?format=json",
json!({
"cypher": "MATCH (n:HP) WHERE n.age = $age RETURN n",
"params": {"age": 30}
}),
),
)
.await;
assert_eq!(status, StatusCode::OK);
let v = parse_json(&body);
let rows = v["rows"].as_array().expect("rows array");
assert_eq!(rows.len(), 1, "must match exactly the node with age=30");
let row_str = rows[0].to_string();
assert!(
row_str.contains("alice"),
"returned node must be alice: {row_str}"
);
}
#[tokio::test]
async fn http_params_injection_safe() {
let (app, db) = open("http-params-injection");
{
let mut w = db.write();
w.insert_node("HPI", "real_node", vec![]).unwrap();
}
let (status, body, _) = send(
app,
json_req(
"POST",
"/query?format=json",
json!({
"cypher": "MATCH (n:HPI {id: $id}) RETURN n",
"params": {"id": "' RETURN 1//"}
}),
),
)
.await;
assert_eq!(status, StatusCode::OK);
let v = parse_json(&body);
let rows = v["rows"].as_array().expect("rows array");
assert_eq!(
rows.len(),
0,
"injection payload must not return rows: {rows:?}"
);
}
#[tokio::test]
async fn http_params_write_set_is_durable() {
let dir = tmp("http-params-write");
let db = SharedDb::open(&dir).unwrap();
let app = router(db.clone());
{
let mut w = db.write();
w.insert_node("HPW", "target", vec![("score".into(), Value::Int(0))])
.unwrap();
}
let (status, body, _) = send(
app.clone(),
json_req(
"POST",
"/query?format=json",
json!({
"cypher": "MATCH (n:HPW) WHERE n.score = 0 SET n.score = $newval",
"params": {"newval": 99}
}),
),
)
.await;
assert_eq!(
status,
StatusCode::OK,
"write must succeed: {}",
String::from_utf8_lossy(&body)
);
let v = parse_json(&body);
assert_eq!(v["rows"][0][1], json!(1), "properties_set must be 1");
drop(app);
drop(db);
let db2 = SharedDb::open(&dir).unwrap();
let rs = db2
.read()
.query(
"MATCH (n:HPW) RETURN n.score",
&std::collections::BTreeMap::new(),
)
.unwrap();
assert_eq!(rs.len(), 1);
assert_eq!(
rs.get(0, "n.score"),
Some(&Value::Int(99)),
"score must be 99 after re-open"
);
}
#[tokio::test]
async fn query_mask_filters_nodes() {
let (app, db) = open("mask-http");
db.write().insert_node("P", "alice", vec![]).unwrap();
db.write().insert_node("P", "bob", vec![]).unwrap();
db.write().insert_node("P", "carol", vec![]).unwrap();
let req = json_req(
"POST",
"/query?format=json",
json!({"cypher": "MATCH (n:P) RETURN n.id", "mask": ["alice", "bob"]}),
);
let (status, body, _) = send(app.clone(), req).await;
assert_eq!(status, StatusCode::OK);
let v = parse_json(&body);
assert_eq!(v["rows"].as_array().unwrap().len(), 2);
let req = json_req(
"POST",
"/query?format=json",
json!({"cypher": "CREATE (n:P {id: 'evil'})", "mask": ["alice"]}),
);
let (status, body, _) = send(app.clone(), req).await;
assert_eq!(
status,
StatusCode::BAD_REQUEST,
"masked write must be 400: {}",
String::from_utf8_lossy(&body)
);
let req = json_req(
"POST",
"/query?format=json",
json!({"cypher": "MATCH (n:P) RETURN n.id"}),
);
let (status, body, _) = send(app, req).await;
assert_eq!(status, StatusCode::OK);
let v = parse_json(&body);
assert_eq!(v["rows"].as_array().unwrap().len(), 3);
}