forked from getredash/redash
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathtest_parameterized_query.py
More file actions
188 lines (139 loc) · 7.27 KB
/
test_parameterized_query.py
File metadata and controls
188 lines (139 loc) · 7.27 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
from unittest import TestCase
from mock import patch
from collections import namedtuple
import pytest
from redash.utils.parameterized_query import ParameterizedQuery, InvalidParameterError, dropdown_values
class TestParameterizedQuery(TestCase):
def test_returns_empty_list_for_regular_query(self):
query = ParameterizedQuery(u"SELECT 1")
self.assertEqual(set([]), query.missing_params)
def test_finds_all_params_when_missing(self):
query = ParameterizedQuery(u"SELECT {{param}} FROM {{table}}")
self.assertEqual(set(['param', 'table']), query.missing_params)
def test_finds_all_params(self):
query = ParameterizedQuery(u"SELECT {{param}} FROM {{table}}").apply({
'param': 'value',
'table': 'value'
})
self.assertEqual(set([]), query.missing_params)
def test_deduplicates_params(self):
query = ParameterizedQuery(u"SELECT {{param}}, {{param}} FROM {{table}}").apply({
'param': 'value',
'table': 'value'
})
self.assertEqual(set([]), query.missing_params)
def test_handles_nested_params(self):
query = ParameterizedQuery(u"SELECT {{param}}, {{param}} FROM {{table}} -- {{#test}} {{nested_param}} {{/test}}").apply({
'param': 'value',
'table': 'value'
})
self.assertEqual(set(['test', 'nested_param']), query.missing_params)
def test_handles_objects(self):
query = ParameterizedQuery(u"SELECT * FROM USERS WHERE created_at between '{{ created_at.start }}' and '{{ created_at.end }}'").apply({
'created_at': {
'start': 1,
'end': 2
}
})
self.assertEqual(set([]), query.missing_params)
def test_raises_on_parameters_not_in_schema(self):
schema = [{"name": "bar", "type": "text"}]
query = ParameterizedQuery("foo", schema)
with pytest.raises(InvalidParameterError):
query.apply({"qux": 7})
def test_raises_on_invalid_text_parameters(self):
schema = [{"name": "bar", "type": "text"}]
query = ParameterizedQuery("foo", schema)
with pytest.raises(InvalidParameterError):
query.apply({"bar": 7})
def test_validates_text_parameters(self):
schema = [{"name": "bar", "type": "text"}]
query = ParameterizedQuery("foo {{bar}}", schema)
query.apply({"bar": u"baz"})
self.assertEquals("foo baz", query.text)
def test_raises_on_invalid_number_parameters(self):
schema = [{"name": "bar", "type": "number"}]
query = ParameterizedQuery("foo", schema)
with pytest.raises(InvalidParameterError):
query.apply({"bar": "baz"})
def test_validates_number_parameters(self):
schema = [{"name": "bar", "type": "number"}]
query = ParameterizedQuery("foo {{bar}}", schema)
query.apply({"bar": 7})
self.assertEquals("foo 7", query.text)
def test_raises_on_invalid_date_parameters(self):
schema = [{"name": "bar", "type": "date"}]
query = ParameterizedQuery("foo", schema)
with pytest.raises(InvalidParameterError):
query.apply({"bar": "baz"})
def test_validates_date_parameters(self):
schema = [{"name": "bar", "type": "date"}]
query = ParameterizedQuery("foo {{bar}}", schema)
query.apply({"bar": "2000-01-01 12:00:00"})
self.assertEquals("foo 2000-01-01 12:00:00", query.text)
def test_raises_on_invalid_enum_parameters(self):
schema = [{"name": "bar", "type": "enum", "enumOptions": ["baz", "qux"]}]
query = ParameterizedQuery("foo", schema)
with pytest.raises(InvalidParameterError):
query.apply({"bar": 7})
def test_raises_on_unlisted_enum_value_parameters(self):
schema = [{"name": "bar", "type": "enum", "enumOptions": ["baz", "qux"]}]
query = ParameterizedQuery("foo", schema)
with pytest.raises(InvalidParameterError):
query.apply({"bar": "shlomo"})
def test_validates_enum_parameters(self):
schema = [{"name": "bar", "type": "enum", "enumOptions": ["baz", "qux"]}]
query = ParameterizedQuery("foo {{bar}}", schema)
query.apply({"bar": "baz"})
self.assertEquals("foo baz", query.text)
@patch('redash.utils.parameterized_query.dropdown_values')
def test_raises_on_invalid_query_parameters(self, _):
schema = [{"name": "bar", "type": "query", "queryId": 1}]
query = ParameterizedQuery("foo", schema)
with pytest.raises(InvalidParameterError):
query.apply({"bar": 7})
@patch('redash.utils.parameterized_query.dropdown_values', return_value=[{"value": "baz"}])
def test_raises_on_unlisted_query_value_parameters(self, _):
schema = [{"name": "bar", "type": "query", "queryId": 1}]
query = ParameterizedQuery("foo", schema)
with pytest.raises(InvalidParameterError):
query.apply({"bar": "shlomo"})
@patch('redash.utils.parameterized_query.dropdown_values', return_value=[{"value": "baz"}])
def test_validates_query_parameters(self, _):
schema = [{"name": "bar", "type": "query", "queryId": 1}]
query = ParameterizedQuery("foo {{bar}}", schema)
query.apply({"bar": "baz"})
self.assertEquals("foo baz", query.text)
def test_raises_on_invalid_date_range_parameters(self):
schema = [{"name": "bar", "type": "date-range"}]
query = ParameterizedQuery("foo", schema)
with pytest.raises(InvalidParameterError):
query.apply({"bar": "baz"})
def test_validates_date_range_parameters(self):
schema = [{"name": "bar", "type": "date-range"}]
query = ParameterizedQuery("foo {{bar.start}} {{bar.end}}", schema)
query.apply({"bar": {"start": "2000-01-01 12:00:00", "end": "2000-12-31 12:00:00"}})
self.assertEquals("foo 2000-01-01 12:00:00 2000-12-31 12:00:00", query.text)
def test_raises_on_unexpected_param_types(self):
schema = [{"name": "bar", "type": "burrito"}]
query = ParameterizedQuery("foo", schema)
with pytest.raises(InvalidParameterError):
query.apply({"bar": "baz"})
@patch('redash.utils.parameterized_query._load_result', return_value={
"columns": [{"name": "id"}, {"name": "Name"}, {"name": "Value"}],
"rows": [{"id": 5, "Name": "John", "Value": "John Doe"}]})
def test_dropdown_values_prefers_name_and_value_columns(self, _):
values = dropdown_values(1)
self.assertEquals(values, [{"name": "John", "value": "John Doe"}])
@patch('redash.utils.parameterized_query._load_result', return_value={
"columns": [{"name": "id"}, {"name": "fish"}, {"name": "poultry"}],
"rows": [{"fish": "Clown", "id": 5, "poultry": "Hen"}]})
def test_dropdown_values_compromises_for_first_column(self, _):
values = dropdown_values(1)
self.assertEquals(values, [{"name": 5, "value": 5}])
@patch('redash.utils.parameterized_query._load_result', return_value={
"columns": [{"name": "ID"}, {"name": "fish"}, {"name": "poultry"}],
"rows": [{"fish": "Clown", "ID": 5, "poultry": "Hen"}]})
def test_dropdown_supports_upper_cased_columns(self, _):
values = dropdown_values(1)
self.assertEquals(values, [{"name": 5, "value": 5}])