crates/knut-graphdb/tests/live_falkordb.rs
//! **Live FalkorDB gates for the folded `GraphDb`.**
//!
//! These drive a REAL FalkorDB. They are `#[ignore]`d so a container-free `cargo
//! test` stays green, and run explicitly:
//!
//! ```text
//! podman run -d --name knut-graphdb-live -p 6379:6379 falkordb/falkordb:latest
//! KNUT_FALKOR_ENDPOINT=127.0.0.1:6379 \
//! cargo test -p knut-graphdb --test live_falkordb -- --ignored --nocapture
//! ```
//!
//! Every graph they create is namespaced by pid and dropped afterwards, so they can
//! run beside another lane's data on the same endpoint.
use std::time::Duration;
use knut_graphdb::{FalkorBackend, GraphDb, GraphError, UpsertEdge, UpsertNode};
fn endpoint() -> String {
std::env::var("KNUT_FALKOR_ENDPOINT").unwrap_or_else(|_| "127.0.0.1:6379".to_string())
}
fn rt() -> tokio::runtime::Runtime {
tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.unwrap()
}
fn db() -> FalkorBackend {
FalkorBackend::new(endpoint()).with_timeout(Duration::from_secs(20))
}
/// Delete a graph key so a re-run starts clean, and so this lane leaves nothing
/// behind on a container other lanes share. Best effort.
async fn drop_graph(db: &FalkorBackend, name: &str) {
let _ = db.delete_graph(name).await;
}
/// **GATE 2 — the fold loses nothing.** Every behaviour korp's `GraphStore` had off
/// a real FalkorDB still works, now from knut: list the graphs, open one for its
/// catalogue, read its nodes and edges.
///
/// RED when broken: point `graph_names` at a dead port and the list is an `Err`, not
/// an empty `Ok` — asserted below, so a "reachable and empty" fold cannot pass.
#[test]
#[ignore = "needs a live FalkorDB; set KNUT_FALKOR_ENDPOINT"]
fn list_open_read_against_a_real_falkordb() {
let db = db();
let g = format!("knut_fold_{}", std::process::id());
rt().block_on(async {
drop_graph(&db, &g).await;
// Populate through the trait's own upsert — no second client anywhere.
let written = db
.upsert(
&g,
&[
UpsertNode::new(1, "Person"),
UpsertNode::new(2, "Company").with_key("5560000001"),
UpsertNode::new(3, "Vessel").with_key("265547890"),
],
&[UpsertEdge::new(1, 2, "WORKS_AT"), UpsertEdge::new(1, 3, "OWNS")],
)
.await
.expect("upsert against a live FalkorDB");
assert!(written >= 5, "wrote 3 nodes + 2 edges, got {written}");
// 1. LIST — the graph is on the endpoint.
let names = db.graph_names().await.expect("GRAPH.LIST");
assert!(
names.iter().any(|n| n == &g),
"the graph we just wrote must be listed; got {names:?}"
);
// 2. OPEN — a real catalogue, not an empty schema.
let schema = db.open(&g).await.expect("open");
assert_eq!(schema.graph, g);
assert!(
schema.labels.contains(&"Person".to_string())
&& schema.labels.contains(&"Company".to_string()),
"labels came back from db.labels(): {:?}",
schema.labels
);
assert!(
schema
.relationship_types
.contains(&"WORKS_AT".to_string()),
"relationship types: {:?}",
schema.relationship_types
);
assert_eq!(schema.node_count, 3, "MATCH (n) RETURN count(n)");
assert_eq!(schema.edge_count, 2, "MATCH ()-[r]->() RETURN count(r)");
// 3. READ — the scene itself, with the caller's cap honoured.
let scene = db.read(&g, 300).await.expect("read");
assert_eq!(scene.edges.len(), 2, "both edges came back");
assert_eq!(scene.nodes.len(), 3, "three distinct nodes");
let cap1 = db.read(&g, 1).await.expect("read capped at 1");
assert_eq!(
cap1.edges.len(),
1,
"the cap the CALLER asked for is the cap that reached the engine — a \
constant query text with LIMIT baked in would return 2 here"
);
// 4. NEIGHBOURS — expansion off a live node id.
let pid = scene
.nodes
.iter()
.find(|n| n.label == "Person")
.expect("a Person node")
.id;
let nb = db.neighbours(&g, pid, 100).await.expect("neighbours");
assert_eq!(nb.edges.len(), 2, "the Person's two 1-hop links");
drop_graph(&db, &g).await;
});
}
/// **GATE 3 — `open()` LISTS BEFORE IT PROBES.**
///
/// FalkorDB creates a graph key on the first `GRAPH.QUERY` against an unknown name.
/// So an `open` that probed first would make the probe itself the reason the answer
/// is "yes", and would leave a stray empty graph behind.
///
/// RED, seen against the live container on 2026-09-01: a probe planted before the
/// `graph_names` check made `open("knut_absent_2283864")` return
/// `Ok(GraphSchema { node_count: 0, edge_count: 0, .. })` — the "blank pane that was
/// really a 404" shape — and `GRAPH.LIST` afterwards carried `knut_absent_2283864`,
/// a graph nobody asked for. Both the `expect_err` and the stray-key assertion catch
/// it; the stray-key one is kept because a backend that refused correctly while still
/// creating the key would pass the first alone.
#[test]
#[ignore = "needs a live FalkorDB; set KNUT_FALKOR_ENDPOINT"]
fn open_lists_before_it_probes() {
let db = db();
let absent = format!("knut_absent_{}", std::process::id());
rt().block_on(async {
let before = db.graph_names().await.expect("GRAPH.LIST");
assert!(
!before.iter().any(|n| n == &absent),
"the name must not exist before the test: {before:?}"
);
let err = db.open(&absent).await.expect_err("an absent graph must not open");
match &err {
GraphError::NotFound { graph, .. } => assert_eq!(graph, &absent),
other => panic!("an absent graph is NotFound, got {other}"),
}
let after = db.graph_names().await.expect("GRAPH.LIST");
assert!(
!after.iter().any(|n| n == &absent),
"the refused open must NOT have created '{absent}' — FalkorDB creates a \
graph key on the first GRAPH.QUERY, so probing before listing leaves a \
stray empty graph. GRAPH.LIST after: {after:?}"
);
// NOTE: no `before.len() == after.len()` assertion. Another test in this
// binary creates and drops its own graph concurrently, so the endpoint's
// total is not stable — and a count is a weaker statement than the membership
// check above, which is the one that goes red.
for name in &before {
assert!(
after.contains(name),
"a refused open must not remove an existing graph ('{name}' vanished)"
);
}
});
}
/// A dead endpoint is an `Err`, never `Ok(vec![])` — the bug this seam exists to
/// stop. Port 1 on loopback has nothing listening, so no container is needed.
#[test]
fn a_dead_endpoint_is_a_typed_error_never_an_empty_list() {
let db = FalkorBackend::new("127.0.0.1:1").with_timeout(Duration::from_secs(5));
let out = rt().block_on(db.graph_names());
match out {
Ok(names) => panic!(
"a dead endpoint returned Ok({names:?}) — an unreachable FalkorDB must never \
look like an empty graph list"
),
Err(e) => assert!(
matches!(e, GraphError::Connect(_)),
"a refused dial is Connect, distinguishable from an empty list: {e}"
),
}
// The same distinction on open and read.
let e = rt().block_on(db.open("anything")).expect_err("open on a dead endpoint errors");
assert!(matches!(e, GraphError::Connect(_)), "{e}");
let e = match rt().block_on(db.read("anything", 10)) {
Ok(s) => panic!(
"read on a dead endpoint returned Ok({} nodes / {} edges) — an unreachable \
FalkorDB must never look like an empty scene",
s.nodes.len(),
s.edges.len()
),
Err(e) => e,
};
assert!(matches!(e, GraphError::Connect(_)), "{e}");
}
/// The endpoint string is normalised without downgrading TLS, and the backend never
/// prints its identity bytes.
#[test]
fn endpoint_is_normalised_without_downgrading_tls() {
use knut_graphdb::falkor::normalise_endpoint;
assert_eq!(normalise_endpoint("127.0.0.1:6379"), "redis://127.0.0.1:6379");
assert_eq!(normalise_endpoint(" falkor.internal:6379 "), "redis://falkor.internal:6379");
assert_eq!(normalise_endpoint("redis://a:6379"), "redis://a:6379");
assert_eq!(normalise_endpoint("rediss://a:6379"), "rediss://a:6379");
let db = FalkorBackend::new("127.0.0.1:6379");
assert_eq!(db.endpoint(), "redis://127.0.0.1:6379");
assert!(!db.is_mtls());
assert_eq!(db.kind(), "falkordb");
let with_id = FalkorBackend::new("rediss://a:6379").with_identity(graphar_flight::FalkorTls {
client_cert_pem: b"SECRETCERT".to_vec(),
client_key_pem: b"SECRETKEY".to_vec(),
ca_pem: None,
});
assert!(with_id.is_mtls());
let printed = format!("{with_id:?}");
assert!(
!printed.contains("SECRET"),
"the identity bytes must never reach a log line: {printed}"
);
assert!(printed.contains("mtls: true"), "{printed}");
}