mirror of
https://github.com/valitydev/redash.git
synced 2024-11-07 01:25:16 +00:00
68 lines
2.8 KiB
Python
68 lines
2.8 KiB
Python
from mock import patch, call, ANY
|
|
from tests import BaseTestCase
|
|
from redash.tasks import refresh_queries
|
|
from redash.models import Query
|
|
|
|
|
|
class TestRefreshQuery(BaseTestCase):
|
|
def test_enqueues_outdated_queries(self):
|
|
"""
|
|
refresh_queries() launches an execution task for each query returned
|
|
from Query.outdated_queries().
|
|
"""
|
|
query1 = self.factory.create_query()
|
|
query2 = self.factory.create_query(
|
|
query_text="select 42;",
|
|
data_source=self.factory.create_data_source())
|
|
oq = staticmethod(lambda: [query1, query2])
|
|
with patch('redash.tasks.queries.enqueue_query') as add_job_mock, \
|
|
patch.object(Query, 'outdated_queries', oq):
|
|
refresh_queries()
|
|
self.assertEqual(add_job_mock.call_count, 2)
|
|
add_job_mock.assert_has_calls([
|
|
call(query1.query_text, query1.data_source, query1.user_id,
|
|
scheduled_query=query1, metadata=ANY),
|
|
call(query2.query_text, query2.data_source, query2.user_id,
|
|
scheduled_query=query2, metadata=ANY)], any_order=True)
|
|
|
|
def test_doesnt_enqueue_outdated_queries_for_paused_data_source(self):
|
|
"""
|
|
refresh_queries() does not launch execution tasks for queries whose
|
|
data source is paused.
|
|
"""
|
|
query = self.factory.create_query()
|
|
oq = staticmethod(lambda: [query])
|
|
query.data_source.pause()
|
|
with patch.object(Query, 'outdated_queries', oq):
|
|
with patch('redash.tasks.queries.enqueue_query') as add_job_mock:
|
|
refresh_queries()
|
|
add_job_mock.assert_not_called()
|
|
|
|
query.data_source.resume()
|
|
|
|
with patch('redash.tasks.queries.enqueue_query') as add_job_mock:
|
|
refresh_queries()
|
|
add_job_mock.assert_called_with(
|
|
query.query_text, query.data_source, query.user_id,
|
|
scheduled_query=query, metadata=ANY)
|
|
|
|
def test_enqueues_parameterized_queries(self):
|
|
"""
|
|
Scheduled queries with parameters use saved values.
|
|
"""
|
|
query = self.factory.create_query(
|
|
query_text="select {{n}}",
|
|
options={"parameters": [{
|
|
"global": False,
|
|
"type": "text",
|
|
"name": "n",
|
|
"value": "42",
|
|
"title": "n"}]})
|
|
oq = staticmethod(lambda: [query])
|
|
with patch('redash.tasks.queries.enqueue_query') as add_job_mock, \
|
|
patch.object(Query, 'outdated_queries', oq):
|
|
refresh_queries()
|
|
add_job_mock.assert_called_with(
|
|
"select 42", query.data_source, query.user_id,
|
|
scheduled_query=query, metadata=ANY)
|