| """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") | |
| if sys.platform.startswith("openbsd"): | |
| # OpenBSD's readline does not honor "completion-query-items 0". | |
| raise unittest.SkipTest("OpenBSD readline hangs on completion") | |
| 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); | |
| CREATE VIEW _view AS SELECT _col_table AS _col_table_alias | |
| FROM _table; | |
| 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_table_alias", "_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", "dragonfly", "sunos")), | |
| "Two actual tabs are inserted when there are no matching" | |
| " completions in the pseudo-terminal opened by run_pty()" | |
| " on this platform") | |
| 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() |