]> git.jsancho.org Git - datasette-connectors.git/blob - tests/test_api.py
Some tests
[datasette-connectors.git] / tests / test_api.py
1 from .fixtures import app_client
2 from urllib.parse import urlencode
3
4 def test_homepage(app_client):
5     _, response = app_client.get('/.json')
6     assert response.status == 200
7     assert response.json.keys() == {'dummy_tables': 0}.keys()
8     d = response.json['dummy_tables']
9     assert d['name'] == 'dummy_tables'
10     assert d['tables_count'] == 2
11
12 def test_database_page(app_client):
13     response = app_client.get('/dummy_tables.json', gather_request=False)
14     data = response.json
15     assert 'dummy_tables' == data['database']
16     assert [{
17         'name': 'table1',
18         'columns': ['c1', 'c2', 'c3'],
19         'primary_keys': [],
20         'count': 2,
21         'label_column': None,
22         'hidden': False,
23         'fts_table': None,
24         'foreign_keys': {'incoming': [], 'outgoing': []}
25     }, {
26         'name': 'table2',
27         'columns': ['c1', 'c2', 'c3'],
28         'primary_keys': [],
29         'count': 2,
30         'label_column': None,
31         'hidden': False,
32         'fts_table': None,
33         'foreign_keys': {'incoming': [], 'outgoing': []}
34     }] == data['tables']
35
36 def test_custom_sql(app_client):
37     response = app_client.get(
38         '/dummy_tables.json?' + urlencode({
39             'sql': 'select c1 from table1',
40             '_shape': 'objects'
41         }),
42         gather_request=False
43     )
44     data = response.json
45     assert {
46         'sql': 'select c1 from table1',
47         'params': {}
48     } == data['query']
49     assert 2 == len(data['rows'])
50     assert [
51         {'c1': 10},
52         {'c1': 20}
53     ] == data['rows']
54     assert ['c1'] == data['columns']
55     assert 'dummy_tables' == data['database']
56     assert not data['truncated']
57
58 def test_invalid_custom_sql(app_client):
59     response = app_client.get(
60         '/dummy_tables.json?sql=.schema',
61         gather_request=False
62     )
63     assert response.status == 400
64     assert response.json['ok'] is False
65     assert 'Statement must be a SELECT' == response.json['error']
66
67 def test_table_json(app_client):
68     response = app_client.get(
69         '/dummy_tables/table2.json?_shape=objects',
70         gather_request=False
71     )
72     assert response.status == 200
73     data = response.json
74     assert data['query']['sql'] == 'select rowid, * from table2 order by rowid limit 51'
75     assert data['rows'] == [{
76         'rowid': 1,
77         'c1': 100,
78         'c2': 120,
79         'c3': 130
80     }, {
81         'rowid': 2,
82         'c1': 200,
83         'c2': 220,
84         'c3': 230
85     }]
86
87 def test_table_not_exists_json(app_client):
88     assert {
89         'ok': False,
90         'error': 'Table not found: blah',
91         'status': 404,
92         'title': None,
93     } == app_client.get(
94         '/dummy_tables/blah.json', gather_request=False
95     ).json
96
97 def test_table_shape_arrays(app_client):
98     response = app_client.get(
99         '/dummy_tables/table2.json?_shape=arrays',
100         gather_request=False
101     )
102     assert [
103         [1, 100, 120, 130],
104         [2, 200, 220, 230],
105     ] == response.json['rows']
106
107 def test_table_shape_objects(app_client):
108     response = app_client.get(
109         '/dummy_tables/table2.json?_shape=objects',
110         gather_request=False
111     )
112     assert [{
113         'rowid': 1,
114         'c1': 100,
115         'c2': 120,
116         'c3': 130,
117     }, {
118         'rowid': 2,
119         'c1': 200,
120         'c2': 220,
121         'c3': 230,
122     }] == response.json['rows']
123
124 def test_table_shape_array(app_client):
125     response = app_client.get(
126         '/dummy_tables/table2.json?_shape=array',
127         gather_request=False
128     )
129     assert [{
130         'rowid': 1,
131         'c1': 100,
132         'c2': 120,
133         'c3': 130,
134     }, {
135         'rowid': 2,
136         'c1': 200,
137         'c2': 220,
138         'c3': 230,
139     }] == response.json
140
141 def test_table_shape_invalid(app_client):
142     response = app_client.get(
143         '/dummy_tables/table2.json?_shape=invalid',
144         gather_request=False
145     )
146     assert {
147         'ok': False,
148         'error': 'Invalid _shape: invalid',
149         'status': 400,
150         'title': None,
151     } == response.json