本文整理汇总了Python中tests.factories.query_factory.create函数的典型用法代码示例。如果您正苦于以下问题：Python create函数的具体用法？Python create怎么用？Python create使用的例子？那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。
在下文中一共展示了create函数的19个代码示例，这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞，您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: test_doesnt_update_queries_with_different_data_source
def test_doesnt_update_queries_with_different_data_source(self):
query1 = query_factory.create(query=self.query, data_source=self.data_source)
query2 = query_factory.create(query=self.query, data_source=self.data_source)
query3 = query_factory.create(query=self.query, data_source=data_source_factory.create())
query_result_id = data_manager.store_query_result(self.data_source.id, self.query,
self.data, self.runtime, self.utcnow)
self.assertEqual(models.Query.get_by_id(query1.id)._data['latest_query_data'], query_result_id)
self.assertEqual(models.Query.get_by_id(query2.id)._data['latest_query_data'], query_result_id)
self.assertNotEqual(models.Query.get_by_id(query3.id)._data['latest_query_data'], query_result_id)
开发者ID:HasanAboShally，项目名称:redash，代码行数:11，代码来源:test_manager.py
示例2: test_enqueues_only_for_relevant_data_source
def test_enqueues_only_for_relevant_data_source(self):
query = query_factory.create(ttl=60)
query2 = query_factory.create(ttl=3600, query=query.query, query_hash=query.query_hash)
retrieved_at = datetime.datetime.utcnow() - datetime.timedelta(minutes=10)
query_result = query_result_factory.create(retrieved_at=retrieved_at, query=query.query,
query_hash=query.query_hash)
query.latest_query_data = query_result
query2.latest_query_data = query_result
query.save()
query2.save()
with patch('redash.data.Manager.add_job') as add_job_mock:
data_manager.refresh_queries()
add_job_mock.assert_called_once_with(query.query, worker.Job.LOW_PRIORITY, query.data_source)
开发者ID:HasanAboShally，项目名称:redash，代码行数:14，代码来源:test_manager.py
示例3: test_enqueues_only_for_relevant_data_source
def test_enqueues_only_for_relevant_data_source(self):
query = query_factory.create(ttl=60)
query2 = query_factory.create(ttl=3600, query=query.query, query_hash=query.query_hash)
retrieved_at = datetime.datetime.utcnow() - datetime.timedelta(minutes=10)
query_result = query_result_factory.create(retrieved_at=retrieved_at, query=query.query,
query_hash=query.query_hash)
query.latest_query_data = query_result
query2.latest_query_data = query_result
query.save()
query2.save()
with patch('redash.tasks.QueryTask.add_task') as add_job_mock:
refresh_queries()
add_job_mock.assert_called_once_with(query.query, query.data_source, scheduled=True)
开发者ID:MaTriXy，项目名称:redash，代码行数:14，代码来源:test_refresh_queries.py
示例4: test_enqueues_only_for_relevant_data_source
def test_enqueues_only_for_relevant_data_source(self):
query = query_factory.create(schedule="60")
query2 = query_factory.create(schedule="3600", query=query.query, query_hash=query.query_hash)
import psycopg2
retrieved_at = datetime.datetime.utcnow().replace(tzinfo=psycopg2.tz.FixedOffsetTimezone(offset=0, name=None)) - datetime.timedelta(minutes=10)
query_result = query_result_factory.create(retrieved_at=retrieved_at, query=query.query,
query_hash=query.query_hash)
query.latest_query_data = query_result
query2.latest_query_data = query_result
query.save()
query2.save()
with patch('redash.tasks.QueryTask.add_task') as add_job_mock:
refresh_queries()
add_job_mock.assert_called_once_with(query.query, query.data_source, scheduled=True, metadata=ANY)
开发者ID:daamien，项目名称:redash，代码行数:15，代码来源:test_refresh_queries.py
示例5: test_enqueues_query_with_correct_data_source
def test_enqueues_query_with_correct_data_source(self):
query = query_factory.create(schedule="60")
query2 = query_factory.create(schedule="60", query=query.query, query_hash=query.query_hash)
retrieved_at = datetime.datetime.utcnow() - datetime.timedelta(minutes=10)
query_result = query_result_factory.create(retrieved_at=retrieved_at, query=query.query,
query_hash=query.query_hash)
query.latest_query_data = query_result
query2.latest_query_data = query_result
query.save()
query2.save()
with patch('redash.tasks.QueryTask.add_task') as add_job_mock:
refresh_queries()
add_job_mock.assert_has_calls([call(query2.query, query2.data_source, scheduled=True), call(query.query, query.data_source, scheduled=True)], any_order=True)
self.assertEquals(2, add_job_mock.call_count)
开发者ID:waseemahmad，项目名称:redash，代码行数:15，代码来源:test_refresh_queries.py
示例6: test_enqueues_query_with_correct_data_source
def test_enqueues_query_with_correct_data_source(self):
query = query_factory.create(ttl=60)
query2 = query_factory.create(ttl=60, query=query.query, query_hash=query.query_hash)
retrieved_at = datetime.datetime.utcnow() - datetime.timedelta(minutes=10)
query_result = query_result_factory.create(retrieved_at=retrieved_at, query=query.query,
query_hash=query.query_hash)
query.latest_query_data = query_result
query2.latest_query_data = query_result
query.save()
query2.save()
with patch('redash.data.Manager.add_job') as add_job_mock:
data_manager.refresh_queries()
add_job_mock.assert_has_calls([call(query2.query, worker.Job.LOW_PRIORITY, query2.data_source), call(query.query, worker.Job.LOW_PRIORITY, query.data_source)], any_order=True)
self.assertEquals(2, add_job_mock.call_count)
开发者ID:HasanAboShally，项目名称:redash，代码行数:15，代码来源:test_manager.py
示例7: test_enqueues_query_only_once
def test_enqueues_query_only_once(self):
query = query_factory.create(schedule="60")
query2 = query_factory.create(schedule="60", query=query.query, query_hash=query.query_hash,
data_source=query.data_source)
retrieved_at = datetime.datetime.utcnow() - datetime.timedelta(minutes=10)
query_result = query_result_factory.create(retrieved_at=retrieved_at, query=query.query,
query_hash=query.query_hash)
query.latest_query_data = query_result
query2.latest_query_data = query_result
query.save()
query2.save()
with patch('redash.tasks.QueryTask.add_task') as add_job_mock:
refresh_queries()
add_job_mock.assert_called_once_with(query.query, query.data_source, scheduled=True, metadata=ANY)#{'Query ID': query.id, 'Username': 'Scheduled'})
开发者ID:daamien，项目名称:redash，代码行数:15，代码来源:test_refresh_queries.py
示例8: setUp
def setUp(self):
super(CsvQueryResultAPITest, self).setUp()
self.paths = []
self.query_result = query_result_factory.create()
self.query = query_factory.create()
self.path = '/api/queries/{0}/results/{1}.csv'.format(self.query.id, self.query_result.id)
开发者ID:daamien，项目名称:redash，代码行数:7，代码来源:test_controllers.py
示例9: test_get_all_queries
def test_get_all_queries(self):
queries = [query_factory.create() for _ in range(10)]
with app.test_client() as c, authenticated_user(c):
rv = json_request(c.get, '/api/queries')
self.assertEquals(rv.status_code, 200)
self.assertEquals(len(rv.json), 10)
开发者ID:daamien，项目名称:redash，代码行数:8，代码来源:test_controllers.py
示例10: test_get_query
def test_get_query(self):
query = query_factory.create()
with app.test_client() as c, authenticated_user(c):
rv = json_request(c.get, '/api/queries/{0}'.format(query.id))
self.assertEquals(rv.status_code, 200)
self.assertResponseEqual(rv.json, query.to_dict(with_visualizations=True))
开发者ID:daamien，项目名称:redash，代码行数:8，代码来源:test_controllers.py
示例11: test_skips_queries_with_no_ttl
def test_skips_queries_with_no_ttl(self):
query = query_factory.create(schedule=None)
retrieved_at = datetime.datetime.utcnow() - datetime.timedelta(minutes=10)
query_result = query_result_factory.create(retrieved_at=retrieved_at, query=query.query,
query_hash=query.query_hash)
with patch('redash.tasks.QueryTask.add_task') as add_job_mock:
refresh_queries()
self.assertFalse(add_job_mock.called)
开发者ID:daamien，项目名称:redash，代码行数:9，代码来源:test_refresh_queries.py
示例12: test_skips_queries_with_no_ttl
def test_skips_queries_with_no_ttl(self):
query = query_factory.create(ttl=-1)
retrieved_at = datetime.datetime.utcnow() - datetime.timedelta(minutes=10)
query_result = query_result_factory.create(retrieved_at=retrieved_at, query=query.query,
query_hash=query.query_hash)
with patch('redash.data.Manager.add_job') as add_job_mock:
data_manager.refresh_queries()
self.assertFalse(add_job_mock.called)
开发者ID:HasanAboShally，项目名称:redash，代码行数:9，代码来源:test_manager.py
示例13: test_post_result_list
def test_post_result_list(self):
data_source = data_source_factory.create()
query_result = query_result_factory.create()
query = query_factory.create()
with app.test_client() as c, authenticated_user(c):
rv = json_request(
c.post, "/api/query_results", data={"data_source_id": data_source.id, "query": query.query}
)
self.assertEquals(rv.status_code, 200)
开发者ID:none4uhiggins，项目名称:redash，代码行数:10，代码来源:test_controllers.py
示例14: test_update_query
def test_update_query(self):
query = query_factory.create()
other_user = user_factory.create()
with app.test_client() as c, authenticated_user(c, user=other_user):
rv = json_request(c.post, "/api/queries/{0}".format(query.id), data={"name": "Testing"})
self.assertEqual(rv.status_code, 200)
self.assertEqual(rv.json["name"], "Testing")
self.assertEqual(rv.json["last_modified_by"]["id"], other_user.id)
开发者ID:none4uhiggins，项目名称:redash，代码行数:10，代码来源:test_controllers.py
示例15: test_create_visualization
def test_create_visualization(self):
query = query_factory.create()
data = {"query_id": query.id, "name": "Chart", "description": "", "options": {}, "type": "CHART"}
with app.test_client() as c, authenticated_user(c):
rv = json_request(c.post, "/api/visualizations", data=data)
self.assertEquals(rv.status_code, 200)
data.pop("query_id")
self.assertDictContainsSubset(data, rv.json)
开发者ID:none4uhiggins，项目名称:redash，代码行数:10，代码来源:test_controllers.py
示例16: test_enqueues_outdated_queries
def test_enqueues_outdated_queries(self):
query = query_factory.create(schedule="60")
retrieved_at = datetime.datetime.utcnow() - datetime.timedelta(minutes=10)
query_result = query_result_factory.create(retrieved_at=retrieved_at, query=query.query,
query_hash=query.query_hash)
query.latest_query_data = query_result
query.save()
with patch('redash.tasks.QueryTask.add_task') as add_job_mock:
refresh_queries()
add_job_mock.assert_called_with(query.query, query.data_source, scheduled=True, metadata=ANY)
开发者ID:daamien，项目名称:redash，代码行数:11，代码来源:test_refresh_queries.py
示例17: test_create_visualization
def test_create_visualization(self):
query = query_factory.create()
data = {
'query_id': query.id,
'name': 'Chart',
'description':'',
'options': {},
'type': 'CHART'
}
with app.test_client() as c, authenticated_user(c):
rv = json_request(c.post, '/api/visualizations', data=data)
self.assertEquals(rv.status_code, 200)
data.pop('query_id')
self.assertDictContainsSubset(data, rv.json)
开发者ID:daamien，项目名称:redash，代码行数:16，代码来源:test_controllers.py
示例18: test_uses_cache_headers_for_specific_result
def test_uses_cache_headers_for_specific_result(self):
query_result = query_result_factory.create()
query = query_factory.create(latest_query_data=query_result)
with app.test_client() as c, authenticated_user(c):
rv = json_request(c.get, '/api/queries/{}/results/{}.json'.format(query.id, query_result.id))
self.assertIn('Cache-Control', rv.headers)
开发者ID:carezone，项目名称:redash，代码行数:6，代码来源:test_query_results.py
示例19: setUp
def setUp(self):
super(TestHMACAuthentication, self).setUp()
self.api_key = 10
self.query = query_factory.create(api_key=self.api_key)
self.path = '/api/queries/{0}'.format(self.query.id)
self.expires = time.time() + 1800
开发者ID:none4uhiggins，项目名称:redash，代码行数:6，代码来源:test_authentication.py
注：本文中的tests.factories.query_factory.create函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台，相关代码片段筛选自各路编程大神贡献的开源项目，源码版权归原作者所有，传播和使用请参考对应项目的License；未经允许，请勿转载。 |
客服电话
APP下载
官方微信
请发表评论