Skip to main content

Testing a Pub/Sub service

A test runs the service's own app with TestApp, against an in-process Pub/Sub or against the emulator.

The service under test sends refunds. A refund arrives on refunds, and the handler answers it with a Refunded message on refunds-done. A refund without an amount is dropped. A refund above the approval limit waits a minute for approval with retry_after. After five deliveries a refund goes to refunds-review, the dead-letter topic. The app is built by service(broker), so a test chooses the broker and main uses the one for the emulator. The tests need the testing feature of ruststream-gcp-pubsub in [dev-dependencies].

A refund answered​

src/main.rs
/// The body both modes run: a refund is answered on `refunds-done`.
async fn a_refund_is_answered(tb: TestApp<()>) -> Result<(), Box<dyn Error>> {
    tb.broker::<PubSubBroker>()
        .message(&Refund {
            id: 7,
            amount_cents: 1250,
        })
        .publish()
        .await?;

    tb.broker::<PubSubBroker>()
        .subscriber("refunds-workers")
        .assert_called_once()
        .with(&Refund {
            id: 7,
            amount_cents: 1250,
        })
        .settled(HandlerOutcome::ack());
    tb.broker::<PubSubBroker>()
        .published::<Refunded>("refunds-done")
        .assert_called_once()
        .with(&Refunded { refund_id: 7 });
    tb.shutdown().await?;
    Ok(())
}

#[tokio::test]
async fn in_process() -> Result<(), Box<dyn Error>> {
    a_refund_is_answered(TestApp::start(app()).await?).await
}

TestApp::start connects PubSubBroker to an in-process Pub/Sub. Nothing is dialled and no credentials are read. The test addresses the broker by its production type, tb.broker::<PubSubBroker>(). Refund names its topic, so message(..).publish() needs no destination. publish returns once the handler and its reply have settled. So the next lines assert on finished state: how the subscriber settled and what reached refunds-done. The body takes the started app as an argument, so the live test below runs it too.

A refund dropped​

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

    tb.broker::<PubSubBroker>()
        .message(&Refund {
            id: 8,
            amount_cents: 0,
        })
        .publish()
        .await?;

    tb.broker::<PubSubBroker>()
        .subscriber("refunds-workers")
        .assert_called_once()
        .settled(HandlerOutcome::drop());
    tb.broker::<PubSubBroker>()
        .published::<Refunded>("refunds-done")
        .assert_not_called();
    tb.shutdown().await?;
    Ok(())
}

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

A delay, without waiting for it​

src/main.rs
#[tokio::test(start_paused = true)]
async fn a_refund_never_approved_goes_to_review() -> Result<(), Box<dyn Error>> {
    let tb = TestApp::start(app()).await?;
    let large = Refund {
        id: 9,
        amount_cents: 250_000,
    };

    tb.broker::<PubSubBroker>()
        .message(&large)
        .publish()
        .await?;
    tb.broker::<PubSubBroker>()
        .subscriber("refunds-workers")
        .assert_called_once()
        .settled(HandlerOutcome::retry_after(APPROVAL_WAIT));

    // Each wait ends in a redelivery, and the fifth delivery is the last one allowed.
    for deliveries in 2..=5 {
        tb.advance(APPROVAL_WAIT).await?;
        tb.broker::<PubSubBroker>()
            .subscriber("refunds-workers")
            .assert_called(deliveries);
    }
    // The fifth delivery waits too; when its wait ends, Pub/Sub moves the refund to review.
    tb.advance(APPROVAL_WAIT).await?;
    tb.broker::<PubSubBroker>()
        .published::<Refund>("refunds-review")
        .assert_called_once()
        .with(&large);
    tb.shutdown().await?;
    Ok(())
}

The test starts on a paused clock, and advance moves it forward. A minute of APPROVAL_WAIT costs nothing. Each wait ends in a redelivery of the same message, and assert_called counts them. The fifth delivery waits too, and when its wait ends Pub/Sub moves the refund to refunds-review as it arrived. published reads what reached a topic, whether the service or Pub/Sub put it there.

What the in-process Pub/Sub models​

The in-process Pub/Sub reads the production broker's project and every descriptor the service mounts. A publish, a subscription or a declaration Pub/Sub refuses fails the test the same way. It models Pub/Sub rather than a queue:

  • Topics and subscriptions are separate, and each subscription of a topic receives every message published after it was created.
  • The consumers of one subscription share its messages.
  • retry delivers a message again, retry_after waits on the test's clock first, and drop acknowledges.
  • A dead-letter policy counts the deliveries and moves a message that ran out of them.

The topology the service expects to find is taken to exist. A subscription the service only names reads the topic of its own name, and a topic the service publishes to accepts the message. Leases, flow control, ordered delivery by key and exactly-once delivery need the emulator or Pub/Sub itself.

The same test against the emulator​

src/main.rs
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn against_the_emulator() -> Result<(), Box<dyn Error>> {
    let Some(host) = live::host() else {
        return Ok(());
    };
    live::fresh(&host, &["refunds-workers"]).await?;
    create_topic(&host, "refunds-done").await?;
    let broker = PubSubBroker::new("my-project").emulator(host);
    a_refund_is_answered(TestApp::start_live(service(broker)).await?).await
}

TestApp::start_live connects the same app to the emulator. A publish then waits up to ten seconds for every subscription it reaches to handle it; start_live_within sets another deadline. The test reads the emulator's address from PUBSUB_TEST_HOST with live::host, a helper of the example's own, and skips when it is unset. The emulator keeps its subscriptions between runs, so live::fresh deletes refunds-workers first, and the run reads only what it publishes. create_topic creates refunds-done the way operations would, because a reply to a topic that is not there fails. The assertions are the in-process ones, unchanged.

cargo test
PUBSUB_TEST_HOST=localhost:8085 cargo test -- --test-threads=1