Skip to content
Merged
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
46 changes: 22 additions & 24 deletions tests/system/test_blob.py
Original file line number Diff line number Diff line change
Expand Up @@ -425,26 +425,23 @@ def test_blob_acl_w_metageneration_match(

# Exercise blob ACL with metageneration/generation match
acl = blob.acl
blob.reload()
acl.domain("google.com").grant_read()

with pytest.raises(exceptions.PreconditionFailed):
acl.save_predefined(
"publicRead", if_metageneration_match=wrong_metageneration_number
)
assert "READER" not in acl.all().get_roles()
acl.save(if_metageneration_match=wrong_metageneration_number)
assert "READER" not in acl.domain("google.com").get_roles()

acl.save_predefined("publicRead", if_metageneration_match=blob.metageneration)
assert "READER" in acl.all().get_roles()
acl.save(if_metageneration_match=blob.metageneration)

blob.reload()
del acl.entities["allUsers"]
assert "READER" in acl.domain("google.com").get_roles()
acl.domain("google.com").revoke_read()

with pytest.raises(exceptions.PreconditionFailed):
acl.save(if_generation_match=wrong_generation_number)
assert acl.has_entity("allUsers")
assert "READER" in acl.domain("google.com").get_roles()

acl.save(if_generation_match=blob.generation)
assert not acl.has_entity("allUsers")
assert "READER" not in acl.domain("google.com").get_roles()


def test_blob_acl_upload_predefined(
Expand All @@ -453,31 +450,32 @@ def test_blob_acl_upload_predefined(
file_data,
service_account,
):
control = shared_bucket.blob(f"logo{uuid.uuid4().hex}")
control_info = file_data["logo"]
control_blob = shared_bucket.blob(f"logo{uuid.uuid4().hex}")
control_blob_info = file_data["logo"]

blob = shared_bucket.blob(f"SmallFile{uuid.uuid4().hex}")
info = file_data["simple"]

try:
control.upload_from_filename(control_info["path"])
control_blob.upload_from_filename(control_blob_info["path"])
finally:
blobs_to_delete.append(control)

blobs_to_delete.append(control_blob)
try:
blob.upload_from_filename(info["path"], predefined_acl="publicRead")
blob.upload_from_filename(info["path"], predefined_acl="private")
finally:
blobs_to_delete.append(blob)

control_acl = control.acl
assert "READER" not in control_acl.all().get_roles()
control_blob_acl = control_blob.acl

acl = blob.acl
assert "READER" in acl.all().get_roles()

acl.all().revoke_read()
assert acl.all().get_roles() == set()
assert control_acl.all().get_roles() == acl.all().get_roles()
count = 0
for entry in acl:
count += 1
entity = entry["entity"]
assert count == 1
assert entity.lstrip("user-") == service_account.service_account_email

assert sum(1 for _ in control_blob_acl) > 1


def test_blob_patch_metadata(
Expand Down