database page
[datasette-pytables.git] / tests / test_api.py
1 from .fixtures import app_client
2 import pytest
3 from urllib.parse import urlencode
4
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
12
13 def test_database_page(app_client):
14     response = app_client.get('/test_tables.json')
15     data = response.json
16     assert 'test_tables' == data['database']
17     assert [{
18         'name': '/array1',
19         'columns': ['value'],
20         'primary_keys': [],
21         'count': 2,
22         'hidden': False,
23         'fts_table': None,
24         'foreign_keys': {'incoming': [], 'outgoing': []},
25         'private': False,
26     }, {
27         'name': '/group1/array2',
28         'columns': ['value'],
29         'primary_keys': [],
30         'count': 10000,
31         'hidden': False,
32         'fts_table': None,
33         'foreign_keys': {'incoming': [], 'outgoing': []},
34         'private': False,
35     }, {
36         'name': '/group1/table1',
37         'columns': ['identity', 'idnumber', 'speed'],
38         'primary_keys': [],
39         'count': 10000,
40         'hidden': False,
41         'fts_table': None,
42         'foreign_keys': {'incoming': [], 'outgoing': []},
43         'private': False,
44     }, {
45         'name': '/group2/multi',
46         'columns': ['value'],
47         'primary_keys': [],
48         'count': 10,
49         'hidden': False,
50         'fts_table': None,
51         'foreign_keys': {'incoming': [], 'outgoing': []},
52         'private': False,
53     }, {
54         'name': '/group2/table2',
55         'columns': ['identity', 'idnumber', 'speed'],
56         'primary_keys': [],
57         'count': 10000,
58         'hidden': False,
59         'fts_table': None,
60         'foreign_keys': {'incoming': [], 'outgoing': []},
61         'private': False,
62     }] == data['tables']
63
64 def test_custom_sql(app_client):
65     response = app_client.get(
66         '/test_tables.json?' + urlencode({
67             'sql': 'select identity from [/group1/table1]',
68             '_shape': 'objects'
69         }),
70         gather_request=False
71     )
72     data = response.json
73     assert {
74         'sql': 'select identity from [/group1/table1]',
75         'params': {}
76     } == data['query']
77     assert 1000 == len(data['rows'])
78     assert [
79         {'identity': 'This is particle:  0'},
80         {'identity': 'This is particle:  1'},
81         {'identity': 'This is particle:  2'},
82         {'identity': 'This is particle:  3'}
83     ] == data['rows'][:4]
84     assert ['identity'] == data['columns']
85     assert 'test_tables' == data['database']
86     assert data['truncated']
87
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',
92             '_shape': 'objects'
93         }),
94         gather_request=False
95     )
96     data = response.json
97     assert {
98         'sql': 'select identity from [/group1/table1] where speed > 100 and idnumber < 55',
99         'params': {}
100     } == data['query']
101     assert 4 == len(data['rows'])
102     assert [
103         {'identity': 'This is particle: 51'},
104         {'identity': 'This is particle: 52'},
105         {'identity': 'This is particle: 53'},
106         {'identity': 'This is particle: 54'}
107     ] == data['rows']
108     assert ['identity'] == data['columns']
109     assert 'test_tables' == data['database']
110     assert not data['truncated']
111
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)',
116             '_shape': 'objects'
117             }),
118         gather_request=False
119     )
120     data = response.json
121     assert {
122         'sql': 'select identity from [/group1/table1] where (speed > 100) & (speed < 500)',
123         'params': {}
124     } == data['query']
125     assert 199 == len(data['rows'])
126     assert [
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']
134
135 def test_invalid_custom_sql(app_client):
136     response = app_client.get(
137         '/test_tables.json?sql=.schema',
138         gather_request=False
139     )
140     assert response.status == 400
141     assert response.json['ok'] is False
142     assert 'Statement must be a SELECT' == response.json['error']
143
144 def test_table_json(app_client):
145     response = app_client.get(
146         '/test_tables/%2Fgroup2%2Ftable2.json?_shape=objects',
147         gather_request=False
148     )
149     assert response.status == 200
150     data = response.json
151     assert data['query']['sql'] == 'select rowid, * from [/group2/table2] order by rowid limit 51'
152     assert data['rows'][3:6] == [{
153         'rowid': 3,
154         'identity': 'This is particle:  3',
155         'idnumber': 3,
156         'speed': 6.0
157     }, {
158         'rowid': 4,
159         'identity': 'This is particle:  4',
160         'idnumber': 4,
161         'speed': 8.0
162     }, {
163         'rowid': 5,
164         'identity': 'This is particle:  5',
165         'idnumber': 5,
166         'speed': 10.0
167     }]
168
169 def test_table_not_exists_json(app_client):
170     assert {
171         'ok': False,
172         'error': 'Table not found: blah',
173         'status': 404,
174         'title': None,
175     } == app_client.get(
176         '/test_tables/blah.json', gather_request=False
177     ).json
178
179 def test_table_shape_arrays(app_client):
180     response = app_client.get(
181         '/test_tables/%2Fgroup2%2Ftable2.json?_shape=arrays',
182         gather_request=False
183     )
184     assert [
185         [6, 'This is particle:  6', 6, 12.0],
186         [7, 'This is particle:  7', 7, 14.0],
187     ] == response.json['rows'][6:8]
188
189 def test_table_shape_objects(app_client):
190     response = app_client.get(
191         '/test_tables/%2Fgroup2%2Ftable2.json?_shape=objects',
192         gather_request=False
193     )
194     assert [{
195         'rowid': 6,
196         'identity': 'This is particle:  6',
197         'idnumber': 6,
198         'speed': 12.0,
199     }, {
200         'rowid': 7,
201         'identity': 'This is particle:  7',
202         'idnumber': 7,
203         'speed': 14.0,
204     }] == response.json['rows'][6:8]
205
206 def test_table_shape_array(app_client):
207     response = app_client.get(
208         '/test_tables/%2Fgroup2%2Ftable2.json?_shape=array',
209         gather_request=False
210     )
211     assert [{
212         'rowid': 6,
213         'identity': 'This is particle:  6',
214         'idnumber': 6,
215         'speed': 12.0,
216     }, {
217         'rowid': 7,
218         'identity': 'This is particle:  7',
219         'idnumber': 7,
220         'speed': 14.0,
221     }] == response.json[6:8]
222
223 def test_table_shape_invalid(app_client):
224     response = app_client.get(
225         '/test_tables/%2Fgroup2%2Ftable2.json?_shape=invalid',
226         gather_request=False
227     )
228     assert {
229         'ok': False,
230         'error': 'Invalid _shape: invalid',
231         'status': 400,
232         'title': None,
233     } == response.json
234
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),
240 ])
241 def test_paginate_tables_and_arrays(app_client, path, expected_rows, expected_pages):
242     fetched = []
243     count = 0
244     while path:
245         response = app_client.get(path, gather_request=False)
246         print("*****", response.json)
247         assert 200 == response.status
248         count += 1
249         fetched.extend(response.json['rows'])
250         path = response.json['next_url']
251         if path:
252             assert response.json['next']
253             assert '_next={}'.format(response.json['next']) in path
254
255     assert expected_rows == len(fetched)
256     assert expected_pages == count