Skip to content

Instantly share code, notes, and snippets.

@wojtossfm
Created April 18, 2015 12:10
Show Gist options
  • Save wojtossfm/55c924857e2caf6c3522 to your computer and use it in GitHub Desktop.
Save wojtossfm/55c924857e2caf6c3522 to your computer and use it in GitHub Desktop.
Created this gist basing on test code that I put together for a project I'm working on. Some of it is a bit rough, but it's meant to be a gist to help others save time .
#!/usr/bin/python2
# -*- coding: utf-8 -*-
import unittest
import webtest
import base64
import datetime
import hashlib
import os
from google.appengine.ext import testbed
import webapp2
from google.appengine.ext.webapp import blobstore_handlers
from google.appengine.ext import blobstore
import cloudstorage
BUCKET_NAME = "testbucket"
import tempfile
class GCSHandlerPrepare(webapp2.RequestHandler):
def get(self):
gs_upload_key = os.path.join(BUCKET_NAME)
target = blobstore.create_upload_url("/uploaded", gs_bucket_name=gs_upload_key)
self.response.write(target)
class GCSHandlerUploaded(webapp2.RequestHandler):
def post(self):
upload_handler = blobstore_handlers.BlobstoreUploadHandler()
upload_handler.initialize(self.request, self.response)
file_info = upload_handler.get_file_infos("file")[0]
blob_key = blobstore.create_gs_key(file_info.gs_object_name.strip())
encoded = base64.encodestring(blob_key).strip()
self.redirect("/download?blob_key={}".format(encoded))
class GCSHandlerDownload(blobstore_handlers.BlobstoreDownloadHandler):
def get(self):
encoded = self.request.get("blob_key")
blob_key = base64.decodestring(encoded)
self.send_blob(blob_key, "application/octet-stream")
def calculate_md5(filepath):
with open(filepath) as file_handle:
md5_hash = hashlib.md5()
for line in file_handle:
md5_hash.update(line)
return md5_hash.hexdigest()
class AppTest(unittest.TestCase):
def setUp(self):
app = webapp2.WSGIApplication([
('/prepare', GCSHandlerPrepare),
('/uploaded', GCSHandlerUploaded),
('/download', GCSHandlerDownload),
])
self.testapp = webtest.TestApp(app)
self.testbed = testbed.Testbed()
self.testbed.activate()
self.testbed.init_memcache_stub()
self.testbed.init_app_identity_stub()
self.testbed.init_blobstore_stub()
self.testbed.init_datastore_v3_stub()
self.testbed.init_urlfetch_stub()
def tearDown(self):
self.testbed.deactivate()
def do_upload(self, prep_url, post_url, field, filepath):
'''
Seems stubs don't handle uploads fully or I'm
setting it up improperly. Worked out the following
mock upload mechanism which seems to work nicely.
:return: Response instance for the request
:rtype: webtest.TestResponse
'''
target = self.testapp.get(prep_url).normal_body
self.assertTrue(target)
filename = os.path.basename(filepath)
creation = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S") + ".000000"
md5_raw = calculate_md5(filepath)
md5 = base64.encodestring(md5_raw)
bucket_target = "/{}/{}".format(BUCKET_NAME, filename)
contents = "Content-Type: text/plain\n" \
"Content-Length:1\n" \
"X-AppEngine-Cloud-Storage-Object: /gs{}\n".format(bucket_target) + \
"X-AppEngine-Upload-Creation: "+creation+"\n" \
"content-md5: " + md5 +"\n"
# seems we have to upload directly as stub isn't working?
# found a partial workaround to simulate a post-successful upload call to the request
with cloudstorage.open(bucket_target, "w") as file_handle:
with open(filepath) as local_handle:
data = local_handle.read()
file_handle.write(data)
blobkey = blobstore.create_gs_key("/gs"+bucket_target)
content_type = 'text/plain; blob-key={key};'.format(key=blobkey)
params =[
(field, webtest.forms.Upload(filename, contents, content_type))
]
response = self.testapp.post(post_url, params=params, content_type=b"multipart/form-data")
return response
def test_upload_download(self):
# Create temporary file to upload
file_descriptor, filepath = tempfile.mkstemp(text=True)
sample_content = "This is a sample test file"
os.write(file_descriptor, sample_content)
os.close(file_descriptor)
# Upload the file
response = self.do_upload("/prepare", "/uploaded", "file", filepath)
self.assertEqual(302, response.status_code)
self.assertIn("/download", response.location)
# Download the file
response = self.testapp.get(response.location)
# md5_raw = calculate_md5(filepath)
self.assertEqual(200, response.status_code)
self.assertEqual("application/octet-stream", response.content_type)
'''
When I looked into the code it seems that the mechanism that
would insert the actual blob data into the body isn't part of the
stubs so we need to do that ourselves.
'''
blobkey = response.headers["X-AppEngine-BlobKey"]
blob_contents = blobstore.fetch_data(blobkey, 0, 100)
self.assertEqual(sample_content, blob_contents)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment