From 97ef74bfe3b6841aa5b0ea0b51d544145eb92cc5 Mon Sep 17 00:00:00 2001 From: mibrohimsulaeman-a11y <236154532+mibrohimsulaeman-a11y@users.noreply.github.com> Date: Sat, 6 Jun 2026 17:03:39 +0700 Subject: [PATCH] test(mq): add unit and ignored broker integration tests for MQ connectors Add pure unit tests (config defaults, builder overrides, serde round-trip + defaults, message construction, fault variant shapes) for vil_mq_rabbitmq, vil_mq_pulsar, vil_mq_sqs, vil_mq_pubsub. Add broker-dependent integration tests marked #[ignore] with run instructions in tests/integration.rs. Add serde_json (all four) and bytes (rabbitmq) dev-dependencies. --- crates/vil_mq_pubsub/Cargo.toml | 1 + crates/vil_mq_pubsub/tests/integration.rs | 37 +++++++++ crates/vil_mq_pubsub/tests/unit.rs | 62 +++++++++++++++ crates/vil_mq_pulsar/Cargo.toml | 1 + crates/vil_mq_pulsar/tests/integration.rs | 22 ++++++ crates/vil_mq_pulsar/tests/unit.rs | 64 ++++++++++++++++ crates/vil_mq_rabbitmq/Cargo.toml | 2 + crates/vil_mq_rabbitmq/tests/integration.rs | 29 +++++++ crates/vil_mq_rabbitmq/tests/unit.rs | 85 +++++++++++++++++++++ crates/vil_mq_sqs/Cargo.toml | 1 + crates/vil_mq_sqs/tests/integration.rs | 36 +++++++++ crates/vil_mq_sqs/tests/unit.rs | 72 +++++++++++++++++ 12 files changed, 412 insertions(+) create mode 100644 crates/vil_mq_pubsub/tests/integration.rs create mode 100644 crates/vil_mq_pubsub/tests/unit.rs create mode 100644 crates/vil_mq_pulsar/tests/integration.rs create mode 100644 crates/vil_mq_pulsar/tests/unit.rs create mode 100644 crates/vil_mq_rabbitmq/tests/integration.rs create mode 100644 crates/vil_mq_rabbitmq/tests/unit.rs create mode 100644 crates/vil_mq_sqs/tests/integration.rs create mode 100644 crates/vil_mq_sqs/tests/unit.rs diff --git a/crates/vil_mq_pubsub/Cargo.toml b/crates/vil_mq_pubsub/Cargo.toml index a9cbcfb7..73a95dc2 100644 --- a/crates/vil_mq_pubsub/Cargo.toml +++ b/crates/vil_mq_pubsub/Cargo.toml @@ -24,3 +24,4 @@ bytes = "1.0" [dev-dependencies] tokio = { version = "1", features = ["full"] } +serde_json = "1" diff --git a/crates/vil_mq_pubsub/tests/integration.rs b/crates/vil_mq_pubsub/tests/integration.rs new file mode 100644 index 00000000..5209d491 --- /dev/null +++ b/crates/vil_mq_pubsub/tests/integration.rs @@ -0,0 +1,37 @@ +//! Google Cloud Pub/Sub integration tests (require the emulator or GCP creds). +//! +//! `#[ignore]`d by default. Run manually (emulator example): +//! ```bash +//! gcloud beta emulators pubsub start --host-port=localhost:8085 & +//! export PUBSUB_EMULATOR_HOST="localhost:8085" +//! export PUBSUB_PROJECT="vil-test" PUBSUB_TOPIC="vil-test-topic" PUBSUB_SUBSCRIPTION="vil-test-sub" +//! cargo test -p vil_mq_pubsub --test integration -- --ignored +//! ``` + +use vil_mq_pubsub::{PubSubClient, PubSubConfig}; + +#[tokio::test] +#[ignore = "requires the Pub/Sub emulator or GCP credentials; see module docs and run with --ignored"] +async fn publish_subscribe_roundtrip() { + let project = std::env::var("PUBSUB_PROJECT").expect("PUBSUB_PROJECT must be set"); + let topic = std::env::var("PUBSUB_TOPIC").expect("PUBSUB_TOPIC must be set"); + let subscription = std::env::var("PUBSUB_SUBSCRIPTION").expect("PUBSUB_SUBSCRIPTION must be set"); + + let mut cfg = PubSubConfig::new(&project, &topic, &subscription); + if let Ok(host) = std::env::var("PUBSUB_EMULATOR_HOST") { + cfg = cfg.with_emulator(&host); + } + + let client = match PubSubClient::new(cfg).await { + Ok(c) => c, + Err(_) => panic!("could not initialize Pub/Sub client"), + }; + + assert!(client.publish(b"ping").await.is_ok()); + + if let Ok(messages) = client.subscribe().await { + for m in messages { + assert!(m.ack().await.is_ok()); + } + } +} diff --git a/crates/vil_mq_pubsub/tests/unit.rs b/crates/vil_mq_pubsub/tests/unit.rs new file mode 100644 index 00000000..97fdf0e0 --- /dev/null +++ b/crates/vil_mq_pubsub/tests/unit.rs @@ -0,0 +1,62 @@ +//! Pure unit tests for `vil_mq_pubsub` (no GCP calls required). + +use vil_mq_pubsub::{PubSubConfig, PubSubFault}; + +#[test] +fn config_new_applies_defaults() { + let c = PubSubConfig::new("proj", "topic", "sub"); + assert_eq!(c.project_id, "proj"); + assert_eq!(c.topic, "topic"); + assert_eq!(c.subscription, "sub"); + assert!(c.emulator_host.is_none()); + assert_eq!(c.max_messages, 10); + assert_eq!(c.ack_deadline_secs, 60); +} + +#[test] +fn with_emulator_sets_host() { + let c = PubSubConfig::new("p", "t", "s").with_emulator("localhost:8085"); + assert_eq!(c.emulator_host.as_deref(), Some("localhost:8085")); +} + +#[test] +fn resource_paths_are_well_formed() { + let c = PubSubConfig::new("proj", "orders", "orders-sub"); + assert_eq!(c.topic_path(), "projects/proj/topics/orders"); + assert_eq!(c.subscription_path(), "projects/proj/subscriptions/orders-sub"); +} + +#[test] +fn config_serde_roundtrip_and_defaults() { + let c = PubSubConfig::new("p", "t", "s").with_emulator("localhost:8085"); + let json = serde_json::to_string(&c).expect("serialize"); + let back: PubSubConfig = serde_json::from_str(&json).expect("deserialize"); + assert_eq!(back.project_id, "p"); + assert_eq!(back.emulator_host.as_deref(), Some("localhost:8085")); + + let partial = r#"{"project_id":"p","topic":"t","subscription":"s"}"#; + let d: PubSubConfig = serde_json::from_str(partial).expect("deserialize"); + assert!(d.emulator_host.is_none()); + assert_eq!(d.max_messages, 10); + assert_eq!(d.ack_deadline_secs, 60); +} + +#[test] +fn fault_variants_carry_expected_fields() { + let f = PubSubFault::PublishFailed { topic_hash: 4, status_code: 13 }; + match f { + PubSubFault::PublishFailed { topic_hash, status_code } => { + assert_eq!(topic_hash, 4); + assert_eq!(status_code, 13); + } + _ => panic!("unexpected variant"), + } + + let s = PubSubFault::SubscriberFailed { subscription_hash: 99 }; + match s { + PubSubFault::SubscriberFailed { subscription_hash } => { + assert_eq!(subscription_hash, 99); + } + _ => panic!("unexpected variant"), + } +} diff --git a/crates/vil_mq_pulsar/Cargo.toml b/crates/vil_mq_pulsar/Cargo.toml index 0b2b2b56..8146759a 100644 --- a/crates/vil_mq_pulsar/Cargo.toml +++ b/crates/vil_mq_pulsar/Cargo.toml @@ -23,3 +23,4 @@ futures = "0.3" [dev-dependencies] tokio = { version = "1", features = ["full"] } +serde_json = "1" diff --git a/crates/vil_mq_pulsar/tests/integration.rs b/crates/vil_mq_pulsar/tests/integration.rs new file mode 100644 index 00000000..5ec0f9db --- /dev/null +++ b/crates/vil_mq_pulsar/tests/integration.rs @@ -0,0 +1,22 @@ +//! Broker integration tests for `vil_mq_pulsar` (require a live Pulsar broker). +//! +//! `#[ignore]`d by default. Run manually: +//! ```bash +//! docker run -d --rm -p 6650:6650 -p 8080:8080 apachepulsar/pulsar:3.2.0 bin/pulsar standalone +//! export PULSAR_URL="pulsar://localhost:6650" +//! cargo test -p vil_mq_pulsar --test integration -- --ignored +//! ``` + +use vil_mq_pulsar::{PulsarClient, PulsarConfig}; + +#[tokio::test] +#[ignore = "requires a live Pulsar broker; see module docs and run with --ignored"] +async fn connect_smoke() { + let url = std::env::var("PULSAR_URL").unwrap_or_else(|_| "pulsar://localhost:6650".to_string()); + let cfg = PulsarConfig::new(&url, "public", "default"); + + match PulsarClient::connect(cfg).await { + Ok(_) => {} + Err(_) => panic!("could not connect to Pulsar broker at {url}"), + } +} diff --git a/crates/vil_mq_pulsar/tests/unit.rs b/crates/vil_mq_pulsar/tests/unit.rs new file mode 100644 index 00000000..83ae0cbf --- /dev/null +++ b/crates/vil_mq_pulsar/tests/unit.rs @@ -0,0 +1,64 @@ +//! Pure unit tests for `vil_mq_pulsar` (no live broker required). + +use vil_mq_pulsar::{PulsarConfig, PulsarFault}; + +#[test] +fn config_new_applies_defaults() { + let c = PulsarConfig::new("pulsar://localhost:6650", "public", "default"); + assert_eq!(c.url, "pulsar://localhost:6650"); + assert_eq!(c.tenant, "public"); + assert_eq!(c.namespace, "default"); + assert!(c.auth_token.is_none()); + assert_eq!(c.operation_timeout_ms, 30_000); + assert_eq!(c.connection_timeout_ms, 5_000); +} + +#[test] +fn config_with_token_sets_auth() { + let c = PulsarConfig::new("pulsar://h", "t", "n").with_token("secret"); + assert_eq!(c.auth_token.as_deref(), Some("secret")); +} + +#[test] +fn topic_fqn_builds_persistent_path() { + let c = PulsarConfig::new("pulsar://h", "acme", "ns1"); + assert_eq!(c.topic_fqn("orders"), "persistent://acme/ns1/orders"); +} + +#[test] +fn config_serde_roundtrip_and_defaults() { + let c = PulsarConfig::new("pulsar://h", "t", "n").with_token("tok"); + let json = serde_json::to_string(&c).expect("serialize"); + let back: PulsarConfig = serde_json::from_str(&json).expect("deserialize"); + assert_eq!(back.url, "pulsar://h"); + assert_eq!(back.tenant, "t"); + assert_eq!(back.namespace, "n"); + assert_eq!(back.auth_token.as_deref(), Some("tok")); + + let partial = r#"{"url":"pulsar://h","tenant":"t","namespace":"n"}"#; + let d: PulsarConfig = serde_json::from_str(partial).expect("deserialize"); + assert!(d.auth_token.is_none()); + assert_eq!(d.operation_timeout_ms, 30_000); + assert_eq!(d.connection_timeout_ms, 5_000); +} + +#[test] +fn fault_variants_carry_expected_fields() { + let f = PulsarFault::ConsumerFailed { topic_hash: 11, subscription_hash: 22 }; + match f { + PulsarFault::ConsumerFailed { topic_hash, subscription_hash } => { + assert_eq!(topic_hash, 11); + assert_eq!(subscription_hash, 22); + } + _ => panic!("unexpected variant"), + } + + let s = PulsarFault::SendFailed { topic_hash: 1, error_code: 7 }; + match s { + PulsarFault::SendFailed { topic_hash, error_code } => { + assert_eq!(topic_hash, 1); + assert_eq!(error_code, 7); + } + _ => panic!("unexpected variant"), + } +} diff --git a/crates/vil_mq_rabbitmq/Cargo.toml b/crates/vil_mq_rabbitmq/Cargo.toml index be85b164..a2d97452 100644 --- a/crates/vil_mq_rabbitmq/Cargo.toml +++ b/crates/vil_mq_rabbitmq/Cargo.toml @@ -23,3 +23,5 @@ futures = "0.3" [dev-dependencies] tokio = { version = "1", features = ["full"] } +serde_json = "1" +bytes = "1" diff --git a/crates/vil_mq_rabbitmq/tests/integration.rs b/crates/vil_mq_rabbitmq/tests/integration.rs new file mode 100644 index 00000000..0eeb7a3a --- /dev/null +++ b/crates/vil_mq_rabbitmq/tests/integration.rs @@ -0,0 +1,29 @@ +//! Broker integration tests for `vil_mq_rabbitmq`. +//! +//! These require a live RabbitMQ broker and are `#[ignore]`d by default so they +//! never run in normal CI. +//! +//! Run them manually: +//! ```bash +//! docker run -d --rm -p 5672:5672 rabbitmq:3-management +//! export RABBITMQ_URI="amqp://guest:guest@localhost:5672/%2F" +//! cargo test -p vil_mq_rabbitmq --test integration -- --ignored +//! ``` + +use vil_mq_rabbitmq::{RabbitClient, RabbitConfig}; + +#[tokio::test] +#[ignore = "requires a live RabbitMQ broker; see module docs and run with --ignored"] +async fn publish_smoke() { + let uri = std::env::var("RABBITMQ_URI") + .unwrap_or_else(|_| "amqp://guest:guest@localhost:5672/%2F".to_string()); + let cfg = RabbitConfig::new(&uri, "", "vil.test.q"); + + let client = match RabbitClient::connect(cfg).await { + Ok(c) => c, + Err(_) => panic!("could not connect to RabbitMQ broker at {uri}"), + }; + + assert!(client.publish("", "vil.test.q", b"ping").await.is_ok()); + client.close().await; +} diff --git a/crates/vil_mq_rabbitmq/tests/unit.rs b/crates/vil_mq_rabbitmq/tests/unit.rs new file mode 100644 index 00000000..ce6b0aa5 --- /dev/null +++ b/crates/vil_mq_rabbitmq/tests/unit.rs @@ -0,0 +1,85 @@ +//! Pure unit tests for `vil_mq_rabbitmq` (no live broker required). +//! +//! Covers config construction, builder defaults/overrides, serde round-trip +//! and serde defaults, message construction, and fault variant shapes. + +use bytes::Bytes; +use vil_mq_rabbitmq::{RabbitConfig, RabbitFault, RabbitMessage}; + +#[test] +fn config_new_applies_defaults() { + let c = RabbitConfig::new("amqp://guest:guest@localhost:5672/%2F", "ex", "q"); + assert_eq!(c.uri, "amqp://guest:guest@localhost:5672/%2F"); + assert_eq!(c.exchange, "ex"); + assert_eq!(c.queue, "q"); + assert_eq!(c.consumer_tag, "vil-consumer"); + assert_eq!(c.connection_timeout_ms, 5_000); + assert_eq!(c.prefetch_count, 10); +} + +#[test] +fn config_builders_override_defaults() { + let c = RabbitConfig::new("amqp://h", "ex", "q") + .with_prefetch(64) + .with_consumer_tag("worker-1"); + assert_eq!(c.prefetch_count, 64); + assert_eq!(c.consumer_tag, "worker-1"); +} + +#[test] +fn config_serde_roundtrip_preserves_fields() { + let c = RabbitConfig::new("amqp://h", "ex", "q").with_prefetch(7); + let json = serde_json::to_string(&c).expect("serialize"); + let back: RabbitConfig = serde_json::from_str(&json).expect("deserialize"); + assert_eq!(back.uri, "amqp://h"); + assert_eq!(back.exchange, "ex"); + assert_eq!(back.queue, "q"); + assert_eq!(back.prefetch_count, 7); +} + +#[test] +fn config_serde_uses_defaults_for_missing_fields() { + let json = r#"{"uri":"amqp://h","exchange":"ex","queue":"q"}"#; + let c: RabbitConfig = serde_json::from_str(json).expect("deserialize"); + assert_eq!(c.consumer_tag, "vil-consumer"); + assert_eq!(c.connection_timeout_ms, 5_000); + assert_eq!(c.prefetch_count, 10); +} + +#[test] +fn message_holds_payload_and_metadata() { + let m = RabbitMessage { + payload: Bytes::from_static(b"hello"), + delivery_tag: 42, + routing_key_hash: 7, + exchange_hash: 9, + }; + assert_eq!(&m.payload[..], &b"hello"[..]); + assert_eq!(m.payload.len(), 5); + assert_eq!(m.delivery_tag, 42); + assert_eq!(m.routing_key_hash, 7); + assert_eq!(m.exchange_hash, 9); +} + +#[test] +fn fault_variants_carry_expected_fields() { + let f = RabbitFault::ConnectionFailed { uri_hash: 1, elapsed_ms: 2 }; + match f { + RabbitFault::ConnectionFailed { uri_hash, elapsed_ms } => { + assert_eq!(uri_hash, 1); + assert_eq!(elapsed_ms, 2); + } + _ => panic!("unexpected variant"), + } + + let p = RabbitFault::PublishFailed { exchange_hash: 3, routing_key_hash: 4 }; + match p { + RabbitFault::PublishFailed { exchange_hash, routing_key_hash } => { + assert_eq!(exchange_hash, 3); + assert_eq!(routing_key_hash, 4); + } + _ => panic!("unexpected variant"), + } + + let _ = RabbitFault::NotConnected; +} diff --git a/crates/vil_mq_sqs/Cargo.toml b/crates/vil_mq_sqs/Cargo.toml index bb852a2c..374c54f5 100644 --- a/crates/vil_mq_sqs/Cargo.toml +++ b/crates/vil_mq_sqs/Cargo.toml @@ -23,3 +23,4 @@ serde = { version = "1.0", features = ["derive"] } [dev-dependencies] tokio = { version = "1", features = ["full"] } +serde_json = "1" diff --git a/crates/vil_mq_sqs/tests/integration.rs b/crates/vil_mq_sqs/tests/integration.rs new file mode 100644 index 00000000..26f66c80 --- /dev/null +++ b/crates/vil_mq_sqs/tests/integration.rs @@ -0,0 +1,36 @@ +//! AWS SQS integration tests (require AWS SQS or a LocalStack endpoint). +//! +//! `#[ignore]`d by default. Run manually (LocalStack example): +//! ```bash +//! docker run -d --rm -p 4566:4566 localstack/localstack +//! export AWS_ACCESS_KEY_ID=test AWS_SECRET_ACCESS_KEY=test AWS_REGION=us-east-1 +//! export SQS_ENDPOINT="http://localhost:4566" +//! export SQS_QUEUE_URL="http://localhost:4566/000000000000/vil-test-q" +//! cargo test -p vil_mq_sqs --test integration -- --ignored +//! ``` + +use vil_mq_sqs::{SqsClient, SqsConfig}; + +#[tokio::test] +#[ignore = "requires AWS SQS or LocalStack; see module docs and run with --ignored"] +async fn send_receive_delete_roundtrip() { + let region = std::env::var("AWS_REGION").unwrap_or_else(|_| "us-east-1".to_string()); + let queue_url = std::env::var("SQS_QUEUE_URL").expect("SQS_QUEUE_URL must be set"); + let mut cfg = SqsConfig::new(®ion, &queue_url); + if let Ok(ep) = std::env::var("SQS_ENDPOINT") { + cfg = cfg.with_endpoint(&ep); + } + + let client = match SqsClient::from_config(cfg).await { + Ok(c) => c, + Err(_) => panic!("could not build SQS client"), + }; + + assert!(client.send_message(b"ping").await.is_ok()); + + if let Ok(messages) = client.receive_messages().await { + for m in messages { + assert!(client.delete_message(&m.receipt_handle).await.is_ok()); + } + } +} diff --git a/crates/vil_mq_sqs/tests/unit.rs b/crates/vil_mq_sqs/tests/unit.rs new file mode 100644 index 00000000..aabe09bb --- /dev/null +++ b/crates/vil_mq_sqs/tests/unit.rs @@ -0,0 +1,72 @@ +//! Pure unit tests for `vil_mq_sqs` (no AWS calls required). + +use vil_mq_sqs::{SqsConfig, SqsFault, SqsMessage}; + +#[test] +fn config_new_applies_defaults() { + let c = SqsConfig::new("us-east-1", "https://sqs.us-east-1.amazonaws.com/123/q"); + assert_eq!(c.region, "us-east-1"); + assert_eq!(c.queue_url, "https://sqs.us-east-1.amazonaws.com/123/q"); + assert!(c.endpoint.is_none()); + assert_eq!(c.max_messages, 10); + assert_eq!(c.visibility_timeout_secs, 30); + assert_eq!(c.wait_time_secs, 20); +} + +#[test] +fn with_endpoint_sets_custom_endpoint() { + let c = SqsConfig::new("us-east-1", "q").with_endpoint("http://localhost:4566"); + assert_eq!(c.endpoint.as_deref(), Some("http://localhost:4566")); +} + +#[test] +fn with_max_messages_clamps_to_valid_range() { + assert_eq!(SqsConfig::new("r", "q").with_max_messages(50).max_messages, 10); + assert_eq!(SqsConfig::new("r", "q").with_max_messages(0).max_messages, 1); + assert_eq!(SqsConfig::new("r", "q").with_max_messages(5).max_messages, 5); +} + +#[test] +fn config_serde_roundtrip_and_defaults() { + let c = SqsConfig::new("eu-west-1", "q").with_endpoint("http://e"); + let json = serde_json::to_string(&c).expect("serialize"); + let back: SqsConfig = serde_json::from_str(&json).expect("deserialize"); + assert_eq!(back.region, "eu-west-1"); + assert_eq!(back.endpoint.as_deref(), Some("http://e")); + + let partial = r#"{"region":"us-east-1","queue_url":"q"}"#; + let d: SqsConfig = serde_json::from_str(partial).expect("deserialize"); + assert!(d.endpoint.is_none()); + assert_eq!(d.max_messages, 10); + assert_eq!(d.visibility_timeout_secs, 30); + assert_eq!(d.wait_time_secs, 20); +} + +#[test] +fn message_holds_body_and_metadata() { + let m = SqsMessage { + body: b"payload".to_vec(), + receipt_handle: "rh-123".to_string(), + queue_hash: 5, + receive_count: 2, + }; + let cloned = m.clone(); + assert_eq!(cloned.body, b"payload".to_vec()); + assert_eq!(cloned.receipt_handle, "rh-123"); + assert_eq!(cloned.queue_hash, 5); + assert_eq!(cloned.receive_count, 2); +} + +#[test] +fn fault_variants_carry_expected_fields() { + let f = SqsFault::SendFailed { queue_hash: 9, error_code: 3 }; + match f { + SqsFault::SendFailed { queue_hash, error_code } => { + assert_eq!(queue_hash, 9); + assert_eq!(error_code, 3); + } + _ => panic!("unexpected variant"), + } + + let _ = SqsFault::InvalidMessage; +}