本地单元测试 Google 云存储签名 URL

Question

我正在使用 App Engine 编写一个新应用程序，正如文档建议的那样不要使用 Blobstore API，我正在使用 Google 云存储客户端 (GCS)。一切都很好，但我希望能够 return "signed urls" 给客户，这样他们就可以在不通过应用程序的情况下获得 GCS 资源。我相信这就是 signet url 的用途。

但是如何测试呢？我可以成功测试来自客户端的 GCS 调用，但我不知道如何使用 urlfetch 测试客户端的 HTTP 调用。

下面是一个完整的测试用例来说明我的问题：

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

您可以在您的 SDK 中测试 GCS 和 service_accounts，但是当您使用已签名的 url 时，您没有本地 Appengine GCS 服务。

但您可以使用服务帐户和 google 云服务测试您的本地应用程序。

服务帐户使得向其他 Google API 和服务授权应用引擎请求变得非常容易。

要在 Appengine SDK 中使用服务帐户，您必须在 run the development server 时添加两个未记录的选项：

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

更多信息在this request for documentation issue

您可以在您的 appengine 云项目的 developers console permissions section 中创建或找到服务帐户。
您可以为服务帐户创建和下载 p12 密钥。

使用 OpenSSL 将此 p12 密钥转换为 RSA pem 密钥。

我用 this OpenSSL installer 代替 Windows。

要在 Windows 中创建 pem 密钥文件，请使用：

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

现在您可以在开发服务器中使用您的云应用服务帐户，并使用app_identity 对请求进行签名和授权。

本地单元测试 Google 云存储签名 URL

Local unit testing Google Cloud Storage signed URL

google-app-engine

google-cloud-storage

google-app-engine-python