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