diff --git a/ui/Makefile b/ui/Makefile index 198c823..5da86c0 100644 --- a/ui/Makefile +++ b/ui/Makefile @@ -1,6 +1,8 @@ +MODULES=cli.py kleintest.py mtree.py server.py + test: fixtures/sample.data.pasn1 fixtures/sample.mtree (. ./p/bin/activate && \ - (ls cli.py mtree.py | entr sh -c 'python -m coverage run -m unittest cli mtree && coverage report -m --omit=p/\*')) + (ls $(MODULES) | entr sh -c 'python -m coverage run -m unittest $(basename $(MODULES)) && coverage report -m --omit=p/\*')) env: virtualenv p diff --git a/ui/kleintest.py b/ui/kleintest.py new file mode 100644 index 0000000..265f5b2 --- /dev/null +++ b/ui/kleintest.py @@ -0,0 +1,162 @@ +#!/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) + diff --git a/ui/requirements.txt b/ui/requirements.txt index 06ea053..eedeccc 100644 --- a/ui/requirements.txt +++ b/ui/requirements.txt @@ -2,3 +2,5 @@ urwid -e git://github.com/jmgurney/pasn1@27ff594a2609c07205753ce24f74d8f45a7ea418#egg=pasn1 coverage mock +klein +cryptography diff --git a/ui/server.py b/ui/server.py new file mode 100644 index 0000000..09cd01c --- /dev/null +++ b/ui/server.py @@ -0,0 +1,47 @@ +#!/usr/bin/env python + +# Notes: +# Python requests: https://2.python-requests.org/en/master/ +# IRequest interface: https://twistedmatrix.com/documents/current/api/twisted.web.iweb.IRequest.html +# IResource interface: https://twistedmatrix.com/documents/current/api/twisted.web.resource.IResource.html +# Twisted TDD: https://twistedmatrix.com/documents/current/core/howto/trial.html +# Hypothesis: https://hypothesis.readthedocs.io/en/latest/ +# Going Async from Flask to Twisted Klein: https://crossbario.com/blog/Going-Asynchronous-from-Flask-to-Twisted-Klein/ +# Klein POST docs: https://klein.readthedocs.io/en/latest/examples/handlingpost.html + +from klein import Klein +from twisted.trial import unittest +from twisted.web.iweb import IRequest +from kleintest import * + +import hashlib +import os.path +import shutil +import tempfile + +class MEDAStore: + app = Klein() + + @app.route('/') + def home(request): + return 'hello' + +# twistd support +#medastore = MEDAStore() +#resource = medastore.app.resource + +class Test(unittest.TestCase): + def setUp(self): + d = os.path.realpath(tempfile.mkdtemp()) + self.basetempdir = d + self.medastore = MEDAStore() + self.requests = FakeRequests(self.medastore.app) + + def tearDown(self): + shutil.rmtree(self.basetempdir) + self.basetempdir = None + + def test_404(self): + h = hashlib.sha256(open('fixtures/testfiles/test.txt').read()).hexdigest() + r = self.requests.get('/chash/%s' % h) + self.assertEqual(r.status_code, 404)