Local Testing of Google Cloud Storage Modules

Question

I am writing a new application using the App Engine, and since the docs suggest not using the Blobstore API, I am using the Google Cloud Storage (GCS) client. Everything is fine, but I want to be able to return "signed URLs" to clients so that they can receive GCS resources without going through the application. I believe that this is what the URL is written for.

But how to check it? I can successfully test GCS calls from the client, but I do not know how to test client HTTP calls using urlfetch.

The following is a complete test case illustrating my problem:

import base64
import mimetypes
import urllib
import urllib2
from datetime import datetime, timedelta
import time

from google.appengine.api import app_identity
from google.appengine.datastore import datastore_stub_util
from google.appengine.ext import testbed
from google.appengine.ext import ndb
import unittest

import cloudstorage

# IS THIS RIGHT ?
GCS_API_ACCESS_ENDPOINT = 'http://localhost:8000/_ah/gcs'


def sign_url(bucket_object, expires_after_seconds=60):
    """ cloudstorage signed url to download cloudstorage object without login
        Docs : https://cloud.google.com/storage/docs/access-control?hl=bg#Signed-URLs
        API : https://cloud.google.com/storage/docs/reference-methods?hl=bg#getobject
    """
    # source: https://github.com/voscausa/appengine-gcs-signed-url/blob/05b8a93e2777679d40af62cc5ffce933216e6a85/sign_url.py
    method = 'GET'
    gcs_filename = urllib.quote(bucket_object)
    content_md5, content_type = None, None

    # expiration : number of seconds since epoch
    expiration_dt = datetime.utcnow() + timedelta(seconds=expires_after_seconds)
    expiration = int(time.mktime(expiration_dt.timetuple()))

    # Generate the string to sign.
    signature_string = '\n'.join([
        method,
        content_md5 or '',
        content_type or '',
        str(expiration),
        gcs_filename])

    signature_bytes = app_identity.sign_blob(signature_string)[1]
    google_access_id = app_identity.get_service_account_name()

    # Set the right query parameters. we use a gae service account for the id
    query_params = {'GoogleAccessId': google_access_id,
                    'Expires': str(expiration),
                    'Signature': base64.b64encode(signature_bytes)}

    # Return the built URL.
    result = '{endpoint}{resource}?{querystring}'.format(endpoint=GCS_API_ACCESS_ENDPOINT,
                                                         resource=gcs_filename,
                                                         querystring=urllib.urlencode(query_params))
    return result


FILE_DATA = "This is file contents."
MIME = "text/plain"


class TestGCS(unittest.TestCase):
    def setUp(self):
        self.testbed = testbed.Testbed()
        self.testbed.activate()
        self.policy = datastore_stub_util.PseudoRandomHRConsistencyPolicy(probability=0)
        self.testbed.init_datastore_v3_stub(consistency_policy=self.policy)
        self.testbed.init_app_identity_stub()
        self.testbed.init_memcache_stub()
        self.testbed.init_urlfetch_stub()
        self.testbed.init_blobstore_stub()
        ndb.get_context().clear_cache()

    def tearDown(self):
        self.testbed.deactivate()

    def test_gcs_works(self):
        with cloudstorage.open('/mybucket/test.txt', 'w', content_type=MIME) as f:
            f.write(FILE_DATA)
        with cloudstorage.open('/mybucket/test.txt', 'r') as f:
            data = f.read()
        print(data)
        self.assertEqual(data, FILE_DATA)

    def test_signurl(self):
        url = sign_url('/mybucket/test.txt')
        # FIXME: Not yet working as we have no idea on how to access local GCS during the test.
        result = urllib2.urlopen(url)
        self.assertEqual(200, result.code)
        self.assertEqual(FILE_DATA, result.read())

Answer 1

GCS service_accounts SDK, GCS , URL.

google.

API Google.

SDK appengine, , :

• --appidentity_email_address=<SERVICE_ACCOUNT_EMAIL_ADDRESS>
• --appidentity_private_key_path=<PEM_KEY_PATH>

.
p12 .

OpenSSL p12 RSA.

OpenSSL Windows.

pem Windows, :

openssl pkcs12 -in <P12_KEY_PATH> -nocerts -nodes -passin pass:notasecret | openssl rsa -out <PEM_KEY_PATH>

app_identity .