mirror of
https://github.com/webrecorder/pywb.git
synced 2025-03-15 00:03:28 +01:00
coverage work! add additional test for wsgi_wrappers
additional test for zipnum bad location for now, not testing cli interfaces which depend on opt params
This commit is contained in:
parent
d702a98bbc
commit
202f6101e0
@ -8,4 +8,5 @@ omit =
|
||||
|
||||
[report]
|
||||
exclude_lines =
|
||||
pragma: no cover
|
||||
if __name__ == .__main__.:
|
||||
|
58
pywb/framework/test/test_wsgi_wrapper.py
Normal file
58
pywb/framework/test/test_wsgi_wrapper.py
Normal file
@ -0,0 +1,58 @@
|
||||
from pywb.framework.wsgi_wrappers import init_app
|
||||
from pywb.framework.wsgi_wrappers import start_wsgi_server
|
||||
|
||||
from pywb.utils.wbexception import AccessException
|
||||
|
||||
import webtest
|
||||
|
||||
class TestOkApp:
|
||||
def __call__(self, env):
|
||||
def response(env, start_response):
|
||||
start_response('200 OK', [])
|
||||
return ['Test']
|
||||
return response
|
||||
|
||||
class TestErrApp:
|
||||
def __call__(self, env):
|
||||
raise Exception('Test Error')
|
||||
|
||||
class TestCustomErrApp:
|
||||
def __call__(self, env):
|
||||
raise AccessException('Forbidden Test')
|
||||
|
||||
|
||||
def initer(app_class):
|
||||
def init():
|
||||
return app_class()
|
||||
return init
|
||||
|
||||
def test_ok_app():
|
||||
the_app = init_app(initer(TestOkApp), load_yaml=False)
|
||||
|
||||
testapp = webtest.TestApp(the_app)
|
||||
resp = testapp.get('/')
|
||||
|
||||
assert resp.status_int == 200
|
||||
assert 'Test' in resp.body
|
||||
|
||||
def test_err_app():
|
||||
the_app = init_app(initer(TestErrApp), load_yaml=False)
|
||||
|
||||
testapp = webtest.TestApp(the_app)
|
||||
resp = testapp.get('/abc', expect_errors=True)
|
||||
|
||||
assert resp.status_int == 400
|
||||
assert '400 Bad Request Error: Test Error' in resp.body
|
||||
|
||||
def test_custom_err_app():
|
||||
the_app = init_app(initer(TestCustomErrApp), load_yaml=False)
|
||||
|
||||
testapp = webtest.TestApp(the_app)
|
||||
resp = testapp.get('/abc', expect_errors=True)
|
||||
|
||||
assert resp.status_int == 403
|
||||
assert '403 Access Denied Error: Forbidden Test' in resp.body
|
||||
|
||||
|
||||
|
||||
|
@ -9,7 +9,7 @@ class NotFoundException(WbException):
|
||||
# Exceptions that effect a specific capture and result in a retry
|
||||
class CaptureException(WbException):
|
||||
def status(self):
|
||||
return '500 Internal Server Error'
|
||||
return '502 Internal Server Error'
|
||||
|
||||
|
||||
class InternalRedirect(WbException):
|
||||
|
@ -71,16 +71,20 @@ class WSGIApp(object):
|
||||
response = WbResponse(StatusAndHeaders(ir.status, ir.httpHeaders))
|
||||
|
||||
except WbException as e:
|
||||
response = handle_exception(env, wb_router.error_view, e, False)
|
||||
response = handle_exception(env, wb_router, e, False)
|
||||
|
||||
except Exception as e:
|
||||
response = handle_exception(env, wb_router.error_view, e, True)
|
||||
response = handle_exception(env, wb_router, e, True)
|
||||
|
||||
return response(env, start_response)
|
||||
|
||||
|
||||
#=================================================================
|
||||
def handle_exception(env, error_view, exc, print_trace):
|
||||
def handle_exception(env, wb_router, exc, print_trace):
|
||||
error_view = None
|
||||
if hasattr(wb_router, 'error_view'):
|
||||
error_view = wb_router.error_view
|
||||
|
||||
if hasattr(exc, 'status'):
|
||||
status = exc.status()
|
||||
else:
|
||||
@ -138,7 +142,7 @@ def init_app(init_func, load_yaml=True, config_file=None):
|
||||
|
||||
|
||||
#=================================================================
|
||||
def start_wsgi_server(the_app):
|
||||
def start_wsgi_server(the_app): # pragma: no cover
|
||||
from wsgiref.simple_server import make_server
|
||||
from optparse import OptionParser
|
||||
|
||||
@ -149,7 +153,8 @@ def start_wsgi_server(the_app):
|
||||
|
||||
port = options.port
|
||||
|
||||
port = the_app.port
|
||||
if not port:
|
||||
port = the_app.port
|
||||
|
||||
if not port:
|
||||
port = DEFAULT_PORT
|
||||
@ -161,5 +166,5 @@ def start_wsgi_server(the_app):
|
||||
httpd.serve_forever()
|
||||
except KeyboardInterrupt as ex:
|
||||
pass
|
||||
|
||||
logging.debug('Stopping CDX Server')
|
||||
finally:
|
||||
logging.debug('Stopping CDX Server')
|
||||
|
@ -70,10 +70,10 @@ def get_rewritten(url, urlrewriter, urlkey=None, head_insert_func=None):
|
||||
return (status_headers, buff)
|
||||
|
||||
#=================================================================
|
||||
def main():
|
||||
def main(): # pragma: no cover
|
||||
if len(sys.argv) < 2:
|
||||
print 'Usage: {0} url-to-fetch [wb-url-target] [extra-prefix]'.format(sys.argv[0])
|
||||
exit(1)
|
||||
return 1
|
||||
else:
|
||||
url = sys.argv[1]
|
||||
|
||||
@ -93,8 +93,9 @@ def main():
|
||||
status_headers, buff = get_rewritten(url, urlrewriter)
|
||||
|
||||
sys.stdout.write(buff)
|
||||
return 0
|
||||
|
||||
|
||||
#=================================================================
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
exit(main())
|
||||
|
@ -1 +1 @@
|
||||
zipnum ./sample_archive/zipcdx/zipnum-sample.cdx.gz
|
||||
zipnum ./sample_archive/x-bad-path-to-ignore-x ./sample_archive/zipcdx/zipnum-sample.cdx.gz
|
||||
|
Loading…
x
Reference in New Issue
Block a user