custom pytables sql
[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     )
71     data = response.json
72     assert {
73         'sql': 'select identity from [/group1/table1]',
74         'params': {}
75     } == data['query']
76     assert 1000 == len(data['rows'])
77     assert [
78         {'identity': 'This is particle:  0'},
79         {'identity': 'This is particle:  1'},
80         {'identity': 'This is particle:  2'},
81         {'identity': 'This is particle:  3'}
82     ] == data['rows'][:4]
83     assert ['identity'] == data['columns']
84     assert 'test_tables' == data['database']
85     assert data['truncated']
86
87 def test_custom_complex_sql(app_client):
88     response = app_client.get(
89         '/test_tables.json?' + urlencode({
90             'sql': 'select identity from [/group1/table1] where speed > 100 and idnumber < 55',
91             '_shape': 'objects',
92         }),
93     )
94     data = response.json
95     assert {
96         'sql': 'select identity from [/group1/table1] where speed > 100 and idnumber < 55',
97         'params': {},
98     } == data['query']
99     assert 4 == len(data['rows'])
100     assert [
101         {'identity': 'This is particle: 51'},
102         {'identity': 'This is particle: 52'},
103         {'identity': 'This is particle: 53'},
104         {'identity': 'This is particle: 54'},
105     ] == data['rows']
106     assert ['identity'] == data['columns']
107     assert 'test_tables' == data['database']
108     assert not data['truncated']
109
110 def test_custom_pytables_sql(app_client):
111     response = app_client.get(
112         '/test_tables.json?' + urlencode({
113             'sql': 'select identity from [/group1/table1] where (speed > 100) & (speed < 500)',
114             '_shape': 'objects',
115         }),
116     )
117     data = response.json
118     assert {
119         'sql': 'select identity from [/group1/table1] where (speed > 100) & (speed < 500)',
120         'params': {},
121     } == data['query']
122     assert 199 == len(data['rows'])
123     assert [
124         {'identity': 'This is particle: 51'},
125         {'identity': 'This is particle: 52'},
126         {'identity': 'This is particle: 53'},
127     ] == data['rows'][:3]
128     assert ['identity'] == data['columns']
129     assert 'test_tables' == data['database']
130     assert not data['truncated']
131
132 def test_invalid_custom_sql(app_client):
133     response = app_client.get(
134         '/test_tables.json?sql=.schema',
135     )
136     assert response.status == 400
137     assert response.json['ok'] is False
138     assert 'Statement must be a SELECT' == response.json['error']
139
140 def test_table_json(app_client):
141     response = app_client.get(
142         '/test_tables/%2Fgroup2%2Ftable2.json?_shape=objects',
143     )
144     assert response.status == 200
145     data = response.json
146     assert data['query']['sql'] == 'select rowid, * from [/group2/table2] order by rowid limit 51'
147     assert data['rows'][3:6] == [{
148         'rowid': 3,
149         'identity': 'This is particle:  3',
150         'idnumber': 3,
151         'speed': 6.0
152     }, {
153         'rowid': 4,
154         'identity': 'This is particle:  4',
155         'idnumber': 4,
156         'speed': 8.0
157     }, {
158         'rowid': 5,
159         'identity': 'This is particle:  5',
160         'idnumber': 5,
161         'speed': 10.0
162     }]
163
164 def test_table_not_exists_json(app_client):
165     assert {
166         'ok': False,
167         'error': 'Table not found: blah',
168         'status': 404,
169         'title': None,
170     } == app_client.get(
171         '/test_tables/blah.json',
172     ).json
173
174 def test_table_shape_arrays(app_client):
175     response = app_client.get(
176         '/test_tables/%2Fgroup2%2Ftable2.json?_shape=arrays',
177     )
178     assert [
179         [6, 'This is particle:  6', 6, 12.0],
180         [7, 'This is particle:  7', 7, 14.0],
181     ] == response.json['rows'][6:8]
182
183 def test_table_shape_objects(app_client):
184     response = app_client.get(
185         '/test_tables/%2Fgroup2%2Ftable2.json?_shape=objects',
186     )
187     assert [{
188         'rowid': 6,
189         'identity': 'This is particle:  6',
190         'idnumber': 6,
191         'speed': 12.0,
192     }, {
193         'rowid': 7,
194         'identity': 'This is particle:  7',
195         'idnumber': 7,
196         'speed': 14.0,
197     }] == response.json['rows'][6:8]
198
199 def test_table_shape_array(app_client):
200     response = app_client.get(
201         '/test_tables/%2Fgroup2%2Ftable2.json?_shape=array',
202     )
203     assert [{
204         'rowid': 6,
205         'identity': 'This is particle:  6',
206         'idnumber': 6,
207         'speed': 12.0,
208     }, {
209         'rowid': 7,
210         'identity': 'This is particle:  7',
211         'idnumber': 7,
212         'speed': 14.0,
213     }] == response.json[6:8]
214
215 def test_table_shape_invalid(app_client):
216     response = app_client.get(
217         '/test_tables/%2Fgroup2%2Ftable2.json?_shape=invalid',
218     )
219     assert {
220         'ok': False,
221         'error': 'Invalid _shape: invalid',
222         'status': 400,
223         'title': None,
224     } == response.json
225
226 @pytest.mark.parametrize('path, expected_rows, expected_pages', [
227     ('/test_tables/%2Farray1.json', 2, 1),
228     ('/test_tables/%2Farray1.json?_size=1', 2, 2),
229     ('/test_tables/%2Fgroup1%2Farray2.json?_size=1000', 10000, 10),
230     ('/test_tables/%2Fgroup2%2Fmulti.json?_size=5', 10, 2),
231 ])
232 def test_paginate_tables_and_arrays(app_client, path, expected_rows, expected_pages):
233     fetched = []
234     count = 0
235     while path:
236         response = app_client.get(path)
237         print("*****", response.json)
238         assert 200 == response.status
239         count += 1
240         fetched.extend(response.json['rows'])
241         path = response.json['next_url']
242         if path:
243             assert response.json['next']
244             assert '_next={}'.format(response.json['next']) in path
245
246     assert expected_rows == len(fetched)
247     assert expected_pages == count