nordisk/skade

serve/tests/serve_roundtrip.rs

//! Functional test for `skade-katalog serve`: one lock-owning catalog fronted
//! over BOTH wires at once. Binds REST + gRPC on ephemeral ports over a single
//! `RedbCatalog`, creates a table over the **gRPC** front, then loads it over
//! the **REST** front — proving both fronts share the one backing catalog (the
//! serve-level guarantee), and that the binds actually happen.

use std::collections::HashMap;
use std::sync::Arc;

use iceberg::io::LocalFsStorageFactory;
use iceberg::spec::{NestedField, PrimitiveType, Schema, Type};
use iceberg::{Catalog, CatalogBuilder, NamespaceIdent, TableIdent};
use iceberg_catalog_rest::{
    REST_CATALOG_PROP_URI, REST_CATALOG_PROP_WAREHOUSE, RestCatalogBuilder,
};
use skade_katalog_grpc::pb;
use skade_katalog_serve::ServeConfig;
use tempfile::TempDir;

fn schema() -> Schema {
    Schema::builder()
        .with_schema_id(0)
        .with_fields(vec![
            NestedField::required(1, "id", Type::Primitive(PrimitiveType::Long)).into(),
        ])
        .build()
        .expect("schema")
}

#[tokio::test]
async fn serve_fronts_one_catalog_over_rest_and_grpc() {
    let tmp = TempDir::new().unwrap();
    let mut cfg = ServeConfig::new(tmp.path().join("wh"));
    cfg.rest = Some("127.0.0.1:0".parse().unwrap());
    cfg.grpc = Some("127.0.0.1:0".parse().unwrap());

    let (rest_addr, grpc_addr) = skade_katalog_serve::spawn(&cfg).await.expect("serve spawn");
    let rest_addr = rest_addr.expect("rest bound");
    let grpc_addr = grpc_addr.expect("grpc bound");

    // --- create a table over the gRPC front ---
    let mut grpc = pb::catalog_client::CatalogClient::connect(format!("http://{grpc_addr}"))
        .await
        .expect("grpc connect");
    grpc.create_namespace(pb::CreateNamespaceReq {
        levels: vec!["db".into()],
        properties: HashMap::new(),
    })
    .await
    .expect("grpc create_namespace");
    let created = grpc
        .create_table(pb::CreateTableReq {
            namespace: vec!["db".into()],
            name: "shared".into(),
            schema_json: serde_json::to_string(&schema()).unwrap(),
            location: String::new(),
        })
        .await
        .expect("grpc create_table")
        .into_inner();

    // --- load the SAME table over the REST front ---
    let rest = RestCatalogBuilder::default()
        .with_storage_factory(Arc::new(LocalFsStorageFactory))
        .load(
            "rest",
            HashMap::from([
                (
                    REST_CATALOG_PROP_URI.to_string(),
                    format!("http://{rest_addr}"),
                ),
                (
                    REST_CATALOG_PROP_WAREHOUSE.to_string(),
                    "warehouse".to_string(),
                ),
            ]),
        )
        .await
        .expect("rest client");

    let ident = TableIdent::new(NamespaceIdent::new("db".to_string()), "shared".to_string());
    let loaded = rest
        .load_table(&ident)
        .await
        .expect("REST load of a gRPC-created table");
    assert_eq!(
        loaded.metadata_location().unwrap(),
        created.metadata_location,
        "REST and gRPC fronts see the one shared catalog's pointer"
    );
}