1 from .fixtures import app_client
3 from urllib.parse import urlencode
5 pytest.fixture(scope='module')(app_client)
7 def test_homepage(app_client):
8 _, response = app_client.get('/.json')
9 assert response.status == 200
10 assert response.json.keys() == {'test_tables': 0}.keys()
11 d = response.json['test_tables']
12 assert d['name'] == 'test_tables'
13 assert d['tables_count'] == 5
15 def test_database_page(app_client):
16 response = app_client.get('/test_tables.json', gather_request=False)
18 assert 'test_tables' == data['database']
27 'foreign_keys': {'incoming': [], 'outgoing': []}
29 'name': '/group1/array2',
36 'foreign_keys': {'incoming': [], 'outgoing': []}
38 'name': '/group1/table1',
39 'columns': ['identity', 'idnumber', 'speed'],
45 'foreign_keys': {'incoming': [], 'outgoing': []}
47 'name': '/group2/multi',
54 'foreign_keys': {'incoming': [], 'outgoing': []}
56 'name': '/group2/table2',
57 'columns': ['identity', 'idnumber', 'speed'],
63 'foreign_keys': {'incoming': [], 'outgoing': []}
66 def test_custom_sql(app_client):
67 response = app_client.get(
68 '/test_tables.json?' + urlencode({
69 'sql': 'select identity from [/group1/table1]',
76 'sql': 'select identity from [/group1/table1]',
79 assert 1000 == len(data['rows'])
81 {'identity': 'This is particle: 0'},
82 {'identity': 'This is particle: 1'},
83 {'identity': 'This is particle: 2'},
84 {'identity': 'This is particle: 3'}
86 assert ['identity'] == data['columns']
87 assert 'test_tables' == data['database']
88 assert data['truncated']
90 def test_custom_complex_sql(app_client):
91 response = app_client.get(
92 '/test_tables.json?' + urlencode({
93 'sql': 'select identity from [/group1/table1] where speed > 100 and idnumber < 55',
100 'sql': 'select identity from [/group1/table1] where speed > 100 and idnumber < 55',
103 assert 4 == len(data['rows'])
105 {'identity': 'This is particle: 51'},
106 {'identity': 'This is particle: 52'},
107 {'identity': 'This is particle: 53'},
108 {'identity': 'This is particle: 54'}
110 assert ['identity'] == data['columns']
111 assert 'test_tables' == data['database']
112 assert not data['truncated']
114 def test_custom_pytables_sql(app_client):
115 response = app_client.get(
116 '/test_tables.json?' + urlencode({
117 'sql': 'select identity from [/group1/table1] where (speed > 100) & (speed < 500)',
124 'sql': 'select identity from [/group1/table1] where (speed > 100) & (speed < 500)',
127 assert 199 == len(data['rows'])
129 {'identity': 'This is particle: 51'},
130 {'identity': 'This is particle: 52'},
131 {'identity': 'This is particle: 53'}
132 ] == data['rows'][:3]
133 assert ['identity'] == data['columns']
134 assert 'test_tables' == data['database']
135 assert not data['truncated']
137 def test_invalid_custom_sql(app_client):
138 response = app_client.get(
139 '/test_tables.json?sql=.schema',
142 assert response.status == 400
143 assert response.json['ok'] is False
144 assert 'Statement must be a SELECT' == response.json['error']
146 def test_table_json(app_client):
147 response = app_client.get(
148 '/test_tables/%2Fgroup2%2Ftable2.json?_shape=objects',
151 assert response.status == 200
153 assert data['query']['sql'] == 'select rowid, * from [/group2/table2] order by rowid limit 51'
154 assert data['rows'][3:6] == [{
156 'identity': 'This is particle: 3',
161 'identity': 'This is particle: 4',
166 'identity': 'This is particle: 5',
171 def test_table_not_exists_json(app_client):
174 'error': 'Table not found: blah',
178 '/test_tables/blah.json', gather_request=False
181 def test_table_shape_arrays(app_client):
182 response = app_client.get(
183 '/test_tables/%2Fgroup2%2Ftable2.json?_shape=arrays',
187 [6, 'This is particle: 6', 6, 12.0],
188 [7, 'This is particle: 7', 7, 14.0],
189 ] == response.json['rows'][6:8]
191 def test_table_shape_objects(app_client):
192 response = app_client.get(
193 '/test_tables/%2Fgroup2%2Ftable2.json?_shape=objects',
198 'identity': 'This is particle: 6',
203 'identity': 'This is particle: 7',
206 }] == response.json['rows'][6:8]
208 def test_table_shape_array(app_client):
209 response = app_client.get(
210 '/test_tables/%2Fgroup2%2Ftable2.json?_shape=array',
215 'identity': 'This is particle: 6',
220 'identity': 'This is particle: 7',
223 }] == response.json[6:8]
225 def test_table_shape_invalid(app_client):
226 response = app_client.get(
227 '/test_tables/%2Fgroup2%2Ftable2.json?_shape=invalid',
232 'error': 'Invalid _shape: invalid',
237 @pytest.mark.parametrize('path, expected_rows, expected_pages', [
238 ('/test_tables/%2Farray1.json', 2, 1),
239 ('/test_tables/%2Farray1.json?_size=1', 2, 2),
240 ('/test_tables/%2Fgroup1%2Farray2.json?_size=1000', 10000, 10),
241 ('/test_tables/%2Fgroup2%2Fmulti.json?_size=5', 10, 2),
243 def test_paginate_tables_and_arrays(app_client, path, expected_rows, expected_pages):
247 response = app_client.get(path, gather_request=False)
248 print("*****", response.json)
249 assert 200 == response.status
251 fetched.extend(response.json['rows'])
252 path = response.json['next_url']
254 assert response.json['next']
255 assert '_next={}'.format(response.json['next']) in path
257 assert expected_rows == len(fetched)
258 assert expected_pages == count