google_cloud_storage/storage/
read_object.rs

1// Copyright 2025 Google LLC
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     https://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15mod non_resumable;
16mod parse_http_response;
17mod resumable;
18
19use super::client::*;
20use super::*;
21use crate::model_ext::KeyAes256;
22use crate::read_object::ReadObjectResponse;
23use crate::read_resume_policy::ReadResumePolicy;
24use crate::storage::checksum::details::Md5;
25use crate::storage::request_options::RequestOptions;
26use gaxi::attempt_info::AttemptInfo;
27use gaxi::http::HttpRequestBuilder;
28use gaxi::http::reqwest::{HeaderValue, Method, Response};
29use google_cloud_gax::options::internal::{PathTemplate, RequestOptionsExt, ResourceName};
30
31/// The request builder for [Storage::read_object][crate::client::Storage::read_object] calls.
32///
33/// # Example: accumulate the contents of an object into a vector
34/// ```
35/// use google_cloud_storage::{client::Storage, builder::storage::ReadObject};
36/// async fn sample(client: &Storage) -> anyhow::Result<()> {
37///     let builder: ReadObject = client.read_object("projects/_/buckets/my-bucket", "my-object");
38///     let mut reader = builder.send().await?;
39///     let mut contents = Vec::new();
40///     while let Some(chunk) = reader.next().await.transpose()? {
41///         contents.extend_from_slice(&chunk);
42///     }
43///     println!("object contents={:?}", contents);
44///     Ok(())
45/// }
46/// ```
47///
48/// # Example: read part of an object
49/// ```
50/// use google_cloud_storage::{client::Storage, builder::storage::ReadObject};
51/// use google_cloud_storage::model_ext::ReadRange;
52/// async fn sample(client: &Storage) -> anyhow::Result<()> {
53///     const MIB: u64 = 1024 * 1024;
54///     let mut contents = Vec::new();
55///     let mut reader = client
56///         .read_object("projects/_/buckets/my-bucket", "my-object")
57///         .set_read_range(ReadRange::segment(4 * MIB, 2 * MIB))
58///         .send()
59///         .await?;
60///     while let Some(chunk) = reader.next().await.transpose()? {
61///         contents.extend_from_slice(&chunk);
62///     }
63///     println!("range contents={:?}", contents);
64///     Ok(())
65/// }
66/// ```
67#[derive(Debug)]
68pub struct ReadObject<S = crate::storage::transport::Storage>
69where
70    S: crate::storage::stub::Storage + 'static,
71{
72    stub: std::sync::Arc<S>,
73    request: crate::model::ReadObjectRequest,
74    options: RequestOptions,
75}
76
77impl<S> Clone for ReadObject<S>
78where
79    S: crate::storage::stub::Storage + 'static,
80{
81    fn clone(&self) -> Self {
82        Self {
83            stub: self.stub.clone(),
84            request: self.request.clone(),
85            options: self.options.clone(),
86        }
87    }
88}
89
90impl<S> ReadObject<S>
91where
92    S: crate::storage::stub::Storage + 'static,
93{
94    pub(crate) fn new<B, O>(
95        stub: std::sync::Arc<S>,
96        bucket: B,
97        object: O,
98        options: RequestOptions,
99    ) -> Self
100    where
101        B: Into<String>,
102        O: Into<String>,
103    {
104        ReadObject {
105            stub,
106            request: crate::model::ReadObjectRequest::new()
107                .set_bucket(bucket)
108                .set_object(object),
109            options,
110        }
111    }
112
113    /// Enables computation of MD5 hashes.
114    ///
115    /// Crc32c hashes are checked by default.
116    ///
117    /// Checksum validation is supported iff:
118    /// 1. The full content is requested.
119    /// 2. All of the content is returned (status != PartialContent).
120    /// 3. The server sent a checksum header.
121    /// 4. The http stack did not uncompress the file.
122    /// 5. The server did not uncompress data on read.
123    ///
124    /// # Example
125    /// ```
126    /// # use google_cloud_storage::client::Storage;
127    /// # async fn sample(client: &Storage) -> anyhow::Result<()> {
128    /// let builder =  client
129    ///     .read_object("projects/_/buckets/my-bucket", "my-object")
130    ///     .compute_md5();
131    /// let mut reader = builder
132    ///     .send()
133    ///     .await?;
134    /// let mut contents = Vec::new();
135    /// while let Some(chunk) = reader.next().await.transpose()? {
136    ///     contents.extend_from_slice(&chunk);
137    /// }
138    /// println!("object contents={:?}", contents);
139    /// # Ok(()) }
140    /// ```
141    pub fn compute_md5(self) -> Self {
142        let mut this = self;
143        this.options.checksum.md5_hash = Some(Md5::default());
144        this
145    }
146
147    /// Enables computation of CRC32C checksums.
148    ///
149    /// Note that the library computes and verifies (if available) CRC32C checksums at the end of
150    /// the download. Use `compute_crc32c(false)` to disable the computation, but note
151    /// that this reduces the data integrity guarantees. Data *can* be corrupted even when
152    /// downloaded over HTTPS or other encrypted channels.
153    ///
154    /// # Example
155    /// ```
156    /// # use google_cloud_storage::client::Storage;
157    /// # async fn sample(client: &Storage) -> anyhow::Result<()> {
158    /// let builder =  client
159    ///     .read_object("projects/_/buckets/my-bucket", "my-object")
160    ///     .compute_crc32c(false);
161    /// let mut reader = builder
162    ///     .send()
163    ///     .await?;
164    /// let mut contents = Vec::new();
165    /// while let Some(chunk) = reader.next().await.transpose()? {
166    ///     contents.extend_from_slice(&chunk);
167    /// }
168    /// println!("object contents={:?}", contents);
169    /// # Ok(()) }
170    /// ```
171    pub fn compute_crc32c(mut self, enable: bool) -> Self {
172        if enable {
173            self.options
174                .checksum
175                .crc32c
176                .get_or_insert_with(Default::default);
177        } else {
178            self.options.checksum.crc32c = None;
179        }
180        self
181    }
182
183    /// If present, selects a specific revision of this object (as
184    /// opposed to the latest version, the default).
185    pub fn set_generation<T: Into<i64>>(mut self, v: T) -> Self {
186        self.request.generation = v.into();
187        self
188    }
189
190    /// Makes the operation conditional on whether the object's current generation
191    /// matches the given value. Setting to 0 makes the operation succeed only if
192    /// there are no live versions of the object.
193    pub fn set_if_generation_match<T>(mut self, v: T) -> Self
194    where
195        T: Into<i64>,
196    {
197        self.request.if_generation_match = Some(v.into());
198        self
199    }
200
201    /// Makes the operation conditional on whether the object's live generation
202    /// does not match the given value. If no live object exists, the precondition
203    /// fails. Setting to 0 makes the operation succeed only if there is a live
204    /// version of the object.
205    pub fn set_if_generation_not_match<T>(mut self, v: T) -> Self
206    where
207        T: Into<i64>,
208    {
209        self.request.if_generation_not_match = Some(v.into());
210        self
211    }
212
213    /// Makes the operation conditional on whether the object's current
214    /// metageneration matches the given value.
215    pub fn set_if_metageneration_match<T>(mut self, v: T) -> Self
216    where
217        T: Into<i64>,
218    {
219        self.request.if_metageneration_match = Some(v.into());
220        self
221    }
222
223    /// Makes the operation conditional on whether the object's current
224    /// metageneration does not match the given value.
225    pub fn set_if_metageneration_not_match<T>(mut self, v: T) -> Self
226    where
227        T: Into<i64>,
228    {
229        self.request.if_metageneration_not_match = Some(v.into());
230        self
231    }
232
233    /// The range of bytes to return in the read.
234    ///
235    /// This can be all the bytes starting at a given offset
236    /// (`ReadRange::offset()`), all the bytes in an explicit range
237    /// (`Range::segment`), or the last N bytes of the object
238    /// (`ReadRange::tail`).
239    ///
240    /// # Examples
241    ///
242    /// Read starting at 100 bytes to end of file.
243    /// ```
244    /// # use google_cloud_storage::client::Storage;
245    /// # use google_cloud_storage::model_ext::ReadRange;
246    /// # async fn sample(client: &Storage) -> anyhow::Result<()> {
247    /// let response = client
248    ///     .read_object("projects/_/buckets/my-bucket", "my-object")
249    ///     .set_read_range(ReadRange::offset(100))
250    ///     .send()
251    ///     .await?;
252    /// println!("response details={response:?}");
253    /// # Ok(()) }
254    /// ```
255    ///
256    /// Read last 100 bytes of file:
257    /// ```
258    /// # use google_cloud_storage::client::Storage;
259    /// # use google_cloud_storage::model_ext::ReadRange;
260    /// # async fn sample(client: &Storage) -> anyhow::Result<()> {
261    /// let response = client
262    ///     .read_object("projects/_/buckets/my-bucket", "my-object")
263    ///     .set_read_range(ReadRange::tail(100))
264    ///     .send()
265    ///     .await?;
266    /// println!("response details={response:?}");
267    /// # Ok(()) }
268    /// ```
269    ///
270    /// Read bytes 1000 to 1099.
271    /// ```
272    /// # use google_cloud_storage::client::Storage;
273    /// # use google_cloud_storage::model_ext::ReadRange;
274    /// # async fn sample(client: &Storage) -> anyhow::Result<()> {
275    /// let response = client
276    ///     .read_object("projects/_/buckets/my-bucket", "my-object")
277    ///     .set_read_range(ReadRange::segment(1000, 100))
278    ///     .send()
279    ///     .await?;
280    /// println!("response details={response:?}");
281    /// # Ok(()) }
282    /// ```
283    pub fn set_read_range(mut self, range: crate::model_ext::ReadRange) -> Self {
284        self.request.with_range(range);
285        self
286    }
287
288    /// The encryption key used with the Customer-Supplied Encryption Keys
289    /// feature. In raw bytes format (not base64-encoded).
290    ///
291    /// Example:
292    /// ```
293    /// # use google_cloud_storage::{model_ext::KeyAes256, client::Storage};
294    /// # async fn sample(client: &Storage) -> anyhow::Result<()> {
295    /// let key: &[u8] = &[97; 32];
296    /// let response = client
297    ///     .read_object("projects/_/buckets/my-bucket", "my-object")
298    ///     .set_key(KeyAes256::new(key)?)
299    ///     .send()
300    ///     .await?;
301    /// println!("response details={response:?}");
302    /// # Ok(()) }
303    /// ```
304    pub fn set_key(mut self, v: KeyAes256) -> Self {
305        self.request.common_object_request_params = Some(v.into());
306        self
307    }
308
309    /// The retry policy used for this request.
310    ///
311    /// # Example
312    /// ```
313    /// # use google_cloud_storage::client::Storage;
314    /// # async fn sample(client: &Storage) -> anyhow::Result<()> {
315    /// use google_cloud_storage::retry_policy::RetryableErrors;
316    /// use std::time::Duration;
317    /// use google_cloud_gax::retry_policy::RetryPolicyExt;
318    /// let response = client
319    ///     .read_object("projects/_/buckets/my-bucket", "my-object")
320    ///     .with_retry_policy(
321    ///         RetryableErrors
322    ///             .with_attempt_limit(5)
323    ///             .with_time_limit(Duration::from_secs(10)),
324    ///     )
325    ///     .send()
326    ///     .await?;
327    /// println!("response details={response:?}");
328    /// # Ok(()) }
329    /// ```
330    pub fn with_retry_policy<V: Into<google_cloud_gax::retry_policy::RetryPolicyArg>>(
331        mut self,
332        v: V,
333    ) -> Self {
334        self.options.retry_policy = v.into().into();
335        self
336    }
337
338    /// The backoff policy used for this request.
339    ///
340    /// # Example
341    /// ```
342    /// # use google_cloud_storage::client::Storage;
343    /// # async fn sample(client: &Storage) -> anyhow::Result<()> {
344    /// use std::time::Duration;
345    /// use google_cloud_gax::exponential_backoff::ExponentialBackoff;
346    /// let response = client
347    ///     .read_object("projects/_/buckets/my-bucket", "my-object")
348    ///     .with_backoff_policy(ExponentialBackoff::default())
349    ///     .send()
350    ///     .await?;
351    /// println!("response details={response:?}");
352    /// # Ok(()) }
353    /// ```
354    pub fn with_backoff_policy<V: Into<google_cloud_gax::backoff_policy::BackoffPolicyArg>>(
355        mut self,
356        v: V,
357    ) -> Self {
358        self.options.backoff_policy = v.into().into();
359        self
360    }
361
362    /// The retry throttler used for this request.
363    ///
364    /// Most of the time you want to use the same throttler for all the requests
365    /// in a client, and even the same throttler for many clients. Rarely it
366    /// may be necessary to use an custom throttler for some subset of the
367    /// requests.
368    ///
369    /// # Example
370    /// ```
371    /// # use google_cloud_storage::client::Storage;
372    /// # async fn sample(client: &Storage) -> anyhow::Result<()> {
373    /// let response = client
374    ///     .read_object("projects/_/buckets/my-bucket", "my-object")
375    ///     .with_retry_throttler(adhoc_throttler())
376    ///     .send()
377    ///     .await?;
378    /// println!("response details={response:?}");
379    /// fn adhoc_throttler() -> google_cloud_gax::retry_throttler::SharedRetryThrottler {
380    ///     # panic!();
381    /// }
382    /// # Ok(()) }
383    /// ```
384    pub fn with_retry_throttler<V: Into<google_cloud_gax::retry_throttler::RetryThrottlerArg>>(
385        mut self,
386        v: V,
387    ) -> Self {
388        self.options.retry_throttler = v.into().into();
389        self
390    }
391
392    /// Configure the resume policy for read requests.
393    ///
394    /// The Cloud Storage client library can automatically resume a read that is
395    /// interrupted by a transient error. Applications may want to limit the
396    /// number of read attempts, or may wish to expand the type of errors
397    /// treated as retryable.
398    ///
399    /// # Example
400    /// ```
401    /// # use google_cloud_storage::client::Storage;
402    /// # async fn sample(client: &Storage) -> anyhow::Result<()> {
403    /// use google_cloud_storage::read_resume_policy::{AlwaysResume, ReadResumePolicyExt};
404    /// let response = client
405    ///     .read_object("projects/_/buckets/my-bucket", "my-object")
406    ///     .with_read_resume_policy(AlwaysResume.with_attempt_limit(3))
407    ///     .send()
408    ///     .await?;
409    /// # Ok(()) }
410    /// ```
411    pub fn with_read_resume_policy<V>(mut self, v: V) -> Self
412    where
413        V: ReadResumePolicy + 'static,
414    {
415        self.options.set_read_resume_policy(std::sync::Arc::new(v));
416        self
417    }
418
419    /// Enables automatic decompression.
420    ///
421    /// The Cloud Storage service [automatically decompresses] objects
422    /// with `content_encoding == "gzip"` during reads. The client library
423    /// disables this behavior by default, as it is not possible to
424    /// perform ranged reads or to resume interrupted downloads if automatic
425    /// decompression is enabled.
426    ///
427    /// Use this option to enable automatic decompression.
428    ///
429    /// # Example
430    /// ```
431    /// # use google_cloud_storage::client::Storage;
432    /// # async fn sample(client: &Storage) -> anyhow::Result<()> {
433    /// let response = client
434    ///     .read_object("projects/_/buckets/my-bucket", "my-object")
435    ///     .with_automatic_decompression(true)
436    ///     .send()
437    ///     .await?;
438    /// println!("response details={response:?}");
439    /// # Ok(()) }
440    /// ```
441    pub fn with_automatic_decompression(mut self, v: bool) -> Self {
442        self.options.automatic_decompression = v;
443        self
444    }
445
446    /// Sets the `User-Agent` header for this request.
447    ///
448    /// # Example
449    /// ```
450    /// # use google_cloud_storage::client::Storage;
451    /// # async fn sample(client: &Storage) -> anyhow::Result<()> {
452    /// let response = client
453    ///     .read_object("projects/_/buckets/my-bucket", "my-object")
454    ///     .with_user_agent("my-app/1.0.0")
455    ///     .send()
456    ///     .await?;
457    /// println!("response details={response:?}");
458    /// # Ok(()) }
459    /// ```
460    pub fn with_user_agent(mut self, user_agent: impl Into<String>) -> Self {
461        self.options.user_agent = Some(user_agent.into());
462        self
463    }
464
465    /// Sets the project that will be billed for this request.
466    ///
467    /// Required for [Requester Pays] buckets. The value overrides any
468    /// `quota_project_id` configured on the credentials; the credential-level
469    /// header is suppressed for this RPC.
470    ///
471    /// # Example
472    /// ```
473    /// # use google_cloud_storage::client::Storage;
474    /// # async fn sample(client: &Storage) -> anyhow::Result<()> {
475    /// let response = client
476    ///     .read_object("projects/_/buckets/my-bucket", "my-object")
477    ///     .with_quota_project("my-billing-project")
478    ///     .send()
479    ///     .await?;
480    /// # Ok(()) }
481    /// ```
482    ///
483    /// [Requester Pays]: https://cloud.google.com/storage/docs/requester-pays
484    pub fn with_quota_project(mut self, project: impl Into<String>) -> Self {
485        self.options.set_quota_project(project);
486        self
487    }
488
489    /// Sends the request.
490    pub async fn send(self) -> Result<ReadObjectResponse> {
491        self.stub.read_object(self.request, self.options).await
492    }
493}
494
495// A convenience struct that saves the request conditions and performs the read.
496#[derive(Clone, Debug)]
497pub(crate) struct Reader {
498    pub inner: std::sync::Arc<StorageInner>,
499    pub request: crate::model::ReadObjectRequest,
500    pub options: RequestOptions,
501}
502
503impl Reader {
504    async fn read(self) -> Result<Response> {
505        let throttler = self.options.retry_throttler.clone();
506        let retry = self.options.retry_policy.clone();
507        let backoff = self.options.backoff_policy.clone();
508        let mut count = 0;
509        let inner = async move |_| {
510            let current = count;
511            count += 1;
512            self.read_attempt(current).await
513        };
514
515        google_cloud_gax::retry_loop_internal::retry_loop(
516            inner,
517            async |duration| tokio::time::sleep(duration).await,
518            true,
519            throttler,
520            retry,
521            backoff,
522        )
523        .await
524    }
525
526    async fn read_attempt(&self, attempt_count: u32) -> Result<Response> {
527        let builder = self.http_request_builder().await?;
528        let options = self
529            .options
530            .gax()
531            .insert_extension(PathTemplate("/storage/v1/b/{bucket}/o/{object}"))
532            .insert_extension(ResourceName(format!(
533                "//storage.googleapis.com/{}",
534                self.request.bucket
535            )));
536        let response = builder
537            .send(options, AttemptInfo::new(attempt_count))
538            .await?;
539        if !response.status().is_success() {
540            return gaxi::http::to_http_error(response).await;
541        }
542        Ok(response)
543    }
544
545    async fn http_request_builder(&self) -> Result<HttpRequestBuilder> {
546        // Collect the required bucket and object parameters.
547        let bucket = &self.request.bucket;
548        let bucket_id = bucket
549            .as_str()
550            .strip_prefix("projects/_/buckets/")
551            .ok_or_else(|| {
552                Error::binding(format!(
553                    "malformed bucket name, it must start with `projects/_/buckets/`: {bucket}"
554                ))
555            })?;
556        let object = &self.request.object;
557
558        // Build the request.
559        let builder = self
560            .inner
561            .client
562            .http_builder(
563                Method::GET,
564                &format!("/storage/v1/b/{bucket_id}/o/{}", enc(object)),
565            )
566            .query("alt", "media")
567            .header(
568                "x-goog-api-client",
569                HeaderValue::from_static(&self::info::X_GOOG_API_CLIENT_HEADER),
570            );
571
572        let builder = if self.options.automatic_decompression {
573            builder
574        } else {
575            // Disable decompressive transcoding: https://cloud.google.com/storage/docs/transcoding
576            //
577            // The default is to decompress objects that have `contentEncoding == "gzip"`. This header
578            // tells Cloud Storage to disable automatic decompression. It has no effect on objects
579            // with a different `contentEncoding` value.
580            builder.header("accept-encoding", HeaderValue::from_static("gzip"))
581        };
582
583        // Add the optional query parameters.
584        let builder = if self.request.generation != 0 {
585            builder.query("generation", self.request.generation)
586        } else {
587            builder
588        };
589        let builder = self
590            .request
591            .if_generation_match
592            .iter()
593            .fold(builder, |b, v| b.query("ifGenerationMatch", v));
594        let builder = self
595            .request
596            .if_generation_not_match
597            .iter()
598            .fold(builder, |b, v| b.query("ifGenerationNotMatch", v));
599        let builder = self
600            .request
601            .if_metageneration_match
602            .iter()
603            .fold(builder, |b, v| b.query("ifMetagenerationMatch", v));
604        let builder = self
605            .request
606            .if_metageneration_not_match
607            .iter()
608            .fold(builder, |b, v| b.query("ifMetagenerationNotMatch", v));
609
610        let builder = apply_customer_supplied_encryption_headers(
611            builder,
612            &self.request.common_object_request_params,
613        );
614
615        // Apply "range" header for read limits and offsets.
616        let builder = match (self.request.read_offset, self.request.read_limit) {
617            // read_limit can't be negative.
618            (_, l) if l < 0 => {
619                unreachable!("ReadObject build never sets a negative read_limit value")
620            }
621            // negative offset can't also have a read_limit.
622            (o, l) if o < 0 && l > 0 => unreachable!(
623                "ReadObject builder never sets a positive read_offset value with a negative read_limit value"
624            ),
625            // If both are zero, we use default implementation (no range header).
626            (0, 0) => builder,
627            // negative offset with no limit means the last N bytes.
628            (o, 0) if o < 0 => builder.header("range", format!("bytes={o}")),
629            // read_limit is zero, means no limit. Read from offset to end of file.
630            // This handles cases like (5, 0) -> "bytes=5-"
631            (o, 0) => builder.header("range", format!("bytes={o}-")),
632            // General case: non-negative offset and positive limit.
633            // This covers cases like (0, 100) -> "bytes=0-99", (5, 100) -> "bytes=5-104"
634            (o, l) => builder.header("range", format!("bytes={o}-{}", o + l - 1)),
635        };
636
637        Ok(builder)
638    }
639
640    fn is_gunzipped(response: &Response) -> bool {
641        // Cloud Storage automatically [decompresses gzip-compressed][transcoding]
642        // objects. Reading such objects comes with a number of restrictions:
643        // - Ranged reads do not work.
644        // - The size of the decompressed data is not known.
645        // - Checksums do not work because the object checksums correspond to the
646        //   compressed data and the client library receives the decompressed data.
647        //
648        // Because ranged reads do not work, resuming a read does not work. Consequently,
649        // the implementation of `ReadObjectResponse` is substantially different for
650        // objects that are gunzipped.
651        //
652        // [transcoding]: https://cloud.google.com/storage/docs/transcoding
653        const TRANSFORMATION: &str = "x-guploader-response-body-transformations";
654        use http::header::WARNING;
655        if response
656            .headers()
657            .get(TRANSFORMATION)
658            .is_some_and(|h| h.as_bytes() == "gunzipped".as_bytes())
659        {
660            return true;
661        }
662        response
663            .headers()
664            .get(WARNING)
665            .is_some_and(|h| h.as_bytes() == "214 UploadServer gunzipped".as_bytes())
666    }
667
668    pub(crate) async fn response(self) -> Result<ReadObjectResponse> {
669        let response = self.clone().read().await?;
670        if Self::is_gunzipped(&response) {
671            return Ok(ReadObjectResponse::new(Box::new(
672                non_resumable::NonResumableResponse::new(response)?,
673            )));
674        }
675        Ok(ReadObjectResponse::new(Box::new(
676            resumable::ResumableResponse::new(self, response)?,
677        )))
678    }
679}
680
681#[cfg(test)]
682mod resume_tests;
683
684#[cfg(test)]
685mod tests {
686    use super::client::tests::{test_builder, test_inner_client};
687    use super::*;
688    use crate::error::{ChecksumMismatch, ReadError};
689    use crate::model_ext::{KeyAes256, ReadRange, tests::create_key_helper};
690    use base64::Engine;
691    use futures::TryStreamExt;
692    use google_cloud_auth::credentials::{
693        CacheableResource, Credentials, EntityTag, anonymous::Builder as Anonymous,
694        testing::error_credentials,
695    };
696    use httptest::{Expectation, Server, all_of, cycle, matchers::*, responders::status_code};
697    use pretty_assertions::assert_eq;
698    use std::collections::HashMap;
699    use std::error::Error;
700    use std::sync::Arc;
701    use test_case::test_case;
702
703    type Result = anyhow::Result<()>;
704
705    mockall::mock! {
706        #[derive(Debug)]
707        Credentials {}
708        impl google_cloud_auth::credentials::CredentialsProvider for Credentials {
709            async fn headers(
710                &self,
711                extensions: http::Extensions,
712            ) -> std::result::Result<
713                google_cloud_auth::credentials::CacheableResource<http::HeaderMap>,
714                google_cloud_gax::error::CredentialsError,
715            >;
716            async fn universe_domain(&self) -> Option<String>;
717        }
718    }
719
720    async fn http_request_builder(
721        inner: Arc<StorageInner>,
722        builder: ReadObject,
723    ) -> crate::Result<HttpRequestBuilder> {
724        let reader = Reader {
725            inner,
726            request: builder.request,
727            options: builder.options,
728        };
729        reader.http_request_builder().await
730    }
731
732    #[tokio::test]
733    async fn test_clone() {
734        let inner = test_inner_client(test_builder()).await;
735        let stub = crate::storage::transport::Storage::new_test(inner.clone());
736        let options = {
737            let mut o = RequestOptions::new();
738            o.set_resumable_upload_threshold(12345_usize);
739            o
740        };
741        let builder = ReadObject::new(stub, "projects/_/buckets/bucket", "object", options);
742
743        let clone = builder.clone();
744        assert!(Arc::ptr_eq(&clone.stub, &builder.stub));
745        assert_eq!(clone.request, builder.request);
746        assert_eq!(clone.options.resumable_upload_threshold(), 12345_usize);
747    }
748
749    // Verify `read_object()` meets normal Send, Sync, requirements.
750    #[tokio::test]
751    async fn test_read_is_send_and_static() -> Result {
752        let client = Storage::builder()
753            .with_credentials(Anonymous::new().build())
754            .build()
755            .await?;
756
757        fn need_send<T: Send>(_val: &T) {}
758        fn need_sync<T: Sync>(_val: &T) {}
759        fn need_static<T: 'static>(_val: &T) {}
760
761        let read = client.read_object("projects/_/buckets/test-bucket", "test-object");
762        need_send(&read);
763        need_sync(&read);
764        need_static(&read);
765
766        let read = client
767            .read_object("projects/_/buckets/test-bucket", "test-object")
768            .send();
769        need_send(&read);
770        need_static(&read);
771
772        Ok(())
773    }
774
775    #[tokio::test]
776    async fn read_object_normal() -> Result {
777        let server = Server::run();
778        server.expect(
779            Expectation::matching(all_of![
780                request::method_path("GET", "/storage/v1/b/test-bucket/o/test-object"),
781                request::headers(contains(("accept-encoding", "gzip"))),
782                request::query(url_decoded(contains(("alt", "media")))),
783            ])
784            .respond_with(
785                status_code(200)
786                    .body("hello world")
787                    .append_header("x-goog-generation", 123456),
788            ),
789        );
790
791        let client = Storage::builder()
792            .with_endpoint(format!("http://{}", server.addr()))
793            .with_credentials(Anonymous::new().build())
794            .build()
795            .await?;
796        let mut reader = client
797            .read_object("projects/_/buckets/test-bucket", "test-object")
798            .send()
799            .await?;
800        let mut got = Vec::new();
801        while let Some(b) = reader.next().await.transpose()? {
802            got.extend_from_slice(&b);
803        }
804        assert_eq!(got, b"hello world");
805
806        Ok(())
807    }
808
809    #[tokio::test]
810    async fn read_object_stream() -> Result {
811        let server = Server::run();
812        server.expect(
813            Expectation::matching(all_of![
814                request::method_path("GET", "/storage/v1/b/test-bucket/o/test-object"),
815                request::query(url_decoded(contains(("alt", "media")))),
816            ])
817            .respond_with(
818                status_code(200)
819                    .append_header("x-goog-generation", 123456)
820                    .body("hello world"),
821            ),
822        );
823
824        let client = Storage::builder()
825            .with_endpoint(format!("http://{}", server.addr()))
826            .with_credentials(Anonymous::new().build())
827            .build()
828            .await?;
829        let response = client
830            .read_object("projects/_/buckets/test-bucket", "test-object")
831            .send()
832            .await?;
833        let result: Vec<_> = response.into_stream().try_collect().await?;
834        assert_eq!(result, vec![bytes::Bytes::from_static(b"hello world")]);
835
836        Ok(())
837    }
838
839    #[tokio::test]
840    async fn read_object_next_then_consume_response() -> Result {
841        // Create a large enough file that will require multiple chunks to read.
842        const BLOCK_SIZE: usize = 500;
843        let mut contents = Vec::new();
844        for i in 0..50 {
845            contents.extend_from_slice(&[i as u8; BLOCK_SIZE]);
846        }
847
848        // Calculate and serialize the crc32c checksum
849        let u = crc32c::crc32c(&contents);
850        let value = base64::prelude::BASE64_STANDARD.encode(u.to_be_bytes());
851
852        let server = Server::run();
853        server.expect(
854            Expectation::matching(all_of![
855                request::method_path("GET", "/storage/v1/b/test-bucket/o/test-object"),
856                request::query(url_decoded(contains(("alt", "media")))),
857            ])
858            .times(1)
859            .respond_with(
860                status_code(200)
861                    .body(contents.clone())
862                    .append_header("x-goog-hash", format!("crc32c={value}"))
863                    .append_header("x-goog-generation", 123456),
864            ),
865        );
866
867        let client = Storage::builder()
868            .with_endpoint(format!("http://{}", server.addr()))
869            .with_credentials(Anonymous::new().build())
870            .build()
871            .await?;
872
873        // Read some bytes, then remainder with stream.
874        let mut response = client
875            .read_object("projects/_/buckets/test-bucket", "test-object")
876            .send()
877            .await?;
878
879        let mut all_bytes = bytes::BytesMut::new();
880        let chunk = response.next().await.transpose()?.unwrap();
881        assert!(!chunk.is_empty());
882        all_bytes.extend(chunk);
883        use futures::StreamExt;
884        let mut stream = response.into_stream();
885        while let Some(chunk) = stream.next().await.transpose()? {
886            all_bytes.extend(chunk);
887        }
888        assert_eq!(all_bytes, contents);
889
890        Ok(())
891    }
892
893    #[tokio::test]
894    async fn read_object_not_found() -> Result {
895        let server = Server::run();
896        server.expect(
897            Expectation::matching(all_of![
898                request::method_path("GET", "/storage/v1/b/test-bucket/o/test-object"),
899                request::query(url_decoded(contains(("alt", "media")))),
900            ])
901            .respond_with(status_code(404).body("NOT FOUND")),
902        );
903
904        let client = Storage::builder()
905            .with_endpoint(format!("http://{}", server.addr()))
906            .with_credentials(Anonymous::new().build())
907            .build()
908            .await?;
909        let err = client
910            .read_object("projects/_/buckets/test-bucket", "test-object")
911            .send()
912            .await
913            .expect_err("expected a not found error");
914        assert_eq!(err.http_status_code(), Some(404));
915
916        Ok(())
917    }
918
919    #[tokio::test]
920    async fn read_object_incorrect_crc32c_check() -> Result {
921        // Calculate and serialize the crc32c checksum
922        let u = crc32c::crc32c("goodbye world".as_bytes());
923        let value = base64::prelude::BASE64_STANDARD.encode(u.to_be_bytes());
924
925        let server = Server::run();
926        server.expect(
927            Expectation::matching(all_of![
928                request::method_path("GET", "/storage/v1/b/test-bucket/o/test-object"),
929                request::query(url_decoded(contains(("alt", "media")))),
930            ])
931            .times(3)
932            .respond_with(
933                status_code(200)
934                    .body("hello world")
935                    .append_header("x-goog-hash", format!("crc32c={value}"))
936                    .append_header("x-goog-generation", 123456),
937            ),
938        );
939
940        let client = Storage::builder()
941            .with_endpoint(format!("http://{}", server.addr()))
942            .with_credentials(Anonymous::new().build())
943            .build()
944            .await?;
945        let mut response = client
946            .read_object("projects/_/buckets/test-bucket", "test-object")
947            .send()
948            .await?;
949        let mut partial = Vec::new();
950        let mut err = None;
951        while let Some(r) = response.next().await {
952            match r {
953                Ok(b) => partial.extend_from_slice(&b),
954                Err(e) => err = Some(e),
955            };
956        }
957        assert_eq!(partial, b"hello world");
958        let err = err.expect("expect error on incorrect crc32c");
959        let source = err.source().and_then(|e| e.downcast_ref::<ReadError>());
960        assert!(
961            matches!(
962                source,
963                Some(&ReadError::ChecksumMismatch(
964                    ChecksumMismatch::Crc32c { .. }
965                ))
966            ),
967            "err={err:?}"
968        );
969
970        let mut response = client
971            .read_object("projects/_/buckets/test-bucket", "test-object")
972            .send()
973            .await?;
974        let err: crate::Error = async {
975            {
976                while (response.next().await.transpose()?).is_some() {}
977                Ok(())
978            }
979        }
980        .await
981        .expect_err("expect error on incorrect crc32c");
982        let source = err.source().and_then(|e| e.downcast_ref::<ReadError>());
983        assert!(
984            matches!(
985                source,
986                Some(&ReadError::ChecksumMismatch(
987                    ChecksumMismatch::Crc32c { .. }
988                ))
989            ),
990            "err={err:?}"
991        );
992
993        use futures::TryStreamExt;
994        let err = client
995            .read_object("projects/_/buckets/test-bucket", "test-object")
996            .send()
997            .await?
998            .into_stream()
999            .try_collect::<Vec<bytes::Bytes>>()
1000            .await
1001            .expect_err("expect error on incorrect crc32c");
1002        let source = err.source().and_then(|e| e.downcast_ref::<ReadError>());
1003        assert!(
1004            matches!(
1005                source,
1006                Some(&ReadError::ChecksumMismatch(
1007                    ChecksumMismatch::Crc32c { .. }
1008                ))
1009            ),
1010            "err={err:?}"
1011        );
1012        Ok(())
1013    }
1014
1015    #[tokio::test]
1016    async fn read_object_disable_crc32c() -> Result {
1017        let inner = test_inner_client(test_builder()).await;
1018        let stub = crate::storage::transport::Storage::new_test(inner.clone());
1019
1020        let builder = ReadObject::new(
1021            stub,
1022            "projects/_/buckets/bucket",
1023            "object",
1024            inner.options.clone(),
1025        )
1026        .compute_crc32c(false);
1027
1028        assert!(builder.options.checksum.crc32c.is_none());
1029        Ok(())
1030    }
1031
1032    #[tokio::test]
1033    async fn read_object_enable_crc32c() -> Result {
1034        let inner = test_inner_client(test_builder()).await;
1035        let stub = crate::storage::transport::Storage::new_test(inner.clone());
1036
1037        let builder = ReadObject::new(
1038            stub,
1039            "projects/_/buckets/bucket",
1040            "object",
1041            inner.options.clone(),
1042        )
1043        .compute_crc32c(false) // Disable it first, because by default it's enabled
1044        .compute_crc32c(true);
1045
1046        assert!(builder.options.checksum.crc32c.is_some());
1047        Ok(())
1048    }
1049
1050    #[tokio::test]
1051    async fn read_object_disable_crc32c_ignores_mismatch() -> Result {
1052        let server = Server::run();
1053        // Calculate and serialize the incorrect crc32c checksum
1054        let u = crc32c::crc32c("goodbye world".as_bytes());
1055        let value = base64::prelude::BASE64_STANDARD.encode(u.to_be_bytes());
1056        server.expect(
1057            Expectation::matching(all_of![
1058                request::method_path("GET", "/storage/v1/b/test-bucket/o/test-object"),
1059                request::query(url_decoded(contains(("alt", "media")))),
1060            ])
1061            .respond_with(
1062                status_code(200)
1063                    .body("hello world")
1064                    .append_header("x-goog-hash", format!("crc32c={value}"))
1065                    .append_header("x-goog-generation", 123456),
1066            ),
1067        );
1068
1069        let client = Storage::builder()
1070            .with_endpoint(format!("http://{}", server.addr()))
1071            .with_credentials(Anonymous::new().build())
1072            .build()
1073            .await?;
1074        let mut response = client
1075            .read_object("projects/_/buckets/test-bucket", "test-object")
1076            .compute_crc32c(false)
1077            .send()
1078            .await?;
1079        let mut got = Vec::new();
1080        while let Some(b) = response.next().await.transpose()? {
1081            got.extend_from_slice(&b);
1082        }
1083
1084        assert_eq!(got, b"hello world");
1085        Ok(())
1086    }
1087
1088    #[tokio::test]
1089    async fn read_object_incorrect_md5_check() -> Result {
1090        // Calculate and serialize the md5 checksum
1091        let digest = md5::compute("goodbye world".as_bytes());
1092        let value = base64::prelude::BASE64_STANDARD.encode(digest.as_ref());
1093
1094        let server = Server::run();
1095        server.expect(
1096            Expectation::matching(all_of![
1097                request::method_path("GET", "/storage/v1/b/test-bucket/o/test-object"),
1098                request::query(url_decoded(contains(("alt", "media")))),
1099            ])
1100            .times(1)
1101            .respond_with(
1102                status_code(200)
1103                    .body("hello world")
1104                    .append_header("x-goog-hash", format!("md5={value}"))
1105                    .append_header("x-goog-generation", 123456),
1106            ),
1107        );
1108
1109        let client = Storage::builder()
1110            .with_endpoint(format!("http://{}", server.addr()))
1111            .with_credentials(Anonymous::new().build())
1112            .build()
1113            .await?;
1114        let mut response = client
1115            .read_object("projects/_/buckets/test-bucket", "test-object")
1116            .compute_md5()
1117            .send()
1118            .await?;
1119        let mut partial = Vec::new();
1120        let mut err = None;
1121        while let Some(r) = response.next().await {
1122            match r {
1123                Ok(b) => partial.extend_from_slice(&b),
1124                Err(e) => err = Some(e),
1125            };
1126        }
1127        assert_eq!(partial, b"hello world");
1128        let err = err.expect("expect error on incorrect md5");
1129        let source = err.source().and_then(|e| e.downcast_ref::<ReadError>());
1130        assert!(
1131            matches!(
1132                source,
1133                Some(&ReadError::ChecksumMismatch(ChecksumMismatch::Md5 { .. }))
1134            ),
1135            "err={err:?}"
1136        );
1137
1138        Ok(())
1139    }
1140
1141    #[tokio::test]
1142    async fn read_object_with_user_agent() -> Result {
1143        use http::header::USER_AGENT;
1144
1145        let user_agent = "quick_fox_lazy_dog/1.2.3";
1146        let server = Server::run();
1147        server.expect(
1148            Expectation::matching(all_of![
1149                request::method_path("GET", "/storage/v1/b/test-bucket/o/test-object"),
1150                request::headers(contains(("accept-encoding", "gzip"))),
1151                request::headers(contains((USER_AGENT.as_str(), user_agent))),
1152                request::query(url_decoded(contains(("alt", "media")))),
1153            ])
1154            .respond_with(
1155                status_code(200)
1156                    .body("hello world")
1157                    .append_header("x-goog-generation", 123456),
1158            ),
1159        );
1160
1161        let client = Storage::builder()
1162            .with_endpoint(format!("http://{}", server.addr()))
1163            .with_credentials(Anonymous::new().build())
1164            .build()
1165            .await?;
1166        let mut reader = client
1167            .read_object("projects/_/buckets/test-bucket", "test-object")
1168            .with_user_agent(user_agent)
1169            .send()
1170            .await?;
1171        let mut got = Vec::new();
1172        while let Some(b) = reader.next().await.transpose()? {
1173            got.extend_from_slice(&b);
1174        }
1175        assert_eq!(got, b"hello world");
1176
1177        Ok(())
1178    }
1179
1180    #[tokio::test]
1181    async fn read_object_with_quota_project() -> Result {
1182        const PROJECT_NAME: &str = "project_lazy_dog";
1183        let server = Server::run();
1184        server.expect(
1185            Expectation::matching(all_of![
1186                request::method_path("GET", "/storage/v1/b/test-bucket/o/test-object"),
1187                request::headers(contains(("x-goog-user-project", PROJECT_NAME))),
1188            ])
1189            .respond_with(
1190                status_code(200)
1191                    .body("hello world")
1192                    .append_header("x-goog-generation", 123456),
1193            ),
1194        );
1195
1196        let client = Storage::builder()
1197            .with_endpoint(format!("http://{}", server.addr()))
1198            .with_credentials(Anonymous::new().build())
1199            .build()
1200            .await?;
1201        let mut reader = client
1202            .read_object("projects/_/buckets/test-bucket", "test-object")
1203            .with_quota_project(PROJECT_NAME)
1204            .send()
1205            .await?;
1206        let mut got = Vec::new();
1207        while let Some(b) = reader.next().await.transpose()? {
1208            got.extend_from_slice(&b);
1209        }
1210        assert_eq!(got, b"hello world");
1211
1212        Ok(())
1213    }
1214
1215    #[tokio::test]
1216    async fn read_object_strips_credential_quota_project() -> Result {
1217        const PROJECT_NAME: &str = "project_lazy_dog";
1218        const CRED_QUOTA_PROJECT: &str = "cred_quota_project";
1219        let server = Server::run();
1220        server.expect(
1221            Expectation::matching(all_of![
1222                request::method_path("GET", "/storage/v1/b/test-bucket/o/test-object"),
1223                request::headers(contains(("x-goog-user-project", PROJECT_NAME))),
1224                request::headers(not(contains(("x-goog-user-project", CRED_QUOTA_PROJECT)))),
1225            ])
1226            .times(1)
1227            .respond_with(
1228                status_code(200)
1229                    .body("hello world")
1230                    .append_header("x-goog-generation", 123456),
1231            ),
1232        );
1233
1234        let mut mock = MockCredentials::new();
1235        mock.expect_headers().returning(|_exts: http::Extensions| {
1236            let mut map = http::HeaderMap::new();
1237            map.insert(
1238                http::header::AUTHORIZATION,
1239                http::HeaderValue::from_static("Bearer test-token"),
1240            );
1241            map.insert(
1242                "x-goog-user-project",
1243                http::HeaderValue::from_static(CRED_QUOTA_PROJECT),
1244            );
1245            Ok(CacheableResource::New {
1246                data: map,
1247                entity_tag: EntityTag::default(),
1248            })
1249        });
1250        mock.expect_universe_domain().returning(|| None);
1251
1252        let client = Storage::builder()
1253            .with_endpoint(format!("http://{}", server.addr()))
1254            .with_credentials(Credentials::from(mock))
1255            .build()
1256            .await?;
1257        let mut reader = client
1258            .read_object("projects/_/buckets/test-bucket", "test-object")
1259            .with_quota_project(PROJECT_NAME)
1260            .send()
1261            .await?;
1262        let mut got = Vec::new();
1263        while let Some(b) = reader.next().await.transpose()? {
1264            got.extend_from_slice(&b);
1265        }
1266        assert_eq!(got, b"hello world");
1267
1268        Ok(())
1269    }
1270
1271    #[tokio::test]
1272    async fn read_object_retry_preserves_quota_project() -> Result {
1273        const PROJECT_NAME: &str = "project_lazy_dog";
1274        let server = Server::run();
1275        server.expect(
1276            Expectation::matching(all_of![
1277                request::method_path("GET", "/storage/v1/b/test-bucket/o/test-object"),
1278                request::headers(contains(("x-goog-user-project", PROJECT_NAME))),
1279            ])
1280            .times(2)
1281            .respond_with(cycle![
1282                status_code(503),
1283                status_code(200)
1284                    .body("hello")
1285                    .append_header("x-goog-generation", 1)
1286            ]),
1287        );
1288
1289        let client = Storage::builder()
1290            .with_endpoint(format!("http://{}", server.addr()))
1291            .with_credentials(Anonymous::new().build())
1292            .build()
1293            .await?;
1294        let mut reader = client
1295            .read_object("projects/_/buckets/test-bucket", "test-object")
1296            .with_quota_project(PROJECT_NAME)
1297            .send()
1298            .await?;
1299        let mut got = Vec::new();
1300        while let Some(b) = reader.next().await.transpose()? {
1301            got.extend_from_slice(&b);
1302        }
1303        assert_eq!(got, b"hello");
1304
1305        Ok(())
1306    }
1307
1308    #[tokio::test]
1309    async fn read_object() -> Result {
1310        let inner = test_inner_client(test_builder()).await;
1311        let stub = crate::storage::transport::Storage::new_test(inner.clone());
1312        let builder = ReadObject::new(
1313            stub,
1314            "projects/_/buckets/bucket",
1315            "object",
1316            inner.options.clone(),
1317        );
1318        let request = http_request_builder(inner, builder)
1319            .await?
1320            .build_for_tests()
1321            .await?;
1322
1323        assert_eq!(request.method(), Method::GET);
1324        assert_eq!(
1325            request.url().as_str(),
1326            "http://private.googleapis.com/storage/v1/b/bucket/o/object?alt=media"
1327        );
1328        Ok(())
1329    }
1330
1331    #[tokio::test]
1332    async fn read_object_error_credentials() -> Result {
1333        let inner =
1334            test_inner_client(test_builder().with_credentials(error_credentials(false))).await;
1335        let stub = crate::storage::transport::Storage::new_test(inner.clone());
1336        let builder = ReadObject::new(
1337            stub,
1338            "projects/_/buckets/bucket",
1339            "object",
1340            inner.options.clone(),
1341        );
1342        let _ = http_request_builder(inner, builder)
1343            .await?
1344            .build_for_tests()
1345            .await
1346            .inspect_err(|e| assert!(e.is_authentication()))
1347            .expect_err("invalid credentials should err");
1348        Ok(())
1349    }
1350
1351    #[tokio::test]
1352    async fn read_object_bad_bucket() -> Result {
1353        let inner = test_inner_client(test_builder()).await;
1354        let stub = crate::storage::transport::Storage::new_test(inner.clone());
1355        let builder = ReadObject::new(stub, "malformed", "object", inner.options.clone());
1356        let _ = http_request_builder(inner, builder)
1357            .await
1358            .expect_err("malformed bucket string should error");
1359        Ok(())
1360    }
1361
1362    #[tokio::test]
1363    async fn read_object_query_params() -> Result {
1364        let inner = test_inner_client(test_builder()).await;
1365        let stub = crate::storage::transport::Storage::new_test(inner.clone());
1366        let builder = ReadObject::new(
1367            stub,
1368            "projects/_/buckets/bucket",
1369            "object",
1370            inner.options.clone(),
1371        )
1372        .set_generation(5)
1373        .set_if_generation_match(10)
1374        .set_if_generation_not_match(20)
1375        .set_if_metageneration_match(30)
1376        .set_if_metageneration_not_match(40);
1377        let request = http_request_builder(inner, builder)
1378            .await?
1379            .build_for_tests()
1380            .await?;
1381
1382        assert_eq!(request.method(), Method::GET);
1383        let want_pairs: HashMap<String, String> = [
1384            ("alt", "media"),
1385            ("generation", "5"),
1386            ("ifGenerationMatch", "10"),
1387            ("ifGenerationNotMatch", "20"),
1388            ("ifMetagenerationMatch", "30"),
1389            ("ifMetagenerationNotMatch", "40"),
1390        ]
1391        .iter()
1392        .map(|(k, v)| (k.to_string(), v.to_string()))
1393        .collect();
1394        let query_pairs: HashMap<String, String> = request
1395            .url()
1396            .query_pairs()
1397            .map(|param| (param.0.to_string(), param.1.to_string()))
1398            .collect();
1399        assert_eq!(query_pairs.len(), want_pairs.len());
1400        assert_eq!(query_pairs, want_pairs);
1401        Ok(())
1402    }
1403
1404    #[tokio::test]
1405    async fn read_object_default_headers() -> Result {
1406        // The API takes the unencoded byte array.
1407        let inner = test_inner_client(test_builder()).await;
1408        let stub = crate::storage::transport::Storage::new_test(inner.clone());
1409        let builder = ReadObject::new(
1410            stub,
1411            "projects/_/buckets/bucket",
1412            "object",
1413            inner.options.clone(),
1414        );
1415        let request = http_request_builder(inner, builder)
1416            .await?
1417            .build_for_tests()
1418            .await?;
1419
1420        assert_eq!(request.method(), Method::GET);
1421        assert_eq!(
1422            request.url().as_str(),
1423            "http://private.googleapis.com/storage/v1/b/bucket/o/object?alt=media"
1424        );
1425
1426        let want = [("accept-encoding", "gzip")];
1427        let headers = request.headers();
1428        for (name, value) in want {
1429            assert_eq!(
1430                headers.get(name).and_then(|h| h.to_str().ok()),
1431                Some(value),
1432                "{request:?}"
1433            );
1434        }
1435        Ok(())
1436    }
1437
1438    #[tokio::test]
1439    async fn read_object_automatic_decompression_headers() -> Result {
1440        // The API takes the unencoded byte array.
1441        let inner = test_inner_client(test_builder()).await;
1442        let stub = crate::storage::transport::Storage::new_test(inner.clone());
1443        let builder = ReadObject::new(
1444            stub,
1445            "projects/_/buckets/bucket",
1446            "object",
1447            inner.options.clone(),
1448        )
1449        .with_automatic_decompression(true);
1450        let request = http_request_builder(inner, builder)
1451            .await?
1452            .build_for_tests()
1453            .await?;
1454
1455        assert_eq!(request.method(), Method::GET);
1456        assert_eq!(
1457            request.url().as_str(),
1458            "http://private.googleapis.com/storage/v1/b/bucket/o/object?alt=media"
1459        );
1460
1461        let headers = request.headers();
1462        assert!(headers.get("accept-encoding").is_none(), "{request:?}");
1463        Ok(())
1464    }
1465
1466    #[tokio::test]
1467    async fn read_object_encryption_headers() -> Result {
1468        // Make a 32-byte key.
1469        let (key, key_base64, _, key_sha256_base64) = create_key_helper();
1470
1471        // The API takes the unencoded byte array.
1472        let inner = test_inner_client(test_builder()).await;
1473        let stub = crate::storage::transport::Storage::new_test(inner.clone());
1474        let builder = ReadObject::new(
1475            stub,
1476            "projects/_/buckets/bucket",
1477            "object",
1478            inner.options.clone(),
1479        )
1480        .set_key(KeyAes256::new(&key)?);
1481        let request = http_request_builder(inner, builder)
1482            .await?
1483            .build_for_tests()
1484            .await?;
1485
1486        assert_eq!(request.method(), Method::GET);
1487        assert_eq!(
1488            request.url().as_str(),
1489            "http://private.googleapis.com/storage/v1/b/bucket/o/object?alt=media"
1490        );
1491
1492        let want = [
1493            ("x-goog-encryption-algorithm", "AES256".to_string()),
1494            ("x-goog-encryption-key", key_base64),
1495            ("x-goog-encryption-key-sha256", key_sha256_base64),
1496        ];
1497
1498        let headers = request.headers();
1499        for (name, value) in want {
1500            assert_eq!(
1501                headers.get(name).and_then(|h| h.to_str().ok()),
1502                Some(value.as_str())
1503            );
1504        }
1505        Ok(())
1506    }
1507
1508    #[test_case(ReadRange::all(), None; "no headers needed")]
1509    #[test_case(ReadRange::offset(10), Some(&http::HeaderValue::from_static("bytes=10-")); "offset only")]
1510    #[test_case(ReadRange::tail(2000), Some(&http::HeaderValue::from_static("bytes=-2000")); "negative offset")]
1511    #[test_case(ReadRange::segment(0, 100), Some(&http::HeaderValue::from_static("bytes=0-99")); "limit only")]
1512    #[test_case(ReadRange::segment(1000, 100), Some(&http::HeaderValue::from_static("bytes=1000-1099")); "offset and limit")]
1513    #[tokio::test]
1514    async fn range_header(input: ReadRange, want: Option<&http::HeaderValue>) -> Result {
1515        let inner = test_inner_client(test_builder()).await;
1516        let stub = crate::storage::transport::Storage::new_test(inner.clone());
1517        let builder = ReadObject::new(
1518            stub,
1519            "projects/_/buckets/bucket",
1520            "object",
1521            inner.options.clone(),
1522        )
1523        .set_read_range(input.clone());
1524        let request = http_request_builder(inner, builder)
1525            .await?
1526            .build_for_tests()
1527            .await?;
1528
1529        assert_eq!(request.method(), Method::GET);
1530        assert_eq!(
1531            request.url().as_str(),
1532            "http://private.googleapis.com/storage/v1/b/bucket/o/object?alt=media"
1533        );
1534
1535        assert_eq!(request.headers().get("range"), want);
1536        Ok(())
1537    }
1538
1539    #[test_case("projects/p", "projects%2Fp")]
1540    #[test_case("kebab-case", "kebab-case")]
1541    #[test_case("dot.name", "dot.name")]
1542    #[test_case("under_score", "under_score")]
1543    #[test_case("tilde~123", "tilde~123")]
1544    #[test_case("exclamation!point!", "exclamation%21point%21")]
1545    #[test_case("spaces   spaces", "spaces%20%20%20spaces")]
1546    #[test_case("encode%percent%21", "encode%25percent%2521")]
1547    #[test_case("100%25", "100%2525")]
1548    #[test_case("foo%2Fbar", "foo%252Fbar")]
1549    #[test_case(".%2e", ".%252e")]
1550    #[test_case("%2e%2e", "%252e%252e")]
1551    #[test_case("double\"quote", "double%22quote")]
1552    #[test_case("less<than", "less%3Cthan")]
1553    #[test_case("greater>than", "greater%3Ethan")]
1554    #[test_case("back\\slash", "back%5Cslash")]
1555    #[test_case(
1556        "tenant-a/uploads\\..\\..\\tenant-b/secret",
1557        "tenant-a%2Fuploads%5C..%5C..%5Ctenant-b%2Fsecret"
1558    )]
1559    #[test_case("caret^", "caret%5E")]
1560    #[test_case("back`tick", "back%60tick")]
1561    #[test_case("curly{brace}", "curly%7Bbrace%7D")]
1562    #[test_case("vertical|bar", "vertical%7Cbar")]
1563    #[test_case("test-\u{e9}", "test-%C3%A9")]
1564    #[test_case(
1565        "testall !#$&'()*+,/:;=?@[]",
1566        "testall%20%21%23%24%26%27%28%29%2A%2B%2C%2F%3A%3B%3D%3F%40%5B%5D"
1567    )]
1568    #[tokio::test]
1569    async fn test_percent_encoding_object_name(name: &str, want: &str) -> Result {
1570        // Arrange
1571        let inner = test_inner_client(test_builder()).await;
1572        let stub = crate::storage::transport::Storage::new_test(inner.clone());
1573        let builder = ReadObject::new(
1574            stub,
1575            "projects/_/buckets/bucket",
1576            name,
1577            inner.options.clone(),
1578        );
1579        // Act
1580        let request = http_request_builder(inner, builder)
1581            .await?
1582            .build_for_tests()
1583            .await?;
1584        // Assert
1585        assert_eq!(
1586            request.url().path(),
1587            format!("/storage/v1/b/bucket/o/{want}")
1588        );
1589        Ok(())
1590    }
1591
1592    #[test_case("x-guploader-response-body-transformations", "gunzipped", true)]
1593    #[test_case("x-guploader-response-body-transformations", "no match", false)]
1594    #[test_case("warning", "214 UploadServer gunzipped", true)]
1595    #[test_case("warning", "no match", false)]
1596    #[test_case("unused", "unused", false)]
1597    fn test_is_gunzipped(name: &'static str, value: &'static str, want: bool) -> Result {
1598        let response = http::Response::builder()
1599            .status(200)
1600            .header(name, value)
1601            .body(Vec::new())?;
1602        let response = Response::from(response);
1603        let got = Reader::is_gunzipped(&response);
1604        assert_eq!(got, want, "{response:?}");
1605        Ok(())
1606    }
1607}
google_cloud_storage/storage/read_object.rs

google_cloud_storage/storage/
read_object.rs
