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