Search code examples
pythonunit-testinggoogle-app-engineblobstore

Unit testing BlobStore create_upload_url uploads


I'm using the using the GAE BlobStore API to create an upload URL, as follows:

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

Everything works fine in production, but when using testbed to unit test this API, I get the following URL back:

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

Uploading to this URL from the test doesn't work, I get a 404. I should have initialized all stubs properly, amongst others:

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

What am I doing wrong? How do I unit test file uploads using the blobstore create_upload_url API?


Solution

  • From the comments and looking at the code, I'm assuming you can't do it out of the box.

    My solution to the problem is to implement the uploading myself, using the google.appengine.api.datastore.Get to extract information from the session pointed to by the create_upload_url url.

    Below is some test code. It's using post from WebTest, and a stub for GCS, although the latter is probably unnecessary.

    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\"")
    ])