test(mq): add unit + ignored broker integration tests for MQ connectors

crates/vil_mq_pubsub/Cargo.toml

@@ -24,3 +24,4 @@ bytes                      = "1.0"
 
 [dev-dependencies]
 tokio = { version = "1", features = ["full"] }
+serde_json = "1"

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());
+        }
+    }
+}

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"),
+    }
+}

crates/vil_mq_pulsar/Cargo.toml

@@ -23,3 +23,4 @@ futures     = "0.3"
 
 [dev-dependencies]
 tokio = { version = "1", features = ["full"] }
+serde_json = "1"

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}"),
+    }
+}

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"),
+    }
+}

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"

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;
+}

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;
+}

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"

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(&region, &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());
+        }
+    }
+}