data = response.json
assert 'test_tables' == data['database']
assert [{
- 'name': '/array1',
+ 'name': '%array1',
'columns': ['value'],
'primary_keys': [],
'count': 2,
'foreign_keys': {'incoming': [], 'outgoing': []},
'private': False,
}, {
- 'name': '/group1/array2',
+ 'name': '%group1%array2',
'columns': ['value'],
'primary_keys': [],
'count': 10000,
'foreign_keys': {'incoming': [], 'outgoing': []},
'private': False,
}, {
- 'name': '/group1/table1',
+ 'name': '%group1%table1',
'columns': ['identity', 'idnumber', 'speed'],
'primary_keys': [],
'count': 10000,
'foreign_keys': {'incoming': [], 'outgoing': []},
'private': False,
}, {
- 'name': '/group2/multi',
+ 'name': '%group2%multi',
'columns': ['value'],
'primary_keys': [],
'count': 10,
'foreign_keys': {'incoming': [], 'outgoing': []},
'private': False,
}, {
- 'name': '/group2/table2',
+ 'name': '%group2%table2',
'columns': ['identity', 'idnumber', 'speed'],
'primary_keys': [],
'count': 10000,
def test_custom_sql(app_client):
response = app_client.get(
'/test_tables.json?' + urlencode({
- 'sql': 'select identity from [/group1/table1]',
+ 'sql': 'select identity from [%group1%table1]',
'_shape': 'objects'
}),
)
data = response.json
assert {
- 'sql': 'select identity from [/group1/table1]',
+ 'sql': 'select identity from [%group1%table1]',
'params': {}
} == data['query']
assert 1000 == len(data['rows'])
def test_custom_complex_sql(app_client):
response = app_client.get(
'/test_tables.json?' + urlencode({
- 'sql': 'select identity from [/group1/table1] where speed > 100 and idnumber < 55',
+ 'sql': 'select identity from [%group1%table1] where speed > 100 and idnumber < 55',
'_shape': 'objects',
}),
)
data = response.json
assert {
- 'sql': 'select identity from [/group1/table1] where speed > 100 and idnumber < 55',
+ 'sql': 'select identity from [%group1%table1] where speed > 100 and idnumber < 55',
'params': {},
} == data['query']
assert 4 == len(data['rows'])
def test_custom_pytables_sql(app_client):
response = app_client.get(
'/test_tables.json?' + urlencode({
- 'sql': 'select identity from [/group1/table1] where (speed > 100) & (speed < 500)',
+ 'sql': 'select identity from [%group1%table1] where (speed > 100) & (speed < 500)',
'_shape': 'objects',
}),
)
data = response.json
assert {
- 'sql': 'select identity from [/group1/table1] where (speed > 100) & (speed < 500)',
+ 'sql': 'select identity from [%group1%table1] where (speed > 100) & (speed < 500)',
'params': {},
} == data['query']
assert 199 == len(data['rows'])
def test_table_json(app_client):
response = app_client.get(
- '/test_tables/%2Fgroup2%2Ftable2.json?_shape=objects',
+ '/test_tables/%group2%table2.json?_shape=objects',
)
assert response.status == 200
data = response.json
- assert data['query']['sql'] == 'select identity, idnumber, speed from [/group2/table2] limit 51'
+ assert data['query']['sql'] == 'select identity, idnumber, speed from [%group2%table2] limit 51'
assert data['rows'][3:6] == [{
'identity': 'This is particle: 3',
'idnumber': 3,
def test_table_shape_arrays(app_client):
response = app_client.get(
- '/test_tables/%2Fgroup2%2Ftable2.json?_shape=arrays',
+ '/test_tables/%group2%table2.json?_shape=arrays',
)
assert [
['This is particle: 6', 6, 12.0],
def test_table_shape_objects(app_client):
response = app_client.get(
- '/test_tables/%2Fgroup2%2Ftable2.json?_shape=objects',
+ '/test_tables/%group2%table2.json?_shape=objects',
)
assert [{
'identity': 'This is particle: 6',
def test_table_shape_array(app_client):
response = app_client.get(
- '/test_tables/%2Fgroup2%2Ftable2.json?_shape=array',
+ '/test_tables/%group2%table2.json?_shape=array',
)
assert [{
'identity': 'This is particle: 6',
def test_table_shape_invalid(app_client):
response = app_client.get(
- '/test_tables/%2Fgroup2%2Ftable2.json?_shape=invalid',
+ '/test_tables/%group2%table2.json?_shape=invalid',
)
assert {
'ok': False,
} == response.json
@pytest.mark.parametrize('path, expected_rows, expected_pages', [
- ('/test_tables/%2Farray1.json', 2, 1),
- ('/test_tables/%2Farray1.json?_size=1', 2, 2),
- ('/test_tables/%2Fgroup1%2Farray2.json?_size=1000', 10000, 10),
- ('/test_tables/%2Fgroup2%2Fmulti.json?_size=5', 10, 2),
+ ('/test_tables/%array1.json', 2, 1),
+ ('/test_tables/%array1.json?_size=1', 2, 2),
+ ('/test_tables/%group1%array2.json?_size=1000', 10000, 10),
+ ('/test_tables/%group2%multi.json?_size=5', 10, 2),
])
def test_paginate_tables_and_arrays(app_client, path, expected_rows, expected_pages):
fetched = []
fetched.extend(response.json['rows'])
path = response.json['next_url']
if path:
+ path = path.replace('http://localhost', '')
assert response.json['next']
assert '_next={}'.format(response.json['next']) in path