1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
|
# fmt: off
import pytest
import subprocess
import sys
from typing import List
from conftest import ShellTest
from conftest import autocomplete_extension
import os
# 'autocomplete_extension' is a fixture which will skip the test if 'autocomplete' is not loaded
def test_autocomplete_select(shell, autocomplete_extension):
test = (
ShellTest(shell)
.statement("CALL sql_auto_complete('SEL')")
)
result = test.run()
result.check_stdout('SELECT')
def test_autocomplete_first_from(shell, autocomplete_extension):
test = (
ShellTest(shell)
.statement("CALL sql_auto_complete('FRO')")
)
result = test.run()
result.check_stdout('FROM')
def test_autocomplete_column(shell, autocomplete_extension):
test = (
ShellTest(shell)
.statement("CREATE TABLE my_table(my_column INTEGER);")
.statement("SELECT * FROM sql_auto_complete('SELECT my_') LIMIT 1;")
)
result = test.run()
result.check_stdout('my_column')
def test_autocomplete_where(shell, autocomplete_extension):
test = (
ShellTest(shell)
.statement("CREATE TABLE my_table(my_column INTEGER);")
.statement("SELECT * FROM sql_auto_complete('SELECT my_column FROM my_table WH') LIMIT 1;")
)
result = test.run()
result.check_stdout('WHERE')
def test_autocomplete_insert(shell, autocomplete_extension):
test = (
ShellTest(shell)
.statement("CREATE TABLE my_table(my_column INTEGER);")
.statement("SELECT * FROM sql_auto_complete('INS') LIMIT 1;")
)
result = test.run()
result.check_stdout('INSERT')
def test_autocomplete_into(shell, autocomplete_extension):
test = (
ShellTest(shell)
.statement("CREATE TABLE my_table(my_column INTEGER);")
.statement("SELECT * FROM sql_auto_complete('INSERT IN') LIMIT 1;")
)
result = test.run()
result.check_stdout('INTO')
def test_autocomplete_into_table(shell, autocomplete_extension):
test = (
ShellTest(shell)
.statement("CREATE TABLE my_table(my_column INTEGER);")
.statement("SELECT * FROM sql_auto_complete('INSERT INTO my_t') LIMIT 1;")
)
result = test.run()
result.check_stdout('my_table')
def test_autocomplete_values(shell, autocomplete_extension):
test = (
ShellTest(shell)
.statement("CREATE TABLE my_table(my_column INTEGER);")
.statement("SELECT * FROM sql_auto_complete('INSERT INTO my_table VAL') LIMIT 1;")
)
result = test.run()
result.check_stdout('VALUES')
def test_autocomplete_delete(shell, autocomplete_extension):
test = (
ShellTest(shell)
.statement("CREATE TABLE my_table(my_column INTEGER);")
.statement("SELECT * FROM sql_auto_complete('DEL') LIMIT 1;")
)
result = test.run()
result.check_stdout('DELETE')
def test_autocomplete_delete_from(shell, autocomplete_extension):
test = (
ShellTest(shell)
.statement("CREATE TABLE my_table(my_column INTEGER);")
.statement("SELECT * FROM sql_auto_complete('DELETE F') LIMIT 1;")
)
result = test.run()
result.check_stdout('FROM')
def test_autocomplete_from_table(shell, autocomplete_extension):
test = (
ShellTest(shell)
.statement("CREATE TABLE my_table(my_column INTEGER);")
.statement("SELECT * FROM sql_auto_complete('DELETE FROM m') LIMIT 1;")
)
result = test.run()
result.check_stdout('my_table')
def test_autocomplete_update(shell, autocomplete_extension):
test = (
ShellTest(shell)
.statement("CREATE TABLE my_table(my_column INTEGER);")
.statement("SELECT * FROM sql_auto_complete('UP') LIMIT 1;")
)
result = test.run()
result.check_stdout('UPDATE')
def test_autocomplete_update_table(shell, autocomplete_extension):
test = (
ShellTest(shell)
.statement("CREATE TABLE my_table(my_column INTEGER);")
.statement("SELECT * FROM sql_auto_complete('UPDATE m') LIMIT 1;")
)
result = test.run()
result.check_stdout('my_table')
test = (
ShellTest(shell)
.statement("CREATE TABLE my_table(my_column INTEGER);")
.statement("""SELECT * FROM sql_auto_complete('UPDATE "m') LIMIT 1;""")
)
result = test.run()
result.check_stdout('my_table')
def test_autocomplete_update_column(shell, autocomplete_extension):
test = (
ShellTest(shell)
.statement("CREATE TABLE my_table(my_column INTEGER);")
.statement("SELECT * FROM sql_auto_complete('UPDATE my_table SET m') LIMIT 1;")
)
result = test.run()
result.check_stdout('my_column')
def test_autocomplete_funky_table(shell, autocomplete_extension):
test = (
ShellTest(shell)
.statement("""CREATE TABLE "Funky Table With Spaces"(my_column INTEGER);""")
.statement("SELECT suggestion FROM sql_auto_complete('SELECT * FROM F') LIMIT 1;")
)
result = test.run()
result.check_stdout('"Funky Table With Spaces"')
test = (
ShellTest(shell)
.statement("""CREATE TABLE "Funky Table With Spaces"("Funky Column" int);""")
.statement("""SELECT suggestion FROM sql_auto_complete('select "Funky Column" FROM f') LIMIT 1;""")
)
result = test.run()
result.check_stdout('"Funky Table With Spaces"')
def test_autocomplete_funky_column(shell, autocomplete_extension):
test = (
ShellTest(shell)
.statement("""CREATE TABLE "Funky Table With Spaces"("Funky Column" int);""")
.statement("SELECT * FROM sql_auto_complete('select f') LIMIT 1;")
)
result = test.run()
result.check_stdout('"Funky Column"')
def test_autocomplete_semicolon(shell, autocomplete_extension):
test = (
ShellTest(shell)
.statement("SELECT * FROM sql_auto_complete('SELECT 42; SEL') LIMIT 1;")
)
result = test.run()
result.check_stdout('SELECT')
def test_autocomplete_comments(shell, autocomplete_extension):
test = (
ShellTest(shell)
.statement("""
SELECT * FROM sql_auto_complete('--SELECT * FROM
SEL') LIMIT 1;""")
)
result = test.run()
result.check_stdout('SELECT')
def test_autocomplete_scalar_functions(shell, autocomplete_extension):
test = (
ShellTest(shell)
.statement("SELECT * FROM sql_auto_complete('SELECT regexp_m') LIMIT 1;")
)
result = test.run()
result.check_stdout('regexp_matches')
def test_autocomplete_aggregates(shell, autocomplete_extension):
test = (
ShellTest(shell)
.statement("SELECT * FROM sql_auto_complete('SELECT approx_c') LIMIT 1;")
)
result = test.run()
result.check_stdout('approx_count_distinct')
def test_autocomplete_builtin_views(shell, autocomplete_extension):
test = (
ShellTest(shell)
.statement("SELECT * FROM sql_auto_complete('SELECT * FROM sqlite_ma') LIMIT 1;")
)
result = test.run()
result.check_stdout('sqlite_master')
def test_autocomplete_table_function(shell, autocomplete_extension):
test = (
ShellTest(shell)
.statement("SELECT * FROM sql_auto_complete('SELECT * FROM read_csv_a') LIMIT 1;")
)
result = test.run()
result.check_stdout('read_csv_auto')
def test_autocomplete_tpch(shell, autocomplete_extension):
test = (
ShellTest(shell)
.statement("CREATE TABLE partsupp(ps_suppkey int);")
.statement("CREATE TABLE supplier(s_suppkey int);")
.statement("CREATE TABLE nation(n_nationkey int);")
.statement("SELECT * FROM sql_auto_complete('DROP TABLE na') LIMIT 1;")
)
result = test.run()
result.check_stdout('nation')
test = (
ShellTest(shell)
.statement("CREATE TABLE partsupp(ps_suppkey int);")
.statement("CREATE TABLE supplier(s_suppkey int);")
.statement("CREATE TABLE nation(n_nationkey int);")
.statement("SELECT * FROM sql_auto_complete('SELECT s_supp') LIMIT 1;")
)
result = test.run()
result.check_stdout('s_suppkey')
test = (
ShellTest(shell)
.statement("CREATE TABLE partsupp(ps_suppkey int);")
.statement("CREATE TABLE supplier(s_suppkey int);")
.statement("CREATE TABLE nation(n_nationkey int);")
.statement("SELECT * FROM sql_auto_complete('SELECT * FROM partsupp JOIN supp') LIMIT 1;")
)
result = test.run()
result.check_stdout('supplier')
test = (
ShellTest(shell)
.statement("CREATE TABLE partsupp(ps_suppkey int);")
.statement("CREATE TABLE supplier(s_suppkey int);")
.statement("CREATE TABLE nation(n_nationkey int);")
.statement(".mode csv")
.statement("SELECT l,l FROM sql_auto_complete('SELECT * FROM partsupp JOIN supplier ON (s_supp') t(l) LIMIT 1;")
)
result = test.run()
result.check_stdout('s_suppkey,s_suppkey')
test = (
ShellTest(shell)
.statement("CREATE TABLE partsupp(ps_suppkey int);")
.statement("CREATE TABLE supplier(s_suppkey int);")
.statement("CREATE TABLE nation(n_nationkey int);")
.statement("SELECT * FROM sql_auto_complete('SELECT * FROM partsupp JOIN supplier USING (ps_su') LIMIT 1;")
)
result = test.run()
result.check_stdout('ps_suppkey')
def test_autocomplete_from(shell, autocomplete_extension):
test = (
ShellTest(shell)
.statement("SELECT * FROM sql_auto_complete('SELECT * FR') LIMIT 1;")
)
result = test.run()
result.check_stdout('FROM')
def test_autocomplete_disambiguation_column(shell, autocomplete_extension):
test = (
ShellTest(shell)
.statement("CREATE TABLE MyTable(MyColumn Varchar);")
.statement("SELECT * FROM sql_auto_complete('SELECT My') LIMIT 1;")
)
result = test.run()
result.check_stdout('MyColumn')
def test_autocomplete_disambiguation_table(shell, autocomplete_extension):
test = (
ShellTest(shell)
.statement("CREATE TABLE MyTable(MyColumn Varchar);")
.statement("SELECT * FROM sql_auto_complete('SELECT MyColumn FROM My') LIMIT 1;")
)
result = test.run()
result.check_stdout('MyTable')
def test_autocomplete_directory(shell, autocomplete_extension, tmp_path):
shell_test_dir = tmp_path / 'shell_test_dir'
extra_path = tmp_path / 'shell_test_dir' / 'extra_path'
shell_test_dir.mkdir()
extra_path.mkdir()
# Create the files
base_files = ['extra.parquet', 'extra.file']
for fname in base_files:
with open(shell_test_dir / fname, 'w+') as f:
f.write('')
# Complete the directory
partial_directory = tmp_path / 'shell_test'
test = (
ShellTest(shell)
.statement("CREATE TABLE MyTable(MyColumn Varchar);")
.statement(f"SELECT * FROM sql_auto_complete('SELECT * FROM ''{partial_directory.as_posix()}') LIMIT 1;")
)
result = test.run()
result.check_stdout("shell_test_dir")
# Complete the sub directory as well
partial_subdirectory = tmp_path / 'shell_test_dir' / 'extra'
test = (
ShellTest(shell)
.statement("CREATE TABLE MyTable(MyColumn Varchar);")
.statement(f"SELECT * FROM sql_auto_complete('SELECT * FROM ''{partial_subdirectory.as_posix()}') LIMIT 1;")
)
result = test.run()
result.check_stdout("extra_path")
# Complete the parquet file in the sub directory
partial_parquet = tmp_path / 'shell_test_dir' / 'extra.par'
test = (
ShellTest(shell)
.statement("CREATE TABLE MyTable(MyColumn Varchar);")
.statement(f"SELECT * FROM sql_auto_complete('SELECT * FROM ''{partial_parquet.as_posix()}') LIMIT 1;")
)
result = test.run()
result.check_stdout("extra.parquet")
# fmt: on
|
