#![cfg(test)]
use std::sync::Arc;
use crate::generation::{CatalogManifest, ManifestColumn, ManifestTable};
use crate::runtime::catalog::{CatalogManager, DEFAULT_PROJECT_ID};
fn manifest_for(message_name: &str, schema: &str, table: &str, checksum: &str) -> CatalogManifest {
CatalogManifest {
checksum_sha256: checksum.to_string(),
generator_version: "u4-test".to_string(),
tables: vec![ManifestTable {
checksum_sha256: checksum.to_string(),
message_name: message_name.to_string(),
schema: schema.to_string(),
table: table.to_string(),
primary_key: vec!["id".to_string()],
columns: vec![ManifestColumn {
field_name: "id".into(),
column_name: "id".into(),
proto_type: "string".into(),
sql_type: "uuid".into(),
is_primary: true,
..Default::default()
}],
..Default::default()
}],
..Default::default()
}
}
async fn make_manager_with_two_projects() -> Arc<CatalogManager> {
let initial = manifest_for(
"default.system.v1.Heartbeat",
"system",
"heartbeats",
"default-initial",
);
let manager = Arc::new(CatalogManager::new(initial));
let billing = manifest_for(
"acme.billing.v1.Customer",
"billing",
"customers",
"billing-checksum",
);
manager
.stage_catalog(billing, "billing".into(), "1.0.0".into(), "backward".into())
.await
.expect("stage billing");
manager
.activate_catalog_for("billing", "1.0.0")
.await
.expect("activate billing");
let analytics = manifest_for(
"acme.analytics.v1.Event",
"analytics",
"events",
"analytics-checksum",
);
manager
.stage_catalog(
analytics,
"analytics".into(),
"1.0.0".into(),
"backward".into(),
)
.await
.expect("stage analytics");
manager
.activate_catalog_for("analytics", "1.0.0")
.await
.expect("activate analytics");
manager
}
#[tokio::test]
async fn two_projects_serve_distinct_manifests() {
let manager = make_manager_with_two_projects().await;
let billing = manager.active_for("billing");
let analytics = manager.active_for("analytics");
let billing_tables: Vec<&str> = billing
.manifest
.tables
.iter()
.map(|t| t.message_name.as_str())
.collect();
let analytics_tables: Vec<&str> = analytics
.manifest
.tables
.iter()
.map(|t| t.message_name.as_str())
.collect();
assert_eq!(billing_tables, vec!["acme.billing.v1.Customer"]);
assert_eq!(analytics_tables, vec!["acme.analytics.v1.Event"]);
assert_eq!(billing.metadata.checksum, "billing-checksum");
assert_eq!(analytics.metadata.checksum, "analytics-checksum");
}
#[tokio::test]
async fn project_a_manifest_does_not_expose_project_b_messages() {
let manager = make_manager_with_two_projects().await;
let billing = manager.active_for("billing");
let leak_attempt =
crate::broker::table_for_message(&billing.manifest, "acme.analytics.v1.Event");
assert!(
leak_attempt.is_none(),
"billing manifest must NOT see analytics messages, but found {:?}",
leak_attempt.map(|t| t.message_name.clone())
);
let analytics = manager.active_for("analytics");
let found = crate::broker::table_for_message(&analytics.manifest, "acme.analytics.v1.Event");
assert!(
found.is_some(),
"analytics manifest must serve its own messages"
);
}
#[tokio::test]
async fn activating_one_project_does_not_disturb_another() {
let manager = make_manager_with_two_projects().await;
let analytics_before = manager.active_for("analytics").metadata.checksum.clone();
assert_eq!(analytics_before, "analytics-checksum");
let new_billing = manifest_for(
"acme.billing.v1.Invoice",
"billing",
"invoices",
"billing-v2",
);
manager
.stage_catalog(
new_billing,
"billing".into(),
"2.0.0".into(),
"backward".into(),
)
.await
.unwrap();
manager
.activate_catalog_for("billing", "2.0.0")
.await
.unwrap();
assert_eq!(
manager.active_for("billing").metadata.checksum,
"billing-v2"
);
assert_eq!(
manager.active_for("analytics").metadata.checksum,
analytics_before,
"analytics catalog must be untouched by billing rotation"
);
let billing = manager.active_for("billing");
assert!(
crate::broker::table_for_message(&billing.manifest, "acme.billing.v1.Customer").is_none(),
"old billing message must be gone after rotation"
);
assert!(
crate::broker::table_for_message(&billing.manifest, "acme.billing.v1.Invoice").is_some(),
"new billing message must be present after rotation"
);
let analytics = manager.active_for("analytics");
assert!(
crate::broker::table_for_message(&analytics.manifest, "acme.analytics.v1.Event").is_some(),
"analytics message must be untouched"
);
}
#[tokio::test]
async fn blank_project_id_resolves_to_default_not_any_other_project() {
let manager = make_manager_with_two_projects().await;
let blank = manager.active_for("");
let default = manager.active_for(DEFAULT_PROJECT_ID);
assert_eq!(blank.metadata.checksum, "default-initial");
assert_eq!(default.metadata.checksum, "default-initial");
assert_ne!(blank.metadata.checksum, "billing-checksum");
assert_ne!(blank.metadata.checksum, "analytics-checksum");
}
#[tokio::test]
async fn staging_slots_are_per_project() {
let manager = make_manager_with_two_projects().await;
let pending_billing = manifest_for(
"acme.billing.v1.PendingInvoice",
"billing",
"pending_invoices",
"billing-pending",
);
manager
.stage_catalog(
pending_billing,
"billing".into(),
"4.0.0".into(),
"backward".into(),
)
.await
.unwrap();
assert!(manager.staged_for("billing").await.is_some());
assert!(manager.staged_for("analytics").await.is_none());
assert!(manager.staged_for(DEFAULT_PROJECT_ID).await.is_none());
manager.rollback_catalog_for("billing").await.unwrap();
assert!(manager.staged_for("billing").await.is_none());
assert_eq!(
manager.active_for("analytics").metadata.checksum,
"analytics-checksum"
);
}