Created
April 18, 2015 12:10
-
-
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 .
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
#!/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