1 from .fixtures import app_client
4 pytest.fixture(scope='module')(app_client)
6 def test_homepage(app_client):
7 _, response = app_client.get('/.json')
8 assert response.status == 200
9 assert response.json.keys() == {'test_tables': 0}.keys()
10 d = response.json['test_tables']
11 assert d['name'] == 'test_tables'
12 assert d['tables_count'] == 4
14 def test_database_page(app_client):
15 response = app_client.get('/test_tables.json', gather_request=False)
17 assert 'test_tables' == data['database']
26 'foreign_keys': {'incoming': [], 'outgoing': []}
28 'name': '/group1/array2',
35 'foreign_keys': {'incoming': [], 'outgoing': []}
37 'name': '/group1/table1',
38 'columns': ['identity', 'idnumber', 'speed'],
44 'foreign_keys': {'incoming': [], 'outgoing': []}
46 'name': '/group2/table2',
47 'columns': ['identity', 'idnumber', 'speed'],
53 'foreign_keys': {'incoming': [], 'outgoing': []}
56 def test_custom_sql(app_client):
57 response = app_client.get(
58 '/test_tables.json?sql=select+identity+from+[/group1/table1]&_shape=objects',
63 'sql': 'select identity from [/group1/table1]',
66 assert 1000 == len(data['rows'])
68 {'identity': 'This is particle: 0'},
69 {'identity': 'This is particle: 1'},
70 {'identity': 'This is particle: 2'}
72 assert ['identity'] == data['columns']
73 assert 'test_tables' == data['database']
74 assert data['truncated']
76 def test_invalid_custom_sql(app_client):
77 response = app_client.get(
78 '/test_tables.json?sql=.schema',
81 assert response.status == 400
82 assert response.json['ok'] is False
83 assert 'Statement must be a SELECT' == response.json['error']
85 def test_table_json(app_client):
86 response = app_client.get(
87 '/test_tables/%2Fgroup2%2Ftable2.json?_shape=objects',
90 assert response.status == 200
92 assert data['query']['sql'] == 'select rowid, * from [/group2/table2] order by rowid limit 51'
93 assert data['rows'][3:6] == [{
95 'identity': 'This is particle: 3',
100 'identity': 'This is particle: 4',
105 'identity': 'This is particle: 5',
110 def test_table_not_exists_json(app_client):
113 'error': 'Table not found: blah',
117 '/test_tables/blah.json', gather_request=False
120 def test_table_shape_arrays(app_client):
121 response = app_client.get(
122 '/test_tables/%2Fgroup2%2Ftable2.json?_shape=arrays',
126 [6, 'This is particle: 6', 6, 12.0],
127 [7, 'This is particle: 7', 7, 14.0],
128 ] == response.json['rows'][6:8]
130 def test_table_shape_objects(app_client):
131 response = app_client.get(
132 '/test_tables/%2Fgroup2%2Ftable2.json?_shape=objects',
137 'identity': 'This is particle: 6',
142 'identity': 'This is particle: 7',
145 }] == response.json['rows'][6:8]
147 def test_table_shape_array(app_client):
148 response = app_client.get(
149 '/test_tables/%2Fgroup2%2Ftable2.json?_shape=array',
154 'identity': 'This is particle: 6',
159 'identity': 'This is particle: 7',
162 }] == response.json[6:8]
164 def test_table_shape_invalid(app_client):
165 response = app_client.get(
166 '/test_tables/%2Fgroup2%2Ftable2.json?_shape=invalid',
171 'error': 'Invalid _shape: invalid',
176 @pytest.mark.parametrize('path, expected_rows, expected_pages', [
177 ('/test_tables/%2Farray1.json', 2, 1),
178 ('/test_tables/%2Farray1.json?_size=1', 2, 2),
179 ('/test_tables/%2Fgroup1%2Farray2.json?_size=1000', 10000, 10),
181 def test_paginate_tables_and_arrays(app_client, path, expected_rows, expected_pages):
185 response = app_client.get(path, gather_request=False)
186 print("*****", response.json)
187 assert 200 == response.status
189 fetched.extend(response.json['rows'])
190 path = response.json['next_url']
192 assert response.json['next']
193 assert '_next={}'.format(response.json['next']) in path
195 assert expected_rows == len(fetched)
196 assert expected_pages == count