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