Local unit testing Google Cloud Storage signed URL

Question

I am writing a new application using App Engine and, as the docs suggest to not use Blobstore API, I'm using the Google Cloud Storage client (GCS). All is good but I want to be able to return "signed urls" to clients so they can get the GCS resources without passing through the application. I believe that is what signet urls are for.

But how to test that? I can sucessfully test GCS calls from the client, but I have no idea how to test the client's HTTP calls using urlfetch.

Below is a full test case that illustrates my issue:

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())

GCS_API_ACCESS_ENDPOIND has no local endpoint. You can only use: storage.googleapis.com. localhost:8000/_ah/gcs is the endpont for the local / SDK Google Google Cloud Storage Client Library, which is used by your test_gcs_works() — voscausa
@voscausa you mean I cannot test GCS HTTP access from local development? test_gcs_works uses only the cloudstorage api, but I need to test HTTP access. What I intend to do in the app is redirect the user to the signed url I created so I do not have to serve the file directly from the application request cycle. — marc.fargas
You can test GCS and service_accounts in your SDK, but you do not have a local appengine GCS service when you use a signed url. — voscausa
newbie-meta question: how do I mark this comment as the right answer for the post? :( Or can you post an answer stating GCS URLs can't be tested, only API calls? thx. — marc.fargas

voscausa voscausa · Accepted Answer · 2016-02-07T15:24:05

You can test GCS and service_accounts in your SDK, but you do not have a local appengine GCS service when you use a signed url.

But you can test your local app with service accounts and google cloud services.

Service accounts make it very easy to authorize appengine requests to other Google APIs and services.

To use a service account in the appengine SDK, you have to add two undocumented options when you run the development server:

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

More info in this request for documentation issue

You can create or find the service account in the developers console permissions section of your appengine cloud project.
And you can create and download a p12 key for the service account.

Use OpenSSL to convert this p12 key in a RSA pem key.

I used this OpenSSL installer for Windows.

To create the pem key file in Windows use:

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

Now you can use your cloud app service accounts in the development server and use app_identity to sign and authorize requests.

Local unit testing Google Cloud Storage signed URL

1 Answers