crates/nornir-release-ledger/src/testkit.rs
//! Test-only support for the warehouse-SEAM ledger core (feature `testkit`).
//!
//! The ledger (`release_changes` / `release_events` / `undo`) reasons over the
//! engine-agnostic [`nornir_warehouse_trait::Warehouse`] trait, never a concrete
//! `IcebergWarehouse` (a warehouse-free core may not name one — it would invert
//! the dep graph). So the round-trip tests need a stand-in: [`MockWarehouse`], a
//! tiny in-memory `HashMap<table, Vec<RecordBatch>>` that honours only the Arrow
//! core the ledger drives — `append_arrow`/`scan_arrow` and their async twins.
//! Every other trait method is `unimplemented!()` (the ledger never calls them).
//!
//! [`block_on`] is a std-only single-future executor (no tokio dep in the ledger
//! build): the mock's async methods complete without ever yielding, so a poll-loop
//! drives `append_release_events(..).await` to completion.
//!
//! Hosted in the leaf rather than in a consumer so there is ONE test double for
//! the seam — a second hand-written mock is how a seam re-forks.
use std::collections::HashMap;
use std::sync::Arc;
use std::sync::Mutex;
use anyhow::Result;
use arrow::array::{Array, Int32Array, RecordBatch, StringArray, TimestampMicrosecondArray};
use arrow::datatypes::{DataType, Field, Schema, TimeUnit};
use uuid::Uuid;
use nornir_bench::BenchRun;
use nornir_warehouse_trait::{
BenchFilter, CallEdgeRow, ColumnInfo, GitHeatScan, ScanFilter, SymbolRow, TablePreview,
Warehouse,
};
use crate::schema::{TABLE_CALL_EDGES, TABLE_SYMBOL_FACTS};
/// In-memory [`Warehouse`] test double: stores appended batches per table name and
/// replays them on scan. Recasts NOTHING (unlike the Iceberg backend) — the batch
/// is built against the canonical schema and read back as-is, which is exactly what
/// the ledger's positional downcasts expect.
#[derive(Default)]
pub struct MockWarehouse {
tables: Mutex<HashMap<String, Vec<RecordBatch>>>,
}
impl MockWarehouse {
pub fn new() -> Self {
Self::default()
}
/// Seed one knowledge snapshot for `repo` by appending real `symbol_facts` +
/// `call_edges` Arrow batches (unique `snapshot_id`) — the shape nornir's
/// deleted syn scan wrote, i.e. what a warehouse made BEFORE Mímir v2 still
/// holds — so the relocated `load_latest` discovers this snapshot and decodes
/// it back. The batches carry only the columns the reader reads, in the
/// schema order it downcasts by name.
pub fn seed_symbol_scan(&self, repo: &str, symbols: &[SymbolRow], calls: &[CallEdgeRow]) {
let snap = Uuid::new_v4().to_string();
let ts = chrono::Utc::now().timestamp_micros();
if !symbols.is_empty() {
let n = symbols.len();
let schema = Arc::new(Schema::new(vec![
Field::new("snapshot_id", DataType::Utf8, false),
Field::new("ts_micros", DataType::Timestamp(TimeUnit::Microsecond, None), false),
Field::new("repo", DataType::Utf8, false),
Field::new("crate_name", DataType::Utf8, false),
Field::new("module_path", DataType::Utf8, false),
Field::new("item_kind", DataType::Utf8, false),
Field::new("item_name", DataType::Utf8, false),
Field::new("visibility", DataType::Utf8, false),
Field::new("file", DataType::Utf8, false),
Field::new("line", DataType::Int32, false),
Field::new("doc_lines", DataType::Int32, false),
Field::new("signature", DataType::Utf8, true),
]));
let cols: Vec<Arc<dyn Array>> = vec![
Arc::new(StringArray::from(vec![snap.clone(); n])),
Arc::new(TimestampMicrosecondArray::from(vec![ts; n])),
Arc::new(StringArray::from(vec![repo.to_string(); n])),
Arc::new(StringArray::from(symbols.iter().map(|r| r.crate_name.clone()).collect::<Vec<_>>())),
Arc::new(StringArray::from(symbols.iter().map(|r| r.module_path.clone()).collect::<Vec<_>>())),
Arc::new(StringArray::from(symbols.iter().map(|r| r.item_kind.clone()).collect::<Vec<_>>())),
Arc::new(StringArray::from(symbols.iter().map(|r| r.item_name.clone()).collect::<Vec<_>>())),
Arc::new(StringArray::from(symbols.iter().map(|r| r.visibility.clone()).collect::<Vec<_>>())),
Arc::new(StringArray::from(symbols.iter().map(|r| r.file.clone()).collect::<Vec<_>>())),
Arc::new(Int32Array::from(symbols.iter().map(|r| r.line as i32).collect::<Vec<_>>())),
Arc::new(Int32Array::from(symbols.iter().map(|r| r.doc_lines as i32).collect::<Vec<_>>())),
Arc::new(StringArray::from(symbols.iter().map(|r| r.signature.clone()).collect::<Vec<_>>())),
];
self.append_arrow(TABLE_SYMBOL_FACTS, RecordBatch::try_new(schema, cols).unwrap()).unwrap();
}
if !calls.is_empty() {
let n = calls.len();
let schema = Arc::new(Schema::new(vec![
Field::new("snapshot_id", DataType::Utf8, false),
Field::new("ts_micros", DataType::Timestamp(TimeUnit::Microsecond, None), false),
Field::new("repo", DataType::Utf8, false),
Field::new("crate_name", DataType::Utf8, false),
Field::new("caller_path", DataType::Utf8, false),
Field::new("callee_ident", DataType::Utf8, false),
Field::new("call_kind", DataType::Utf8, false),
Field::new("file", DataType::Utf8, false),
Field::new("line", DataType::Int32, false),
]));
let cols: Vec<Arc<dyn Array>> = vec![
Arc::new(StringArray::from(vec![snap.clone(); n])),
Arc::new(TimestampMicrosecondArray::from(vec![ts; n])),
Arc::new(StringArray::from(vec![repo.to_string(); n])),
Arc::new(StringArray::from(calls.iter().map(|r| r.crate_name.clone()).collect::<Vec<_>>())),
Arc::new(StringArray::from(calls.iter().map(|r| r.caller_path.clone()).collect::<Vec<_>>())),
Arc::new(StringArray::from(calls.iter().map(|r| r.callee_ident.clone()).collect::<Vec<_>>())),
Arc::new(StringArray::from(calls.iter().map(|r| r.call_kind.clone()).collect::<Vec<_>>())),
Arc::new(StringArray::from(calls.iter().map(|r| r.file.clone()).collect::<Vec<_>>())),
Arc::new(Int32Array::from(calls.iter().map(|r| r.line as i32).collect::<Vec<_>>())),
];
self.append_arrow(TABLE_CALL_EDGES, RecordBatch::try_new(schema, cols).unwrap()).unwrap();
}
}
}
#[async_trait::async_trait]
impl Warehouse for MockWarehouse {
fn append_arrow(&self, table: &str, batch: RecordBatch) -> Result<()> {
self.tables.lock().unwrap().entry(table.to_string()).or_default().push(batch);
Ok(())
}
fn scan_arrow(&self, table: &str) -> Result<Vec<RecordBatch>> {
Ok(self.tables.lock().unwrap().get(table).cloned().unwrap_or_default())
}
fn scan_filtered(
&self,
table: &str,
_filter: &ScanFilter,
_columns: &[&str],
) -> Result<Vec<RecordBatch>> {
// Replay ALL stored batches for the table (no file/row-group pruning). The
// knowledge readers keep RESIDUAL per-row guards (`repo == …`,
// `snapshot_id == …`), so an unfiltered replay yields byte-identical
// results — the pushdown is a scan optimization, not a correctness gate.
Ok(self.tables.lock().unwrap().get(table).cloned().unwrap_or_default())
}
fn scan_limited(&self, _table: &str, _max_rows: usize) -> Result<Vec<RecordBatch>> {
unimplemented!("MockWarehouse: scan_limited not needed by the ledger tests")
}
fn ensure_table(
&self,
_table: &str,
_schema: ::iceberg::spec::Schema,
_partition_cols: &[&str],
) -> Result<()> {
Ok(())
}
fn ensure_columns(&self, _table: &str, _canonical: &::iceberg::spec::Schema) -> Result<()> {
Ok(())
}
fn table_names(&self) -> Result<Vec<String>> {
let mut names: Vec<String> = self.tables.lock().unwrap().keys().cloned().collect();
names.sort();
Ok(names)
}
fn scan_preview(&self, _table: &str, _limit: usize) -> Result<TablePreview> {
unimplemented!("MockWarehouse: scan_preview not needed by the ledger tests")
}
/// `unimplemented!` on purpose: a mock that answered `Ok(VectorScan::default())`
/// would be indistinguishable from a warehouse holding no embeddings, and that
/// exact confusion is what the typed vector route was added to end.
fn scan_vectors(
&self,
_table: &str,
_key_col: &str,
_vec_col: &str,
_limit: usize,
) -> Result<nornir_warehouse_trait::VectorScan> {
unimplemented!("MockWarehouse: scan_vectors not needed by the ledger tests")
}
fn describe_columns(&self, _table: &str) -> Result<Vec<ColumnInfo>> {
unimplemented!("MockWarehouse: describe_columns not needed by the ledger tests")
}
fn append_bench_run(&self, _repo: &str, _run: &BenchRun) -> Result<Uuid> {
unimplemented!("MockWarehouse: append_bench_run not needed by the ledger tests")
}
fn query_bench_runs(&self, _filter: &BenchFilter) -> Result<Vec<BenchRun>> {
unimplemented!("MockWarehouse: query_bench_runs not needed by the ledger tests")
}
async fn append_arrow_async(&self, table: &str, batch: RecordBatch) -> Result<()> {
self.append_arrow(table, batch)
}
async fn scan_arrow_async(&self, table: &str) -> Result<Vec<RecordBatch>> {
self.scan_arrow(table)
}
async fn append_bench_run_async(&self, _repo: &str, _run: &BenchRun) -> Result<Uuid> {
unimplemented!("MockWarehouse: append_bench_run_async not needed by the ledger tests")
}
async fn query_bench_runs_async(&self, _filter: &BenchFilter) -> Result<Vec<BenchRun>> {
unimplemented!("MockWarehouse: query_bench_runs_async not needed by the ledger tests")
}
async fn append_git_heat_scan_async(&self, _scan: &GitHeatScan) -> Result<()> {
unimplemented!("MockWarehouse: append_git_heat_scan_async not needed by the ledger tests")
}
fn as_any(&self) -> &dyn std::any::Any {
self
}
}
/// Minimal std-only executor: poll one future to completion. The ledger's async
/// path over [`MockWarehouse`] never yields (the mock's async methods return
/// immediately-ready), so a poll-loop with a no-op waker terminates on the first
/// `Ready`. Keeps the `release` test build free of a tokio dependency.
pub fn block_on<F: std::future::Future>(fut: F) -> F::Output {
use std::pin::pin;
use std::task::{Context, Poll, RawWaker, RawWakerVTable, Waker};
fn raw() -> RawWaker {
fn clone(_: *const ()) -> RawWaker {
raw()
}
fn noop(_: *const ()) {}
static V: RawWakerVTable = RawWakerVTable::new(clone, noop, noop, noop);
RawWaker::new(std::ptr::null(), &V)
}
let waker = unsafe { Waker::from_raw(raw()) };
let mut cx = Context::from_waker(&waker);
let mut fut = pin!(fut);
loop {
match fut.as_mut().poll(&mut cx) {
Poll::Ready(v) => return v,
Poll::Pending => std::hint::spin_loop(),
}
}
}