camel-test 0.28.0

Testing utilities for rust-camel
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144

//! Integration tests for the MQTT component (rumqttc).
//!
//! Uses testcontainers to spin up an Eclipse Mosquitto broker.
//! **Requires Docker** and the `integration-tests` feature.

#![cfg(feature = "integration-tests")]

mod support;
use support::install_crypto_provider;

use camel_builder::{RouteBuilder, StepAccumulator};
use camel_component_mqtt::MqttComponent;
use camel_component_mqtt::config::{MqttBrokerConfig, MqttConfig};
use camel_component_mqtt::headers::CAMEL_MQTT_TOPIC;
use camel_test::CamelTestContext;
use support::mosquitto::shared_mosquitto;
use support::wait::wait_until;

fn mqtt_component_for(host_port: &str) -> MqttComponent {
    let config = MqttConfig {
        brokers: [(
            "test".to_string(),
            MqttBrokerConfig {
                url: format!("mqtt://{host_port}"),
                username: None,
                password: None,
                tls_ca_cert: None,
            },
        )]
        .into_iter()
        .collect(),
        ..Default::default()
    };
    MqttComponent::with_config(config).expect("mqtt config")
}

// ===========================================================================
// Producer test — timer → mqtt, assert the exchange reaches mock:result
// ===========================================================================
#[tokio::test(flavor = "multi_thread")]
async fn mqtt_producer_publishes_without_error() {
    install_crypto_provider();
    let (_container, host_port) = shared_mosquitto().await;

    let h = CamelTestContext::builder()
        .with_timer()
        .with_mock()
        .with_component(mqtt_component_for(host_port))
        .build()
        .await;

    let route = RouteBuilder::from("timer:tick?period=50&repeatCount=1")
        .set_body(camel_api::Value::String("hello-mqtt".to_string()))
        .to("mqtt://test/sensors/temp")
        .to("mock:result")
        .route_id("mqtt-producer-test")
        .build()
        .unwrap();

    h.add_route(route).await.unwrap();
    h.start().await;

    let endpoint = h.mock().get_endpoint("result").unwrap();
    wait_until(
        "mqtt producer route delivery",
        std::time::Duration::from_secs(5),
        std::time::Duration::from_millis(100),
        || {
            let endpoint = endpoint.clone();
            async move { Ok(!endpoint.get_received_exchanges().await.is_empty()) }
        },
    )
    .await
    .unwrap();

    h.stop().await;

    // Check no errors reached the error endpoint
    if let Some(error_ep) = h.mock().get_endpoint("error") {
        let errors = error_ep.get_received_exchanges().await;
        if !errors.is_empty() {
            panic!("Producer route had errors: {:?}", errors[0].error);
        }
    }

    endpoint.assert_exchange_count(1).await;
}

// ===========================================================================
// Consumer test — consumer subscribes; producer publishes; consumer receives
// ===========================================================================
#[tokio::test(flavor = "multi_thread")]
async fn mqtt_consumer_receives_published_message() {
    install_crypto_provider();
    let (_container, host_port) = shared_mosquitto().await;

    let h = CamelTestContext::builder()
        .with_timer()
        .with_mock()
        .with_component(mqtt_component_for(host_port))
        .build()
        .await;

    // Consumer: subscribe to sensors/# (wildcard) → mock:consumed
    let consumer_route = RouteBuilder::from("mqtt://test/sensors/#")
        .to("mock:consumed")
        .route_id("mqtt-consumer-test")
        .build()
        .unwrap();

    // Producer: delayed timer so the consumer connects + subscribes first.
    let producer_route = RouteBuilder::from("timer:produce?period=3000&delay=3000&repeatCount=1")
        .set_body(camel_api::Value::String("hello-mqtt".to_string()))
        .to("mqtt://test/sensors/temp")
        .route_id("mqtt-producer-for-consumer-test")
        .build()
        .unwrap();

    h.add_route(consumer_route).await.unwrap();
    h.add_route(producer_route).await.unwrap();
    h.start().await;

    let endpoint = h.mock().get_endpoint("consumed").unwrap();
    wait_until(
        "mqtt consumer receives message",
        std::time::Duration::from_secs(20),
        std::time::Duration::from_millis(200),
        || {
            let endpoint = endpoint.clone();
            async move { Ok(!endpoint.get_received_exchanges().await.is_empty()) }
        },
    )
    .await
    .unwrap();

    h.stop().await;

    // The received exchange carries the MQTT topic header
    let ex = &endpoint.get_received_exchanges().await[0];
    assert_eq!(
        ex.input.header(CAMEL_MQTT_TOPIC),
        Some(&camel_api::Value::String("sensors/temp".to_string())),
    );
}