439 lines
16 KiB
Python
439 lines
16 KiB
Python
"""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 = "sqlite> "
|
|
PS2 = "... "
|
|
|
|
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("\x1b[1;35msqlite> \x1b[0m", out)
|
|
self.assertIn("\x1b[1;35m ... \x1b[0m\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):
|
|
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()
|