Adding upstream version 1.31.1.
Signed-off-by: Daniel Baumann <daniel@debian.org>
This commit is contained in:
parent
f38abee4bc
commit
9aaa9c7fe0
GPG key ID:
FBB4F0E80A80222F
27 changed files with 387 additions and 394 deletions
|
|
@ -65,7 +65,7 @@ def before_all(context):
|
|||
_, my_cnf = mkstemp()
|
||||
with open(my_cnf, "w") as f:
|
||||
f.write(
|
||||
"[client]\n" "pager={0} {1} {2}\n".format(
|
||||
"[client]\npager={0} {1} {2}\n".format(
|
||||
sys.executable, os.path.join(context.package_root, "test/features/wrappager.py"), context.conf["pager_boundary"]
|
||||
)
|
||||
)
|
||||
|
|
|
|||
|
|
@ -41,7 +41,7 @@ def step_see_small_results(context):
|
|||
@then("we see large results in vertical format")
|
||||
def step_see_large_results(context):
|
||||
rows = ["{n:3}| {n}".format(n=str(n)) for n in range(1, 50)]
|
||||
expected = "***************************[ 1. row ]" "***************************\r\n" + "{}\r\n".format("\r\n".join(rows) + "\r\n")
|
||||
expected = "***************************[ 1. row ]***************************\r\n" + "{}\r\n".format("\r\n".join(rows) + "\r\n")
|
||||
|
||||
wrappers.expect_pager(context, expected, timeout=10)
|
||||
wrappers.expect_exact(context, "1 row in set", timeout=2)
|
||||
|
|
|
|||
|
|
@ -32,7 +32,7 @@ def status_contains(context, expression):
|
|||
|
||||
@when("we create my.cnf file")
|
||||
def step_create_my_cnf_file(context):
|
||||
my_cnf = "[client]\n" f"host = {HOST}\n" f"port = {PORT}\n" f"user = {USER}\n" f"password = {PASSWORD}\n"
|
||||
my_cnf = f"[client]\nhost = {HOST}\nport = {PORT}\nuser = {USER}\npassword = {PASSWORD}\n"
|
||||
with open(MY_CNF_PATH, "w") as f:
|
||||
f.write(my_cnf)
|
||||
|
||||
|
|
@ -40,7 +40,7 @@ def step_create_my_cnf_file(context):
|
|||
@when("we create mylogin.cnf file")
|
||||
def step_create_mylogin_cnf_file(context):
|
||||
os.environ.pop("MYSQL_TEST_LOGIN_FILE", None)
|
||||
mylogin_cnf = f"[{TEST_LOGIN_PATH}]\n" f"host = {HOST}\n" f"port = {PORT}\n" f"user = {USER}\n" f"password = {PASSWORD}\n"
|
||||
mylogin_cnf = f"[{TEST_LOGIN_PATH}]\nhost = {HOST}\nport = {PORT}\nuser = {USER}\npassword = {PASSWORD}\n"
|
||||
with open(MYLOGIN_CNF_PATH, "wb") as f:
|
||||
input_file = io.StringIO(mylogin_cnf)
|
||||
f.write(encrypt_mylogin_cnf(input_file).read())
|
||||
|
|
|
|||
|
|
@ -81,9 +81,7 @@ def run_cli(context, run_args=None, exclude_args=None):
|
|||
try:
|
||||
cli_cmd = context.conf["cli_command"]
|
||||
except KeyError:
|
||||
cli_cmd = ('{0!s} -c "' "import coverage ; " "coverage.process_startup(); " "import mycli.main; " "mycli.main.cli()" '"').format(
|
||||
sys.executable
|
||||
)
|
||||
cli_cmd = ('{0!s} -c "import coverage ; coverage.process_startup(); import mycli.main; mycli.main.cli()"').format(sys.executable)
|
||||
|
||||
cmd_parts = [cli_cmd] + rendered_args
|
||||
cmd = " ".join(cmd_parts)
|
||||
|
|
|
|||
|
|
@ -9,26 +9,22 @@ def sorted_dicts(dicts):
|
|||
|
||||
def test_select_suggests_cols_with_visible_table_scope():
|
||||
suggestions = suggest_type("SELECT FROM tabl", "SELECT ")
|
||||
assert sorted_dicts(suggestions) == sorted_dicts(
|
||||
[
|
||||
{"type": "alias", "aliases": ["tabl"]},
|
||||
{"type": "column", "tables": [(None, "tabl", None)]},
|
||||
{"type": "function", "schema": []},
|
||||
{"type": "keyword"},
|
||||
]
|
||||
)
|
||||
assert sorted_dicts(suggestions) == sorted_dicts([
|
||||
{"type": "alias", "aliases": ["tabl"]},
|
||||
{"type": "column", "tables": [(None, "tabl", None)]},
|
||||
{"type": "function", "schema": []},
|
||||
{"type": "keyword"},
|
||||
])
|
||||
|
||||
|
||||
def test_select_suggests_cols_with_qualified_table_scope():
|
||||
suggestions = suggest_type("SELECT FROM sch.tabl", "SELECT ")
|
||||
assert sorted_dicts(suggestions) == sorted_dicts(
|
||||
[
|
||||
{"type": "alias", "aliases": ["tabl"]},
|
||||
{"type": "column", "tables": [("sch", "tabl", None)]},
|
||||
{"type": "function", "schema": []},
|
||||
{"type": "keyword"},
|
||||
]
|
||||
)
|
||||
assert sorted_dicts(suggestions) == sorted_dicts([
|
||||
{"type": "alias", "aliases": ["tabl"]},
|
||||
{"type": "column", "tables": [("sch", "tabl", None)]},
|
||||
{"type": "function", "schema": []},
|
||||
{"type": "keyword"},
|
||||
])
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
|
|
@ -48,14 +44,12 @@ def test_select_suggests_cols_with_qualified_table_scope():
|
|||
)
|
||||
def test_where_suggests_columns_functions(expression):
|
||||
suggestions = suggest_type(expression, expression)
|
||||
assert sorted_dicts(suggestions) == sorted_dicts(
|
||||
[
|
||||
{"type": "alias", "aliases": ["tabl"]},
|
||||
{"type": "column", "tables": [(None, "tabl", None)]},
|
||||
{"type": "function", "schema": []},
|
||||
{"type": "keyword"},
|
||||
]
|
||||
)
|
||||
assert sorted_dicts(suggestions) == sorted_dicts([
|
||||
{"type": "alias", "aliases": ["tabl"]},
|
||||
{"type": "column", "tables": [(None, "tabl", None)]},
|
||||
{"type": "function", "schema": []},
|
||||
{"type": "keyword"},
|
||||
])
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
|
|
@ -67,27 +61,23 @@ def test_where_suggests_columns_functions(expression):
|
|||
)
|
||||
def test_where_in_suggests_columns(expression):
|
||||
suggestions = suggest_type(expression, expression)
|
||||
assert sorted_dicts(suggestions) == sorted_dicts(
|
||||
[
|
||||
{"type": "alias", "aliases": ["tabl"]},
|
||||
{"type": "column", "tables": [(None, "tabl", None)]},
|
||||
{"type": "function", "schema": []},
|
||||
{"type": "keyword"},
|
||||
]
|
||||
)
|
||||
assert sorted_dicts(suggestions) == sorted_dicts([
|
||||
{"type": "alias", "aliases": ["tabl"]},
|
||||
{"type": "column", "tables": [(None, "tabl", None)]},
|
||||
{"type": "function", "schema": []},
|
||||
{"type": "keyword"},
|
||||
])
|
||||
|
||||
|
||||
def test_where_equals_any_suggests_columns_or_keywords():
|
||||
text = "SELECT * FROM tabl WHERE foo = ANY("
|
||||
suggestions = suggest_type(text, text)
|
||||
assert sorted_dicts(suggestions) == sorted_dicts(
|
||||
[
|
||||
{"type": "alias", "aliases": ["tabl"]},
|
||||
{"type": "column", "tables": [(None, "tabl", None)]},
|
||||
{"type": "function", "schema": []},
|
||||
{"type": "keyword"},
|
||||
]
|
||||
)
|
||||
assert sorted_dicts(suggestions) == sorted_dicts([
|
||||
{"type": "alias", "aliases": ["tabl"]},
|
||||
{"type": "column", "tables": [(None, "tabl", None)]},
|
||||
{"type": "function", "schema": []},
|
||||
{"type": "keyword"},
|
||||
])
|
||||
|
||||
|
||||
def test_lparen_suggests_cols():
|
||||
|
|
@ -107,14 +97,12 @@ def test_operand_inside_function_suggests_cols2():
|
|||
|
||||
def test_select_suggests_cols_and_funcs():
|
||||
suggestions = suggest_type("SELECT ", "SELECT ")
|
||||
assert sorted_dicts(suggestions) == sorted_dicts(
|
||||
[
|
||||
{"type": "alias", "aliases": []},
|
||||
{"type": "column", "tables": []},
|
||||
{"type": "function", "schema": []},
|
||||
{"type": "keyword"},
|
||||
]
|
||||
)
|
||||
assert sorted_dicts(suggestions) == sorted_dicts([
|
||||
{"type": "alias", "aliases": []},
|
||||
{"type": "column", "tables": []},
|
||||
{"type": "function", "schema": []},
|
||||
{"type": "keyword"},
|
||||
])
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
|
|
@ -170,14 +158,12 @@ def test_distinct_suggests_cols():
|
|||
|
||||
def test_col_comma_suggests_cols():
|
||||
suggestions = suggest_type("SELECT a, b, FROM tbl", "SELECT a, b,")
|
||||
assert sorted_dicts(suggestions) == sorted_dicts(
|
||||
[
|
||||
{"type": "alias", "aliases": ["tbl"]},
|
||||
{"type": "column", "tables": [(None, "tbl", None)]},
|
||||
{"type": "function", "schema": []},
|
||||
{"type": "keyword"},
|
||||
]
|
||||
)
|
||||
assert sorted_dicts(suggestions) == sorted_dicts([
|
||||
{"type": "alias", "aliases": ["tbl"]},
|
||||
{"type": "column", "tables": [(None, "tbl", None)]},
|
||||
{"type": "function", "schema": []},
|
||||
{"type": "keyword"},
|
||||
])
|
||||
|
||||
|
||||
def test_table_comma_suggests_tables_and_schemas():
|
||||
|
|
@ -207,50 +193,42 @@ def test_insert_into_lparen_comma_suggests_cols():
|
|||
|
||||
def test_partially_typed_col_name_suggests_col_names():
|
||||
suggestions = suggest_type("SELECT * FROM tabl WHERE col_n", "SELECT * FROM tabl WHERE col_n")
|
||||
assert sorted_dicts(suggestions) == sorted_dicts(
|
||||
[
|
||||
{"type": "alias", "aliases": ["tabl"]},
|
||||
{"type": "column", "tables": [(None, "tabl", None)]},
|
||||
{"type": "function", "schema": []},
|
||||
{"type": "keyword"},
|
||||
]
|
||||
)
|
||||
assert sorted_dicts(suggestions) == sorted_dicts([
|
||||
{"type": "alias", "aliases": ["tabl"]},
|
||||
{"type": "column", "tables": [(None, "tabl", None)]},
|
||||
{"type": "function", "schema": []},
|
||||
{"type": "keyword"},
|
||||
])
|
||||
|
||||
|
||||
def test_dot_suggests_cols_of_a_table_or_schema_qualified_table():
|
||||
suggestions = suggest_type("SELECT tabl. FROM tabl", "SELECT tabl.")
|
||||
assert sorted_dicts(suggestions) == sorted_dicts(
|
||||
[
|
||||
{"type": "column", "tables": [(None, "tabl", None)]},
|
||||
{"type": "table", "schema": "tabl"},
|
||||
{"type": "view", "schema": "tabl"},
|
||||
{"type": "function", "schema": "tabl"},
|
||||
]
|
||||
)
|
||||
assert sorted_dicts(suggestions) == sorted_dicts([
|
||||
{"type": "column", "tables": [(None, "tabl", None)]},
|
||||
{"type": "table", "schema": "tabl"},
|
||||
{"type": "view", "schema": "tabl"},
|
||||
{"type": "function", "schema": "tabl"},
|
||||
])
|
||||
|
||||
|
||||
def test_dot_suggests_cols_of_an_alias():
|
||||
suggestions = suggest_type("SELECT t1. FROM tabl1 t1, tabl2 t2", "SELECT t1.")
|
||||
assert sorted_dicts(suggestions) == sorted_dicts(
|
||||
[
|
||||
{"type": "table", "schema": "t1"},
|
||||
{"type": "view", "schema": "t1"},
|
||||
{"type": "column", "tables": [(None, "tabl1", "t1")]},
|
||||
{"type": "function", "schema": "t1"},
|
||||
]
|
||||
)
|
||||
assert sorted_dicts(suggestions) == sorted_dicts([
|
||||
{"type": "table", "schema": "t1"},
|
||||
{"type": "view", "schema": "t1"},
|
||||
{"type": "column", "tables": [(None, "tabl1", "t1")]},
|
||||
{"type": "function", "schema": "t1"},
|
||||
])
|
||||
|
||||
|
||||
def test_dot_col_comma_suggests_cols_or_schema_qualified_table():
|
||||
suggestions = suggest_type("SELECT t1.a, t2. FROM tabl1 t1, tabl2 t2", "SELECT t1.a, t2.")
|
||||
assert sorted_dicts(suggestions) == sorted_dicts(
|
||||
[
|
||||
{"type": "column", "tables": [(None, "tabl2", "t2")]},
|
||||
{"type": "table", "schema": "t2"},
|
||||
{"type": "view", "schema": "t2"},
|
||||
{"type": "function", "schema": "t2"},
|
||||
]
|
||||
)
|
||||
assert sorted_dicts(suggestions) == sorted_dicts([
|
||||
{"type": "column", "tables": [(None, "tabl2", "t2")]},
|
||||
{"type": "table", "schema": "t2"},
|
||||
{"type": "view", "schema": "t2"},
|
||||
{"type": "function", "schema": "t2"},
|
||||
])
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
|
|
@ -306,34 +284,31 @@ def test_sub_select_table_name_completion(expression):
|
|||
|
||||
def test_sub_select_col_name_completion():
|
||||
suggestions = suggest_type("SELECT * FROM (SELECT FROM abc", "SELECT * FROM (SELECT ")
|
||||
assert sorted_dicts(suggestions) == sorted_dicts(
|
||||
[
|
||||
{"type": "alias", "aliases": ["abc"]},
|
||||
{"type": "column", "tables": [(None, "abc", None)]},
|
||||
{"type": "function", "schema": []},
|
||||
{"type": "keyword"},
|
||||
]
|
||||
)
|
||||
assert sorted_dicts(suggestions) == sorted_dicts([
|
||||
{"type": "alias", "aliases": ["abc"]},
|
||||
{"type": "column", "tables": [(None, "abc", None)]},
|
||||
{"type": "function", "schema": []},
|
||||
{"type": "keyword"},
|
||||
])
|
||||
|
||||
|
||||
@pytest.mark.xfail
|
||||
def test_sub_select_multiple_col_name_completion():
|
||||
suggestions = suggest_type("SELECT * FROM (SELECT a, FROM abc", "SELECT * FROM (SELECT a, ")
|
||||
assert sorted_dicts(suggestions) == sorted_dicts(
|
||||
[{"type": "column", "tables": [(None, "abc", None)]}, {"type": "function", "schema": []}]
|
||||
)
|
||||
assert sorted_dicts(suggestions) == sorted_dicts([
|
||||
{"type": "column", "tables": [(None, "abc", None)]},
|
||||
{"type": "function", "schema": []},
|
||||
])
|
||||
|
||||
|
||||
def test_sub_select_dot_col_name_completion():
|
||||
suggestions = suggest_type("SELECT * FROM (SELECT t. FROM tabl t", "SELECT * FROM (SELECT t.")
|
||||
assert sorted_dicts(suggestions) == sorted_dicts(
|
||||
[
|
||||
{"type": "column", "tables": [(None, "tabl", "t")]},
|
||||
{"type": "table", "schema": "t"},
|
||||
{"type": "view", "schema": "t"},
|
||||
{"type": "function", "schema": "t"},
|
||||
]
|
||||
)
|
||||
assert sorted_dicts(suggestions) == sorted_dicts([
|
||||
{"type": "column", "tables": [(None, "tabl", "t")]},
|
||||
{"type": "table", "schema": "t"},
|
||||
{"type": "view", "schema": "t"},
|
||||
{"type": "function", "schema": "t"},
|
||||
])
|
||||
|
||||
|
||||
@pytest.mark.parametrize("join_type", ["", "INNER", "LEFT", "RIGHT OUTER"])
|
||||
|
|
@ -353,14 +328,12 @@ def test_join_suggests_tables_and_schemas(tbl_alias, join_type):
|
|||
)
|
||||
def test_join_alias_dot_suggests_cols1(sql):
|
||||
suggestions = suggest_type(sql, sql)
|
||||
assert sorted_dicts(suggestions) == sorted_dicts(
|
||||
[
|
||||
{"type": "column", "tables": [(None, "abc", "a")]},
|
||||
{"type": "table", "schema": "a"},
|
||||
{"type": "view", "schema": "a"},
|
||||
{"type": "function", "schema": "a"},
|
||||
]
|
||||
)
|
||||
assert sorted_dicts(suggestions) == sorted_dicts([
|
||||
{"type": "column", "tables": [(None, "abc", "a")]},
|
||||
{"type": "table", "schema": "a"},
|
||||
{"type": "view", "schema": "a"},
|
||||
{"type": "function", "schema": "a"},
|
||||
])
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
|
|
@ -372,14 +345,12 @@ def test_join_alias_dot_suggests_cols1(sql):
|
|||
)
|
||||
def test_join_alias_dot_suggests_cols2(sql):
|
||||
suggestions = suggest_type(sql, sql)
|
||||
assert sorted_dicts(suggestions) == sorted_dicts(
|
||||
[
|
||||
{"type": "column", "tables": [(None, "def", "d")]},
|
||||
{"type": "table", "schema": "d"},
|
||||
{"type": "view", "schema": "d"},
|
||||
{"type": "function", "schema": "d"},
|
||||
]
|
||||
)
|
||||
assert sorted_dicts(suggestions) == sorted_dicts([
|
||||
{"type": "column", "tables": [(None, "def", "d")]},
|
||||
{"type": "table", "schema": "d"},
|
||||
{"type": "view", "schema": "d"},
|
||||
{"type": "function", "schema": "d"},
|
||||
])
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
|
|
@ -445,14 +416,12 @@ def test_join_using_suggests_common_columns(col_list):
|
|||
)
|
||||
def test_two_join_alias_dot_suggests_cols1(sql):
|
||||
suggestions = suggest_type(sql, sql)
|
||||
assert sorted_dicts(suggestions) == sorted_dicts(
|
||||
[
|
||||
{"type": "column", "tables": [(None, "ghi", "g")]},
|
||||
{"type": "table", "schema": "g"},
|
||||
{"type": "view", "schema": "g"},
|
||||
{"type": "function", "schema": "g"},
|
||||
]
|
||||
)
|
||||
assert sorted_dicts(suggestions) == sorted_dicts([
|
||||
{"type": "column", "tables": [(None, "ghi", "g")]},
|
||||
{"type": "table", "schema": "g"},
|
||||
{"type": "view", "schema": "g"},
|
||||
{"type": "function", "schema": "g"},
|
||||
])
|
||||
|
||||
|
||||
def test_2_statements_2nd_current():
|
||||
|
|
@ -460,14 +429,12 @@ def test_2_statements_2nd_current():
|
|||
assert sorted_dicts(suggestions) == sorted_dicts([{"type": "table", "schema": []}, {"type": "view", "schema": []}, {"type": "schema"}])
|
||||
|
||||
suggestions = suggest_type("select * from a; select from b", "select * from a; select ")
|
||||
assert sorted_dicts(suggestions) == sorted_dicts(
|
||||
[
|
||||
{"type": "alias", "aliases": ["b"]},
|
||||
{"type": "column", "tables": [(None, "b", None)]},
|
||||
{"type": "function", "schema": []},
|
||||
{"type": "keyword"},
|
||||
]
|
||||
)
|
||||
assert sorted_dicts(suggestions) == sorted_dicts([
|
||||
{"type": "alias", "aliases": ["b"]},
|
||||
{"type": "column", "tables": [(None, "b", None)]},
|
||||
{"type": "function", "schema": []},
|
||||
{"type": "keyword"},
|
||||
])
|
||||
|
||||
# Should work even if first statement is invalid
|
||||
suggestions = suggest_type("select * from; select * from ", "select * from; select * from ")
|
||||
|
|
@ -479,14 +446,12 @@ def test_2_statements_1st_current():
|
|||
assert sorted_dicts(suggestions) == sorted_dicts([{"type": "table", "schema": []}, {"type": "view", "schema": []}, {"type": "schema"}])
|
||||
|
||||
suggestions = suggest_type("select from a; select * from b", "select ")
|
||||
assert sorted_dicts(suggestions) == sorted_dicts(
|
||||
[
|
||||
{"type": "alias", "aliases": ["a"]},
|
||||
{"type": "column", "tables": [(None, "a", None)]},
|
||||
{"type": "function", "schema": []},
|
||||
{"type": "keyword"},
|
||||
]
|
||||
)
|
||||
assert sorted_dicts(suggestions) == sorted_dicts([
|
||||
{"type": "alias", "aliases": ["a"]},
|
||||
{"type": "column", "tables": [(None, "a", None)]},
|
||||
{"type": "function", "schema": []},
|
||||
{"type": "keyword"},
|
||||
])
|
||||
|
||||
|
||||
def test_3_statements_2nd_current():
|
||||
|
|
@ -494,14 +459,12 @@ def test_3_statements_2nd_current():
|
|||
assert sorted_dicts(suggestions) == sorted_dicts([{"type": "table", "schema": []}, {"type": "view", "schema": []}, {"type": "schema"}])
|
||||
|
||||
suggestions = suggest_type("select * from a; select from b; select * from c", "select * from a; select ")
|
||||
assert sorted_dicts(suggestions) == sorted_dicts(
|
||||
[
|
||||
{"type": "alias", "aliases": ["b"]},
|
||||
{"type": "column", "tables": [(None, "b", None)]},
|
||||
{"type": "function", "schema": []},
|
||||
{"type": "keyword"},
|
||||
]
|
||||
)
|
||||
assert sorted_dicts(suggestions) == sorted_dicts([
|
||||
{"type": "alias", "aliases": ["b"]},
|
||||
{"type": "column", "tables": [(None, "b", None)]},
|
||||
{"type": "function", "schema": []},
|
||||
{"type": "keyword"},
|
||||
])
|
||||
|
||||
|
||||
def test_create_db_with_template():
|
||||
|
|
|
|||
|
|
@ -93,7 +93,7 @@ def test_batch_mode(executor):
|
|||
run(executor, """create table test(a text)""")
|
||||
run(executor, """insert into test values('abc'), ('def'), ('ghi')""")
|
||||
|
||||
sql = "select count(*) from test;\n" "select * from test limit 1;"
|
||||
sql = "select count(*) from test;\nselect * from test limit 1;"
|
||||
|
||||
runner = CliRunner()
|
||||
result = runner.invoke(cli, args=CLI_ARGS, input=sql)
|
||||
|
|
@ -107,7 +107,7 @@ def test_batch_mode_table(executor):
|
|||
run(executor, """create table test(a text)""")
|
||||
run(executor, """insert into test values('abc'), ('def'), ('ghi')""")
|
||||
|
||||
sql = "select count(*) from test;\n" "select * from test limit 1;"
|
||||
sql = "select count(*) from test;\nselect * from test limit 1;"
|
||||
|
||||
runner = CliRunner()
|
||||
result = runner.invoke(cli, args=CLI_ARGS + ["-t"], input=sql)
|
||||
|
|
@ -543,7 +543,7 @@ def test_init_command_arg(executor):
|
|||
@dbtest
|
||||
def test_init_command_multiple_arg(executor):
|
||||
init_command = "set sql_select_limit=2000; set max_join_size=20000"
|
||||
sql = 'show variables like "sql_select_limit";\n' 'show variables like "max_join_size"'
|
||||
sql = 'show variables like "sql_select_limit";\nshow variables like "max_join_size"'
|
||||
runner = CliRunner()
|
||||
result = runner.invoke(cli, args=CLI_ARGS + ["--init-command", init_command], input=sql)
|
||||
|
||||
|
|
@ -554,6 +554,7 @@ def test_init_command_multiple_arg(executor):
|
|||
assert expected_sql_select_limit in result.output
|
||||
assert expected_max_join_size in result.output
|
||||
|
||||
|
||||
@dbtest
|
||||
def test_global_init_commands(executor):
|
||||
"""Tests that global init-commands from config are executed by default."""
|
||||
|
|
|
|||
|
|
@ -122,24 +122,24 @@ def test_query_starts_with_comment():
|
|||
|
||||
|
||||
def test_queries_start_with():
|
||||
sql = "# comment\n" "show databases;" "use foo;"
|
||||
sql = "# comment\nshow databases;use foo;"
|
||||
assert queries_start_with(sql, ("show", "select")) is True
|
||||
assert queries_start_with(sql, ("use", "drop")) is True
|
||||
assert queries_start_with(sql, ("delete", "update")) is False
|
||||
|
||||
|
||||
def test_is_destructive():
|
||||
sql = "use test;\n" "show databases;\n" "drop database foo;"
|
||||
sql = "use test;\nshow databases;\ndrop database foo;"
|
||||
assert is_destructive(sql) is True
|
||||
|
||||
|
||||
def test_is_destructive_update_with_where_clause():
|
||||
sql = "use test;\n" "show databases;\n" "UPDATE test SET x = 1 WHERE id = 1;"
|
||||
sql = "use test;\nshow databases;\nUPDATE test SET x = 1 WHERE id = 1;"
|
||||
assert is_destructive(sql) is False
|
||||
|
||||
|
||||
def test_is_destructive_update_without_where_clause():
|
||||
sql = "use test;\n" "show databases;\n" "UPDATE test SET x = 1;"
|
||||
sql = "use test;\nshow databases;\nUPDATE test SET x = 1;"
|
||||
assert is_destructive(sql) is True
|
||||
|
||||
|
||||
|
|
@ -167,7 +167,7 @@ def test_query_has_where_clause(sql, has_where_clause):
|
|||
("drop database foo; create database bar", "foo", True),
|
||||
("select bar from foo; drop database bazz", "foo", False),
|
||||
("select bar from foo; drop database bazz", "bazz", True),
|
||||
("-- dropping database \n " "drop -- really dropping \n " "schema abc -- now it is dropped", "abc", True),
|
||||
("-- dropping database \n drop -- really dropping \n schema abc -- now it is dropped", "abc", True),
|
||||
],
|
||||
)
|
||||
def test_is_dropping_database(sql, dbname, is_dropping):
|
||||
|
|
|
|||
|
|
@ -72,33 +72,29 @@ def test_table_completion(completer, complete_event):
|
|||
text = "SELECT * FROM "
|
||||
position = len(text)
|
||||
result = completer.get_completions(Document(text=text, cursor_position=position), complete_event)
|
||||
assert list(result) == list(
|
||||
[
|
||||
Completion(text="users", start_position=0),
|
||||
Completion(text="orders", start_position=0),
|
||||
Completion(text="`select`", start_position=0),
|
||||
Completion(text="`réveillé`", start_position=0),
|
||||
]
|
||||
)
|
||||
assert list(result) == list([
|
||||
Completion(text="users", start_position=0),
|
||||
Completion(text="orders", start_position=0),
|
||||
Completion(text="`select`", start_position=0),
|
||||
Completion(text="`réveillé`", start_position=0),
|
||||
])
|
||||
|
||||
|
||||
def test_function_name_completion(completer, complete_event):
|
||||
text = "SELECT MA"
|
||||
position = len("SELECT MA")
|
||||
result = completer.get_completions(Document(text=text, cursor_position=position), complete_event)
|
||||
assert list(result) == list(
|
||||
[
|
||||
Completion(text="MAX", start_position=-2),
|
||||
Completion(text="CHANGE MASTER TO", start_position=-2),
|
||||
Completion(text="CURRENT_TIMESTAMP", start_position=-2),
|
||||
Completion(text="DECIMAL", start_position=-2),
|
||||
Completion(text="FORMAT", start_position=-2),
|
||||
Completion(text="MASTER", start_position=-2),
|
||||
Completion(text="PRIMARY", start_position=-2),
|
||||
Completion(text="ROW_FORMAT", start_position=-2),
|
||||
Completion(text="SMALLINT", start_position=-2),
|
||||
]
|
||||
)
|
||||
assert list(result) == list([
|
||||
Completion(text="MAX", start_position=-2),
|
||||
Completion(text="CHANGE MASTER TO", start_position=-2),
|
||||
Completion(text="CURRENT_TIMESTAMP", start_position=-2),
|
||||
Completion(text="DECIMAL", start_position=-2),
|
||||
Completion(text="FORMAT", start_position=-2),
|
||||
Completion(text="MASTER", start_position=-2),
|
||||
Completion(text="PRIMARY", start_position=-2),
|
||||
Completion(text="ROW_FORMAT", start_position=-2),
|
||||
Completion(text="SMALLINT", start_position=-2),
|
||||
])
|
||||
|
||||
|
||||
def test_suggested_column_names(completer, complete_event):
|
||||
|
|
@ -138,15 +134,13 @@ def test_suggested_column_names_in_function(completer, complete_event):
|
|||
text = "SELECT MAX( from users"
|
||||
position = len("SELECT MAX(")
|
||||
result = completer.get_completions(Document(text=text, cursor_position=position), complete_event)
|
||||
assert list(result) == list(
|
||||
[
|
||||
Completion(text="*", start_position=0),
|
||||
Completion(text="id", start_position=0),
|
||||
Completion(text="email", start_position=0),
|
||||
Completion(text="first_name", start_position=0),
|
||||
Completion(text="last_name", start_position=0),
|
||||
]
|
||||
)
|
||||
assert list(result) == list([
|
||||
Completion(text="*", start_position=0),
|
||||
Completion(text="id", start_position=0),
|
||||
Completion(text="email", start_position=0),
|
||||
Completion(text="first_name", start_position=0),
|
||||
Completion(text="last_name", start_position=0),
|
||||
])
|
||||
|
||||
|
||||
def test_suggested_column_names_with_table_dot(completer, complete_event):
|
||||
|
|
@ -160,15 +154,13 @@ def test_suggested_column_names_with_table_dot(completer, complete_event):
|
|||
text = "SELECT users. from users"
|
||||
position = len("SELECT users.")
|
||||
result = list(completer.get_completions(Document(text=text, cursor_position=position), complete_event))
|
||||
assert result == list(
|
||||
[
|
||||
Completion(text="*", start_position=0),
|
||||
Completion(text="id", start_position=0),
|
||||
Completion(text="email", start_position=0),
|
||||
Completion(text="first_name", start_position=0),
|
||||
Completion(text="last_name", start_position=0),
|
||||
]
|
||||
)
|
||||
assert result == list([
|
||||
Completion(text="*", start_position=0),
|
||||
Completion(text="id", start_position=0),
|
||||
Completion(text="email", start_position=0),
|
||||
Completion(text="first_name", start_position=0),
|
||||
Completion(text="last_name", start_position=0),
|
||||
])
|
||||
|
||||
|
||||
def test_suggested_column_names_with_alias(completer, complete_event):
|
||||
|
|
@ -182,15 +174,13 @@ def test_suggested_column_names_with_alias(completer, complete_event):
|
|||
text = "SELECT u. from users u"
|
||||
position = len("SELECT u.")
|
||||
result = list(completer.get_completions(Document(text=text, cursor_position=position), complete_event))
|
||||
assert result == list(
|
||||
[
|
||||
Completion(text="*", start_position=0),
|
||||
Completion(text="id", start_position=0),
|
||||
Completion(text="email", start_position=0),
|
||||
Completion(text="first_name", start_position=0),
|
||||
Completion(text="last_name", start_position=0),
|
||||
]
|
||||
)
|
||||
assert result == list([
|
||||
Completion(text="*", start_position=0),
|
||||
Completion(text="id", start_position=0),
|
||||
Completion(text="email", start_position=0),
|
||||
Completion(text="first_name", start_position=0),
|
||||
Completion(text="last_name", start_position=0),
|
||||
])
|
||||
|
||||
|
||||
def test_suggested_multiple_column_names(completer, complete_event):
|
||||
|
|
@ -231,15 +221,13 @@ def test_suggested_multiple_column_names_with_alias(completer, complete_event):
|
|||
text = "SELECT u.id, u. from users u"
|
||||
position = len("SELECT u.id, u.")
|
||||
result = list(completer.get_completions(Document(text=text, cursor_position=position), complete_event))
|
||||
assert result == list(
|
||||
[
|
||||
Completion(text="*", start_position=0),
|
||||
Completion(text="id", start_position=0),
|
||||
Completion(text="email", start_position=0),
|
||||
Completion(text="first_name", start_position=0),
|
||||
Completion(text="last_name", start_position=0),
|
||||
]
|
||||
)
|
||||
assert result == list([
|
||||
Completion(text="*", start_position=0),
|
||||
Completion(text="id", start_position=0),
|
||||
Completion(text="email", start_position=0),
|
||||
Completion(text="first_name", start_position=0),
|
||||
Completion(text="last_name", start_position=0),
|
||||
])
|
||||
|
||||
|
||||
def test_suggested_multiple_column_names_with_dot(completer, complete_event):
|
||||
|
|
@ -254,77 +242,65 @@ def test_suggested_multiple_column_names_with_dot(completer, complete_event):
|
|||
text = "SELECT users.id, users. from users u"
|
||||
position = len("SELECT users.id, users.")
|
||||
result = list(completer.get_completions(Document(text=text, cursor_position=position), complete_event))
|
||||
assert result == list(
|
||||
[
|
||||
Completion(text="*", start_position=0),
|
||||
Completion(text="id", start_position=0),
|
||||
Completion(text="email", start_position=0),
|
||||
Completion(text="first_name", start_position=0),
|
||||
Completion(text="last_name", start_position=0),
|
||||
]
|
||||
)
|
||||
assert result == list([
|
||||
Completion(text="*", start_position=0),
|
||||
Completion(text="id", start_position=0),
|
||||
Completion(text="email", start_position=0),
|
||||
Completion(text="first_name", start_position=0),
|
||||
Completion(text="last_name", start_position=0),
|
||||
])
|
||||
|
||||
|
||||
def test_suggested_aliases_after_on(completer, complete_event):
|
||||
text = "SELECT u.name, o.id FROM users u JOIN orders o ON "
|
||||
position = len("SELECT u.name, o.id FROM users u JOIN orders o ON ")
|
||||
result = list(completer.get_completions(Document(text=text, cursor_position=position), complete_event))
|
||||
assert result == list(
|
||||
[
|
||||
Completion(text="u", start_position=0),
|
||||
Completion(text="o", start_position=0),
|
||||
]
|
||||
)
|
||||
assert result == list([
|
||||
Completion(text="u", start_position=0),
|
||||
Completion(text="o", start_position=0),
|
||||
])
|
||||
|
||||
|
||||
def test_suggested_aliases_after_on_right_side(completer, complete_event):
|
||||
text = "SELECT u.name, o.id FROM users u JOIN orders o ON o.user_id = "
|
||||
position = len("SELECT u.name, o.id FROM users u JOIN orders o ON o.user_id = ")
|
||||
result = list(completer.get_completions(Document(text=text, cursor_position=position), complete_event))
|
||||
assert result == list(
|
||||
[
|
||||
Completion(text="u", start_position=0),
|
||||
Completion(text="o", start_position=0),
|
||||
]
|
||||
)
|
||||
assert result == list([
|
||||
Completion(text="u", start_position=0),
|
||||
Completion(text="o", start_position=0),
|
||||
])
|
||||
|
||||
|
||||
def test_suggested_tables_after_on(completer, complete_event):
|
||||
text = "SELECT users.name, orders.id FROM users JOIN orders ON "
|
||||
position = len("SELECT users.name, orders.id FROM users JOIN orders ON ")
|
||||
result = list(completer.get_completions(Document(text=text, cursor_position=position), complete_event))
|
||||
assert result == list(
|
||||
[
|
||||
Completion(text="users", start_position=0),
|
||||
Completion(text="orders", start_position=0),
|
||||
]
|
||||
)
|
||||
assert result == list([
|
||||
Completion(text="users", start_position=0),
|
||||
Completion(text="orders", start_position=0),
|
||||
])
|
||||
|
||||
|
||||
def test_suggested_tables_after_on_right_side(completer, complete_event):
|
||||
text = "SELECT users.name, orders.id FROM users JOIN orders ON orders.user_id = "
|
||||
position = len("SELECT users.name, orders.id FROM users JOIN orders ON orders.user_id = ")
|
||||
result = list(completer.get_completions(Document(text=text, cursor_position=position), complete_event))
|
||||
assert result == list(
|
||||
[
|
||||
Completion(text="users", start_position=0),
|
||||
Completion(text="orders", start_position=0),
|
||||
]
|
||||
)
|
||||
assert result == list([
|
||||
Completion(text="users", start_position=0),
|
||||
Completion(text="orders", start_position=0),
|
||||
])
|
||||
|
||||
|
||||
def test_table_names_after_from(completer, complete_event):
|
||||
text = "SELECT * FROM "
|
||||
position = len("SELECT * FROM ")
|
||||
result = list(completer.get_completions(Document(text=text, cursor_position=position), complete_event))
|
||||
assert result == list(
|
||||
[
|
||||
Completion(text="users", start_position=0),
|
||||
Completion(text="orders", start_position=0),
|
||||
Completion(text="`select`", start_position=0),
|
||||
Completion(text="`réveillé`", start_position=0),
|
||||
]
|
||||
)
|
||||
assert result == list([
|
||||
Completion(text="users", start_position=0),
|
||||
Completion(text="orders", start_position=0),
|
||||
Completion(text="`select`", start_position=0),
|
||||
Completion(text="`réveillé`", start_position=0),
|
||||
])
|
||||
|
||||
|
||||
def test_auto_escaped_col_names(completer, complete_event):
|
||||
|
|
|
|||
|
|
@ -44,7 +44,7 @@ def test_bools(executor):
|
|||
@dbtest
|
||||
def test_binary(executor):
|
||||
run(executor, """create table bt(geom linestring NOT NULL)""")
|
||||
run(executor, "INSERT INTO bt VALUES " "(ST_GeomFromText('LINESTRING(116.37604 39.73979,116.375 39.73965)'));")
|
||||
run(executor, "INSERT INTO bt VALUES (ST_GeomFromText('LINESTRING(116.37604 39.73979,116.375 39.73965)'));")
|
||||
results = run(executor, """select * from bt""")
|
||||
|
||||
geom = (
|
||||
|
|
@ -139,7 +139,7 @@ def test_favorite_query_multiple_statement(executor):
|
|||
run(executor, "insert into test values('abc')")
|
||||
run(executor, "insert into test values('def')")
|
||||
|
||||
results = run(executor, "\\fs test-ad select * from test where a like 'a%'; " "select * from test where a like 'd%'")
|
||||
results = run(executor, "\\fs test-ad select * from test where a like 'a%'; select * from test where a like 'd%'")
|
||||
assert_result_equal(results, status="Saved.")
|
||||
|
||||
results = run(executor, "\\f test-ad")
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue