Python webtest 模块，TestApp() 实例源码

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)

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/')

def test_internal_redirect_with_after_hook(self):
        run_hook = []

        class RootController(object):
            @expose()
            def internal(self):
                redirect('/testing', internal=True)

            @expose()
            def testing(self):
                return 'it worked!'

        class SimpleHook(PecanHook):
            def after(self, state):
                run_hook.append('after')

        app = TestApp(make_app(RootController(), hooks=[SimpleHook()]))
        response = app.get('/internal')
        assert response.body == b_('it worked!')

        assert len(run_hook) == 1

def test_getall_with_trailing_slash(self):

        class ThingsController(RestController):

            data = ['zero', 'one', 'two', 'three']

            @expose('json')
            def get_all(self):
                return dict(items=self.data)

        class RootController(object):
            things = ThingsController()

        # create the app
        app = TestApp(make_app(RootController()))

        # test get_all
        r = app.get('/things/')
        assert r.status_int == 200
        assert r.body == b_(dumps(dict(items=ThingsController.data)))

def test_post_with_kwargs_only(self):

        class RootController(RestController):

            @expose()
            def get_all(self):
                return 'INDEX'

            @expose('json')
            def post(self, **kw):
                return kw

        # create the app
        app = TestApp(make_app(RootController()))

        r = app.get('/')
        assert r.status_int == 200
        assert r.body == b_('INDEX')

        kwargs = {'foo': 'bar', 'spam': 'eggs'}
        r = app.post('/', kwargs)
        assert r.status_int == 200
        assert r.namespace['foo'] == 'bar'
        assert r.namespace['spam'] == 'eggs'

def test_nested_rest_with_default(self):

        class FooController(RestController):

            @expose()
            def _default(self, *remainder):
                return "DEFAULT %s" % remainder

        class RootController(RestController):
            foo = FooController()

        app = TestApp(make_app(RootController()))

        r = app.get('/foo/missing')
        assert r.status_int == 200
        assert r.body == b_("DEFAULT missing")

def test_rest_with_non_utf_8_body(self):
        if PY3:
            # webob+PY3 doesn't suffer from this bug; the POST parsing in PY3
            # seems to more gracefully detect the bytestring
            return

        class FooController(RestController):

            @expose()
            def post(self):
                return "POST"

        class RootController(RestController):
            foo = FooController()

        app = TestApp(make_app(RootController()))

        data = struct.pack('255h', *range(0, 255))
        r = app.post('/foo/', data, expect_errors=True)
        assert r.status_int == 400

def setUp(self):
        """set up the test
        """
        import transaction
        from pyramid import paster, testing
        from webtest import TestApp
        from stalker import db
        from stalker.db.session import DBSession

        testing.setUp()
        import os
        import stalker_pyramid
        app = paster.get_app(
            os.path.join(
                os.path.dirname(
                    stalker_pyramid.__path__[0],
                ),
                'testing.ini'
            ).replace('\\', '/')
        )

        self.test_app = TestApp(app)

        # patch DBSession commit, to let the db.init() work
        # with its calls to DBSession.commit()
        _orig_commit = DBSession.commit
        DBSession.commit = transaction.commit
        db.setup(app.registry.settings)
        db.init()
        # restore DBSession.commit
        DBSession.commit = _orig_commit

        from stalker import User
        self.admin = User.query.filter(User.name == 'admin').first()

def setUp(self):
        secrets = SecretsStore("example_secrets.json")
        self.app = SocketServer(
            metrics=Mock(),
            dispatcher=Mock(),
            secrets=secrets,
            error_reporter=None,
            ping_interval=1,
            admin_auth='test-auth',
            conn_shed_rate=5,
        )
        self.test_app = webtest.TestApp(self.app)

def setUp(self):
        self.under_test = webtest.TestApp(routes.app)
        self.testbed = testbed.Testbed()
        self.testbed.activate()
        self.testbed.init_memcache_stub()
        self.testbed.init_app_identity_stub()

def setUp(self):
        patch('googleapiclient.discovery.build').start()
        self.testbed = testbed.Testbed()
        self.testbed.activate()
        self.testbed.init_memcache_stub()
        app = webapp2.WSGIApplication(
            [('/createModels', ModelCreatorHandler)]
        )
        self.under_test = webtest.TestApp(app)

def init_webtest(self):
        self.under_test = webtest.TestApp(routes.app)
        self.testbed = testbed.Testbed()
        self.testbed.activate()
        self.testbed.init_memcache_stub()

        path = os.path.join(os.path.dirname(__file__), '../config')
        logging.debug("queue.yaml path: %s", path)
        self.testbed.init_taskqueue_stub(root_path=path)
        self.taskqueue_stub = self.testbed.get_stub(
            testbed.TASKQUEUE_SERVICE_NAME)
        self.testbed.init_app_identity_stub()

def snapshot(args, test=False):
    urlpath = '/VolumeDriver.Snapshot'
    param = json.dumps({'Name': args.name[0]})
    if test:
        resp = TestApp(app).post(urlpath, param)
    else:
        resp = Session().post(
            'http+unix://{}{}'
            .format(urllib.parse.quote_plus(SOCKET), urlpath),
            param)
    res = get_from(resp, 'Snapshot')
    if res:
        print(res)
    return res

def send(args, test=False):
    urlpath = '/VolumeDriver.Snapshot.Send'
    param = {'Name': args.snapshot[0], 'Host': args.host[0]}
    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 sync(args, test=False):
    urlpath = '/VolumeDriver.Volume.Sync'
    param = {'Volumes': args.volumes, 'Hosts': args.hosts}
    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 _make_app(conf):
        return webtest.TestApp(app.load_app(conf, appname='panko+keystone'))

def _make_app(self, conf):
        content = ('{"context_is_admin": "role:admin",'
                   '"telemetry:events:index": "rule:context_is_admin",'
                   '"telemetry:events:show": "rule:context_is_admin"}')
        if six.PY3:
            content = content.encode('utf-8')
        self.tempfile = fileutils.write_to_tempfile(content=content,
                                                    prefix='policy',
                                                    suffix='.json')

        conf.set_override("policy_file", self.tempfile, group='oslo_policy')
        return webtest.TestApp(app.load_app(conf, appname='panko+noauth'))

def _make_app(conf):
        return webtest.TestApp(app.load_app(conf, appname='panko+noauth'))

def setUp(self):
        from eitbapi import main
        app = main({})
        from webtest import TestApp
        self.testapp = TestApp(app)

def setUp(self):
        from eitbapi import main
        app = main({})
        from webtest import TestApp
        self.testapp = TestApp(app)

def setUp(self):
        from eitbapi import main
        app = main({})
        from webtest import TestApp
        self.testapp = TestApp(app)

def setUp(self):
        from eitbapi import main
        self.app = main({})
        self.testapp = TestApp(self.app)