Перейти к основному содержимому
Нормативная версия документации - английская

Эта страница переведена с английского языковой моделью. При любом расхождении верен английский оригинал.

Тестирование сервиса на Pub/Sub

Тест запускает собственное приложение сервиса через TestApp на внутрипроцессном Pub/Sub или на эмуляторе.

Тестируемый сервис проводит возвраты. Возврат приходит в refunds, и обработчик отвечает на него сообщением Refunded в refunds-done. Возврат без суммы отбрасывается. Возврат сверх лимита согласования ждёт согласования минуту с retry_after. После пяти доставок возврат уходит в refunds-review, тему dead-letter. Приложение строит service(broker), поэтому тест выбирает брокер сам, а main берёт брокер для эмулятора. Тестам нужна фича testing крейта ruststream-gcp-pubsub в [dev-dependencies].

Возврат с ответом​

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 подключает PubSubBroker к внутрипроцессному Pub/Sub. Соединения не устанавливаются, и учётные данные не читаются. Тест обращается к брокеру по его рабочему типу: tb.broker::<PubSubBroker>(). Refund называет свою тему, поэтому message(..).publish() не нужен адрес. publish возвращает управление, когда обработчик и его ответ завершены. Поэтому следующие строки проверяют уже готовое состояние: как подписчик завершил доставку и что дошло до refunds-done. Тело теста принимает запущенное приложение аргументом, поэтому его же выполняет и тест на эмуляторе ниже.

Отброшенный возврат​

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 проверяет исход, который вернул обработчик. assert_not_called проверяет, что ответ не опубликован.

Задержка без ожидания​

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(())
}

Тест начинается с остановленными часами, а advance сдвигает их вперёд. Минута APPROVAL_WAIT не стоит ничего. Каждое ожидание заканчивается повторной доставкой того же сообщения, и assert_called их считает. Пятая доставка тоже ждёт, а когда её ожидание заканчивается, Pub/Sub переносит возврат в refunds-review в том виде, в каком он пришёл. published читает то, что дошло до темы, кто бы его туда ни поместил: сервис или Pub/Sub.

Что моделирует внутрипроцессный Pub/Sub​

Внутрипроцессный Pub/Sub читает проект рабочего брокера и каждый дескриптор, который монтирует сервис. Публикация, подписка или объявление, которые отклонил бы Pub/Sub, так же прерывают тест. Он моделирует Pub/Sub, а не очередь:

  • Темы и подписки разделены, и каждая подписка на тему получает каждое сообщение, опубликованное после её создания.
  • Потребители одной подписки делят её сообщения между собой.
  • retry доставляет сообщение снова, retry_after сначала ждёт на часах теста, а drop подтверждает.
  • Политика dead-letter считает доставки и переносит сообщение, у которого они кончились.

Топология, которую сервис ожидает найти, считается существующей. Подписка, которую сервис только называет, читает одноимённую тему, а тема, в которую публикует сервис, принимает сообщение. Для аренды, управления потоком, упорядоченной доставки по ключу и доставки exactly-once нужен эмулятор или сам Pub/Sub.

Тот же тест на эмуляторе​

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 подключает то же приложение к эмулятору. Тогда публикация ждёт до десяти секунд, пока её обработает каждая подписка, до которой она дошла. Другой срок задаёт start_live_within. Тест читает адрес эмулятора из PUBSUB_TEST_HOST через live::host, собственную вспомогательную функцию примера, и пропускается, если переменная не задана. Эмулятор хранит подписки между запусками, поэтому live::fresh сначала удаляет refunds-workers, и запуск читает только то, что публикует сам. create_topic создаёт refunds-done так, как это сделала бы эксплуатация, потому что ответ в несуществующую тему возвращает ошибку. Проверки те же, что внутри процесса, без изменений.

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