Google 会使用 AI 技术将内容翻译成您偏好的语言。AI 翻译可能包含错误。

为长时间运行的操作编写测试

Google Cloud Rust 客户端库提供了可简化与长时间运行的操作 (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)

编写测试

定义实现 speech::stub::Speech trait 的模拟结构体：

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>>;
    }
}

在测试中，创建模拟对象并设置预期：

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

使用中间元数据测试手动轮询

假设您的应用代码手动轮询并处理部分更新：

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

编写测试

定义模拟结构体，该结构体实现 speech::stub::Speech trait 并替换 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>>;
    }
}

在测试中，创建模拟对象并设置预期，使其返回部分结果（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 与后续 RPC：第一个预期在 batch_recognize() 上设置，而所有后续预期在 get_operation() 上设置。初始 BatchRecognize RPC 会在服务器上启动 LRO，并在 name 字段中返回操作标识符（为简洁起见，测试代码中省略了该标识符）。此后，客户端库会使用 GetOperation RPC 轮询该 LRO 的状态。
- 调用顺序：预期在 Sequence 中注册。这样，mockall 就可以验证调用的顺序，并确定哪个 expect_get_operation 预期与每个轮询匹配。

通过模拟创建客户端，调用函数，并验证响应：

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

为长时间运行的操作编写测试 使用集合让一切井井有条 根据您的偏好保存内容并对其进行分类。

前提条件

测试自动轮询

创建 longrunning::model::Operation

编写测试

使用中间元数据测试手动轮询

创建 longrunning::model::Operation

编写测试

模拟错误

模拟启动 LRO 时出错

模拟导致错误的 LRO

模拟轮询错误

完整测试代码

自动轮询：完整测试

使用中间元数据进行手动轮询：完整测试

模拟错误：完整测试

为长时间运行的操作编写测试

创建 `longrunning::model::Operation`

创建 `longrunning::model::Operation`