test http server for unit testing
Actual code to test:
import cStringIO
import json
import os.path
from urlparse import urlparse
import pycurl
from flask import current_app
from PIL import Image
def upload_image(fileobj, mimetype):
"""
Upload a jpeg image to the image upload service. Return 2-tuple
(<response code="code">, <image path="path">)
Note: pycurl was used instead of urllib2 because urllib2 does not easily
support POSTing binary data.
"""
# TODO: set the maximum size
url = 'http://{host}/upload/image?purpose=dp&maxsize;=2000'.format(
host=current_app.config['IMAGE_API_HOST'])
current_app.logger.debug('Uploading to: ' + url)
c = pycurl.Curl()
c.setopt(pycurl.VERBOSE, 1)
c.setopt(pycurl.URL, url)
fout = cStringIO.StringIO()
c.setopt(pycurl.WRITEFUNCTION, fout.write)
c.setopt(pycurl.POST, 1)
c.setopt(pycurl.HTTPHEADER, ['Content-Type: %s' % mimetype])
# Get the size of the file
fileobj.seek(0, 2)
filesize = fileobj.tell()
fileobj.seek(0, 0)
c.setopt(pycurl.POSTFIELDSIZE, filesize)
c.setopt(pycurl.READFUNCTION, fileobj.read)
c.perform()
response_code = c.getinfo(pycurl.RESPONSE_CODE)
response_data = fout.getvalue()
if response_code == 200:
image_path = json.loads(response_data)['path']
else:
current_app.logger.error(
'Upload failed: POST %s: HTTP %d: %s' % (
url, response_code, response_data))
image_path = None
c.close()
return response_code, image_path
def download_image(url):
"""
Download an image from the image service. Return 2-tuple
(</image></response><response code="code">, <file object="object">)
"""
current_app.logger.debug('Downloading: ' + url)
c = pycurl.Curl()
c.setopt(pycurl.VERBOSE, 1)
c.setopt(pycurl.URL, url)
fout = cStringIO.StringIO()
c.setopt(pycurl.WRITEFUNCTION, fout.write)
c.perform()
response_code = c.getinfo(pycurl.RESPONSE_CODE)
c.close()
fout.seek(0, 0)
return response_code, fout
</file></response>
Test HTTP server:
import BaseHTTPServer
import imghdr
import os.path
import textwrap
TEST_IMAGE_DIR = os.path.join(
os.path.abspath(os.path.dirname(__file__)), 'testimages')
TEST_SERVER_HOST = 'localhost'
TEST_SERVER_PORT_IMAGE_UPLOAD = 8555
TEST_SERVER_PORT_IMAGE_DOWNLOAD = 8666
TEST_SERVER_IMAGE_UPLOAD = '{}:{}'.format(
TEST_SERVER_HOST, TEST_SERVER_PORT_IMAGE_UPLOAD)
TEST_SERVER_IMAGE_DOWNLOAD = '{}:{}'.format(
TEST_SERVER_HOST, TEST_SERVER_PORT_IMAGE_DOWNLOAD)
class ImageUploadAPIHandler(BaseHTTPServer.BaseHTTPRequestHandler):
def do_POST(self):
ctype = self.headers.getheader('content-type').lower()
if ctype != 'image/jpeg':
self.send_response(400)
self.end_headers()
self.wfile.write(textwrap.dedent('''
<html>
<head><title>400 Bad Request</title></head>
<body bgcolor="white">
<center><h1>400 Bad Request</h1></center>
<hr /><center>Python Test Server</center>
</body>
</html>
'''))
return
length = int(self.headers.getheader('content-length'))
fstream = self.rfile.read(length)
imgtype = imghdr.what('', h=fstream)
if imgtype != 'jpeg':
self.send_response(415)
self.end_headers()
self.wfile.write(textwrap.dedent('''
<html>
<head><title>415 Unsupported Media Type</title></head>
<body bgcolor="white">
<center><h1>415 Unsupported Media Type</h1></center>
<hr /><center>Python Test Server</center>
</body>
</html>
'''))
return
self.send_response(200)
self.send_header('Content-Type', 'application/json')
self.end_headers()
self.wfile.write(textwrap.dedent('''
{
"path":"/path/to/uploaded/test/image.jpg",
"width":"335",
"height":"500",
"type":"jpg"
}
'''))
class ImageDownloadHandler(BaseHTTPServer.BaseHTTPRequestHandler):
def do_GET(self):
imageobj = open(os.path.join(TEST_IMAGE_DIR, 'tower-crane.jpg'))
self.send_response(200)
self.end_headers()
self.wfile.write(imageobj.read())
httpd_image_upload = BaseHTTPServer.HTTPServer(
(TEST_SERVER_HOST, TEST_SERVER_PORT_IMAGE_UPLOAD), ImageUploadAPIHandler)
httpd_image_download = BaseHTTPServer.HTTPServer(
(TEST_SERVER_HOST, TEST_SERVER_PORT_IMAGE_DOWNLOAD), ImageDownloadHandler)
Test code:
import imghdr
import os.path
import textwrap
import threading
from flask import current_app
from mock import Mock
from dp.blueprints.profile import imageutil
from tests.blueprints.profile.asset_service_test_server import (
httpd_image_upload, httpd_image_download, TEST_SERVER_IMAGE_UPLOAD,
TEST_SERVER_IMAGE_DOWNLOAD, TEST_IMAGE_DIR)
from tests.util import _BaseTestCase
class UploadImageTestCase(_BaseTestCase):
def setUp(self):
super(UploadImageTestCase, self).setUp()
self.app.logger.error = Mock()
self.app.config['IMAGE_API_HOST'] = TEST_SERVER_IMAGE_UPLOAD
# Start the asset service test server in another thread
httpd_thread = threading.Thread(target=httpd_image_upload.serve_forever)
httpd_thread.daemon = True
httpd_thread.start()
def test_success(self):
fileobj = open(os.path.join(TEST_IMAGE_DIR, 'tower-crane.jpg'))
response_code, image_path = imageutil.upload_image(fileobj, 'image/jpeg')
self.assertEqual(response_code, 200)
self.assertEqual(image_path, '/path/to/uploaded/test/image.jpg')
def test_error_400(self):
fileobj = open(os.path.join(TEST_IMAGE_DIR, 'tower-crane.jpg'))
response_code, image_path = imageutil.upload_image(fileobj, 'image/png')
current_app.logger.error.assert_called_once_with(
'Upload failed: POST http://{}/upload/image?purpose=dp&maxsize;=2000: HTTP 400: '.format(
TEST_SERVER_IMAGE_UPLOAD)
+ textwrap.dedent('''
<html>
<head><title>400 Bad Request</title></head>
<body bgcolor="white">
<center><h1>400 Bad Request</h1></center>
<hr /><center>Python Test Server</center>
</body>
</html>
'''))
self.assertEqual(response_code, 400)
self.assertEqual(image_path, None)
def test_error_415(self):
fileobj = open(os.path.join(TEST_IMAGE_DIR, 'Tropical-Fish.png'))
response_code, image_path = imageutil.upload_image(fileobj, 'image/jpeg')
current_app.logger.error.assert_called_once_with(
'Upload failed: POST http://{}/upload/image?purpose=dp&maxsize;=2000: HTTP 415: '.format(
TEST_SERVER_IMAGE_UPLOAD)
+ textwrap.dedent('''
<html>
<head><title>415 Unsupported Media Type</title></head>
<body bgcolor="white">
<center><h1>415 Unsupported Media Type</h1></center>
<hr /><center>Python Test Server</center>
</body>
</html>
'''))
self.assertEqual(response_code, 415)
self.assertEqual(image_path, None)
class DownloadImageTestCase(_BaseTestCase):
def setUp(self):
super(DownloadImageTestCase, self).setUp()
self.app.logger.error = Mock()
# Start the asset service test server in another thread
httpd_thread = threading.Thread(target=httpd_image_download.serve_forever)
httpd_thread.daemon = True
httpd_thread.start()
def test_success(self):
image_url = 'http://{}/path/to/download/image.jpg'.format(
TEST_SERVER_IMAGE_DOWNLOAD)
response_code, fout = imageutil.download_image(image_url)
imgtype = imghdr.what('', h=fout.read())
self.assertEqual(response_code, 200)
self.assertEqual(imgtype, 'jpeg')
