-
-
Notifications
You must be signed in to change notification settings - Fork 34.3k
Expand file tree
/
Copy pathtest_cli.py
More file actions
442 lines (383 loc) · 16.9 KB
/
test_cli.py
File metadata and controls
442 lines (383 loc) · 16.9 KB
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
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
"""sqlite3 CLI tests."""
import sqlite3
import sys
import textwrap
import unittest
import unittest.mock
import os
from sqlite3.__main__ import main as cli
from test.support.import_helper import import_module
from test.support.os_helper import TESTFN, unlink
from test.support.pty_helper import run_pty
from test.support import (
captured_stdout,
captured_stderr,
captured_stdin,
force_not_colorized_test_class,
requires_subprocess,
verbose,
)
@force_not_colorized_test_class
class CommandLineInterface(unittest.TestCase):
def _do_test(self, *args, expect_success=True):
with (
captured_stdout() as out,
captured_stderr() as err,
self.assertRaises(SystemExit) as cm
):
cli(args)
return out.getvalue(), err.getvalue(), cm.exception.code
def expect_success(self, *args):
out, err, code = self._do_test(*args)
self.assertEqual(code, 0,
"\n".join([f"Unexpected failure: {args=}", out, err]))
self.assertEqual(err, "")
return out
def expect_failure(self, *args):
out, err, code = self._do_test(*args, expect_success=False)
self.assertNotEqual(code, 0,
"\n".join([f"Unexpected failure: {args=}", out, err]))
self.assertEqual(out, "")
return err
def test_cli_help(self):
out = self.expect_success("-h")
self.assertIn("usage: ", out)
self.assertIn(" [-h] [-v] [filename] [sql]", out)
self.assertIn("Python sqlite3 CLI", out)
def test_cli_version(self):
out = self.expect_success("-v")
self.assertIn(sqlite3.sqlite_version, out)
def test_cli_execute_sql(self):
out = self.expect_success(":memory:", "select 1")
self.assertIn("(1,)", out)
def test_cli_execute_too_much_sql(self):
stderr = self.expect_failure(":memory:", "select 1; select 2")
err = "ProgrammingError: You can only execute one statement at a time"
self.assertIn(err, stderr)
def test_cli_execute_incomplete_sql(self):
stderr = self.expect_failure(":memory:", "sel")
self.assertIn("OperationalError (SQLITE_ERROR)", stderr)
def test_cli_on_disk_db(self):
self.addCleanup(unlink, TESTFN)
out = self.expect_success(TESTFN, "create table t(t)")
self.assertEqual(out, "")
out = self.expect_success(TESTFN, "select count(t) from t")
self.assertIn("(0,)", out)
@force_not_colorized_test_class
class InteractiveSession(unittest.TestCase):
MEMORY_DB_MSG = "Connected to a transient in-memory database"
PS1 = "\001\002sqlite> \001\002"
PS2 = "\001\002 ... \001\002"
def run_cli(self, *args, commands=()):
with (
captured_stdin() as stdin,
captured_stdout() as stdout,
captured_stderr() as stderr,
self.assertRaises(SystemExit) as cm
):
for cmd in commands:
stdin.write(cmd + "\n")
stdin.seek(0)
cli(args)
out = stdout.getvalue()
err = stderr.getvalue()
self.assertEqual(cm.exception.code, 0,
f"Unexpected failure: {args=}\n{out}\n{err}")
return out, err
def test_interact(self):
out, err = self.run_cli()
self.assertIn(self.MEMORY_DB_MSG, err)
self.assertIn(self.MEMORY_DB_MSG, err)
self.assertEndsWith(out, self.PS1)
self.assertEqual(out.count(self.PS1), 1)
self.assertEqual(out.count(self.PS2), 0)
def test_interact_quit(self):
out, err = self.run_cli(commands=(".quit",))
self.assertIn(self.MEMORY_DB_MSG, err)
self.assertEndsWith(out, self.PS1)
self.assertEqual(out.count(self.PS1), 1)
self.assertEqual(out.count(self.PS2), 0)
def test_interact_version(self):
out, err = self.run_cli(commands=(".version",))
self.assertIn(self.MEMORY_DB_MSG, err)
self.assertIn(sqlite3.sqlite_version + "\n", out)
self.assertEndsWith(out, self.PS1)
self.assertEqual(out.count(self.PS1), 2)
self.assertEqual(out.count(self.PS2), 0)
self.assertIn(sqlite3.sqlite_version, out)
def test_interact_empty_source(self):
out, err = self.run_cli(commands=("", " "))
self.assertIn(self.MEMORY_DB_MSG, err)
self.assertEndsWith(out, self.PS1)
self.assertEqual(out.count(self.PS1), 3)
self.assertEqual(out.count(self.PS2), 0)
def test_interact_dot_commands_unknown(self):
out, err = self.run_cli(commands=(".unknown_command", ))
self.assertIn(self.MEMORY_DB_MSG, err)
self.assertEndsWith(out, self.PS1)
self.assertEqual(out.count(self.PS1), 2)
self.assertEqual(out.count(self.PS2), 0)
self.assertIn('Error: unknown command: "', err)
# test "unknown_command" is pointed out in the error message
self.assertIn("unknown_command", err)
def test_interact_dot_commands_empty(self):
out, err = self.run_cli(commands=("."))
self.assertIn(self.MEMORY_DB_MSG, err)
self.assertEndsWith(out, self.PS1)
self.assertEqual(out.count(self.PS1), 2)
self.assertEqual(out.count(self.PS2), 0)
def test_interact_dot_commands_with_whitespaces(self):
out, err = self.run_cli(commands=(".version ", ". version"))
self.assertIn(self.MEMORY_DB_MSG, err)
self.assertEqual(out.count(sqlite3.sqlite_version + "\n"), 2)
self.assertEndsWith(out, self.PS1)
self.assertEqual(out.count(self.PS1), 3)
self.assertEqual(out.count(self.PS2), 0)
def test_interact_valid_sql(self):
out, err = self.run_cli(commands=("SELECT 1;",))
self.assertIn(self.MEMORY_DB_MSG, err)
self.assertIn("(1,)\n", out)
self.assertEndsWith(out, self.PS1)
self.assertEqual(out.count(self.PS1), 2)
self.assertEqual(out.count(self.PS2), 0)
def test_interact_incomplete_multiline_sql(self):
out, err = self.run_cli(commands=("SELECT 1",))
self.assertIn(self.MEMORY_DB_MSG, err)
self.assertEndsWith(out, self.PS2)
self.assertEqual(out.count(self.PS1), 1)
self.assertEqual(out.count(self.PS2), 1)
def test_interact_valid_multiline_sql(self):
out, err = self.run_cli(commands=("SELECT 1\n;",))
self.assertIn(self.MEMORY_DB_MSG, err)
self.assertIn(self.PS2, out)
self.assertIn("(1,)\n", out)
self.assertEndsWith(out, self.PS1)
self.assertEqual(out.count(self.PS1), 2)
self.assertEqual(out.count(self.PS2), 1)
def test_interact_invalid_sql(self):
out, err = self.run_cli(commands=("sel;",))
self.assertIn(self.MEMORY_DB_MSG, err)
self.assertIn("OperationalError (SQLITE_ERROR)", err)
self.assertEndsWith(out, self.PS1)
self.assertEqual(out.count(self.PS1), 2)
self.assertEqual(out.count(self.PS2), 0)
def test_interact_on_disk_file(self):
self.addCleanup(unlink, TESTFN)
out, err = self.run_cli(TESTFN, commands=("CREATE TABLE t(t);",))
self.assertIn(TESTFN, err)
self.assertEndsWith(out, self.PS1)
out, _ = self.run_cli(TESTFN, commands=("SELECT count(t) FROM t;",))
self.assertIn("(0,)\n", out)
def test_color(self):
with unittest.mock.patch("_colorize.can_colorize", return_value=True):
out, err = self.run_cli(commands="TEXT\n")
self.assertIn("\x01\x1b[1;35m\x02sqlite> \x01\x1b[0m\x02", out)
self.assertIn("\x01\x1b[1;35m\x02 ... \x01\x1b[0m\x02\x01\x1b", out)
out, err = self.run_cli(commands=("sel;",))
self.assertIn('\x1b[1;35mOperationalError (SQLITE_ERROR)\x1b[0m: '
'\x1b[35mnear "sel": syntax error\x1b[0m', err)
@requires_subprocess()
@force_not_colorized_test_class
class Completion(unittest.TestCase):
# run_pty() creates a real terminal environment, where sqlite3 CLI
# SqliteInteractiveConsole invokes GNU Readline for input. Readline's
# _rl_strip_prompt() strips \001 and \002 from the output, so test
# assertions use the plain prompt.
PS1 = "sqlite> "
@classmethod
def setUpClass(cls):
readline = import_module("readline")
if readline.backend == "editline":
raise unittest.SkipTest("libedit readline is not supported")
def write_input(self, input_, env=None):
script = textwrap.dedent("""
import readline
from sqlite3.__main__ import main
# Configure readline to ...:
# - hide control sequences surrounding each candidate
# - hide "Display all xxx possibilities? (y or n)"
# - show candidates one per line
readline.parse_and_bind("set colored-completion-prefix off")
readline.parse_and_bind("set completion-query-items 0")
readline.parse_and_bind("set page-completions off")
readline.parse_and_bind("set completion-display-width 0")
main()
""")
return run_pty(script, input_, env)
def test_complete_sql_keywords(self):
_sqlite3 = import_module("_sqlite3")
if not hasattr(_sqlite3, "SQLITE_KEYWORDS"):
raise unittest.SkipTest("unable to determine SQLite keywords")
# List candidates starting with 'S', there should be multiple matches.
input_ = b"S\t\tEL\t 1;\n.quit\n"
output = self.write_input(input_)
self.assertIn(b"SELECT", output)
self.assertIn(b"SET", output)
self.assertIn(b"SAVEPOINT", output)
self.assertIn(b"(1,)", output)
# Keywords are completed in upper case for even lower case user input.
input_ = b"sel\t\t 1;\n.quit\n"
output = self.write_input(input_)
self.assertIn(b"SELECT", output)
self.assertIn(b"(1,)", output)
# .commands are completed without changing case
input_ = b".ver\t\n.quit\n"
output = self.write_input(input_)
self.assertIn(b".version", output)
def test_complete_table_indexes_triggers_views(self):
input_ = textwrap.dedent("""\
CREATE TABLE _Table (id);
CREATE INDEX _Index ON _table (id);
CREATE TRIGGER _Trigger BEFORE INSERT
ON _Table BEGIN SELECT 1; END;
CREATE VIEW _View AS SELECT 1;
CREATE TEMP TABLE _Temp_table (id);
CREATE INDEX temp._Temp_index ON _Temp_table (id);
CREATE TEMP TRIGGER _Temp_trigger BEFORE INSERT
ON _Table BEGIN SELECT 1; END;
CREATE TEMP VIEW _Temp_view AS SELECT 1;
ATTACH ':memory:' AS attached;
CREATE TABLE attached._Attached_table (id);
CREATE INDEX attached._Attached_index ON _Attached_table (id);
CREATE TRIGGER attached._Attached_trigger BEFORE INSERT
ON _Attached_table BEGIN SELECT 1; END;
CREATE VIEW attached._Attached_view AS SELECT 1;
SELECT id FROM _\t\tta\t;
.quit\n""").encode()
output = self.write_input(input_)
lines = output.decode().splitlines()
indices = [i for i, line in enumerate(lines)
if line.startswith(self.PS1)]
start, end = indices[-3], indices[-2]
candidates = [l.strip() for l in lines[start+1:end]]
self.assertEqual(candidates,
[
"_Attached_index",
"_Attached_table",
"_Attached_trigger",
"_Attached_view",
"_Index",
"_Table",
"_Temp_index",
"_Temp_table",
"_Temp_trigger",
"_Temp_view",
"_Trigger",
"_View",
],
)
start, end = indices[-2], indices[-1]
# direct match with '_Table' completed, no candidates displayed
candidates = [l.strip() for l in lines[start+1:end]]
self.assertEqual(len(candidates), 0)
@unittest.skipIf(sqlite3.sqlite_version_info < (3, 16, 0),
"PRAGMA table-valued function is not available until "
"SQLite 3.16.0")
def test_complete_columns(self):
input_ = textwrap.dedent("""\
CREATE TABLE _table (_col_table);
CREATE TEMP TABLE _temp_table (_col_temp);
ATTACH ':memory:' AS attached;
CREATE TABLE attached._attached_table (_col_attached);
SELECT _col_\t\tta\tFROM _table;
.quit\n""").encode()
output = self.write_input(input_)
lines = output.decode().splitlines()
indices = [
i for i, line in enumerate(lines) if line.startswith(self.PS1)
]
start, end = indices[-3], indices[-2]
candidates = [l.strip() for l in lines[start+1:end]]
self.assertEqual(
candidates, ["_col_attached", "_col_table", "_col_temp"]
)
@unittest.skipIf(sqlite3.sqlite_version_info < (3, 30, 0),
"PRAGMA function_list is not available until "
"SQLite 3.30.0")
def test_complete_functions(self):
input_ = b"SELECT AV\t1);\n.quit\n"
output = self.write_input(input_)
self.assertIn(b"AVG(1);", output)
self.assertIn(b"(1.0,)", output)
# Functions are completed in upper case for even lower case user input.
input_ = b"SELECT av\t1);\n.quit\n"
output = self.write_input(input_)
self.assertIn(b"AVG(1);", output)
self.assertIn(b"(1.0,)", output)
def test_complete_schemata(self):
input_ = textwrap.dedent("""\
ATTACH ':memory:' AS MixedCase;
-- Test '_' is escaped in Like pattern filtering
ATTACH ':memory:' AS _underscore;
-- Let database_list pragma have a 'temp' schema entry
CREATE TEMP TABLE _table (id);
SELECT * FROM \t\tmIX\t.sqlite_master;
SELECT * FROM _und\t.sqlite_master;
.quit\n""").encode()
output = self.write_input(input_)
lines = output.decode().splitlines()
indices = [
i for i, line in enumerate(lines) if line.startswith(self.PS1)
]
start, end = indices[-4], indices[-3]
candidates = [l.strip() for l in lines[start+1:end]]
self.assertIn("MixedCase", candidates)
self.assertIn("_underscore", candidates)
self.assertIn("main", candidates)
self.assertIn("temp", candidates)
@unittest.skipIf(sys.platform.startswith("freebsd"),
"Two actual tabs are inserted when there are no matching"
" completions in the pseudo-terminal opened by run_pty()"
" on FreeBSD")
def test_complete_no_match(self):
input_ = b"xyzzy\t\t\b\b\b\b\b\b\b.quit\n"
# Set NO_COLOR to disable coloring for self.PS1.
output = self.write_input(input_, env={**os.environ, "NO_COLOR": "1"})
lines = output.decode().splitlines()
indices = (
i for i, line in enumerate(lines, 1)
if line.startswith(f"{self.PS1}xyzzy")
)
line_num = next(indices, -1)
self.assertNotEqual(line_num, -1)
# Completions occupy lines, assert no extra lines when there is nothing
# to complete.
self.assertEqual(line_num, len(lines))
def test_complete_no_input(self):
script = textwrap.dedent("""
import readline
from sqlite3.__main__ import main
# Configure readline to ...:
# - hide control sequences surrounding each candidate
# - hide "Display all xxx possibilities? (y or n)"
# - hide "--More--"
# - show candidates one per line
readline.parse_and_bind("set colored-completion-prefix off")
readline.parse_and_bind("set colored-stats off")
readline.parse_and_bind("set completion-query-items 0")
readline.parse_and_bind("set page-completions off")
readline.parse_and_bind("set completion-display-width 0")
readline.parse_and_bind("set show-all-if-ambiguous off")
readline.parse_and_bind("set show-all-if-unmodified off")
main()
""")
input_ = b"\t\t.quit\n"
output = run_pty(script, input_, env={**os.environ, "NO_COLOR": "1"})
try:
lines = output.decode().splitlines()
indices = [
i for i, line in enumerate(lines)
if line.startswith(self.PS1)
]
self.assertEqual(len(indices), 2)
start, end = indices
candidates = [l.strip() for l in lines[start+1:end]]
self.assertEqual(candidates, sorted(candidates))
except:
if verbose:
print(' PTY output: '.center(30, '-'))
print(output.decode(errors='replace'))
print(' end PTY output '.center(30, '-'))
raise
if __name__ == "__main__":
unittest.main()