編寫長時間執行作業的測試

Rust 的 Google Cloud 用戶端程式庫提供輔助程式,可簡化與長時間執行的作業 (LRO) 的互動。如要在單元測試中模擬 LRO 的行為,必須瞭解這些輔助函式抽象化的基礎細節。本指南說明如何為自動和手動 LRO 輪詢編寫單元測試,包括錯誤情境。

必要條件

本指南假設您已熟悉下列指南:

測試自動輪詢

假設應用程式程式碼等待 lro::Poller::until_done() (自動輪詢):

// An example application function that automatically polls.
//
// It starts an LRO, awaits the result, and processes it.
pub async fn my_automatic_poller(
    client: &speech::client::Speech,
    project_id: &str,
) -> Result<Option<wkt::Duration>> {
    use google_cloud_lro::Poller;
    client
        .batch_recognize()
        .set_recognizer(format!(
            "projects/{project_id}/locations/global/recognizers/_"
        ))
        .poller()
        .until_done()
        .await
        .map(|r| r.total_billed_duration)
}

由於應用程式只會使用 LRO 的最終結果,因此您不需要測試應用程式如何處理中繼輪詢結果。測試可以直接從模擬傳回 LRO 的最終結果。

建立 longrunning::model::Operation

假設您希望呼叫傳回下列回應:

fn expected_duration() -> Option<wkt::Duration> {
    Some(wkt::Duration::clamp(100, 0))
}

fn expected_response() -> BatchRecognizeResponse {
    BatchRecognizeResponse::new().set_or_clear_total_billed_duration(expected_duration())
}

虛設常式會傳回 longrunning::model::Operation,而非 BatchRecognizeResponse。將預期回應封裝至 Operation::result:

fn make_finished_operation(response: &BatchRecognizeResponse) -> Result<Response<Operation>> {
    let any = wkt::Any::from_msg(response).expect("test message should succeed");
    let operation = Operation::new()
        .set_done(true)
        .set_result(OperationResult::Response(any.into()));
    Ok(Response::from(operation))
}

將 done 欄位設為 true,向 Poller 指出作業已完成,結束輪詢迴圈:

.set_done(true)

編寫測試

  1. 定義模擬結構,實作 speech::stub::Speech 特徵:

    mockall::mock! {
        #[derive(Debug)]
        Speech {}
        impl speech::stub::Speech for Speech {
            async fn batch_recognize(&self, req: BatchRecognizeRequest, _options: gax::options::RequestOptions) -> Result<Response<Operation>>;
        }
    }

  2. 在測試中建立模擬物件,並設定預期行為:

    let mut mock = MockSpeech::new();
    mock.expect_batch_recognize()
        .return_once(|_, _| make_finished_operation(&expected_response()));

  3. 從模擬物件建立用戶端、呼叫函式,並驗證回應:

    // Create a client, implemented by our mock.
    let client = speech::client::Speech::from_stub(mock);
    
    // Call our function which automatically polls.
    let billed_duration = my_automatic_poller(&client, "my-project").await?;
    
    // Verify the final result of the LRO.
    assert_eq!(billed_duration, expected_duration());

使用中繼中繼資料測試手動輪詢

假設應用程式程式碼會手動輪詢並處理部分更新:

pub struct BatchRecognizeResult {
    pub progress_updates: Vec<i32>,
    pub billed_duration: Result<Option<wkt::Duration>>,
}

// An example application function that manually polls.
//
// It starts an LRO. It consolidates the polling results, whether full or
// partial.
//
// In this case, it is the `BatchRecognize` RPC. If we get a partial update,
// we extract the `progress_percent` field. If we get a final result, we
// extract the `total_billed_duration` field.
pub async fn my_manual_poller(
    client: &speech::client::Speech,
    project_id: &str,
) -> BatchRecognizeResult {
    use google_cloud_lro::{Poller, PollingResult};
    let mut progress_updates = Vec::new();
    let mut poller = client
        .batch_recognize()
        .set_recognizer(format!(
            "projects/{project_id}/locations/global/recognizers/_"
        ))
        .poller();
    while let Some(p) = poller.poll().await {
        match p {
            PollingResult::Completed(r) => {
                let billed_duration = r.map(|r| r.total_billed_duration);
                return BatchRecognizeResult {
                    progress_updates,
                    billed_duration,
                };
            }
            PollingResult::InProgress(m) => {
                if let Some(metadata) = m {
                    // This is a silly application. Your application likely
                    // performs some task immediately with the partial
                    // update, instead of storing it for after the operation
                    // has completed.
                    progress_updates.push(metadata.progress_percent);
                }
            }
            PollingResult::PollingError(e) => {
                return BatchRecognizeResult {
                    progress_updates,
                    billed_duration: Err(e),
                };
            }
        }
        tokio::time::sleep(std::time::Duration::from_millis(500)).await;
    }

    // We can only get here if `poll()` returns `None`, but it only returns
    // `None` after it returned `PollingResult::Completed`. Therefore this
    // is never reached.
    unreachable!("loop should exit via the `Completed` branch.");
}

您可以從模擬傳回進行中的作業,模擬應用程式收到中繼中繼資料時的行為。

建立 longrunning::model::Operation

BatchRecognize RPC 會以 speech::model::OperationMetadata 形式傳回部分結果。將這項中繼資料封裝到傳回的 longrunning::model::Operation 的 Operation::metadata 欄位中:

fn make_partial_operation(progress: i32) -> Result<Response<Operation>> {
    let metadata = OperationMetadata::new().set_progress_percent(progress);
    let any = wkt::Any::from_msg(&metadata).expect("test message should succeed");
    let operation = Operation::new().set_metadata(any);
    Ok(Response::from(operation))
}

編寫測試

  1. 定義模擬結構體,實作 speech::stub::Speech 特徵並覆寫 get_operation():

    mockall::mock! {
        #[derive(Debug)]
        Speech {}
        impl speech::stub::Speech for Speech {
            async fn batch_recognize(&self, req: BatchRecognizeRequest, _options: gax::options::RequestOptions) -> Result<Response<Operation>>;
            async fn get_operation(&self, req: GetOperationRequest, _options: gax::options::RequestOptions) -> Result<Response<Operation>>;
        }
    }

  2. 在測試中,建立模擬物件並設定預期結果,讓其傳回部分結果 (25%、50%、75%),然後傳回最終結果:

    let mut seq = mockall::Sequence::new();
    let mut mock = MockSpeech::new();
    mock.expect_batch_recognize()
        .once()
        .in_sequence(&mut seq)
        .returning(|_, _| make_partial_operation(25));
    mock.expect_get_operation()
        .once()
        .in_sequence(&mut seq)
        .returning(|_, _| make_partial_operation(50));
    mock.expect_get_operation()
        .once()
        .in_sequence(&mut seq)
        .returning(|_, _| make_partial_operation(75));
    mock.expect_get_operation()
        .once()
        .in_sequence(&mut seq)
        .returning(|_, _| make_finished_operation(&expected_response()));

    請留意下列預期設定細節:

    • 初始與後續 RPC:第一個預期設定在 batch_recognize(),而所有後續預期則設定在 get_operation()。初始 BatchRecognize RPC 會在伺服器上啟動 LRO,並在 name 欄位中傳回作業 ID (為求簡潔,測試程式碼中已省略)。此後,用戶端程式庫會使用 GetOperation RPC 輪詢該 LRO 的狀態。
    • 呼叫順序:預期會註冊在 Sequence 中。這樣 mockall 就能驗證呼叫順序,並判斷哪個 expect_get_operation 預期與每個輪詢相符。
  3. 從模擬物件建立用戶端、呼叫函式,並驗證回應:

    // Create a client, implemented by our mock.
    let client = speech::client::Speech::from_stub(mock);
    
    // Call our function which manually polls.
    let result = my_manual_poller(&client, "my-project").await;
    
    // Verify the partial metadata updates, and the final result.
    assert_eq!(result.progress_updates, [25, 50, 75]);
    assert_eq!(result.billed_duration?, expected_duration());

模擬錯誤

LRO 期間可能會在多個時間點發生錯誤。

如果應用程式使用自動輪詢,所有錯誤來源都相同:無論錯誤來源為何,until_done() 都會傳回 Result 中的錯誤。模擬啟動 LRO 時發生錯誤,可產生最簡單的測試。

模擬啟動 LRO 時發生錯誤

如要模擬啟動 LRO 時發生錯誤,請將初始要求設為失敗,並顯示錯誤:

mock.expect_batch_recognize().return_once(|_, _| {
    use gax::error::Error;
    use gax::error::rpc::{Code, Status};
    let status = Status::default()
        .set_code(Code::Aborted)
        .set_message("Resource exhausted");
    Err(Error::service(status))
});

如果是手動輪詢,PollingResult::Completed 分支版本會收到啟動 LRO 時發生的錯誤,並結束輪詢迴圈:

PollingResult::Completed(r) => {

模擬 LRO 導致錯誤

如要模擬 LRO 在傳回中繼中繼資料後失敗,請在最終 longrunning::model::Operation 中傳回錯誤:

fn make_failed_operation(status: rpc::model::Status) -> Result<Response<Operation>> {
    let operation = Operation::new()
        .set_done(true)
        .set_result(OperationResult::Error(status.into()));
    Ok(Response::from(operation))
}

設定預期結果,從 get_operation 傳回失敗的 Operation:

mock.expect_get_operation()
    .once()
    .in_sequence(&mut seq)
    .returning(|_, _| {
        // This is a common error for `Create*` RPCs, which are often
        // LROs. It is less applicable to `BatchRecognize` in practice.
        let status = rpc::model::Status::default()
            .set_code(gax::error::rpc::Code::AlreadyExists as i32)
            .set_message("resource already exists");
        make_failed_operation(status)
    });

在手動輪詢中,PollingResult::Completed 分支會收到錯誤結果,並結束輪詢迴圈:

PollingResult::Completed(r) => {

模擬輪詢錯誤

輪詢政策用盡時,輪詢迴圈也會結束。發生這種情況時,用戶端程式庫無法明確判斷 LRO 是否已完成。

如果應用程式有自訂邏輯來處理輪詢錯誤,您可以直接從 get_operation 預期傳回錯誤,藉此測試該邏輯:

mock.expect_get_operation()
    .once()
    .in_sequence(&mut seq)
    .returning(|_, _| {
        use gax::error::Error;
        use gax::error::rpc::{Code, Status};
        let status = Status::default()
            .set_code(Code::Aborted)
            .set_message("Operation was aborted");
        Err(Error::service(status))
    });

在手動輪詢中,PollingResult::PollingError 分支會收到模擬的輪詢錯誤:

PollingResult::PollingError(e) => {

完整測試程式碼

以下各節顯示本指南所述測試的完整程式碼。

自動輪詢:完整測試

use gax::Result;
use gax::response::Response;
use google_cloud_gax as gax;
use google_cloud_longrunning as longrunning;
use google_cloud_speech_v2 as speech;
use google_cloud_wkt as wkt;
use longrunning::model::Operation;
use longrunning::model::operation::Result as OperationResult;
use speech::model::{BatchRecognizeRequest, BatchRecognizeResponse};

// Example application code that is under test
mod my_application {
    use super::*;

    // An example application function that automatically polls.
    //
    // It starts an LRO, awaits the result, and processes it.
    pub async fn my_automatic_poller(
        client: &speech::client::Speech,
        project_id: &str,
    ) -> Result<Option<wkt::Duration>> {
        use google_cloud_lro::Poller;
        client
            .batch_recognize()
            .set_recognizer(format!(
                "projects/{project_id}/locations/global/recognizers/_"
            ))
            .poller()
            .until_done()
            .await
            .map(|r| r.total_billed_duration)
    }
}

#[cfg(test)]
mod tests {
    use super::my_application::*;
    use super::*;

    mockall::mock! {
        #[derive(Debug)]
        Speech {}
        impl speech::stub::Speech for Speech {
            async fn batch_recognize(&self, req: BatchRecognizeRequest, _options: gax::options::RequestOptions) -> Result<Response<Operation>>;
        }
    }

    fn expected_duration() -> Option<wkt::Duration> {
        Some(wkt::Duration::clamp(100, 0))
    }

    fn expected_response() -> BatchRecognizeResponse {
        BatchRecognizeResponse::new().set_or_clear_total_billed_duration(expected_duration())
    }

    fn make_finished_operation(response: &BatchRecognizeResponse) -> Result<Response<Operation>> {
        let any = wkt::Any::from_msg(response).expect("test message should succeed");
        let operation = Operation::new()
            .set_done(true)
            .set_result(OperationResult::Response(any.into()));
        Ok(Response::from(operation))
    }

    #[tokio::test]
    async fn automatic_polling() -> Result<()> {
        // Create a mock, and set expectations on it.
        let mut mock = MockSpeech::new();
        mock.expect_batch_recognize()
            .return_once(|_, _| make_finished_operation(&expected_response()));

        // Create a client, implemented by our mock.
        let client = speech::client::Speech::from_stub(mock);

        // Call our function which automatically polls.
        let billed_duration = my_automatic_poller(&client, "my-project").await?;

        // Verify the final result of the LRO.
        assert_eq!(billed_duration, expected_duration());

        Ok(())
    }
}

使用中繼中繼資料手動輪詢:完整測試

use gax::Result;
use gax::response::Response;
use google_cloud_gax as gax;
use google_cloud_longrunning as longrunning;
use google_cloud_speech_v2 as speech;
use google_cloud_wkt as wkt;
use longrunning::model::operation::Result as OperationResult;
use longrunning::model::{GetOperationRequest, Operation};
use speech::model::{BatchRecognizeRequest, BatchRecognizeResponse, OperationMetadata};

// Example application code that is under test
mod my_application {
    use super::*;

    pub struct BatchRecognizeResult {
        pub progress_updates: Vec<i32>,
        pub billed_duration: Result<Option<wkt::Duration>>,
    }

    // An example application function that manually polls.
    //
    // It starts an LRO. It consolidates the polling results, whether full or
    // partial.
    //
    // In this case, it is the `BatchRecognize` RPC. If we get a partial update,
    // we extract the `progress_percent` field. If we get a final result, we
    // extract the `total_billed_duration` field.
    pub async fn my_manual_poller(
        client: &speech::client::Speech,
        project_id: &str,
    ) -> BatchRecognizeResult {
        use google_cloud_lro::{Poller, PollingResult};
        let mut progress_updates = Vec::new();
        let mut poller = client
            .batch_recognize()
            .set_recognizer(format!(
                "projects/{project_id}/locations/global/recognizers/_"
            ))
            .poller();
        while let Some(p) = poller.poll().await {
            match p {
                PollingResult::Completed(r) => {
                    let billed_duration = r.map(|r| r.total_billed_duration);
                    return BatchRecognizeResult {
                        progress_updates,
                        billed_duration,
                    };
                }
                PollingResult::InProgress(m) => {
                    if let Some(metadata) = m {
                        // This is a silly application. Your application likely
                        // performs some task immediately with the partial
                        // update, instead of storing it for after the operation
                        // has completed.
                        progress_updates.push(metadata.progress_percent);
                    }
                }
                PollingResult::PollingError(e) => {
                    return BatchRecognizeResult {
                        progress_updates,
                        billed_duration: Err(e),
                    };
                }
            }
            tokio::time::sleep(std::time::Duration::from_millis(500)).await;
        }

        // We can only get here if `poll()` returns `None`, but it only returns
        // `None` after it returned `PollingResult::Completed`. Therefore this
        // is never reached.
        unreachable!("loop should exit via the `Completed` branch.");
    }
}

#[cfg(test)]
mod tests {
    use super::my_application::*;
    use super::*;

    mockall::mock! {
        #[derive(Debug)]
        Speech {}
        impl speech::stub::Speech for Speech {
            async fn batch_recognize(&self, req: BatchRecognizeRequest, _options: gax::options::RequestOptions) -> Result<Response<Operation>>;
            async fn get_operation(&self, req: GetOperationRequest, _options: gax::options::RequestOptions) -> Result<Response<Operation>>;
        }
    }

    fn expected_duration() -> Option<wkt::Duration> {
        Some(wkt::Duration::clamp(100, 0))
    }

    fn expected_response() -> BatchRecognizeResponse {
        BatchRecognizeResponse::new().set_or_clear_total_billed_duration(expected_duration())
    }

    fn make_finished_operation(
        response: &BatchRecognizeResponse,
    ) -> Result<gax::response::Response<Operation>> {
        let any = wkt::Any::from_msg(response).expect("test message should succeed");
        let operation = Operation::new()
            .set_done(true)
            .set_result(OperationResult::Response(any.into()));
        Ok(Response::from(operation))
    }

    fn make_partial_operation(progress: i32) -> Result<Response<Operation>> {
        let metadata = OperationMetadata::new().set_progress_percent(progress);
        let any = wkt::Any::from_msg(&metadata).expect("test message should succeed");
        let operation = Operation::new().set_metadata(any);
        Ok(Response::from(operation))
    }

    #[tokio::test]
    async fn manual_polling_with_metadata() -> Result<()> {
        let mut seq = mockall::Sequence::new();
        let mut mock = MockSpeech::new();
        mock.expect_batch_recognize()
            .once()
            .in_sequence(&mut seq)
            .returning(|_, _| make_partial_operation(25));
        mock.expect_get_operation()
            .once()
            .in_sequence(&mut seq)
            .returning(|_, _| make_partial_operation(50));
        mock.expect_get_operation()
            .once()
            .in_sequence(&mut seq)
            .returning(|_, _| make_partial_operation(75));
        mock.expect_get_operation()
            .once()
            .in_sequence(&mut seq)
            .returning(|_, _| make_finished_operation(&expected_response()));

        // Create a client, implemented by our mock.
        let client = speech::client::Speech::from_stub(mock);

        // Call our function which manually polls.
        let result = my_manual_poller(&client, "my-project").await;

        // Verify the partial metadata updates, and the final result.
        assert_eq!(result.progress_updates, [25, 50, 75]);
        assert_eq!(result.billed_duration?, expected_duration());

        Ok(())
    }
}

模擬錯誤:完整測試

use gax::Result;
use gax::response::Response;
use google_cloud_gax as gax;
use google_cloud_longrunning as longrunning;
use google_cloud_rpc as rpc;
use google_cloud_speech_v2 as speech;
use google_cloud_wkt as wkt;
use longrunning::model::operation::Result as OperationResult;
use longrunning::model::{GetOperationRequest, Operation};
use speech::model::{BatchRecognizeRequest, OperationMetadata};

// Example application code that is under test
mod my_application {
    use super::*;

    pub struct BatchRecognizeResult {
        pub progress_updates: Vec<i32>,
        pub billed_duration: Result<Option<wkt::Duration>>,
    }

    // An example application function that manually polls.
    //
    // It starts an LRO. It consolidates the polling results, whether full or
    // partial.
    //
    // In this case, it is the `BatchRecognize` RPC. If we get a partial update,
    // we extract the `progress_percent` field. If we get a final result, we
    // extract the `total_billed_duration` field.
    pub async fn my_manual_poller(
        client: &speech::client::Speech,
        project_id: &str,
    ) -> BatchRecognizeResult {
        use google_cloud_lro::{Poller, PollingResult};
        let mut progress_updates = Vec::new();
        let mut poller = client
            .batch_recognize()
            .set_recognizer(format!(
                "projects/{project_id}/locations/global/recognizers/_"
            ))
            .poller();
        while let Some(p) = poller.poll().await {
            match p {
                PollingResult::Completed(r) => {
                    let billed_duration = r.map(|r| r.total_billed_duration);
                    return BatchRecognizeResult {
                        progress_updates,
                        billed_duration,
                    };
                }
                PollingResult::InProgress(m) => {
                    if let Some(metadata) = m {
                        // This is a silly application. Your application likely
                        // performs some task immediately with the partial
                        // update, instead of storing it for after the operation
                        // has completed.
                        progress_updates.push(metadata.progress_percent);
                    }
                }
                PollingResult::PollingError(e) => {
                    return BatchRecognizeResult {
                        progress_updates,
                        billed_duration: Err(e),
                    };
                }
            }
            tokio::time::sleep(std::time::Duration::from_millis(500)).await;
        }

        // We can only get here if `poll()` returns `None`, but it only returns
        // `None` after it returned `PollingResult::Completed`. Therefore this
        // is never reached.
        unreachable!("loop should exit via the `Completed` branch.");
    }
}

#[cfg(test)]
mod tests {
    use super::my_application::*;
    use super::*;

    mockall::mock! {
        #[derive(Debug)]
        Speech {}
        impl speech::stub::Speech for Speech {
            async fn batch_recognize(&self, req: BatchRecognizeRequest, _options: gax::options::RequestOptions) -> Result<Response<Operation>>;
            async fn get_operation(&self, req: GetOperationRequest, _options: gax::options::RequestOptions) -> Result<Response<Operation>>;
        }
    }

    fn make_partial_operation(progress: i32) -> Result<Response<Operation>> {
        let metadata = OperationMetadata::new().set_progress_percent(progress);
        let any = wkt::Any::from_msg(&metadata).expect("test message should succeed");
        let operation = Operation::new().set_metadata(any);
        Ok(Response::from(operation))
    }

    fn make_failed_operation(status: rpc::model::Status) -> Result<Response<Operation>> {
        let operation = Operation::new()
            .set_done(true)
            .set_result(OperationResult::Error(status.into()));
        Ok(Response::from(operation))
    }

    #[tokio::test]
    async fn error_starting_lro() -> Result<()> {
        let mut mock = MockSpeech::new();
        mock.expect_batch_recognize().return_once(|_, _| {
            use gax::error::Error;
            use gax::error::rpc::{Code, Status};
            let status = Status::default()
                .set_code(Code::Aborted)
                .set_message("Resource exhausted");
            Err(Error::service(status))
        });

        // Create a client, implemented by our mock.
        let client = speech::client::Speech::from_stub(mock);

        // Call our function which manually polls.
        let result = my_manual_poller(&client, "my-project").await;

        // Verify the the final result.
        assert!(
            result.billed_duration.is_err(),
            "{:?}",
            result.billed_duration
        );

        Ok(())
    }

    #[tokio::test]
    async fn lro_ending_in_error() -> Result<()> {
        let mut seq = mockall::Sequence::new();
        let mut mock = MockSpeech::new();
        mock.expect_batch_recognize()
            .once()
            .in_sequence(&mut seq)
            .returning(|_, _| make_partial_operation(25));
        mock.expect_get_operation()
            .once()
            .in_sequence(&mut seq)
            .returning(|_, _| make_partial_operation(50));
        mock.expect_get_operation()
            .once()
            .in_sequence(&mut seq)
            .returning(|_, _| make_partial_operation(75));
        mock.expect_get_operation()
            .once()
            .in_sequence(&mut seq)
            .returning(|_, _| {
                // This is a common error for `Create*` RPCs, which are often
                // LROs. It is less applicable to `BatchRecognize` in practice.
                let status = rpc::model::Status::default()
                    .set_code(gax::error::rpc::Code::AlreadyExists as i32)
                    .set_message("resource already exists");
                make_failed_operation(status)
            });

        // Create a client, implemented by our mock.
        let client = speech::client::Speech::from_stub(mock);

        // Call our function which manually polls.
        let result = my_manual_poller(&client, "my-project").await;

        // Verify the partial metadata updates, and the final result.
        assert_eq!(result.progress_updates, [25, 50, 75]);
        assert!(
            result.billed_duration.is_err(),
            "{:?}",
            result.billed_duration
        );

        Ok(())
    }

    #[tokio::test]
    async fn polling_loop_error() -> Result<()> {
        let mut seq = mockall::Sequence::new();
        let mut mock = MockSpeech::new();
        mock.expect_batch_recognize()
            .once()
            .in_sequence(&mut seq)
            .returning(|_, _| make_partial_operation(25));
        mock.expect_get_operation()
            .once()
            .in_sequence(&mut seq)
            .returning(|_, _| {
                use gax::error::Error;
                use gax::error::rpc::{Code, Status};
                let status = Status::default()
                    .set_code(Code::Aborted)
                    .set_message("Operation was aborted");
                Err(Error::service(status))
            });

        // Create a client, implemented by our mock.
        let client = speech::client::Speech::from_stub(mock);

        // Call our function which manually polls.
        let result = my_manual_poller(&client, "my-project").await;

        // Verify the partial metadata updates, and the final result.
        assert_eq!(result.progress_updates, [25]);
        assert!(
            result.billed_duration.is_err(),
            "{:?}",
            result.billed_duration
        );

        Ok(())
    }
}