dbcli/pgcli
1
0
Fork 0
Code Issues Activity

Merging upstream version 4.0.1.

Browse source 
Signed-off-by: Daniel Baumann <daniel@debian.org>
This commit is contained in:
Daniel Baumann 2025-02-09 20:05:21 +01:00
parent af10454b21
commit 7c65fc707e
Signed by: daniel
GPG key ID: FBB4F0E80A80222F
42 changed files with 955 additions and 184 deletions
Download patch file Download diff file Expand all files Collapse all files

24
tests/features/basic_commands.feature
View file

@ -23,6 +23,30 @@ Feature: run the cli,
When we send "ctrl + d"
then dbcli exits
Scenario: confirm exit when a transaction is ongoing
When we begin transaction
and we try to send "ctrl + d"
then we see ongoing transaction message
when we send "c"
then dbcli exits
Scenario: cancel exit when a transaction is ongoing
When we begin transaction
and we try to send "ctrl + d"
then we see ongoing transaction message
when we send "a"
then we see dbcli prompt
when we rollback transaction
when we send "ctrl + d"
then dbcli exits
Scenario: interrupt current query via "ctrl + c"
When we send sleep query
and we send "ctrl + c"
then we see cancelled query warning
when we check for any non-idle sleep queries
then we don't see any non-idle sleep queries
Scenario: list databases
When we list databases
then we see list of databases

2
tests/features/crud_database.feature
View file

@ -5,7 +5,7 @@ Feature: manipulate databases:
When we create database
then we see database created
when we drop database
then we confirm the destructive warning
then we respond to the destructive warning: y
then we see database dropped
when we connect to dbserver
then we see database connected

29
tests/features/crud_table.feature
View file

@ -8,15 +8,38 @@ Feature: manipulate tables:
then we see table created
when we insert into table
then we see record inserted
when we select from table
then we see data selected: initial
when we update table
then we see record updated
when we select from table
then we see data selected
then we see data selected: updated
when we delete from table
then we confirm the destructive warning
then we respond to the destructive warning: y
then we see record deleted
when we drop table
then we confirm the destructive warning
then we respond to the destructive warning: y
then we see table dropped
when we connect to dbserver
then we see database connected
Scenario: transaction handling, with cancelling on a destructive warning.
When we connect to test database
then we see database connected
when we create table
then we see table created
when we begin transaction
then we see transaction began
when we insert into table
then we see record inserted
when we delete from table
then we respond to the destructive warning: n
when we select from table
then we see data selected: initial
when we rollback transaction
then we see transaction rolled back
when we select from table
then we see select output without data
when we drop table
then we respond to the destructive warning: y
then we see table dropped

14
tests/features/environment.py
View file

@ -164,10 +164,24 @@ def before_step(context, _):
context.atprompt = False
def is_known_problem(scenario):
"""TODO: why is this not working in 3.12?"""
if sys.version_info >= (3, 12):
return scenario.name in (
'interrupt current query via "ctrl + c"',
"run the cli with --username",
"run the cli with --user",
"run the cli with --port",
)
return False
def before_scenario(context, scenario):
if scenario.name == "list databases":
# not using the cli for that
return
if is_known_problem(scenario):
scenario.skip()
currentdb = None
if "pgbouncer" in scenario.feature.tags:
if context.pgbouncer_available:

6
tests/features/expanded.feature
View file

@ -7,7 +7,7 @@ Feature: expanded mode:
and we select from table
then we see expanded data selected
when we drop table
then we confirm the destructive warning
then we respond to the destructive warning: y
then we see table dropped
Scenario: expanded off
@ -16,7 +16,7 @@ Feature: expanded mode:
and we select from table
then we see nonexpanded data selected
when we drop table
then we confirm the destructive warning
then we respond to the destructive warning: y
then we see table dropped
Scenario: expanded auto
@ -25,5 +25,5 @@ Feature: expanded mode:
and we select from table
then we see auto data selected
when we drop table
then we confirm the destructive warning
then we respond to the destructive warning: y
then we see table dropped

97
tests/features/steps/basic_commands.py
View file

@ -64,13 +64,83 @@ def step_ctrl_d(context):
"""
Send Ctrl + D to hopefully exit.
"""
step_try_to_ctrl_d(context)
context.cli.expect(pexpect.EOF, timeout=5)
context.exit_sent = True
@when('we try to send "ctrl + d"')
def step_try_to_ctrl_d(context):
"""
Send Ctrl + D, perhaps exiting, perhaps not (if a transaction is
ongoing).
"""
# turn off pager before exiting
context.cli.sendcontrol("c")
context.cli.sendline(r"\pset pager off")
wrappers.wait_prompt(context)
context.cli.sendcontrol("d")
context.cli.expect(pexpect.EOF, timeout=5)
context.exit_sent = True
@when('we send "ctrl + c"')
def step_ctrl_c(context):
"""Send Ctrl + c to hopefully interrupt."""
context.cli.sendcontrol("c")
@then("we see cancelled query warning")
def step_see_cancelled_query_warning(context):
"""
Make sure we receive the warning that the current query was cancelled.
"""
wrappers.expect_exact(context, "cancelled query", timeout=2)
@then("we see ongoing transaction message")
def step_see_ongoing_transaction_error(context):
"""
Make sure we receive the warning that a transaction is ongoing.
"""
context.cli.expect("A transaction is ongoing.", timeout=2)
@when("we send sleep query")
def step_send_sleep_15_seconds(context):
"""
Send query to sleep for 15 seconds.
"""
context.cli.sendline("select pg_sleep(15)")
@when("we check for any non-idle sleep queries")
def step_check_for_active_sleep_queries(context):
"""
Send query to check for any non-idle pg_sleep queries.
"""
context.cli.sendline(
"select state from pg_stat_activity where query not like '%pg_stat_activity%' and query like '%pg_sleep%' and state != 'idle';"
)
@then("we don't see any non-idle sleep queries")
def step_no_active_sleep_queries(context):
"""Confirm that any pg_sleep queries are either idle or not active."""
wrappers.expect_exact(
context,
context.conf["pager_boundary"]
+ "\r"
+ dedent(
"""
+-------+\r
| state |\r
|-------|\r
+-------+\r
SELECT 0\r
"""
)
+ context.conf["pager_boundary"],
timeout=5,
)
@when(r'we send "\?" command')
@ -131,18 +201,31 @@ def step_see_found(context):
)
@then("we confirm the destructive warning")
def step_confirm_destructive_command(context):
"""Confirm destructive command."""
@then("we respond to the destructive warning: {response}")
def step_resppond_to_destructive_command(context, response):
"""Respond to destructive command."""
wrappers.expect_exact(
context,
"You're about to run a destructive command.\r\nDo you want to proceed? (y/n):",
"You're about to run a destructive command.\r\nDo you want to proceed? [y/N]:",
timeout=2,
)
context.cli.sendline("y")
context.cli.sendline(response.strip())
@then("we send password")
def step_send_password(context):
wrappers.expect_exact(context, "Password for", timeout=5)
context.cli.sendline(context.conf["pass"] or "DOES NOT MATTER")
@when('we send "{text}"')
def step_send_text(context, text):
context.cli.sendline(text)
# Try to detect whether we are exiting. If so, set `exit_sent`
# so that `after_scenario` correctly cleans up.
try:
context.cli.expect(pexpect.EOF, timeout=0.2)
except pexpect.TIMEOUT:
pass
else:
context.exit_sent = True

91
tests/features/steps/crud_table.py
View file

@ -9,6 +9,10 @@ from textwrap import dedent
import wrappers
INITIAL_DATA = "xxx"
UPDATED_DATA = "yyy"
@when("we create table")
def step_create_table(context):
"""
@ -22,7 +26,7 @@ def step_insert_into_table(context):
"""
Send insert into table.
"""
context.cli.sendline("""insert into a(x) values('xxx');""")
context.cli.sendline(f"""insert into a(x) values('{INITIAL_DATA}');""")
@when("we update table")
@ -30,7 +34,9 @@ def step_update_table(context):
"""
Send insert into table.
"""
context.cli.sendline("""update a set x = 'yyy' where x = 'xxx';""")
context.cli.sendline(
f"""update a set x = '{UPDATED_DATA}' where x = '{INITIAL_DATA}';"""
)
@when("we select from table")
@ -46,7 +52,7 @@ def step_delete_from_table(context):
"""
Send deete from table.
"""
context.cli.sendline("""delete from a where x = 'yyy';""")
context.cli.sendline(f"""delete from a where x = '{UPDATED_DATA}';""")
@when("we drop table")
@ -57,6 +63,30 @@ def step_drop_table(context):
context.cli.sendline("drop table a;")
@when("we alter the table")
def step_alter_table(context):
"""
Alter the table by adding a column.
"""
context.cli.sendline("""alter table a add column y varchar;""")
@when("we begin transaction")
def step_begin_transaction(context):
"""
Begin transaction
"""
context.cli.sendline("begin;")
@when("we rollback transaction")
def step_rollback_transaction(context):
"""
Rollback transaction
"""
context.cli.sendline("rollback;")
@then("we see table created")
def step_see_table_created(context):
"""
@ -81,21 +111,42 @@ def step_see_record_updated(context):
wrappers.expect_pager(context, "UPDATE 1\r\n", timeout=2)
@then("we see data selected")
def step_see_data_selected(context):
@then("we see data selected: {data}")
def step_see_data_selected(context, data):
"""
Wait to see select output.
Wait to see select output with initial or updated data.
"""
x = UPDATED_DATA if data == "updated" else INITIAL_DATA
wrappers.expect_pager(
context,
dedent(
f"""\
+-----+\r
| x |\r
|-----|\r
| {x} |\r
+-----+\r
SELECT 1\r
"""
),
timeout=1,
)
@then("we see select output without data")
def step_see_no_data_selected(context):
"""
Wait to see select output without data.
"""
wrappers.expect_pager(
context,
dedent(
"""\
+-----+\r
| x |\r
|-----|\r
| yyy |\r
+-----+\r
SELECT 1\r
+---+\r
| x |\r
|---|\r
+---+\r
SELECT 0\r
"""
),
timeout=1,
@ -116,3 +167,19 @@ def step_see_table_dropped(context):
Wait to see drop output.
"""
wrappers.expect_pager(context, "DROP TABLE\r\n", timeout=2)
@then("we see transaction began")
def step_see_transaction_began(context):
"""
Wait to see transaction began.
"""
wrappers.expect_pager(context, "BEGIN\r\n", timeout=2)
@then("we see transaction rolled back")
def step_see_transaction_rolled_back(context):
"""
Wait to see transaction rollback.
"""
wrappers.expect_pager(context, "ROLLBACK\r\n", timeout=2)

2
tests/features/steps/expanded.py
View file

@ -16,7 +16,7 @@ def step_prepare_data(context):
context.cli.sendline("drop table if exists a;")
wrappers.expect_exact(
context,
"You're about to run a destructive command.\r\nDo you want to proceed? (y/n):",
"You're about to run a destructive command.\r\nDo you want to proceed? [y/N]:",
timeout=2,
)
context.cli.sendline("y")

5
tests/features/steps/wrappers.py
View file

@ -3,10 +3,7 @@ import pexpect
from pgcli.main import COLOR_CODE_REGEX
import textwrap
try:
from StringIO import StringIO
except ImportError:
from io import StringIO
from io import StringIO
def expect_exact(context, expected, timeout):

4
tests/formatter/test_sqlformatter.py
View file

@ -34,7 +34,7 @@ def test_output_sql_insert():
"Jackson",
"jackson_test@gmail.com",
"132454789",
"",
None,
"2022-09-09 19:44:32.712343+08",
"2022-09-09 19:44:32.712343+08",
]
@ -58,7 +58,7 @@ def test_output_sql_insert():
output_list = [l for l in output]
expected = [
'INSERT INTO "user" ("id", "name", "email", "phone", "description", "created_at", "updated_at") VALUES',
" ('1', 'Jackson', 'jackson_test@gmail.com', '132454789', '', "
" ('1', 'Jackson', 'jackson_test@gmail.com', '132454789', NULL, "
+ "'2022-09-09 19:44:32.712343+08', '2022-09-09 19:44:32.712343+08')",
";",
]

56
tests/parseutils/test_parseutils.py
View file

@ -1,5 +1,10 @@
import pytest
from pgcli.packages.parseutils import is_destructive
from pgcli.packages.parseutils import (
is_destructive,
parse_destructive_warning,
BASE_KEYWORDS,
ALL_KEYWORDS,
)
from pgcli.packages.parseutils.tables import extract_tables
from pgcli.packages.parseutils.utils import find_prev_keyword, is_open_quote
@ -263,18 +268,43 @@ def test_is_open_quote__open(sql):
@pytest.mark.parametrize(
("sql", "warning_level", "expected"),
("sql", "keywords", "expected"),
[
("update abc set x = 1", "all", True),
("update abc set x = 1 where y = 2", "all", True),
("update abc set x = 1", "moderate", True),
("update abc set x = 1 where y = 2", "moderate", False),
("select x, y, z from abc", "all", False),
("drop abc", "all", True),
("alter abc", "all", True),
("delete abc", "all", True),
("truncate abc", "all", True),
("update abc set x = 1", ALL_KEYWORDS, True),
("update abc set x = 1 where y = 2", ALL_KEYWORDS, True),
("update abc set x = 1", BASE_KEYWORDS, True),
("update abc set x = 1 where y = 2", BASE_KEYWORDS, False),
("select x, y, z from abc", ALL_KEYWORDS, False),
("drop abc", ALL_KEYWORDS, True),
("alter abc", ALL_KEYWORDS, True),
("delete abc", ALL_KEYWORDS, True),
("truncate abc", ALL_KEYWORDS, True),
("insert into abc values (1, 2, 3)", ALL_KEYWORDS, False),
("insert into abc values (1, 2, 3)", BASE_KEYWORDS, False),
("insert into abc values (1, 2, 3)", ["insert"], True),
("insert into abc values (1, 2, 3)", ["insert"], True),
],
)
def test_is_destructive(sql, warning_level, expected):
assert is_destructive(sql, warning_level=warning_level) == expected
def test_is_destructive(sql, keywords, expected):
assert is_destructive(sql, keywords) == expected
@pytest.mark.parametrize(
("warning_level", "expected"),
[
("true", ALL_KEYWORDS),
("false", []),
("all", ALL_KEYWORDS),
("moderate", BASE_KEYWORDS),
("off", []),
("", []),
(None, []),
(ALL_KEYWORDS, ALL_KEYWORDS),
(BASE_KEYWORDS, BASE_KEYWORDS),
("insert", ["insert"]),
("drop,alter,delete", ["drop", "alter", "delete"]),
(["drop", "alter", "delete"], ["drop", "alter", "delete"]),
],
)
def test_parse_destructive_warning(warning_level, expected):
assert parse_destructive_warning(warning_level) == expected

18
tests/test_main.py
View file

@ -216,7 +216,6 @@ def pset_pager_mocks():
with mock.patch("pgcli.main.click.echo") as mock_echo, mock.patch(
"pgcli.main.click.echo_via_pager"
) as mock_echo_via_pager, mock.patch.object(cli, "prompt_app") as mock_app:
yield cli, mock_echo, mock_echo_via_pager, mock_app
@ -297,6 +296,22 @@ def test_i_works(tmpdir, executor):
run(executor, statement, pgspecial=cli.pgspecial)
@dbtest
def test_echo_works(executor):
cli = PGCli(pgexecute=executor)
statement = r"\echo asdf"
result = run(executor, statement, pgspecial=cli.pgspecial)
assert result == ["asdf"]
@dbtest
def test_qecho_works(executor):
cli = PGCli(pgexecute=executor)
statement = r"\qecho asdf"
result = run(executor, statement, pgspecial=cli.pgspecial)
assert result == ["asdf"]
@dbtest
def test_watch_works(executor):
cli = PGCli(pgexecute=executor)
@ -371,7 +386,6 @@ def test_quoted_db_uri(tmpdir):
def test_pg_service_file(tmpdir):
with mock.patch.object(PGCli, "connect") as mock_connect:
cli = PGCli(pgclirc_file=str(tmpdir.join("rcfile")))
with open(tmpdir.join(".pg_service.conf").strpath, "w") as service_conf:

76
tests/test_pgcompleter.py Normal file
View file

@ -0,0 +1,76 @@
import pytest
from pgcli import pgcompleter
def test_load_alias_map_file_missing_file():
with pytest.raises(
pgcompleter.InvalidMapFile,
match=r"Cannot read alias_map_file - /path/to/non-existent/file.json does not exist$",
):
pgcompleter.load_alias_map_file("/path/to/non-existent/file.json")
def test_load_alias_map_file_invalid_json(tmp_path):
fpath = tmp_path / "foo.json"
fpath.write_text("this is not valid json")
with pytest.raises(pgcompleter.InvalidMapFile, match=r".*is not valid json$"):
pgcompleter.load_alias_map_file(str(fpath))
@pytest.mark.parametrize(
"table_name, alias",
[
("SomE_Table", "SET"),
("SOmeTabLe", "SOTL"),
("someTable", "T"),
],
)
def test_generate_alias_uses_upper_case_letters_from_name(table_name, alias):
assert pgcompleter.generate_alias(table_name) == alias
@pytest.mark.parametrize(
"table_name, alias",
[
("some_tab_le", "stl"),
("s_ome_table", "sot"),
("sometable", "s"),
],
)
def test_generate_alias_uses_first_char_and_every_preceded_by_underscore(
table_name, alias
):
assert pgcompleter.generate_alias(table_name) == alias
@pytest.mark.parametrize(
"table_name, alias_map, alias",
[
("some_table", {"some_table": "my_alias"}, "my_alias"),
],
)
def test_generate_alias_can_use_alias_map(table_name, alias_map, alias):
assert pgcompleter.generate_alias(table_name, alias_map) == alias
@pytest.mark.parametrize(
"table_name, alias_map, alias",
[
("SomeTable", {"SomeTable": "my_alias"}, "my_alias"),
],
)
def test_generate_alias_prefers_alias_over_upper_case_name(
table_name, alias_map, alias
):
assert pgcompleter.generate_alias(table_name, alias_map) == alias
@pytest.mark.parametrize(
"table_name, alias",
[
("Some_tablE", "SE"),
("SomeTab_le", "ST"),
],
)
def test_generate_alias_prefers_upper_case_name_over_underscore_name(table_name, alias):
assert pgcompleter.generate_alias(table_name) == alias

155
tests/test_pgexecute.py
View file

@ -304,9 +304,7 @@ def test_execute_from_commented_file_that_executes_another_file(
@dbtest
def test_execute_commented_first_line_and_special(executor, pgspecial, tmpdir):
# https://github.com/dbcli/pgcli/issues/1362
# just some base caes that should work also
# just some base cases that should work also
statement = "--comment\nselect now();"
result = run(executor, statement, pgspecial=pgspecial)
assert result != None
@ -317,23 +315,43 @@ def test_execute_commented_first_line_and_special(executor, pgspecial, tmpdir):
assert result != None
assert result[1].find("now") >= 0
statement = "/*comment\ncomment line2*/\nselect now();"
result = run(executor, statement, pgspecial=pgspecial)
assert result != None
assert result[1].find("now") >= 0
# https://github.com/dbcli/pgcli/issues/1362
statement = "--comment\n\\h"
result = run(executor, statement, pgspecial=pgspecial)
assert result != None
assert result[1].find("ALTER") >= 0
assert result[1].find("ABORT") >= 0
statement = "--comment1\n--comment2\n\\h"
result = run(executor, statement, pgspecial=pgspecial)
assert result != None
assert result[1].find("ALTER") >= 0
assert result[1].find("ABORT") >= 0
statement = "/*comment*/\n\h;"
result = run(executor, statement, pgspecial=pgspecial)
assert result != None
assert result[1].find("ALTER") >= 0
assert result[1].find("ABORT") >= 0
statement = """/*comment1
comment2*/
\h"""
result = run(executor, statement, pgspecial=pgspecial)
assert result != None
assert result[1].find("ALTER") >= 0
assert result[1].find("ABORT") >= 0
statement = """/*comment1
comment2*/
/*comment 3
comment4*/
\\h"""
result = run(executor, statement, pgspecial=pgspecial)
assert result != None
assert result[1].find("ALTER") >= 0
assert result[1].find("ABORT") >= 0
statement = " /*comment*/\n\h;"
result = run(executor, statement, pgspecial=pgspecial)
assert result != None
@ -352,6 +370,126 @@ def test_execute_commented_first_line_and_special(executor, pgspecial, tmpdir):
assert result[1].find("ALTER") >= 0
assert result[1].find("ABORT") >= 0
statement = """\\h /*comment4 */"""
result = run(executor, statement, pgspecial=pgspecial)
print(result)
assert result != None
assert result[0].find("No help") >= 0
# TODO: we probably don't want to do this but sqlparse is not parsing things well
# we relly want it to find help but right now, sqlparse isn't dropping the /*comment*/
# style comments after command
statement = """/*comment1*/
\h
/*comment4 */"""
result = run(executor, statement, pgspecial=pgspecial)
assert result != None
assert result[0].find("No help") >= 0
# TODO: same for this one
statement = """/*comment1
comment3
comment2*/
\\h
/*comment4
comment5
comment6*/"""
result = run(executor, statement, pgspecial=pgspecial)
assert result != None
assert result[0].find("No help") >= 0
@dbtest
def test_execute_commented_first_line_and_normal(executor, pgspecial, tmpdir):
# https://github.com/dbcli/pgcli/issues/1403
# just some base cases that should work also
statement = "--comment\nselect now();"
result = run(executor, statement, pgspecial=pgspecial)
assert result != None
assert result[1].find("now") >= 0
statement = "/*comment*/\nselect now();"
result = run(executor, statement, pgspecial=pgspecial)
assert result != None
assert result[1].find("now") >= 0
# this simulates the original error (1403) without having to add/drop tables
# since it was just an error on reading input files and not the actual
# command itself
# test that the statement works
statement = """VALUES (1, 'one'), (2, 'two'), (3, 'three');"""
result = run(executor, statement, pgspecial=pgspecial)
assert result != None
assert result[5].find("three") >= 0
# test the statement with a \n in the middle
statement = """VALUES (1, 'one'),\n (2, 'two'), (3, 'three');"""
result = run(executor, statement, pgspecial=pgspecial)
assert result != None
assert result[5].find("three") >= 0
# test the statement with a newline in the middle
statement = """VALUES (1, 'one'),
(2, 'two'), (3, 'three');"""
result = run(executor, statement, pgspecial=pgspecial)
assert result != None
assert result[5].find("three") >= 0
# now add a single comment line
statement = """--comment\nVALUES (1, 'one'), (2, 'two'), (3, 'three');"""
result = run(executor, statement, pgspecial=pgspecial)
assert result != None
assert result[5].find("three") >= 0
# doing without special char \n
statement = """--comment
VALUES (1,'one'),
(2, 'two'), (3, 'three');"""
result = run(executor, statement, pgspecial=pgspecial)
assert result != None
assert result[5].find("three") >= 0
# two comment lines
statement = """--comment\n--comment2\nVALUES (1,'one'), (2, 'two'), (3, 'three');"""
result = run(executor, statement, pgspecial=pgspecial)
assert result != None
assert result[5].find("three") >= 0
# doing without special char \n
statement = """--comment
--comment2
VALUES (1,'one'), (2, 'two'), (3, 'three');
"""
result = run(executor, statement, pgspecial=pgspecial)
assert result != None
assert result[5].find("three") >= 0
# multiline comment + newline in middle of the statement
statement = """/*comment
comment2
comment3*/
VALUES (1,'one'),
(2, 'two'), (3, 'three');"""
result = run(executor, statement, pgspecial=pgspecial)
assert result != None
assert result[5].find("three") >= 0
# multiline comment + newline in middle of the statement
# + comments after the statement
statement = """/*comment
comment2
comment3*/
VALUES (1,'one'),
(2, 'two'), (3, 'three');
--comment4
--comment5"""
result = run(executor, statement, pgspecial=pgspecial)
assert result != None
assert result[5].find("three") >= 0
@dbtest
def test_multiple_queries_same_line(executor):
@ -558,6 +696,7 @@ def test_view_definition(executor):
run(executor, "create view vw1 AS SELECT * FROM tbl1")
run(executor, "create materialized view mvw1 AS SELECT * FROM tbl1")
result = executor.view_definition("vw1")
assert 'VIEW "public"."vw1" AS' in result
assert "FROM tbl1" in result
# import pytest; pytest.set_trace()
result = executor.view_definition("mvw1")

9
tests/test_prompt_utils.py
View file

@ -7,4 +7,11 @@ def test_confirm_destructive_query_notty():
stdin = click.get_text_stream("stdin")
if not stdin.isatty():
sql = "drop database foo;"
assert confirm_destructive_query(sql, "all") is None
assert confirm_destructive_query(sql, [], None) is None
def test_confirm_destructive_query_with_alias():
stdin = click.get_text_stream("stdin")
if not stdin.isatty():
sql = "drop database foo;"
assert confirm_destructive_query(sql, ["drop"], "test") is None