1 from .fixtures import app_client
3 from urllib.parse import urlencode
5 def test_homepage(app_client):
6 response = app_client.get('/.json')
7 assert response.status == 200
8 assert response.json.keys() == {'test_tables': 0}.keys()
9 d = response.json['test_tables']
10 assert d['name'] == 'test_tables'
11 assert d['tables_count'] == 5
13 def test_database_page(app_client):
14 response = app_client.get('/test_tables.json', gather_request=False)
16 assert 'test_tables' == data['database']
25 'foreign_keys': {'incoming': [], 'outgoing': []}
27 'name': '/group1/array2',
34 'foreign_keys': {'incoming': [], 'outgoing': []}
36 'name': '/group1/table1',
37 'columns': ['identity', 'idnumber', 'speed'],
43 'foreign_keys': {'incoming': [], 'outgoing': []}
45 'name': '/group2/multi',
52 'foreign_keys': {'incoming': [], 'outgoing': []}
54 'name': '/group2/table2',
55 'columns': ['identity', 'idnumber', 'speed'],
61 'foreign_keys': {'incoming': [], 'outgoing': []}
64 def test_custom_sql(app_client):
65 response = app_client.get(
66 '/test_tables.json?' + urlencode({
67 'sql': 'select identity from [/group1/table1]',
74 'sql': 'select identity from [/group1/table1]',
77 assert 1000 == len(data['rows'])
79 {'identity': 'This is particle: 0'},
80 {'identity': 'This is particle: 1'},
81 {'identity': 'This is particle: 2'},
82 {'identity': 'This is particle: 3'}
84 assert ['identity'] == data['columns']
85 assert 'test_tables' == data['database']
86 assert data['truncated']
88 def test_custom_complex_sql(app_client):
89 response = app_client.get(
90 '/test_tables.json?' + urlencode({
91 'sql': 'select identity from [/group1/table1] where speed > 100 and idnumber < 55',
98 'sql': 'select identity from [/group1/table1] where speed > 100 and idnumber < 55',
101 assert 4 == len(data['rows'])
103 {'identity': 'This is particle: 51'},
104 {'identity': 'This is particle: 52'},
105 {'identity': 'This is particle: 53'},
106 {'identity': 'This is particle: 54'}
108 assert ['identity'] == data['columns']
109 assert 'test_tables' == data['database']
110 assert not data['truncated']
112 def test_custom_pytables_sql(app_client):
113 response = app_client.get(
114 '/test_tables.json?' + urlencode({
115 'sql': 'select identity from [/group1/table1] where (speed > 100) & (speed < 500)',
122 'sql': 'select identity from [/group1/table1] where (speed > 100) & (speed < 500)',
125 assert 199 == len(data['rows'])
127 {'identity': 'This is particle: 51'},
128 {'identity': 'This is particle: 52'},
129 {'identity': 'This is particle: 53'}
130 ] == data['rows'][:3]
131 assert ['identity'] == data['columns']
132 assert 'test_tables' == data['database']
133 assert not data['truncated']
135 def test_invalid_custom_sql(app_client):
136 response = app_client.get(
137 '/test_tables.json?sql=.schema',
140 assert response.status == 400
141 assert response.json['ok'] is False
142 assert 'Statement must be a SELECT' == response.json['error']
144 def test_table_json(app_client):
145 response = app_client.get(
146 '/test_tables/%2Fgroup2%2Ftable2.json?_shape=objects',
149 assert response.status == 200
151 assert data['query']['sql'] == 'select rowid, * from [/group2/table2] order by rowid limit 51'
152 assert data['rows'][3:6] == [{
154 'identity': 'This is particle: 3',
159 'identity': 'This is particle: 4',
164 'identity': 'This is particle: 5',
169 def test_table_not_exists_json(app_client):
172 'error': 'Table not found: blah',
176 '/test_tables/blah.json', gather_request=False
179 def test_table_shape_arrays(app_client):
180 response = app_client.get(
181 '/test_tables/%2Fgroup2%2Ftable2.json?_shape=arrays',
185 [6, 'This is particle: 6', 6, 12.0],
186 [7, 'This is particle: 7', 7, 14.0],
187 ] == response.json['rows'][6:8]
189 def test_table_shape_objects(app_client):
190 response = app_client.get(
191 '/test_tables/%2Fgroup2%2Ftable2.json?_shape=objects',
196 'identity': 'This is particle: 6',
201 'identity': 'This is particle: 7',
204 }] == response.json['rows'][6:8]
206 def test_table_shape_array(app_client):
207 response = app_client.get(
208 '/test_tables/%2Fgroup2%2Ftable2.json?_shape=array',
213 'identity': 'This is particle: 6',
218 'identity': 'This is particle: 7',
221 }] == response.json[6:8]
223 def test_table_shape_invalid(app_client):
224 response = app_client.get(
225 '/test_tables/%2Fgroup2%2Ftable2.json?_shape=invalid',
230 'error': 'Invalid _shape: invalid',
235 @pytest.mark.parametrize('path, expected_rows, expected_pages', [
236 ('/test_tables/%2Farray1.json', 2, 1),
237 ('/test_tables/%2Farray1.json?_size=1', 2, 2),
238 ('/test_tables/%2Fgroup1%2Farray2.json?_size=1000', 10000, 10),
239 ('/test_tables/%2Fgroup2%2Fmulti.json?_size=5', 10, 2),
241 def test_paginate_tables_and_arrays(app_client, path, expected_rows, expected_pages):
245 response = app_client.get(path, gather_request=False)
246 print("*****", response.json)
247 assert 200 == response.status
249 fetched.extend(response.json['rows'])
250 path = response.json['next_url']
252 assert response.json['next']
253 assert '_next={}'.format(response.json['next']) in path
255 assert expected_rows == len(fetched)
256 assert expected_pages == count