rest/tests/rest_commit.rs
//! End-to-end exercise of the shipped REST front's `commit_table` (update)
//! endpoint — the RED-when-broken conformance test for skade-katalog-rest.
//!
//! Spins the in-process axum front over an embedded `RedbCatalog`, points a
//! stock iceberg-rust `RestCatalog` at it, creates a table, then commits a
//! property-update transaction *through the REST path*. Verifies the commit
//! round-trips: the metadata location advances and the new property is visible
//! both via the REST client's `load_table` and via the embedded catalog
//! underneath (i.e. it was persisted in redb, not just echoed back). A second
//! test drives the raw HTTP `commit_table` with a stale `assert-table-uuid`
//! requirement and asserts the front maps the resulting
//! `CatalogCommitConflicts` to HTTP **409** — the optimistic-concurrency guard.
use std::collections::HashMap;
use std::sync::Arc;
use iceberg::io::LocalFsStorageFactory;
use iceberg::spec::{NestedField, PrimitiveType, Schema, Type};
use iceberg::transaction::{ApplyTransactionAction, Transaction};
use iceberg::{Catalog, CatalogBuilder, NamespaceIdent, TableCreation, TableIdent};
use iceberg_catalog_rest::{
RestCatalogBuilder, REST_CATALOG_PROP_URI, REST_CATALOG_PROP_WAREHOUSE,
};
use skade_katalog::{RedbCatalog, RedbCatalogBuilder};
use tempfile::TempDir;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
#[cfg(feature = "testmatrix")]
macro_rules! matrix {
($check:expr, $ok:expr, $detail:expr) => {
nornir_testmatrix::functional_status("skade-katalog-rest", $check, $ok, $detail);
};
}
#[cfg(not(feature = "testmatrix"))]
macro_rules! matrix {
($check:expr, $ok:expr, $detail:expr) => {{
let _ = ($check, $ok, $detail);
}};
}
fn schema() -> Schema {
Schema::builder()
.with_schema_id(0)
.with_fields(vec![
NestedField::required(1, "id", Type::Primitive(PrimitiveType::Long)).into(),
])
.build()
.expect("schema")
}
async fn embedded(tmp: &TempDir) -> RedbCatalog {
let warehouse = tmp.path().join("warehouse");
std::fs::create_dir_all(&warehouse).unwrap();
RedbCatalogBuilder::default()
.db_path(tmp.path().join("catalog.redb"))
.warehouse_location(format!("file://{}", warehouse.display()))
.with_storage_factory(Arc::new(LocalFsStorageFactory))
.load("skade", HashMap::new())
.await
.expect("embedded catalog")
}
#[tokio::test]
async fn commit_table_round_trips_over_rest() {
let tmp = TempDir::new().unwrap();
// The embedded catalog the front writes through. Keep a clone so we can
// inspect the persisted state after the REST commit.
let cat = embedded(&tmp).await;
let underlying = cat.clone();
let bound = skade_katalog_rest::spawn(cat, "127.0.0.1:0".parse().unwrap())
.await
.expect("front bound");
let rest = RestCatalogBuilder::default()
.with_storage_factory(Arc::new(LocalFsStorageFactory))
.load(
"rest",
HashMap::from([
(REST_CATALOG_PROP_URI.to_string(), format!("http://{bound}")),
(REST_CATALOG_PROP_WAREHOUSE.to_string(), "warehouse".to_string()),
]),
)
.await
.expect("rest client");
let ns = NamespaceIdent::new("db".to_string());
rest.create_namespace(&ns, HashMap::new()).await.unwrap();
let ident = TableIdent::new(ns.clone(), "t".to_string());
let creation = TableCreation::builder()
.name("t".to_string())
.schema(schema())
.build();
let table = rest.create_table(&ns, creation).await.unwrap();
let loc_before = table.metadata_location().unwrap().to_string();
assert!(
table.metadata().properties().get("answer").is_none(),
"fresh table should not carry the property yet"
);
// Commit a property update through the REST commit_table endpoint.
let tx = Transaction::new(&table);
let tx = tx
.update_table_properties()
.set("answer".to_string(), "42".to_string())
.apply(tx)
.unwrap();
let committed = tx.commit(&rest).await.expect("REST commit_table");
// The commit advanced the metadata location and carries the new property.
let loc_after = committed.metadata_location().unwrap().to_string();
assert_ne!(loc_before, loc_after, "commit must write a new metadata file");
assert_eq!(
committed.metadata().properties().get("answer").map(String::as_str),
Some("42"),
"committed table reflects the update"
);
// A fresh REST load sees the committed state (not the create-time metadata).
let reloaded = rest.load_table(&ident).await.unwrap();
assert_eq!(reloaded.metadata_location().unwrap(), loc_after);
assert_eq!(
reloaded.metadata().properties().get("answer").map(String::as_str),
Some("42"),
"REST load_table sees the persisted commit"
);
// And it was persisted in the backing redb catalog, not just echoed: the
// embedded catalog's pointer advanced to the same new location.
let direct = underlying.load_table(&ident).await.unwrap();
assert_eq!(
direct.metadata_location().unwrap(),
loc_after,
"embedded catalog under the front advanced its pointer"
);
assert_eq!(
direct.metadata().properties().get("answer").map(String::as_str),
Some("42"),
);
matrix!("commit_table.round_trip", true, "REST create+commit persisted in redb");
}
/// One-shot HTTP/1.1 POST to `addr` `path` with a JSON body. Returns the numeric
/// status code from the response status line. Pure tokio — no HTTP client dep.
async fn post_json(addr: std::net::SocketAddr, path: &str, body: &str) -> u16 {
let mut s = tokio::net::TcpStream::connect(addr).await.unwrap();
let req = format!(
"POST {path} HTTP/1.1\r\nHost: {addr}\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
s.write_all(req.as_bytes()).await.unwrap();
let mut resp = Vec::new();
s.read_to_end(&mut resp).await.unwrap();
let head = String::from_utf8_lossy(&resp);
// Status line: "HTTP/1.1 409 Conflict"
head.split_whitespace().nth(1).and_then(|c| c.parse().ok()).unwrap_or(0)
}
/// RED-when-broken: a `commit_table` carrying a stale `assert-table-uuid`
/// requirement must be rejected with HTTP **409** (the front's mapping of
/// `ErrorKind::CatalogCommitConflicts`). If the front stopped checking
/// requirements, or stopped mapping the conflict to 409, this flips to 2xx and
/// fails.
#[tokio::test]
async fn commit_table_conflicts_on_stale_requirement() {
let tmp = TempDir::new().unwrap();
let cat = embedded(&tmp).await;
// Seed a namespace + table directly on the backing catalog.
let ns = NamespaceIdent::new("db".to_string());
cat.create_namespace(&ns, HashMap::new()).await.unwrap();
let creation = TableCreation::builder().name("t".to_string()).schema(schema()).build();
cat.create_table(&ns, creation).await.unwrap();
let bound = skade_katalog_rest::spawn(cat, "127.0.0.1:0".parse().unwrap())
.await
.expect("front bound");
// A requirement asserting a UUID the table does not have → CatalogCommitConflicts.
let body = r#"{"requirements":[{"type":"assert-table-uuid","uuid":"00000000-0000-0000-0000-000000000000"}],"updates":[]}"#;
let code = post_json(bound, "/v1/namespaces/db/tables/t", body).await;
assert_eq!(code, 409, "stale requirement must map to HTTP 409 Conflict");
matrix!("commit_table.conflict", code == 409, "stale assert-table-uuid → 409");
}
/// `commit_table` on a missing table surfaces `TableNotFound` from the backing
/// catalog primitive (which the REST front maps to HTTP 404).
#[tokio::test]
async fn commit_table_not_found_for_missing_table() {
let tmp = TempDir::new().unwrap();
let cat = embedded(&tmp).await;
let ns = NamespaceIdent::new("db".to_string());
cat.create_namespace(&ns, HashMap::new()).await.unwrap();
let ghost = TableIdent::new(ns, "ghost".to_string());
let err = cat
.commit_table(ghost, vec![], vec![])
.await
.expect_err("commit on a table that was never created");
assert_eq!(err.kind(), iceberg::ErrorKind::TableNotFound);
matrix!("commit_table.not_found", true, "missing table → TableNotFound");
}