Testing

TestApp::start connects SqsBroker to an in-process account with queues, visibility, receive counts, redrive, FIFO groups and topics.

A test hands the harness the app main runs, unchanged, and addresses the broker by its production type, tb.broker::<SqsBroker>(). The testing feature in [dev-dependencies] adds the in-process mode, and turns on the core's harness with it.

The account has no settings of its own. Its queue URLs are built on the broker's endpoint and region, and every setting a queue has comes from the service's descriptors. Every queue and topic the service names exists in it.

The account keeps SQS's semantics. A queue hands each message to one of its subscriptions, and one receive returns at most ten messages. A received message stays invisible until it is deleted or its visibility timeout lapses; a descriptor that names no visibility gets the 30 seconds of a new queue. The receive count rises with every receive, and a redrive policy moves a message whose receives ran out to the dead-letter queue. A FIFO group waits while one of its messages is in flight, and a deduplication id is remembered for five minutes. With sns, a topic delivers a copy to every queue subscribed to it.

The account refuses what SQS and SNS refuse: more than ten headers, a header name the service does not take, an empty body or header value, a message over the size limit, a queue name over 80 characters, a dead-letter queue of the other kind, a maxReceiveCount over 1000, a visibility that would end more than twelve hours after the receive, and a FIFO queue on a standard topic. A publish through the harness that SQS would refuse fails the test. A standard queue keeps the order of its messages here, which SQS does not promise. A message sent to a queue no subscription reads is dropped.

A publish through the harness returns once every handler it woke has settled. tb.advance(..), on a clock paused with #[tokio::test(start_paused = true)], moves past a visibility timeout or a deferred retry. What belongs to a running service, such as a queue an operator configured, is for a live test.

Example

The dependency line:

[dev-dependencies]
ruststream-sqs-sns = { version = "0.7", features = ["testing"] }

The first test publishes a payment and reads how the handler settled it.

Macros
Manual

examples/sqs_testing.rs
#[tokio::test]
async fn an_accepted_card_is_charged() -> Result<(), Box<dyn Error>> {
    let tb = TestApp::start(app()).await?;
    let payment = Payment {
        id: 7,
        card_accepted: true,
    };
    tb.broker::<SqsBroker>().message(&payment).publish().await?;

    tb.broker::<SqsBroker>()
        .subscriber("payments")
        .assert_called_once()
        .with(&payment)
        .settled(HandlerOutcome::ack());
    tb.shutdown().await?;
    Ok(())
}

examples/manual/sqs_testing.rs
#[tokio::test]
async fn an_accepted_card_is_charged() -> Result<(), Box<dyn Error>> {
    let tb = TestApp::start(app()).await?;
    let payment = Payment {
        id: 7,
        card_accepted: true,
    };
    tb.broker::<SqsBroker>().message(&payment).publish().await?;

    tb.broker::<SqsBroker>()
        .subscriber("payments")
        .assert_called_once()
        .with(&payment)
        .settled(HandlerOutcome::ack());
    tb.shutdown().await?;
    Ok(())
}

A declined card is retried after a minute, and the queue's redrive policy moves the payment to the dead-letter queue after the third receive. The test advances the paused clock past each delay.

Macros
Manual

examples/sqs_testing.rs
#[tokio::test(start_paused = true)]
async fn a_declined_card_ends_in_the_dead_letter_queue() -> Result<(), Box<dyn Error>> {
    let tb = TestApp::start(app()).await?;
    let payment = Payment {
        id: 8,
        card_accepted: false,
    };
    tb.broker::<SqsBroker>().message(&payment).publish().await?;

    // Each delay hands the payment back; after the third receive the queue moves it.
    for _ in 0..3 {
        tb.advance(RETRY_DELAY).await?;
    }

    tb.broker::<SqsBroker>()
        .subscriber("payments")
        .assert_called(3)
        .settled(HandlerOutcome::retry_after(RETRY_DELAY));
    tb.broker::<SqsBroker>()
        .published::<Payment>("payments-dead")
        .assert_called_once()
        .with(&payment);
    tb.shutdown().await?;
    Ok(())
}

examples/manual/sqs_testing.rs
#[tokio::test(start_paused = true)]
async fn a_declined_card_ends_in_the_dead_letter_queue() -> Result<(), Box<dyn Error>> {
    let tb = TestApp::start(app()).await?;
    let payment = Payment {
        id: 8,
        card_accepted: false,
    };
    tb.broker::<SqsBroker>().message(&payment).publish().await?;

    // Each delay hands the payment back; after the third receive the queue moves it.
    for _ in 0..3 {
        tb.advance(RETRY_DELAY).await?;
    }

    tb.broker::<SqsBroker>()
        .subscriber("payments")
        .assert_called(3)
        .settled(HandlerOutcome::retry_after(RETRY_DELAY));
    tb.broker::<SqsBroker>()
        .published::<Payment>("payments-dead")
        .assert_called_once()
        .with(&payment);
    tb.shutdown().await?;
    Ok(())
}

Example​

Example