1# Copyright 2021 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
15import io
16import re
17import tempfile
18
19import pytest
20
21from google.cloud import exceptions
22from test_utils.vpcsc_config import vpcsc_config
23from . import _helpers
24
25
26public_bucket = "gcp-public-data-landsat"
27
28
29@vpcsc_config.skip_if_inside_vpcsc
30def test_anonymous_client_access_to_public_bucket():
31    from google.cloud.storage.client import Client
32
33    anonymous_client = Client.create_anonymous_client()
34    bucket = anonymous_client.bucket(public_bucket)
35    (blob,) = _helpers.retry_429_503(anonymous_client.list_blobs)(
36        bucket, max_results=1,
37    )
38    with tempfile.TemporaryFile() as stream:
39        _helpers.retry_429_503(blob.download_to_file)(stream)
40
41
42def test_get_service_account_email(storage_client, service_account):
43    domain = "gs-project-accounts.iam.gserviceaccount.com"
44    email = storage_client.get_service_account_email()
45
46    new_style = re.compile(r"service-(?P<projnum>[^@]+)@{}".format(domain))
47    old_style = re.compile(r"{}@{}".format(storage_client.project, domain))
48    patterns = [new_style, old_style]
49    matches = [pattern.match(email) for pattern in patterns]
50
51    assert any(match for match in matches if match is not None)
52
53
54def test_create_bucket_simple(storage_client, buckets_to_delete):
55    new_bucket_name = _helpers.unique_name("a-new-bucket")
56
57    with pytest.raises(exceptions.NotFound):
58        storage_client.get_bucket(new_bucket_name)
59
60    created = _helpers.retry_429_503(storage_client.create_bucket)(new_bucket_name)
61    buckets_to_delete.append(created)
62
63    assert created.name == new_bucket_name
64
65
66def test_list_buckets(storage_client, buckets_to_delete):
67    buckets_to_create = [
68        _helpers.unique_name("new"),
69        _helpers.unique_name("newer"),
70        _helpers.unique_name("newest"),
71    ]
72    created_buckets = []
73
74    for bucket_name in buckets_to_create:
75        bucket = _helpers.retry_429_503(storage_client.create_bucket)(bucket_name)
76        buckets_to_delete.append(bucket)
77
78    all_buckets = storage_client.list_buckets()
79
80    created_buckets = [
81        bucket.name for bucket in all_buckets if bucket.name in buckets_to_create
82    ]
83
84    assert sorted(created_buckets) == sorted(buckets_to_create)
85
86
87def test_download_blob_to_file_w_uri(
88    storage_client, shared_bucket, blobs_to_delete, service_account,
89):
90    blob = shared_bucket.blob("MyBuffer")
91    payload = b"Hello World"
92    blob.upload_from_string(payload)
93    blobs_to_delete.append(blob)
94
95    with tempfile.NamedTemporaryFile() as temp_f:
96
97        with open(temp_f.name, "wb") as file_obj:
98            storage_client.download_blob_to_file(
99                "gs://" + shared_bucket.name + "/MyBuffer", file_obj
100            )
101
102        with open(temp_f.name, "rb") as file_obj:
103            stored_contents = file_obj.read()
104
105    assert stored_contents == payload
106
107
108def test_download_blob_to_file_w_etag(
109    storage_client, shared_bucket, blobs_to_delete, service_account,
110):
111    filename = "kittens"
112    blob = shared_bucket.blob(filename)
113    payload = b"fluffy"
114    blob.upload_from_string(payload)
115    blobs_to_delete.append(blob)
116
117    buffer = io.BytesIO()
118    with pytest.raises(exceptions.NotModified):
119        storage_client.download_blob_to_file(
120            "gs://" + shared_bucket.name + "/" + filename,
121            buffer,
122            if_etag_not_match=blob.etag,
123        )
124
125    buffer = io.BytesIO()
126    with pytest.raises(exceptions.PreconditionFailed):
127        storage_client.download_blob_to_file(
128            "gs://" + shared_bucket.name + "/" + filename,
129            buffer,
130            if_etag_match="kittens",
131        )
132
133    buffer = io.BytesIO()
134    storage_client.download_blob_to_file(
135        "gs://" + shared_bucket.name + "/" + filename,
136        buffer,
137        if_etag_not_match="kittens",
138    )
139    assert buffer.getvalue() == payload
140
141    buffer = io.BytesIO()
142    storage_client.download_blob_to_file(
143        "gs://" + shared_bucket.name + "/" + filename, buffer, if_etag_match=blob.etag,
144    )
145    assert buffer.getvalue() == payload
146