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)編寫測試
定義模擬結構,實作
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>>; } }在測試中建立模擬物件,並設定預期行為:
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特徵並覆寫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:第一個預期設定在
batch_recognize(),而所有後續預期則設定在get_operation()。初始BatchRecognizeRPC 會在伺服器上啟動 LRO,並在name欄位中傳回作業 ID (為求簡潔,測試程式碼中已省略)。此後,用戶端程式庫會使用GetOperationRPC 輪詢該 LRO 的狀態。 - 呼叫順序:預期會註冊在
Sequence中。這樣mockall就能驗證呼叫順序,並判斷哪個expect_get_operation預期與每個輪詢相符。
- 初始與後續 RPC:第一個預期設定在
從模擬物件建立用戶端、呼叫函式,並驗證回應:
// 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(())
}
}