Python main 模块，app() 实例源码

def test_oauthed(resource, app, login):
    login()

    mock_http = HttpMock(
        resource('datasets-list.json'),
        {'status': '200'})

    with mock.patch.object(main.decorator, 'http', return_value=mock_http):
        with mock.patch.object(
                main.decorator, 'has_credentials', return_value=True):
            response = app.get('/')

    # Should make the api call
    assert response.status_int == 200
    assert re.search(
        re.compile(r'.*datasets.*datasetReference.*etag.*', re.DOTALL),
        response.body)

def test_app(testbed):
    key_name = 'foo'

    testbed.init_taskqueue_stub(root_path=os.path.dirname(__file__))

    app = webtest.TestApp(main.app)
    app.post('/', {'key': key_name})

    tq_stub = testbed.get_stub(gaetestbed.TASKQUEUE_SERVICE_NAME)
    tasks = tq_stub.get_filtered_tasks()
    assert len(tasks) == 1
    assert tasks[0].name == 'task1'

    with mock.patch('main.update_counter') as mock_update:
        # Force update to fail, otherwise the loop will go forever.
        mock_update.side_effect = RuntimeError()

        app.get('/_ah/start', status=500)

        assert mock_update.called

def test_update_schema(app, testbed):
    reload(models_v1)
    test_model = models_v1.Picture(author='Test', name='Test')
    test_model.put()

    response = app.post('/update_schema')
    assert response.status_int == 200

    # Run the queued task.
    tasks = testbed.taskqueue_stub.get_filtered_tasks()
    assert len(tasks) == 1
    deferred.run(tasks[0].payload)

    # Check the updated items
    reload(models_v2)
    updated_model = test_model.key.get()
    assert updated_model.num_votes == 1
    assert updated_model.avg_rating == 5.0

def test_get_search(self):
        with main.app.test_request_context():
            # Add sample search, and attempt to get the JSON
            # representing it
            sid = self.add_sample_search()
            rv = self.app.get("/search/{0}".format(sid))
            d = json.loads(rv.data.decode("utf-8"))
            # Make sure the search data is still good, and
            # and only 1 search is returned
            assert len(d['searches']) == 1
            search = d['searches'][0]
            assert search['id'] == sid
            assert len(search['data_searches']) == 2
            print ("test_get_search passed")

    # Make sure that we can get a list of searches from the backend in
    # JSON

def test_get_search2(self):
        with main.app.test_request_context():
            # Add sample search, and attempt to get the JSON
            # representing it
            sid = self.add_sample_search()
            sid2 = self.add_sample_search()
            rv = self.app.get("/search")
            d = json.loads(rv.data.decode("utf-8"))
            # Make sure the search data is still good, and
            # and only 1 search is returned
            assert len(d['searches']) == 2
            assert len(d['searches'][1]["data_searches"]) == 2
            print ("test_get_search2 passed")

    # Make sure that we can get the metadata about available data sources
    # to enable users to easily create valid searches

def test_sessions(self):
        request = webapp2.Request.blank("/")
        request.method = "POST"
        request.headers["Content-Type"] = "application/json"
        request.body = json.encode({
            'update': 1,
            'message': {
                u'date': 1450696897,
                u'text': u'??????',
                u'from': {
                    u'username': u'm_messiah',
                    u'first_name': u'Maxim',
                    u'last_name': u'Muzafarov',
                    u'id': 1
                },
                u'message_id': 1,
                u'chat': {
                    u'type': u'group',
                    u'id': -1,
                    u'title': u'??'
                }
            }
        })
        request.get_response(app)
        self.assertIn(-1, SESSIONS)

def send_request(self, path, post=None, json_data=None, headers=None,
                     method=None):
        """
        Send request to main app. Returns response object.
        If json_data is given (dict) it is encoded and sent as post payload.
        """
        if headers is None:
            headers = {}
        if json_data is not None:
            post = json.dumps(json_data)
            headers['Content-Type'] = 'application/json'
        request = webapp2.Request.blank(path, POST=post, headers=headers)
        if method:
            request.method = method
        return request.get_response(main.app)

def get_test_app():

    def test_loader(app):
        return load_themes_from(os.path.join(os.path.dirname(__file__), '../themes/'))

    Themes(main.app, app_identifier='yelplove', loaders=[test_loader])

    return TestApp(main.app)

def addCsrfTokenToSession(self):
        csrf_token = 'MY_TOKEN'
        with self.app.session_transaction() as session:
            session['_csrf_token'] = csrf_token
        return csrf_token

def test_app(testbed):
    app = webtest.TestApp(main.app)
    response = app.get('/')
    assert response.status_int == 200

def make_shell_context():
    return dict(
        app=app,
        db=db,
        User=User,
        Post=Post,
        Tag=Tag,
        Comment=Comment
    )

def make_shell_context():
    return dict(
        app=app,
        db=db,
        User=User,
        Post=Post,
        Tag=Tag,
        Comment=Comment
    )

def make_shell_context():
    return dict(app=app, db=db, User=User, Post=Post, Tag=Tag)

def make_shell_context():
    return dict(app=app)

def test_get():
    app = webtest.TestApp(main.app)

    response = app.get('/')

    assert response.status_int == 200

def test_app(testbed):
    app = webtest.TestApp(main.app)
    response = app.get('/')
    assert response.status_int == 200

def test_app(main):
    app = webtest.TestApp(main.app)
    response = app.get('/')

    assert response.status_int == 200
    assert re.search(
        re.compile(r'.*version.*', re.DOTALL),
        response.body)

def app(testbed):
    return webtest.TestApp(main.app)

def test_url_lib(app):
    response = app.get('/')
    assert 'Google' in response.body

def test_url_fetch(app):
    response = app.get('/url_fetch')
    assert 'Google' in response.body

def app(cloud_config, testbed):
    main.PROJECTID = cloud_config.project
    return webtest.TestApp(main.app)

def test_anonymous(app):
    response = app.get('/')

    # Should redirect to login
    assert response.status_int == 302
    assert re.search(r'.*accounts.*Login.*', response.headers['Location'])

def test_loggedin(app, login):
    login()

    response = app.get('/')

    # Should redirect to oauth2
    assert response.status_int == 302
    assert re.search(r'.*oauth2.*', response.headers['Location'])

def test_get_module_info(modules_mock, app):
    modules_mock.get_current_module_name.return_value = "default"
    modules_mock.get_current_instance_id.return_value = 1
    response = app.get('/')
    assert response.status_int == 200
    results = response.body.split('&')
    assert results[0].split('=')[1] == 'default'
    assert results[1].split('=')[1] == '1'

def test_get_backend(url_open_mock, modules_mock, app):
    url_read_mock = mock.Mock(read=mock.Mock(return_value='hello world'))
    url_open_mock.return_value = url_read_mock
    response = app.get('/access_backend')

    assert response.status_int == 200
    assert response.body == 'Got response hello world'

def app(cloud_config, testbed):
    main.PROJECTID = cloud_config.project
    return webtest.TestApp(main.app)

def test_background(thread, app):
    app.get('/dog')
    response = app.get('/')
    assert response.status_int == 200
    assert response.body == 'Dog'
    app.get('/cat')
    # no stub for system so manually set
    main.val = 'Cat'
    response = app.get('/')
    assert response.status_int == 200
    assert response.body == 'Cat'

def test_background_auto_start(thread, app):
    app.get('/dog')
    response = app.get('/')
    assert response.status_int == 200
    assert response.body == 'Dog'
    app.get('/cat?auto=True')
    # no stub for system so manually set
    main.val = 'Cat'
    response = app.get('/')
    assert response.status_int == 200
    assert response.body == 'Cat'

def test_app(app):
    response = app.get('/')
    assert response.status_int == 200

def test_add_entities(app):
    response = app.post('/add_entities')
    assert response.status_int == 200
    response = app.get('/')
    assert response.status_int == 200
    assert 'Author: Bob' in response.body
    assert 'Name: Sunrise' in response.body
    assert 'Author: Alice' in response.body
    assert 'Name: Sunset' in response.body

def test_app(testbed):
    app = webtest.TestApp(main.app)
    response = app.get('/')
    assert response.status_int == 200

def test_app(testbed, login):
    app = webtest.TestApp(main.app)

    login()
    response = app.get('/')

    assert '/_ah/upload' in response

def test_app(testbed):
    app = webtest.TestApp(main.app)
    response = app.get('/')
    assert response.status_int == 200
    assert 'Verified: True' in response.text

def test_app(testbed):
    app = webtest.TestApp(main.app)

    response = app.get('/', status=403)

    response = app.get('/', headers={
        'X-Appengine-Inbound-Appid': 'other-app-id'
    })
    assert response.status_int == 200

def test_get():
    app = webtest.TestApp(main.app)

    response = app.get('/')

    assert response.status_int == 200
    assert response.body == 'Hello, World!'

def test_app(testbed):
    app = webtest.TestApp(main.app)
    response = app.get('/')
    assert response.status_int == 200
    assert 'No log entries found' in response.text
    assert 'More' not in response.text

def test_app(testbed):
    app = webtest.TestApp(main.app)
    response = app.get('/')
    assert response.status_int == 200
    assert 'Logging example' in response.text

def app():
    return webtest.TestApp(main.app)

def test_get(app):
    response = app.get('/')
    assert response.status_int == 200

def test_get(cloud_config):
    main.BUCKET_NAME = cloud_config.project
    app = webtest.TestApp(main.app)

    response = app.get('/')

    assert response.status_int == 200
    assert re.search(
        re.compile(r'.*.*items.*etag.*', re.DOTALL),
        response.body)

def app():
    return webtest.TestApp(main.app)