medashare/ui/kleintest.py

#!/usr/bin/env python

from klein import Klein
from klein.interfaces import IKleinRequest
from twisted.internet.defer import Deferred
from twisted.trial import unittest
from twisted.web.http_headers import Headers
from zope.interface import implements
from StringIO import StringIO
from requests.structures import CaseInsensitiveDict

__all__ = [ 'FakeRequests', ]

class FakeHTTPRequest(object):
	# https://github.com/twisted/twisted/blob/twisted-19.7.0/src/twisted/web/http.py#L664
	implements(IKleinRequest)

	code = 200

	def __init__(self, meth, uri):

		#self.requestHeaders = Headers()
		self.responseHeaders = Headers()

		self.path = uri
		self.prepath = []
		self.postpath = uri.split('/')
		self.method = meth
		self.notifications = []
		self.finished = False

	def setHeader(self, name, value):
		self.responseHeaders.setRawHeaders(name, [value])

	def setResponseCode(self, code, message=None):
		self.code = code

	def getRequestHostname(self):
		return ''

	def getHost(self):
		return ''

	def isSecure(self):
		return False

	#def processingFailed(self, failure):
	#	print 'f:', `failure`
	#	print 'b:', failure.getTraceback()

	def _cleanup(self):
		for d in self.notifications:
			d.callback(None)
		self.notifications = []

	def finish(self):
		if self.finished: # pragma: no cover
			warnings.warn('Warning! request.finish called twice.', stacklevel=2)

		self.finished = True

		self._cleanup()

	#def processingFailed(self, failure):
	#	print 'pf:', failure.getTraceback()

	def notifyFinish(self):
		self.notifications.append(Deferred())
		return self.notifications[-1]

class FakeRequestsResponse(object):
	def __init__(self, req):
		self._req = req
		self._io = StringIO()
		req.write = self.write

	def _finished(self, arg):
		if arg is not None:
			raise NotImplementedError('cannot handle exceptions yet')

		self.status_code = self._req.code

		self.text = self._io.getvalue()
		self.headers = CaseInsensitiveDict((k.lower(), v[-1]) for k, v in self._req.responseHeaders.getAllRawHeaders())

	def write(self, data):
		self._io.write(data)

class FakeRequests(object):
	'''This class wraps a Klein app into a calling interface that is similar
	to the requests module for testing apps.

	Example test:
	```
	app = Klein()

	@app.route('/')
	def home(request):
		return 'hello'

	class TestFakeRequests(unittest.TestCase):
		def setUp(self):
			self.requests = FakeRequests(app)

		def test_basic(self):
			r = self.requests.get('/')
			self.assertEqual(r.status_code, 200)
			self.assertEqual(r.text, 'hello')
	```
	'''

	def __init__(self, app):
		'''Wrap the passed in app as if it was a server for a requests
		like interface.  The URLs expected will not be complete urls.'''

		self._app = app
		self._res = app.resource()

	def get(self, url):
		'''Return a response for the passed in url.'''

		if url[0] != '/':
			raise ValueError('url must be absolute (start w/ a slash)')

		req = FakeHTTPRequest('GET', url)
		resp = FakeRequestsResponse(req)

		req.notifyFinish().addBoth(resp._finished)

		r = self._res.render(req)

		return resp

class TestFakeRequests(unittest.TestCase):
	def setUp(self):
		app = Klein()

		@app.route('/')
		def home(request):
			request.setHeader('x-testing', 'value')

			return 'hello'

		@app.route('/404')
		def notfound(request):
			request.setResponseCode(404)
			return 'not found'

		self.requests = FakeRequests(app)

	def test_bad(self):
		self.assertRaises(ValueError, self.requests.get, 'foobar')

	def test_basic(self):
		r = self.requests.get('/')
		self.assertEqual(r.status_code, 200)
		self.assertEqual(r.text, 'hello')
		self.assertEqual(r.headers['X-testing'], 'value')

		r = self.requests.get('/404')
		self.assertEqual(r.status_code, 404)