def test_make_app(self):
test = self.Test2(value=10)
app = bottle_api.make_app(('/api', test))
test_app = TestApp(app)
params = {'arg1': 'test'}
response = test_app.post(
'/api/endpoint/', {'data': json.dumps(params)})
expected_response = {
'data': {'works': True, 'arg1': 'test', 'value': 10},
'status': 'OK'
}
self.assertEqual(response.json, expected_response)
params = {'arg1': 'error'}
response = test_app.post(
'/api/endpoint/', {'data': json.dumps(params)})
expected_response = {
'data': {},
'status': 'ERROR'
}
self.assertEqual(response.json, expected_response)
python类TestApp()的实例源码
def get_from(resp, key):
"""get specified key from plugin response output
"""
if resp is None:
return False
try: # bottle
content = resp.content
except: # TestApp
content = resp.body
if resp.status_code == 200:
error = jsonloads(content)['Err']
if error:
log.error(error)
return False
return jsonloads(content).get(key)
else:
log.error('%s: %s', resp.status_code, resp.reason)
return False
def purge(args, test=False):
urlpath = '/VolumeDriver.Snapshots.Purge'
param = {'Name': args.name[0],
'Pattern': args.pattern[0],
'Dryrun': args.dryrun}
if test:
param['Test'] = True
resp = TestApp(app).post(urlpath, json.dumps(param))
else:
resp = Session().post(
'http+unix://{}{}'
.format(urllib.parse.quote_plus(SOCKET), urlpath),
json.dumps(param))
res = get_from(resp, '')
if res:
print(res)
return res
def setup():
it.BUILDER_NAME = os.environ.get('BUILDER_NAME', None)
it.BUILDER_PATH = os.environ.get('BUILDER_PATH', None)
_logger.info("Builder name: %s", it.BUILDER_NAME)
_logger.info("Builder path: %s", it.BUILDER_PATH)
if it.BUILDER_NAME:
it.PROJECT_FILE = p.join(VIM_HDL_EXAMPLES, it.BUILDER_NAME + '.prj')
else:
it.PROJECT_FILE = None
if it.BUILDER_PATH:
it.patch = mock.patch.dict(
'os.environ',
{'PATH' : os.pathsep.join([it.BUILDER_PATH, os.environ['PATH']])})
it.patch.start()
it.app = TestApp(handlers.app)
def test_tween_overriden():
""" In case our tween is overriden by the user config we should not log
rendering """
from ...test_tracer import get_dummy_tracer
from ...util import override_global_tracer
tracer = get_dummy_tracer()
with override_global_tracer(tracer):
config = Configurator(settings={'pyramid.tweens': 'pyramid.tweens.excview_tween_factory'})
trace_pyramid(config)
def json(request):
return {'a': 1}
config.add_route('json', '/json')
config.add_view(json, route_name='json', renderer='json')
app = webtest.TestApp(config.make_wsgi_app())
app.get('/json', status=200)
spans = tracer.writer.pop()
assert not spans
def test_app(backend):
from webtest import TestApp
from vishnu.middleware import SessionMiddleware
from vishnu.session import Config
api = falcon.API()
api.add_route("/private", PrivateHandler())
api.add_route("/public", PublicHandler())
api.add_route("/login/save", LoginHandler())
api.add_route("/logout", LogoutHandler())
config = Config(
secret="OVc1Mbt79AK5Pmi6sWnJnXZvEPNO3BnI",
backend=backend
)
session = SessionMiddleware(api, config)
return TestApp(app=session, extra_environ={'wsgi.url_scheme': 'https'})
def setUp(self):
"""Defines useful variables and initializes database.
After this, following variables will be available-
1. config: Application configuration
2. engine: DB engine
3. session: DB session instance
4. test_app: Test WSGI app
"""
settings = self.get_settings()
app = services.main({}, **settings)
self.test_app = webtest.TestApp(app=app)
self.config = testing.setUp(settings=settings)
self.engine = models.get_engine(settings)
session_factory = models.get_session_factory(self.engine)
self.session = models.get_tm_session(
session_factory,
transaction.manager
)
self.__init_database()
def _get_initialized_app_context(parsed_args):
"""
:param parsed_args: parsed args (eg. from take_action)
:return: (wsgi_app, test_app)
"""
config_file = parsed_args.config_file
config_name = 'config:%s' % config_file
here_dir = os.getcwd()
# Load locals and populate with objects for use in shell
sys.path.insert(0, here_dir)
# Load the wsgi app first so that everything is initialized right
wsgi_app = loadapp(config_name, relative_to=here_dir)
test_app = TestApp(wsgi_app)
# Make available the tg.request and other global variables
tresponse = test_app.get('/_test_vars')
return wsgi_app, test_app
def app(request):
"""py.test fixture to set up a dummy app for Redis testing.
:param request: pytest's FixtureRequest (internal class, cannot be hinted on a signature)
"""
config = testing.setUp()
config.add_route("home", "/")
config.add_route("redis_test", "/redis_test")
config.add_view(redis_test, route_name="redis_test")
# same is in test.ini
config.registry.settings["redis.sessions.url"] = "redis://localhost:6379/14"
def teardown():
testing.tearDown()
config.registry.redis = create_redis(config.registry)
app = TestApp(config.make_wsgi_app())
return app
def throttle_app(request, paster_config):
'''Custom WSGI app with permission test views enabled.'''
class Initializer(websauna.system.Initializer):
def configure_views(self):
self.config.add_route("throttle_sample", "/")
self.config.add_view(throttle_sample, route_name="throttle_sample", decorator=throttled_view(limit=1))
def configure_csrf(self):
"""Disable CSRF for this test run for making testing simpler."""
self.config.set_default_csrf_options(require_csrf=False)
global_config, app_settings = paster_config
init = Initializer(global_config, app_settings)
init.run()
app = TestApp(init.make_wsgi_app())
app.init = init
return app
def test_throttle(throttle_app: TestApp, test_request):
"""Throttling should give us 429."""
app = throttle_app
# clear counter from previous test run
clear_throttle(test_request, "throttle_sample")
app.get("/", status=200)
# We exceeded the limit of 1 request per hour
app.get("/", status=429)
# Let's clear the counter
clear_throttle(test_request, "throttle_sample")
app.get("/", status=200)
def setup_wsgi():
configurator = Configurator()
configurator.include("pyramid_jinja2")
configurator.add_jinja2_renderer('.html')
configurator.add_jinja2_search_path('websauna.tests:templates/viewconfig', name='.html')
configurator.add_route("parent_hello", "/parent_hello")
configurator.add_route("child_hello", "/child_hello")
from websauna.tests.viewconfig import testmodule
configurator.set_root_factory(testmodule.Root)
configurator.scan(testmodule)
wsgi = TestApp(configurator.make_wsgi_app())
return wsgi
def open(self, url_string, follow_redirects=True, **kwargs):
"""GETs the URL in `url_string`.
:param url_string: A string containing the URL to be opened.
:keyword follow_redirects: If False, this method acts as a simple GET request. If True (the default),
the method hebaves like a browser would, by opening redirect responses.
:keyword relative: Set to True to indicate that `url_string` contains a path relative to the current location.
Other keyword arguments are passed directly on to
`WebTest.get <http://webtest.readthedocs.org/en/latest/api.html#webtest.app.TestApp.get>`_.
"""
if self.last_response:
self.history.append(self.last_response.request.url)
relative = not url_string.startswith('/')
if relative:
url_string = self.get_full_path(url_string)
self.last_response = self.testapp.get(url_string, **kwargs)
if follow_redirects:
self.follow_response()
def setUp(self):
super(SecureControllerSharedPermissionsRegression, self).setUp()
class Parent(object):
@expose()
def index(self):
return 'hello'
class UnsecuredChild(Parent):
pass
class SecureChild(Parent, SecureController):
@classmethod
def check_permissions(cls):
return False
class RootController(object):
secured = SecureChild()
unsecured = UnsecuredChild()
self.app = TestApp(make_app(RootController()))
def forward(app):
app = TestApp(RecursiveMiddleware(app))
res = app.get('')
assert res.headers['content-type'] == 'text/plain'
assert res.status == '200 OK'
assert 'requested page returned' in res
res = app.get('/error')
assert res.headers['content-type'] == 'text/plain'
assert res.status == '200 OK'
assert 'Page not found' in res
res = app.get('/not_found')
assert res.headers['content-type'] == 'text/plain'
assert res.status == '200 OK'
assert 'Page not found' in res
try:
res = app.get('/recurse')
except AssertionError as e:
if str(e).startswith('Forwarding loop detected'):
pass
else:
raise AssertionError('Failed to detect forwarding loop')
def app_(self):
class LookupController(object):
def __init__(self, someID):
self.someID = someID
@expose()
def index(self):
return '/%s' % self.someID
@expose()
def name(self):
return '/%s/name' % self.someID
class RootController(object):
@expose()
def index(self):
return '/'
@expose()
def _lookup(self, someID, *remainder):
return LookupController(someID), remainder
return TestApp(Pecan(RootController()))
def app_(self):
class LookupController(object):
def __init__(self, someID):
self.someID = someID
@expose()
def index(self):
return self.someID
class UserController(object):
@expose()
def _lookup(self, someID, *remainder):
return LookupController(someID), remainder
class RootController(object):
users = UserController()
return TestApp(Pecan(RootController()))
def app_(self):
class RootController(object):
@expose()
def index(self):
redirect('/testing')
@expose()
def internal(self):
redirect('/testing', internal=True)
@expose()
def bad_internal(self):
redirect('/testing', internal=True, code=301)
@expose()
def permanent(self):
redirect('/testing', code=301)
@expose()
def testing(self):
return 'it worked!'
return TestApp(make_app(RootController(), debug=False))
def test_x_forward_proto(self):
class ChildController(object):
@expose()
def index(self):
redirect('/testing') # pragma: nocover
class RootController(object):
@expose()
def index(self):
redirect('/testing') # pragma: nocover
@expose()
def testing(self):
return 'it worked!' # pragma: nocover
child = ChildController()
app = TestApp(make_app(RootController(), debug=True))
res = app.get(
'/child', extra_environ=dict(HTTP_X_FORWARDED_PROTO='https')
)
# non-canonical url will redirect, so we won't get a 301
assert res.status_int == 302
# should add trailing / and changes location to https
assert res.location == 'https://localhost/child/'
assert res.request.environ['HTTP_X_FORWARDED_PROTO'] == 'https'
def app_(self):
class RootController(object):
@expose()
def redirect_with_context(self):
request.context['foo'] = 'bar'
redirect('/testing')
@expose()
def internal_with_context(self):
request.context['foo'] = 'bar'
redirect('/testing', internal=True)
@expose('json')
def testing(self):
return request.context
return TestApp(make_app(RootController(), debug=False))
def test_thread_local_dir(self):
"""
Threadlocal proxies for request and response should properly
proxy ``dir()`` calls to the underlying webob class.
"""
class RootController(object):
@expose()
def index(self):
assert 'method' in dir(request)
assert 'status' in dir(response)
return '/'
app = TestApp(Pecan(RootController()))
r = app.get('/')
assert r.status_int == 200
assert r.body == b_('/')
def test_request_state_cleanup(self):
"""
After a request, the state local() should be totally clean
except for state.app (so that objects don't leak between requests)
"""
from pecan.core import state
class RootController(object):
@expose()
def index(self):
return '/'
app = TestApp(Pecan(RootController()))
r = app.get('/')
assert r.status_int == 200
assert r.body == b_('/')
assert state.__dict__ == {}
def test_kajiki(self):
class RootController(object):
@expose('kajiki:kajiki.html')
def index(self, name='Jonathan'):
return dict(name=name)
app = TestApp(
Pecan(RootController(), template_path=self.template_path)
)
r = app.get('/')
assert r.status_int == 200
assert b_("<h1>Hello, Jonathan!</h1>") in r.body
r = app.get('/index.html?name=World')
assert r.status_int == 200
assert b_("<h1>Hello, World!</h1>") in r.body
def test_renderer_not_found(self):
class RootController(object):
@expose('mako3:mako.html')
def index(self, name='Jonathan'):
return dict(name=name)
app = TestApp(
Pecan(RootController(), template_path=self.template_path)
)
try:
r = app.get('/')
except Exception as e:
expected = e
assert 'support for "mako3" was not found;' in str(expected)
def test_json(self):
try:
from simplejson import loads
except:
from json import loads # noqa
expected_result = dict(
name='Jonathan',
age=30, nested=dict(works=True)
)
class RootController(object):
@expose('json')
def index(self):
return expected_result
app = TestApp(Pecan(RootController()))
r = app.get('/')
assert r.status_int == 200
result = dict(loads(r.body.decode()))
assert result == expected_result
def test_alternate_route(self):
class RootController(object):
@expose(route='some-path')
def some_path(self):
return 'Hello, World!'
app = TestApp(Pecan(RootController()))
r = app.get('/some-path/')
assert r.status_int == 200
assert r.body == b_('Hello, World!')
r = app.get('/some_path/', expect_errors=True)
assert r.status_int == 404
def test_manual_route(self):
class SubController(object):
@expose(route='some-path')
def some_path(self):
return 'Hello, World!'
class RootController(object):
pass
route(RootController, 'some-controller', SubController())
app = TestApp(Pecan(RootController()))
r = app.get('/some-controller/some-path/')
assert r.status_int == 200
assert r.body == b_('Hello, World!')
r = app.get('/some-controller/some_path/', expect_errors=True)
assert r.status_int == 404
def test_custom_route_with_attribute_conflict(self):
class RootController(object):
@expose(route='mock')
def greet(self):
return 'Hello, World!'
@expose()
def mock(self):
return 'You are not worthy!'
app = TestApp(Pecan(RootController()))
self.assertRaises(
RuntimeError,
app.get,
'/mock/'
)
def test_conflicting_custom_routes(self):
class RootController(object):
@expose(route='testing')
def foo(self):
return 'Foo!'
@expose(route='testing')
def bar(self):
return 'Bar!'
app = TestApp(Pecan(RootController()))
self.assertRaises(
RuntimeError,
app.get,
'/testing/'
)
def test_conflicting_custom_routes_in_subclass(self):
class BaseController(object):
@expose(route='testing')
def foo(self):
return request.path
class ChildController(BaseController):
pass
class RootController(BaseController):
child = ChildController()
app = TestApp(Pecan(RootController()))
r = app.get('/testing/')
assert r.body == b_('/testing/')
r = app.get('/child/testing/')
assert r.body == b_('/child/testing/')
