repro_gcp_9163.py

import datetime

import requests

from google.api_core import exceptions
from google.cloud import storage

BUCKET_NAME = 'repro-gcp-9163'
BLOB_NAME = 'test.txt'
PAYLOAD = b"DEADBEEF"

def setup(client):
    bucket = client.bucket(BUCKET_NAME)

    try:
        bucket.create()
    except exceptions.Conflict:
        pass

    blob = bucket.blob(BLOB_NAME)

    if not blob.exists():
        blob.upload_from_string(PAYLOAD)

    return blob

def test_signed_url(blob, version, filename=None):
    now = datetime.datetime.utcnow()
    expiration = now + datetime.timedelta(seconds=3600)
    kwargs = {}

    if filename is not None:
        disposition = f"attachment; filename={filename}"
        kwargs["response_disposition"] = disposition

    signed_url = blob.generate_signed_url(
        version=version,
        expiration=expiration,
        **kwargs
    )

    response = requests.get(signed_url)

    if not response.ok:
        print(f"Failed: {version}:{filename}:{signed_url}")
        print(f"{response.status_code}: {response.reason}")
        print(response.content)
    else:
        assert response.content == PAYLOAD

def main(blob):
    for version in ("v2", "v4"):
        for filename in (None, "test.txt"):
            test_signed_url(blob, version, filename)

def teardown(client):
    try:
        client.bucket(BUCKET_NAME).delete(force=True)
    except exceptions.NotFound:
        pass

if __name__ == '__main__':
    client = storage.Client()
    blob = setup(client)
    main(blob)
    teardown(client)