#![allow(
clippy::unwrap_used,
clippy::print_stdout,
clippy::print_stderr,
clippy::missing_assert_message
)] #![allow(clippy::cast_possible_truncation)] #![allow(clippy::map_unwrap_or)] use std::{collections::HashMap, sync::Arc, time::Duration};
use fraiseql_core::{
db::{postgres::PostgresAdapter, traits::DatabaseAdapter},
federation::{
mutation_executor::FederationMutationExecutor,
types::{EntityRepresentation, FederatedType, FederationMetadata, KeyDirective},
},
};
use serde_json::{Value, json};
pub async fn pg_entity_fixture(
table: &str,
column_ddl: &[&str],
rows: &[HashMap<String, Value>],
) -> Option<(fraiseql_test_support::Service, Arc<PostgresAdapter>)> {
let (pg, adapter) = pg_adapter().await?;
adapter
.execute_raw_query(&format!(r#"DROP TABLE IF EXISTS "{table}" CASCADE"#))
.await
.expect("drop fixture table");
adapter
.execute_raw_query(&format!(r#"CREATE TABLE "{table}" ({})"#, column_ddl.join(", ")))
.await
.expect("create fixture table");
let col_names: Vec<&str> = column_ddl
.iter()
.map(|c| c.split_whitespace().next().expect("non-empty column ddl"))
.collect();
let col_list = col_names.iter().map(|c| format!("\"{c}\"")).collect::<Vec<_>>().join(", ");
for row in rows {
let vals = col_names
.iter()
.map(|c| sql_literal(row.get(*c)))
.collect::<Vec<_>>()
.join(", ");
adapter
.execute_raw_query(&format!(r#"INSERT INTO "{table}" ({col_list}) VALUES ({vals})"#))
.await
.expect("seed fixture row");
}
Some((pg, adapter))
}
pub async fn pg_adapter() -> Option<(fraiseql_test_support::Service, Arc<PostgresAdapter>)> {
let pg = fraiseql_test_support::postgres().await?;
let adapter = PostgresAdapter::new(pg.url()).await.expect("connect to harness postgres");
Some((pg, Arc::new(adapter)))
}
pub fn row(pairs: &[(&str, Value)]) -> HashMap<String, Value> {
pairs.iter().map(|(k, v)| ((*k).to_string(), v.clone())).collect()
}
pub fn rep(typename: &str, keys: &[(&str, Value)]) -> EntityRepresentation {
let key_fields = row(keys);
EntityRepresentation {
typename: typename.to_string(),
all_fields: key_fields.clone(),
key_fields,
}
}
fn sql_literal(value: Option<&Value>) -> String {
match value {
Some(Value::String(s)) => format!("'{}'", s.replace('\'', "''")),
Some(Value::Number(n)) => n.to_string(),
Some(Value::Bool(b)) => b.to_string(),
Some(Value::Null) | None => "NULL".to_string(),
Some(other) => format!("'{}'", other.to_string().replace('\'', "''")),
}
}
pub async fn pg_mutation_executor(
metadata: FederationMetadata,
tables: &[(&str, &[&str])],
) -> Option<(fraiseql_test_support::Service, FederationMutationExecutor<PostgresAdapter>)> {
let (pg, adapter) = pg_adapter().await?;
for (table, column_ddl) in tables {
let table = table.to_lowercase();
adapter
.execute_raw_query(&format!(r#"DROP TABLE IF EXISTS "{table}" CASCADE"#))
.await
.expect("drop mutation table");
adapter
.execute_raw_query(&format!(r#"CREATE TABLE "{table}" ({})"#, column_ddl.join(", ")))
.await
.expect("create mutation table");
}
Some((pg, FederationMutationExecutor::new(adapter, metadata, false)))
}
pub fn metadata_single_key(type_name: &str, key_field: &str) -> FederationMetadata {
FederationMetadata {
enabled: true,
version: "v2".to_string(),
types: vec![FederatedType {
name: type_name.to_string(),
keys: vec![KeyDirective {
fields: vec![key_field.to_string()],
resolvable: true,
}],
is_extends: false,
external_fields: vec![],
shareable_fields: vec![],
inaccessible_fields: vec![],
field_directives: std::collections::HashMap::new(),
type_shareable: false,
}],
remote_subscription_fields: std::collections::HashMap::new(),
}
}
pub fn metadata_extended_type(
type_name: &str,
key_field: &str,
external_fields: &[&str],
shareable_fields: &[&str],
) -> FederationMetadata {
FederationMetadata {
enabled: true,
version: "v2".to_string(),
types: vec![FederatedType {
name: type_name.to_string(),
keys: vec![KeyDirective {
fields: vec![key_field.to_string()],
resolvable: true,
}],
is_extends: true,
external_fields: external_fields.iter().map(|s| (*s).to_string()).collect(),
shareable_fields: shareable_fields.iter().map(|s| (*s).to_string()).collect(),
inaccessible_fields: vec![],
field_directives: std::collections::HashMap::new(),
type_shareable: false,
}],
remote_subscription_fields: std::collections::HashMap::new(),
}
}
pub fn metadata_composite_key(type_name: &str, key_fields: &[&str]) -> FederationMetadata {
FederationMetadata {
enabled: true,
version: "v2".to_string(),
types: vec![FederatedType {
name: type_name.to_string(),
keys: vec![KeyDirective {
fields: key_fields.iter().map(|s| (*s).to_string()).collect(),
resolvable: true,
}],
is_extends: false,
external_fields: vec![],
shareable_fields: vec![],
inaccessible_fields: vec![],
field_directives: std::collections::HashMap::new(),
type_shareable: false,
}],
remote_subscription_fields: std::collections::HashMap::new(),
}
}
pub fn enforce_requires(
metadata: &FederationMetadata,
typename: &str,
fields: &[&str],
representation: &EntityRepresentation,
) -> std::result::Result<(), String> {
let federated_type = metadata
.types
.iter()
.find(|t| t.name == typename)
.ok_or_else(|| format!("Type {} not found in federation metadata", typename))?;
for field in fields {
if let Some(directives) = federated_type.get_field_directives(field) {
for required in &directives.requires {
let field_path = required.path.join(".");
if !representation.has_field(&field_path) {
return Err(format!(
"Validation Error: Required field missing\n\
Type: {}\n\
Field: {}\n\
Required: {}\n\
Issue: Field '{}' requires '{}' but it is missing from entity \
representation\n\
Suggestion: Ensure '{}' is requested from the owning subgraph",
typename, field, field_path, field, field_path, field_path
));
}
}
}
}
Ok(())
}
pub const APOLLO_GATEWAY_URL: &str = "http://localhost:4000/graphql";
pub const USERS_SUBGRAPH_URL: &str = "http://localhost:4001/graphql";
pub const ORDERS_SUBGRAPH_URL: &str = "http://localhost:4002/graphql";
pub const PRODUCTS_SUBGRAPH_URL: &str = "http://localhost:4003/graphql";
pub async fn wait_for_service(
url: &str,
max_retries: u32,
) -> std::result::Result<(), Box<dyn std::error::Error>> {
let client = reqwest::Client::new();
let mut retries = 0;
loop {
match client
.post(url)
.json(&json!({ "query": "{ __typename }" }))
.timeout(Duration::from_secs(5))
.send()
.await
{
Ok(response) if response.status().is_success() => {
println!("✓ Service ready: {}", url);
return Ok(());
},
Ok(response) => {
println!("✗ Service {} returned status: {}", url, response.status());
},
Err(e) => {
println!("✗ Service {} connection failed: {}", url, e);
},
}
retries += 1;
if retries >= max_retries {
return Err(format!(
"Service {} failed to become ready after {} retries",
url, max_retries
)
.into());
}
tokio::time::sleep(Duration::from_secs(2)).await;
}
}
pub async fn graphql_query(
url: &str,
query: &str,
) -> std::result::Result<Value, Box<dyn std::error::Error>> {
let client = reqwest::Client::new();
let response = client
.post(url)
.json(&json!({ "query": query }))
.timeout(Duration::from_secs(10))
.send()
.await?;
let body: Value = response.json().await?;
Ok(body)
}
pub fn extract_data(response: &Value) -> Option<&Value> {
response.get("data")
}
pub fn has_errors(response: &Value) -> bool {
response.get("errors").is_some()
}
pub fn get_error_messages(response: &Value) -> String {
response
.get("errors")
.and_then(|e| e.as_array())
.map(|arr| {
arr.iter()
.filter_map(|err| err.get("message")?.as_str())
.collect::<Vec<_>>()
.join("; ")
})
.unwrap_or_else(|| "Unknown error".to_string())
}
pub async fn setup_federation_tests() -> std::result::Result<(), Box<dyn std::error::Error>> {
println!("\n=== Setting up 2-subgraph federation tests ===\n");
println!("Waiting for users subgraph...");
wait_for_service(USERS_SUBGRAPH_URL, 30).await?;
println!("Waiting for orders subgraph...");
wait_for_service(ORDERS_SUBGRAPH_URL, 30).await?;
println!("Waiting for Apollo Router gateway...");
wait_for_service(APOLLO_GATEWAY_URL, 30).await?;
println!("\n✓ All services ready for 2-subgraph federation tests\n");
Ok(())
}
pub async fn setup_three_subgraph_tests() -> std::result::Result<(), Box<dyn std::error::Error>> {
println!("\n=== Setting up 3-subgraph federation tests ===\n");
println!("Waiting for users subgraph (port 4001)...");
wait_for_service(USERS_SUBGRAPH_URL, 30).await?;
println!("Waiting for orders subgraph (port 4002)...");
wait_for_service(ORDERS_SUBGRAPH_URL, 30).await?;
println!("Waiting for products subgraph (port 4003)...");
wait_for_service(PRODUCTS_SUBGRAPH_URL, 30).await?;
println!("Waiting for Apollo Router gateway...");
wait_for_service(APOLLO_GATEWAY_URL, 30).await?;
println!("\n✓ All 3 subgraphs + gateway ready for federation tests\n");
Ok(())
}