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 implementer
from io import StringIO
from requests.structures import CaseInsensitiveDict

__all__ = [ 'FakeRequests', ]

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

	code = 200

	def __init__(self, meth, uri, data):

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

		self.content = StringIO(data)

		self.path = uri
		self.prepath = []
		self.postpath = uri.split(b'/')
		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 b''

	def getHost(self):
		return b''

	def isSecure(self):
		return False

	def processingFailed(self, failure):
		self.setResponseCode(500, 'Internal Server Error')

		#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 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
		self.status_code = 500

	def _finished(self, arg):
		if arg is not None:	# pragma: no cover
			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 _makerequest(self, method, url, data=''):
		if url[0:1] != b'/':
			raise ValueError('url must be absolute (start w/ a slash)')

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

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

		r = self._res.render(req)

		return resp

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

		print('ktg:', repr(url))
		return self._makerequest('GET', url)

	def put(self, url, data=''):
		'''Make a put request to the provied URL w/ the body of data.'''

		return self._makerequest('PUT', url, data)

class TestFakeRequests(unittest.TestCase):
	def setUp(self):
		self.putdata = []

		app = Klein()

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

			return b'hello'

		@app.route('/500')
		def causeerror(request):
			raise ValueError('random exception')

		@app.route('/put')
		def putreq(request):
			self.putdata.append(request.content.read())

			request.setResponseCode(201)
			return ''

		@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, b'foobar')

	def test_basic(self):
		r = self.requests.get(b'/')
		print(repr(r))
		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)

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

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

		body = 'body'
		r = self.requests.put('/put', data=body)
		self.assertEqual(r.status_code, 201)
		self.assertEqual(r.text, '')
		self.assertEqual(''.join(self.putdata), body)