Paginate with offset argument
[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 identity, idnumber, speed from [/group2/table2] limit 51'
147     assert data['rows'][3:6] == [{
148         'identity': 'This is particle:  3',
149         'idnumber': 3,
150         'speed': 6.0
151     }, {
152         'identity': 'This is particle:  4',
153         'idnumber': 4,
154         'speed': 8.0
155     }, {
156         'identity': 'This is particle:  5',
157         'idnumber': 5,
158         'speed': 10.0
159     }]
160
161 def test_table_not_exists_json(app_client):
162     assert {
163         'ok': False,
164         'error': 'Table not found: blah',
165         'status': 404,
166         'title': None,
167     } == app_client.get(
168         '/test_tables/blah.json',
169     ).json
170
171 def test_table_shape_arrays(app_client):
172     response = app_client.get(
173         '/test_tables/%2Fgroup2%2Ftable2.json?_shape=arrays',
174     )
175     assert [
176         ['This is particle:  6', 6, 12.0],
177         ['This is particle:  7', 7, 14.0],
178     ] == response.json['rows'][6:8]
179
180 def test_table_shape_objects(app_client):
181     response = app_client.get(
182         '/test_tables/%2Fgroup2%2Ftable2.json?_shape=objects',
183     )
184     assert [{
185         'identity': 'This is particle:  6',
186         'idnumber': 6,
187         'speed': 12.0,
188     }, {
189         'identity': 'This is particle:  7',
190         'idnumber': 7,
191         'speed': 14.0,
192     }] == response.json['rows'][6:8]
193
194 def test_table_shape_array(app_client):
195     response = app_client.get(
196         '/test_tables/%2Fgroup2%2Ftable2.json?_shape=array',
197     )
198     assert [{
199         'identity': 'This is particle:  6',
200         'idnumber': 6,
201         'speed': 12.0,
202     }, {
203         'identity': 'This is particle:  7',
204         'idnumber': 7,
205         'speed': 14.0,
206     }] == response.json[6:8]
207
208 def test_table_shape_invalid(app_client):
209     response = app_client.get(
210         '/test_tables/%2Fgroup2%2Ftable2.json?_shape=invalid',
211     )
212     assert {
213         'ok': False,
214         'error': 'Invalid _shape: invalid',
215         'status': 400,
216         'title': None,
217     } == response.json
218
219 @pytest.mark.parametrize('path, expected_rows, expected_pages', [
220     ('/test_tables/%2Farray1.json', 2, 1),
221     ('/test_tables/%2Farray1.json?_size=1', 2, 2),
222     ('/test_tables/%2Fgroup1%2Farray2.json?_size=1000', 10000, 10),
223     ('/test_tables/%2Fgroup2%2Fmulti.json?_size=5', 10, 2),
224 ])
225 def test_paginate_tables_and_arrays(app_client, path, expected_rows, expected_pages):
226     fetched = []
227     count = 0
228     while path:
229         response = app_client.get(path)
230         assert 200 == response.status
231         count += 1
232         fetched.extend(response.json['rows'])
233         path = response.json['next_url']
234         if path:
235             path = path.replace('http://localhost', '')
236             assert response.json['next']
237             assert '_next={}'.format(response.json['next']) in path
238
239     assert expected_rows == len(fetched)
240     assert expected_pages == count