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)
python类app()的实例源码
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