Skip to main content

Testing an SQS service

A test runs the service's own app with TestApp, against an in-process SQS or against a running stack.

The service under test is a checkout. An order arrives on checkout-orders, and confirm answers it with a confirmation on checkout-confirmations, which notify reads. An order without an amount is dropped. A back-ordered order waits an hour with retry_after and comes back to be checked again. The app is built by service(broker), so a test chooses the broker and main uses the production one. The tests need the testing feature of ruststream-sqs-sns in [dev-dependencies].

An order and its confirmation​

src/main.rs
#[tokio::test]
async fn an_order_is_confirmed() -> Result<(), Box<dyn Error>> {
    let tb = TestApp::start(service(SqsBroker::new())).await?;
    a_confirmation_follows_the_order(tb).await
}

async fn a_confirmation_follows_the_order(tb: TestApp<()>) -> Result<(), Box<dyn Error>> {
    tb.broker::<SqsBroker>()
        .message(&order(1, 2_500, false))
        .publish()
        .await?;

    tb.broker::<SqsBroker>()
        .subscriber("checkout-orders")
        .assert_called_once()
        .with(&order(1, 2_500, false))
        .settled(HandlerOutcome::ack());
    tb.broker::<SqsBroker>()
        .subscriber("checkout-confirmations")
        .assert_called_once()
        .with(&Confirmation {
            order: 1,
            amount_cents: 2_500,
        });
    tb.shutdown().await?;
    Ok(())
}

TestApp::start connects SqsBroker to an in-process SQS. No endpoint is called and no credentials are needed. The test addresses the broker by its production type, tb.broker::<SqsBroker>(). publish returns once the handlers it reached have settled, the reply and its reader included. So the next lines assert on finished state: how confirm settled the order and what notify received. The body takes the started app as an argument, so the live test below runs it too.

An empty order​

src/main.rs
#[tokio::test]
async fn an_empty_order_is_dropped() -> Result<(), Box<dyn Error>> {
    let tb = TestApp::start(service(SqsBroker::new())).await?;

    tb.broker::<SqsBroker>()
        .message(&order(2, 0, false))
        .publish()
        .await?;

    tb.broker::<SqsBroker>()
        .subscriber("checkout-orders")
        .assert_called_once()
        .settled(HandlerOutcome::drop());
    tb.broker::<SqsBroker>()
        .published::<Confirmation>("checkout-confirmations")
        .assert_not_called();
    tb.shutdown().await?;
    Ok(())
}

settled checks the outcome the handler returned. assert_not_called checks that no confirmation was published.

A delay, without waiting for it​

src/main.rs
#[tokio::test(start_paused = true)]
async fn a_back_ordered_order_comes_back_after_the_restock() -> Result<(), Box<dyn Error>> {
    let tb = TestApp::start(service(SqsBroker::new())).await?;
    let waiting = order(3, 2_500, true);

    tb.broker::<SqsBroker>().message(&waiting).publish().await?;
    tb.advance(RESTOCK).await?;

    tb.broker::<SqsBroker>()
        .subscriber("checkout-orders")
        .assert_called(2)
        .with(&waiting)
        .settled(HandlerOutcome::retry_after(RESTOCK));
    // The same message came back; nothing was published again to bring it.
    tb.broker::<SqsBroker>()
        .published::<Order>("checkout-orders")
        .assert_called_once();
    tb.shutdown().await?;
    Ok(())
}

The test starts on a paused clock, and advance moves it forward. An hour of RESTOCK costs nothing. retry_after makes the message invisible for the delay, as SQS does, and the queue hands it out again when the clock passes it. published shows one order on checkout-orders: the second delivery is the same message, not a copy.

What the in-process SQS models​

The in-process SQS is an account inside the test, and every queue and topic the service names exists in it. It models what a service's logic depends on:

  • A queue hands each message to one of its subscriptions, and a received message stays invisible until it is deleted or its visibility timeout ends.
  • The receive count grows with every receive, and a redrive policy moves a message whose receives ran out to the dead-letter queue.
  • A batch holds at most the ten messages one receive returns.
  • 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 with subscribe_queue_to_topic.

What SQS or SNS refuses is refused there too: more than ten headers, a header name SQS 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 standard queue keeps the order of its messages there, which SQS does not promise. A descriptor that names no visibility gets 30 seconds, the default of a new queue. A message sent to a queue no subscription reads is dropped. The visibility an operator set on a queue, the queues and topics the service expects to find, and credentials are exercised against a running stack.

The same test on a running stack​

src/main.rs
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn an_order_is_confirmed_on_a_live_stack() -> Result<(), Box<dyn Error>> {
    let Some(endpoint) = live::endpoint("SQS_TEST_ENDPOINT") else {
        return Ok(());
    };
    delete_queues(&endpoint).await?;
    let tb = TestApp::start_live(service(live::broker(&endpoint))).await?;
    a_confirmation_follows_the_order(tb).await
}

TestApp::start_live connects the same app to a running SQS. The test reads the endpoint from SQS_TEST_ENDPOINT with live::endpoint, a helper of the example's own, and skips when it is unset. It deletes the service's queues first, so nothing an earlier run left on them is counted; the service creates them again on start. The assertions are the in-process ones, unchanged.

cargo test
SQS_TEST_ENDPOINT=http://localhost:4566 cargo test -- --test-threads=1