本地单元测试签名URL

Question

我正在使用API编写一个新的应用程序，正如文档所建议的，我使用的是(GCS)。这一切都很好，但我希望能够将“签名urls”返回给客户端，这样他们就可以获得GCS资源，而无需通过应用程序。我相信这就是印章网址的用途。

但如何测试呢？我可以成功地测试来自客户机的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时，您没有本地应用程序引擎GCS服务。

但是你可以用服务帐户和谷歌云服务来测试你的本地应用程序。

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

要在appengine中使用服务帐户，必须在运行开发服务器中添加两个未记录的选项：

1. --appidentity_email_address=<SERVICE_ACCOUNT_EMAIL_ADDRESS>
2. --appidentity_private_key_path=<PEM_KEY_PATH>

关于这份文件问题的请求的更多信息

您可以在应用程序云项目的开发人员控制台权限部分中创建或找到服务帐户。

您可以为服务帐户创建和下载一个p12密钥。

使用OpenSSL在RSA密钥中转换此p12密钥。

我在Windows上使用这个OpenSSL安装程序。

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

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

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