camel-test 0.7.5

//! Integration tests for autoStartup and ControlBus component.
//!
//! These tests verify:
//! - Routes with auto_startup(false) do not start when ctx.start() is called
//! - ControlBus component can start/stop routes dynamically

use std::time::Duration;

use camel_api::{RouteStatus, RuntimeCommand};
use camel_builder::{RouteBuilder, StepAccumulator};
use camel_component_controlbus::ControlBusComponent;
use camel_test::CamelTestContext;

async fn route_status(h: &CamelTestContext, route_id: &str) -> Option<RouteStatus> {
    let ctx = h.ctx().lock().await;
    match ctx
        .runtime_route_status(route_id)
        .await
        .expect("runtime route status query failed")
    {
        Some(status) => match status.as_str() {
            "Stopped" => Some(RouteStatus::Stopped),
            // Registered = pre-start state (route added but never started), equivalent to Stopped for assertions
            "Registered" => Some(RouteStatus::Stopped),
            "Starting" => Some(RouteStatus::Starting),
            "Started" => Some(RouteStatus::Started),
            "Stopping" => Some(RouteStatus::Stopping),
            "Suspended" => Some(RouteStatus::Suspended),
            "Failed" => Some(RouteStatus::Failed("failed".to_string())),
            _ => None,
        },
        None => None,
    }
}

async fn start_route(h: &CamelTestContext, route_id: &str) {
    let runtime = {
        let ctx = h.ctx().lock().await;
        ctx.runtime()
    };

    runtime
        .execute(RuntimeCommand::StartRoute {
            route_id: route_id.to_string(),
            command_id: format!("test:start:{route_id}"),
            causation_id: None,
        })
        .await
        .expect("failed to start route");
}

async fn suspend_route(h: &CamelTestContext, route_id: &str) {
    let runtime = {
        let ctx = h.ctx().lock().await;
        ctx.runtime()
    };

    runtime
        .execute(RuntimeCommand::SuspendRoute {
            route_id: route_id.to_string(),
            command_id: format!("test:suspend:{route_id}"),
            causation_id: None,
        })
        .await
        .expect("failed to suspend route");
}

async fn resume_route(h: &CamelTestContext, route_id: &str) {
    let runtime = {
        let ctx = h.ctx().lock().await;
        ctx.runtime()
    };

    runtime
        .execute(RuntimeCommand::ResumeRoute {
            route_id: route_id.to_string(),
            command_id: format!("test:resume:{route_id}"),
            causation_id: None,
        })
        .await
        .expect("failed to resume route");
}

// ---------------------------------------------------------------------------
// Test 1: Route with auto_startup(false) does not start on ctx.start()
// ---------------------------------------------------------------------------

#[tokio::test]
async fn autostartup_false_route_does_not_start() {
    let h = CamelTestContext::builder()
        .with_timer()
        .with_mock()
        .build()
        .await;

    // Create a route with auto_startup(false)
    let route = RouteBuilder::from("timer:lazy?period=50&repeatCount=3")
        .route_id("lazy-route")
        .auto_startup(false)
        .to("mock:result")
        .build()
        .unwrap();

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

    // Start the context - should NOT start the lazy route
    h.start().await;

    // Check route status is Stopped, not Started
    let status = route_status(&h, "lazy-route").await;
    assert_eq!(
        status,
        Some(RouteStatus::Stopped),
        "Route with auto_startup(false) should remain Stopped after ctx.start()"
    );

    // Wait a bit and verify no exchanges were processed
    tokio::time::sleep(Duration::from_millis(200)).await;

    // The mock endpoint should NOT have received any exchanges
    // because the route was never started
    if let Some(endpoint) = h.mock().get_endpoint("result") {
        let exchanges = endpoint.get_received_exchanges().await;
        assert_eq!(
            exchanges.len(),
            0,
            "Lazy route should not have processed any exchanges"
        );
    }

    h.stop().await;
}

// ---------------------------------------------------------------------------
// Test 2: ControlBus starts a lazy route
// ---------------------------------------------------------------------------

#[tokio::test]
async fn controlbus_starts_route() {
    let h = CamelTestContext::builder()
        .with_timer()
        .with_mock()
        .with_component(ControlBusComponent::new())
        .build()
        .await;

    // Create a lazy route that won't start automatically
    let lazy_route = RouteBuilder::from("timer:lazy?period=50&repeatCount=3")
        .route_id("lazy-route")
        .auto_startup(false)
        .to("mock:lazy-result")
        .build()
        .unwrap();

    // Create a trigger route that starts the lazy route via ControlBus
    // The timer fires once after 50ms, then sends to controlbus to start the lazy route
    let trigger_route = RouteBuilder::from("timer:trigger?period=50&repeatCount=1")
        .route_id("trigger-route")
        .to("controlbus:route?routeId=lazy-route&action=start")
        .to("mock:trigger-done")
        .build()
        .unwrap();

    h.add_route(lazy_route).await.unwrap();
    h.add_route(trigger_route).await.unwrap();

    // Before starting, lazy route should be Stopped
    let status_before = route_status(&h, "lazy-route").await;
    assert_eq!(
        status_before,
        Some(RouteStatus::Stopped),
        "Lazy route should be Stopped before context starts"
    );

    // Start the context - only the trigger route should start
    h.start().await;

    // Trigger route should be Started
    let trigger_status = route_status(&h, "trigger-route").await;
    assert_eq!(
        trigger_status,
        Some(RouteStatus::Started),
        "Trigger route should be Started"
    );

    // Wait for the trigger route to fire and start the lazy route
    tokio::time::sleep(Duration::from_millis(150)).await;

    // Now check that the lazy route was started by the ControlBus
    let status_after = route_status(&h, "lazy-route").await;
    assert_eq!(
        status_after,
        Some(RouteStatus::Started),
        "Lazy route should be Started after ControlBus action"
    );

    // The trigger should have completed
    let trigger_endpoint = h.mock().get_endpoint("trigger-done").unwrap();
    trigger_endpoint.assert_exchange_count(1).await;

    // Give the lazy route time to process its timer exchanges
    tokio::time::sleep(Duration::from_millis(250)).await;

    h.stop().await;

    // The lazy route should have processed its exchanges now that it's started
    let lazy_endpoint = h.mock().get_endpoint("lazy-result").unwrap();
    let lazy_exchanges = lazy_endpoint.get_received_exchanges().await;
    assert!(
        lazy_exchanges.len() >= 3,
        "Lazy route should have processed at least 3 exchanges after being started, got {}",
        lazy_exchanges.len()
    );
}

// ---------------------------------------------------------------------------
// Test 3: Direct API test - start_route via RouteController
// ---------------------------------------------------------------------------

/// This is a simpler version of test_controlbus_starts_route that uses
/// the direct API instead of going through a live route. This is more
/// reliable for testing the core functionality.
#[tokio::test]
async fn route_controller_starts_lazy_route() {
    let h = CamelTestContext::builder()
        .with_timer()
        .with_mock()
        .build()
        .await;

    // Create a lazy route
    let lazy_route = RouteBuilder::from("timer:direct-test?period=50&repeatCount=3")
        .route_id("direct-lazy-route")
        .auto_startup(false)
        .to("mock:direct-result")
        .build()
        .unwrap();

    h.add_route(lazy_route).await.unwrap();

    // Start context - lazy route should NOT start
    h.start().await;

    let status = route_status(&h, "direct-lazy-route").await;
    assert_eq!(
        status,
        Some(RouteStatus::Stopped),
        "Lazy route should be Stopped after ctx.start()"
    );

    // Now use the RouteController directly to start the lazy route
    start_route(&h, "direct-lazy-route").await;

    // Verify the route is now started
    let status_after = route_status(&h, "direct-lazy-route").await;
    assert_eq!(
        status_after,
        Some(RouteStatus::Started),
        "Lazy route should be Started after direct start_route call"
    );

    // Wait for timer to fire
    tokio::time::sleep(Duration::from_millis(250)).await;

    h.stop().await;

    // Verify exchanges were processed
    let endpoint = h.mock().get_endpoint("direct-result").unwrap();
    let exchanges = endpoint.get_received_exchanges().await;
    assert!(
        exchanges.len() >= 3,
        "Route should have processed at least 3 exchanges after being started, got {}",
        exchanges.len()
    );
}

// ---------------------------------------------------------------------------
// Test 4: ControlBus stops a running route
// ---------------------------------------------------------------------------

#[tokio::test]
async fn controlbus_stops_route() {
    let h = CamelTestContext::builder()
        .with_timer()
        .with_mock()
        .with_component(ControlBusComponent::new())
        .build()
        .await;

    // Create a route that starts automatically
    let auto_route = RouteBuilder::from("timer:auto?period=100&repeatCount=10")
        .route_id("auto-route")
        .auto_startup(true)
        .to("mock:auto-result")
        .build()
        .unwrap();

    // Create a trigger route that stops the auto route after a delay
    let trigger_route = RouteBuilder::from("timer:stop-trigger?period=50&repeatCount=1")
        .route_id("stop-trigger-route")
        .to("controlbus:route?routeId=auto-route&action=stop")
        .to("mock:stop-done")
        .build()
        .unwrap();

    h.add_route(auto_route).await.unwrap();
    h.add_route(trigger_route).await.unwrap();

    h.start().await;

    // Both routes should start
    let auto_status = route_status(&h, "auto-route").await;
    assert_eq!(
        auto_status,
        Some(RouteStatus::Started),
        "Auto route should be Started"
    );

    // Wait for the stop trigger to fire
    tokio::time::sleep(Duration::from_millis(150)).await;

    // Now the auto route should be stopped
    let status_after = route_status(&h, "auto-route").await;
    assert_eq!(
        status_after,
        Some(RouteStatus::Stopped),
        "Auto route should be Stopped after ControlBus stop action"
    );

    h.stop().await;

    // Verify the stop trigger executed
    let stop_endpoint = h.mock().get_endpoint("stop-done").unwrap();
    stop_endpoint.assert_exchange_count(1).await;
}

// ---------------------------------------------------------------------------
// Test 5: Multiple routes with different startup orders
// ---------------------------------------------------------------------------

#[tokio::test]
async fn startup_order_respected() {
    let h = CamelTestContext::builder()
        .with_timer()
        .with_mock()
        .build()
        .await;

    // Create routes with different startup orders
    // Lower startup_order values start first
    let route1 = RouteBuilder::from("timer:order1?period=50&repeatCount=1")
        .route_id("route-order-1")
        .startup_order(10)
        .to("mock:order1")
        .build()
        .unwrap();

    let route2 = RouteBuilder::from("timer:order2?period=50&repeatCount=1")
        .route_id("route-order-2")
        .startup_order(5) // This should start first (lower value)
        .to("mock:order2")
        .build()
        .unwrap();

    let route3 = RouteBuilder::from("timer:order3?period=50&repeatCount=1")
        .route_id("route-order-3")
        .startup_order(20)
        .to("mock:order3")
        .build()
        .unwrap();

    h.add_route(route1).await.unwrap();
    h.add_route(route2).await.unwrap();
    h.add_route(route3).await.unwrap();

    h.start().await;

    // All routes should be started
    assert_eq!(
        route_status(&h, "route-order-1").await,
        Some(RouteStatus::Started)
    );
    assert_eq!(
        route_status(&h, "route-order-2").await,
        Some(RouteStatus::Started)
    );
    assert_eq!(
        route_status(&h, "route-order-3").await,
        Some(RouteStatus::Started)
    );

    // Wait for timers to fire
    tokio::time::sleep(Duration::from_millis(200)).await;

    h.stop().await;

    // All routes should have processed their exchanges
    let ep1 = h.mock().get_endpoint("order1").unwrap();
    let ep2 = h.mock().get_endpoint("order2").unwrap();
    let ep3 = h.mock().get_endpoint("order3").unwrap();

    ep1.assert_exchange_count(1).await;
    ep2.assert_exchange_count(1).await;
    ep3.assert_exchange_count(1).await;
}

// ---------------------------------------------------------------------------
// Test 6: Mixed auto_startup routes
// ---------------------------------------------------------------------------

#[tokio::test]
async fn mixed_autostartup_routes() {
    let h = CamelTestContext::builder()
        .with_timer()
        .with_mock()
        .build()
        .await;

    // Route that starts automatically
    let auto_route = RouteBuilder::from("timer:auto-start?period=50&repeatCount=2")
        .route_id("auto-start-route")
        .auto_startup(true)
        .to("mock:auto")
        .build()
        .unwrap();

    // Route that does NOT start automatically
    let lazy_route = RouteBuilder::from("timer:lazy-start?period=50&repeatCount=2")
        .route_id("lazy-start-route")
        .auto_startup(false)
        .to("mock:lazy")
        .build()
        .unwrap();

    h.add_route(auto_route).await.unwrap();
    h.add_route(lazy_route).await.unwrap();

    h.start().await;

    // Auto route should be started
    assert_eq!(
        route_status(&h, "auto-start-route").await,
        Some(RouteStatus::Started)
    );

    // Lazy route should be stopped
    assert_eq!(
        route_status(&h, "lazy-start-route").await,
        Some(RouteStatus::Stopped)
    );

    // Wait for auto route to process
    tokio::time::sleep(Duration::from_millis(200)).await;

    // Auto route should have processed exchanges
    let auto_ep = h.mock().get_endpoint("auto").unwrap();
    auto_ep.assert_exchange_count(2).await;

    // Lazy route should NOT have processed any
    if let Some(lazy_ep) = h.mock().get_endpoint("lazy") {
        let exchanges = lazy_ep.get_received_exchanges().await;
        assert_eq!(
            exchanges.len(),
            0,
            "Lazy route should not have processed any exchanges"
        );
    }

    h.stop().await;
}

// ---------------------------------------------------------------------------
// Test 7: Suspend/Resume Lifecycle - Basic State Transitions
// ---------------------------------------------------------------------------

/// Test that suspend_route changes route status to Suspended
#[tokio::test]
async fn suspend_changes_status_to_suspended() {
    let h = CamelTestContext::builder()
        .with_timer()
        .with_mock()
        .build()
        .await;

    // Create a route that starts automatically
    let route = RouteBuilder::from("timer:suspend-test?period=100&repeatCount=10")
        .route_id("suspend-route")
        .auto_startup(true)
        .to("mock:suspend-result")
        .build()
        .unwrap();

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

    // Route should be started
    assert_eq!(
        route_status(&h, "suspend-route").await,
        Some(RouteStatus::Started),
        "Route should be Started initially"
    );

    // Suspend the route
    suspend_route(&h, "suspend-route").await;

    // Status should be Suspended
    assert_eq!(
        route_status(&h, "suspend-route").await,
        Some(RouteStatus::Suspended),
        "Route should be Suspended after suspend_route call"
    );

    h.stop().await;
}

/// Test that resume_route changes status back to Started
#[tokio::test]
async fn resume_changes_status_to_started() {
    let h = CamelTestContext::builder()
        .with_timer()
        .with_mock()
        .build()
        .await;

    let route = RouteBuilder::from("timer:resume-test?period=100&repeatCount=10")
        .route_id("resume-route")
        .auto_startup(true)
        .to("mock:resume-result")
        .build()
        .unwrap();

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

    // Suspend the route
    suspend_route(&h, "resume-route").await;

    assert_eq!(
        route_status(&h, "resume-route").await,
        Some(RouteStatus::Suspended),
        "Route should be Suspended"
    );

    // Resume the route
    resume_route(&h, "resume-route").await;

    // Status should be Started again
    assert_eq!(
        route_status(&h, "resume-route").await,
        Some(RouteStatus::Started),
        "Route should be Started after resume_route call"
    );

    h.stop().await;
}

// ---------------------------------------------------------------------------
// Test 8: Suspend Drains In-Flight Messages
// ---------------------------------------------------------------------------

/// Test that suspend waits for in-flight messages to complete before returning.
///
/// INTENDED BEHAVIOR:
/// When suspend_route is called while messages are being processed, it should:
/// 1. Stop accepting new messages from the consumer
/// 2. Wait for all in-flight pipeline processing to complete
/// 3. Only then return and mark the route as Suspended
///
/// EXPECTED RESULT:
/// This test currently PASSES because the implementation waits for tasks to finish,
/// but it doesn't truly "drain" in a graceful manner - it cancels and waits.
/// The test verifies that at minimum, messages are not lost.
#[tokio::test]
async fn suspend_drains_inflight_messages() {
    // Timing constants
    const TIMER_PERIOD_MS: u64 = 50;
    const INITIAL_FLOW_MS: u64 = 120; // Let ~2 messages through

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

    let route = RouteBuilder::from(&format!(
        "timer:drain-test?period={TIMER_PERIOD_MS}&repeatCount=5"
    ))
    .route_id("drain-route")
    .auto_startup(true)
    .to("mock:drain-result")
    .build()
    .unwrap();

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

    // Let some messages be sent (but not all)
    tokio::time::sleep(Duration::from_millis(INITIAL_FLOW_MS)).await;

    let endpoint = h.mock().get_endpoint("drain-result").unwrap();
    let count_before_suspend = endpoint.get_received_exchanges().await.len();

    // Suspend should wait for in-flight messages to complete
    let suspend_start = std::time::Instant::now();
    suspend_route(&h, "drain-route").await;
    let suspend_duration = suspend_start.elapsed();

    let count_after_suspend = endpoint.get_received_exchanges().await.len();

    // With proper drain: messages complete before suspend returns
    assert!(
        count_after_suspend >= count_before_suspend,
        "Suspend should drain in-flight messages. Before: {}, After: {}, Duration: {:?}",
        count_before_suspend,
        count_after_suspend,
        suspend_duration
    );

    h.stop().await;
}

// ---------------------------------------------------------------------------
// Test 9: Suspend Blocks New Intake Until Resume
// ---------------------------------------------------------------------------

/// Test that a suspended route does not accept new messages until resumed.
///
/// INTENDED BEHAVIOR:
/// When a route is suspended:
/// 1. The consumer should stop producing new messages
/// 2. No new exchanges should enter the pipeline while suspended
/// 3. After resume, the consumer should start producing messages again
///
/// CURRENT IMPLEMENTATION:
/// There's a race condition where messages can slip through during the
/// suspension window. This test accounts for that by:
/// - Waiting for a settling period after suspend
/// - Checking that messages definitively stop after settling
/// - Verifying resume restarts message flow
#[tokio::test]
async fn suspend_blocks_new_intake_until_resume() {
    // Timing constants
    const TIMER_PERIOD_MS: u64 = 20; // Fast timer to ensure continuous flow
    const INITIAL_FLOW_MS: u64 = 100; // Let messages flow before suspend
    const SETTLING_MS: u64 = 50; // Wait for cancellation to fully take effect
    const SUSPENDED_OBSERVATION_MS: u64 = 150; // Observe while suspended
    const POST_RESUME_FLOW_MS: u64 = 100; // Let messages flow after resume

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

    let route = RouteBuilder::from(&format!(
        "timer:block-test?period={TIMER_PERIOD_MS}&repeatCount=50"
    ))
    .route_id("block-route")
    .auto_startup(true)
    .to("mock:block-result")
    .build()
    .unwrap();

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

    // Let messages flow before suspend
    tokio::time::sleep(Duration::from_millis(INITIAL_FLOW_MS)).await;

    let endpoint = h.mock().get_endpoint("block-result").unwrap();
    let count_before_suspend = endpoint.get_received_exchanges().await.len();

    // Suspend the route
    suspend_route(&h, "block-route").await;

    // Wait for consumer cancellation to fully take effect (avoid race window)
    tokio::time::sleep(Duration::from_millis(SETTLING_MS)).await;

    // Get baseline count after settling - messages may have slipped through during race
    let count_after_settling = endpoint.get_received_exchanges().await.len();

    // Wait longer while suspended - no NEW messages should arrive after settling
    tokio::time::sleep(Duration::from_millis(SUSPENDED_OBSERVATION_MS)).await;

    let count_while_suspended = endpoint.get_received_exchanges().await.len();

    // Key assertion: After settling, no new messages should arrive while suspended
    // This is robust to the race condition at suspend time
    assert_eq!(
        count_while_suspended, count_after_settling,
        "After settling, no new messages should arrive while suspended. \
         Before suspend: {}, After settling: {}, While suspended: {}",
        count_before_suspend, count_after_settling, count_while_suspended
    );

    // Resume the route - messages should start flowing again
    resume_route(&h, "block-route").await;

    // Wait for messages after resume
    tokio::time::sleep(Duration::from_millis(POST_RESUME_FLOW_MS)).await;

    let count_after_resume = endpoint.get_received_exchanges().await.len();

    // Verify messages flow again after resume
    assert!(
        count_after_resume > count_while_suspended,
        "Messages should flow again after resume. While suspended: {}, After resume: {}",
        count_while_suspended,
        count_after_resume
    );

    h.stop().await;
}