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/%group2%table2.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/%group2%table2.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/%group2%table2.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/%group2%table2.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/%group2%table2.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/%array1.json', 2, 1),
 221     ('/test_tables/%array1.json?_size=1', 2, 2),
 222     ('/test_tables/%group1%array2.json?_size=1000', 10000, 10),
 223     ('/test_tables/%group2%multi.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