单元测试 BlobStore create_upload_url 上传

Unit testing BlobStore create_upload_url uploads

我正在使用 GAE BlobStore API 创建上传 URL,如下所示:

self.response.write(blobstore.create_upload_url(....))

在生产环境中一切正常,但是当使用 testbed 对此 API 进行单元测试时,我得到以下 URL 返回:

http://testbed.example.com/_ah/upload/agx0ZXN0YmVkLXRlc3RyGwsSFV9fQmxvYlVwbG9hZFNlc3Npb25fXxgDDA

从测试上传到此 URL 无效,我收到 404。我应该正确初始化所有存根,其中包括:

self.testbed.init_datastore_v3_stub()
self.testbed.init_blobstore_stub()
self.testbed.init_files_stub()

我做错了什么?如何使用 blobstore create_upload_url API?

对文件上传进行单元测试

根据我的说法,您不能直接创建 blob 数据,而是通过提交的 Web 表单或 HTTP post 请求间接创建 blob。有关更多详细信息,您可以阅读文档 [1].

[1] https://cloud.google.com/appengine/docs/java/blobstore/#Java_Introducing_the_Blobstore

根据评论和查看代码,我假设您不能开箱即用。

我的解决方案是自己实现上传,使用google.appengine.api.datastore.Getcreate_upload_urlurl指向的会话中提取信息。

下面是一些测试代码。它使用来自 WebTest 的 post 和 GCS 的存根,尽管后者可能是不必要的。

from google.appengine.api import datastore
import uuid
import base64
import md5

def testSomething(self):
  upload_url = self.testapp.post('/path/to/create_upload_url/handler').body
  response = self.upload_blobstore_file(upload_url, "file", "myfile.txt", "MYDATA1").json
  # Test something on the response from the success handler and/or backing storage

def upload_blobstore_file(self, upload_url, field, filename, contents, type = "application/octet-stream"):
  session = datastore.Get(upload_url.split('/')[-1])
  new_object = "/" + session["gs_bucket_name"] + "/" + str(uuid.uuid4())
  self.storage_stub.store(new_object, type, contents)

  message = "Content-Type: application/octet-stream\r\nContent-Length: " + str(len(contents)) + "\r\nContent-MD5: " + base64.b64encode(md5.new(contents).hexdigest()) + "\r\nX-AppEngine-Cloud-Storage-Object: /gs" + new_object.encode("ascii") + "\r\ncontent-disposition: form-data; name=\"" + field + "\"; filename=\"" + filename + "\"\r\nX-AppEngine-Upload-Creation: 2015-07-17 16:19:55.531719\r\n\r\n"
  return self.testapp.post(session["success_path"], upload_files = [
  (field, filename, message, "message/external-body; blob-key=\"encoded_gs_file:blablabla\"; access-type=\"X-AppEngine-BlobKey\"")
])

在测试台中设置http_host:

self.testbed.setup_env(
        http_host='%s:%s' % (SERVER_NAME, PORT),
        overwrite=True,)