Adding upstream version 1.29.2.
Signed-off-by: Daniel Baumann <daniel@debian.org>
This commit is contained in:
parent
5bd6a68e8c
commit
f9065f1bef
GPG key ID:
FBB4F0E80A80222F
68 changed files with 3723 additions and 3336 deletions
|
|
@ -17,11 +17,11 @@ def test_set_get_pager():
|
|||
assert mycli.packages.special.is_pager_enabled()
|
||||
mycli.packages.special.set_pager_enabled(False)
|
||||
assert not mycli.packages.special.is_pager_enabled()
|
||||
mycli.packages.special.set_pager('less')
|
||||
assert os.environ['PAGER'] == "less"
|
||||
mycli.packages.special.set_pager("less")
|
||||
assert os.environ["PAGER"] == "less"
|
||||
mycli.packages.special.set_pager(False)
|
||||
assert os.environ['PAGER'] == "less"
|
||||
del os.environ['PAGER']
|
||||
assert os.environ["PAGER"] == "less"
|
||||
del os.environ["PAGER"]
|
||||
mycli.packages.special.set_pager(False)
|
||||
mycli.packages.special.disable_pager()
|
||||
assert not mycli.packages.special.is_pager_enabled()
|
||||
|
|
@ -42,45 +42,44 @@ def test_set_get_expanded_output():
|
|||
|
||||
|
||||
def test_editor_command():
|
||||
assert mycli.packages.special.editor_command(r'hello\e')
|
||||
assert mycli.packages.special.editor_command(r'\ehello')
|
||||
assert not mycli.packages.special.editor_command(r'hello')
|
||||
assert mycli.packages.special.editor_command(r"hello\e")
|
||||
assert mycli.packages.special.editor_command(r"\ehello")
|
||||
assert not mycli.packages.special.editor_command(r"hello")
|
||||
|
||||
assert mycli.packages.special.get_filename(r'\e filename') == "filename"
|
||||
assert mycli.packages.special.get_filename(r"\e filename") == "filename"
|
||||
|
||||
os.environ['EDITOR'] = 'true'
|
||||
os.environ['VISUAL'] = 'true'
|
||||
os.environ["EDITOR"] = "true"
|
||||
os.environ["VISUAL"] = "true"
|
||||
# Set the editor to Notepad on Windows
|
||||
if os.name != 'nt':
|
||||
mycli.packages.special.open_external_editor(sql=r'select 1') == "select 1"
|
||||
if os.name != "nt":
|
||||
mycli.packages.special.open_external_editor(sql=r"select 1") == "select 1"
|
||||
else:
|
||||
pytest.skip('Skipping on Windows platform.')
|
||||
|
||||
pytest.skip("Skipping on Windows platform.")
|
||||
|
||||
|
||||
def test_tee_command():
|
||||
mycli.packages.special.write_tee(u"hello world") # write without file set
|
||||
mycli.packages.special.write_tee("hello world") # write without file set
|
||||
# keep Windows from locking the file with delete=False
|
||||
with tempfile.NamedTemporaryFile(delete=False) as f:
|
||||
mycli.packages.special.execute(None, u"tee " + f.name)
|
||||
mycli.packages.special.write_tee(u"hello world")
|
||||
if os.name=='nt':
|
||||
mycli.packages.special.execute(None, "tee " + f.name)
|
||||
mycli.packages.special.write_tee("hello world")
|
||||
if os.name == "nt":
|
||||
assert f.read() == b"hello world\r\n"
|
||||
else:
|
||||
assert f.read() == b"hello world\n"
|
||||
|
||||
mycli.packages.special.execute(None, u"tee -o " + f.name)
|
||||
mycli.packages.special.write_tee(u"hello world")
|
||||
mycli.packages.special.execute(None, "tee -o " + f.name)
|
||||
mycli.packages.special.write_tee("hello world")
|
||||
f.seek(0)
|
||||
if os.name=='nt':
|
||||
if os.name == "nt":
|
||||
assert f.read() == b"hello world\r\n"
|
||||
else:
|
||||
assert f.read() == b"hello world\n"
|
||||
|
||||
mycli.packages.special.execute(None, u"notee")
|
||||
mycli.packages.special.write_tee(u"hello world")
|
||||
mycli.packages.special.execute(None, "notee")
|
||||
mycli.packages.special.write_tee("hello world")
|
||||
f.seek(0)
|
||||
if os.name=='nt':
|
||||
if os.name == "nt":
|
||||
assert f.read() == b"hello world\r\n"
|
||||
else:
|
||||
assert f.read() == b"hello world\n"
|
||||
|
|
@ -92,52 +91,49 @@ def test_tee_command():
|
|||
os.remove(f.name)
|
||||
except Exception as e:
|
||||
print(f"An error occurred while attempting to delete the file: {e}")
|
||||
|
||||
|
||||
|
||||
def test_tee_command_error():
|
||||
with pytest.raises(TypeError):
|
||||
mycli.packages.special.execute(None, 'tee')
|
||||
mycli.packages.special.execute(None, "tee")
|
||||
|
||||
with pytest.raises(OSError):
|
||||
with tempfile.NamedTemporaryFile() as f:
|
||||
os.chmod(f.name, stat.S_IRUSR | stat.S_IRGRP | stat.S_IROTH)
|
||||
mycli.packages.special.execute(None, 'tee {}'.format(f.name))
|
||||
mycli.packages.special.execute(None, "tee {}".format(f.name))
|
||||
|
||||
|
||||
@dbtest
|
||||
|
||||
@pytest.mark.skipif(os.name == "nt", reason="Bug: fails on Windows, needs fixing, singleton of FQ not working right")
|
||||
def test_favorite_query():
|
||||
with db_connection().cursor() as cur:
|
||||
query = u'select "✔"'
|
||||
mycli.packages.special.execute(cur, u'\\fs check {0}'.format(query))
|
||||
assert next(mycli.packages.special.execute(
|
||||
cur, u'\\f check'))[0] == "> " + query
|
||||
query = 'select "✔"'
|
||||
mycli.packages.special.execute(cur, "\\fs check {0}".format(query))
|
||||
assert next(mycli.packages.special.execute(cur, "\\f check"))[0] == "> " + query
|
||||
|
||||
|
||||
def test_once_command():
|
||||
with pytest.raises(TypeError):
|
||||
mycli.packages.special.execute(None, u"\\once")
|
||||
mycli.packages.special.execute(None, "\\once")
|
||||
|
||||
with pytest.raises(OSError):
|
||||
mycli.packages.special.execute(None, u"\\once /proc/access-denied")
|
||||
mycli.packages.special.execute(None, "\\once /proc/access-denied")
|
||||
|
||||
mycli.packages.special.write_once(u"hello world") # write without file set
|
||||
mycli.packages.special.write_once("hello world") # write without file set
|
||||
# keep Windows from locking the file with delete=False
|
||||
with tempfile.NamedTemporaryFile(delete=False) as f:
|
||||
mycli.packages.special.execute(None, u"\\once " + f.name)
|
||||
mycli.packages.special.write_once(u"hello world")
|
||||
if os.name=='nt':
|
||||
mycli.packages.special.execute(None, "\\once " + f.name)
|
||||
mycli.packages.special.write_once("hello world")
|
||||
if os.name == "nt":
|
||||
assert f.read() == b"hello world\r\n"
|
||||
else:
|
||||
assert f.read() == b"hello world\n"
|
||||
|
||||
mycli.packages.special.execute(None, u"\\once -o " + f.name)
|
||||
mycli.packages.special.write_once(u"hello world line 1")
|
||||
mycli.packages.special.write_once(u"hello world line 2")
|
||||
mycli.packages.special.execute(None, "\\once -o " + f.name)
|
||||
mycli.packages.special.write_once("hello world line 1")
|
||||
mycli.packages.special.write_once("hello world line 2")
|
||||
f.seek(0)
|
||||
if os.name=='nt':
|
||||
if os.name == "nt":
|
||||
assert f.read() == b"hello world line 1\r\nhello world line 2\r\n"
|
||||
else:
|
||||
assert f.read() == b"hello world line 1\nhello world line 2\n"
|
||||
|
|
@ -151,52 +147,47 @@ def test_once_command():
|
|||
|
||||
def test_pipe_once_command():
|
||||
with pytest.raises(IOError):
|
||||
mycli.packages.special.execute(None, u"\\pipe_once")
|
||||
mycli.packages.special.execute(None, "\\pipe_once")
|
||||
|
||||
with pytest.raises(OSError):
|
||||
mycli.packages.special.execute(
|
||||
None, u"\\pipe_once /proc/access-denied")
|
||||
mycli.packages.special.execute(None, "\\pipe_once /proc/access-denied")
|
||||
|
||||
if os.name == 'nt':
|
||||
mycli.packages.special.execute(None, u'\\pipe_once python -c "import sys; print(len(sys.stdin.read().strip()))"')
|
||||
mycli.packages.special.write_once(u"hello world")
|
||||
if os.name == "nt":
|
||||
mycli.packages.special.execute(None, '\\pipe_once python -c "import sys; print(len(sys.stdin.read().strip()))"')
|
||||
mycli.packages.special.write_once("hello world")
|
||||
mycli.packages.special.unset_pipe_once_if_written()
|
||||
else:
|
||||
mycli.packages.special.execute(None, u"\\pipe_once wc")
|
||||
mycli.packages.special.write_once(u"hello world")
|
||||
mycli.packages.special.unset_pipe_once_if_written()
|
||||
# how to assert on wc output?
|
||||
with tempfile.NamedTemporaryFile() as f:
|
||||
mycli.packages.special.execute(None, "\\pipe_once tee " + f.name)
|
||||
mycli.packages.special.write_pipe_once("hello world")
|
||||
mycli.packages.special.unset_pipe_once_if_written()
|
||||
f.seek(0)
|
||||
assert f.read() == b"hello world\n"
|
||||
|
||||
|
||||
def test_parseargfile():
|
||||
"""Test that parseargfile expands the user directory."""
|
||||
expected = {'file': os.path.join(os.path.expanduser('~'), 'filename'),
|
||||
'mode': 'a'}
|
||||
|
||||
if os.name=='nt':
|
||||
assert expected == mycli.packages.special.iocommands.parseargfile(
|
||||
'~\\filename')
|
||||
else:
|
||||
assert expected == mycli.packages.special.iocommands.parseargfile(
|
||||
'~/filename')
|
||||
expected = {"file": os.path.join(os.path.expanduser("~"), "filename"), "mode": "a"}
|
||||
|
||||
expected = {'file': os.path.join(os.path.expanduser('~'), 'filename'),
|
||||
'mode': 'w'}
|
||||
if os.name=='nt':
|
||||
assert expected == mycli.packages.special.iocommands.parseargfile(
|
||||
'-o ~\\filename')
|
||||
if os.name == "nt":
|
||||
assert expected == mycli.packages.special.iocommands.parseargfile("~\\filename")
|
||||
else:
|
||||
assert expected == mycli.packages.special.iocommands.parseargfile(
|
||||
'-o ~/filename')
|
||||
assert expected == mycli.packages.special.iocommands.parseargfile("~/filename")
|
||||
|
||||
expected = {"file": os.path.join(os.path.expanduser("~"), "filename"), "mode": "w"}
|
||||
if os.name == "nt":
|
||||
assert expected == mycli.packages.special.iocommands.parseargfile("-o ~\\filename")
|
||||
else:
|
||||
assert expected == mycli.packages.special.iocommands.parseargfile("-o ~/filename")
|
||||
|
||||
|
||||
def test_parseargfile_no_file():
|
||||
"""Test that parseargfile raises a TypeError if there is no filename."""
|
||||
with pytest.raises(TypeError):
|
||||
mycli.packages.special.iocommands.parseargfile('')
|
||||
mycli.packages.special.iocommands.parseargfile("")
|
||||
|
||||
with pytest.raises(TypeError):
|
||||
mycli.packages.special.iocommands.parseargfile('-o ')
|
||||
mycli.packages.special.iocommands.parseargfile("-o ")
|
||||
|
||||
|
||||
@dbtest
|
||||
|
|
@ -205,11 +196,9 @@ def test_watch_query_iteration():
|
|||
the desired query and returns the given results."""
|
||||
expected_value = "1"
|
||||
query = "SELECT {0!s}".format(expected_value)
|
||||
expected_title = '> {0!s}'.format(query)
|
||||
expected_title = "> {0!s}".format(query)
|
||||
with db_connection().cursor() as cur:
|
||||
result = next(mycli.packages.special.iocommands.watch_query(
|
||||
arg=query, cur=cur
|
||||
))
|
||||
result = next(mycli.packages.special.iocommands.watch_query(arg=query, cur=cur))
|
||||
assert result[0] == expected_title
|
||||
assert result[2][0] == expected_value
|
||||
|
||||
|
|
@ -230,14 +219,12 @@ def test_watch_query_full():
|
|||
wait_interval = 1
|
||||
expected_value = "1"
|
||||
query = "SELECT {0!s}".format(expected_value)
|
||||
expected_title = '> {0!s}'.format(query)
|
||||
expected_title = "> {0!s}".format(query)
|
||||
expected_results = 4
|
||||
ctrl_c_process = send_ctrl_c(wait_interval)
|
||||
with db_connection().cursor() as cur:
|
||||
results = list(
|
||||
result for result in mycli.packages.special.iocommands.watch_query(
|
||||
arg='{0!s} {1!s}'.format(watch_seconds, query), cur=cur
|
||||
)
|
||||
result for result in mycli.packages.special.iocommands.watch_query(arg="{0!s} {1!s}".format(watch_seconds, query), cur=cur)
|
||||
)
|
||||
ctrl_c_process.join(1)
|
||||
assert len(results) == expected_results
|
||||
|
|
@ -247,14 +234,12 @@ def test_watch_query_full():
|
|||
|
||||
|
||||
@dbtest
|
||||
@patch('click.clear')
|
||||
@patch("click.clear")
|
||||
def test_watch_query_clear(clear_mock):
|
||||
"""Test that the screen is cleared with the -c flag of `watch` command
|
||||
before execute the query."""
|
||||
with db_connection().cursor() as cur:
|
||||
watch_gen = mycli.packages.special.iocommands.watch_query(
|
||||
arg='0.1 -c select 1;', cur=cur
|
||||
)
|
||||
watch_gen = mycli.packages.special.iocommands.watch_query(arg="0.1 -c select 1;", cur=cur)
|
||||
assert not clear_mock.called
|
||||
next(watch_gen)
|
||||
assert clear_mock.called
|
||||
|
|
@ -271,19 +256,20 @@ def test_watch_query_bad_arguments():
|
|||
watch_query = mycli.packages.special.iocommands.watch_query
|
||||
with db_connection().cursor() as cur:
|
||||
with pytest.raises(ProgrammingError):
|
||||
next(watch_query('a select 1;', cur=cur))
|
||||
next(watch_query("a select 1;", cur=cur))
|
||||
with pytest.raises(ProgrammingError):
|
||||
next(watch_query('-a select 1;', cur=cur))
|
||||
next(watch_query("-a select 1;", cur=cur))
|
||||
with pytest.raises(ProgrammingError):
|
||||
next(watch_query('1 -a select 1;', cur=cur))
|
||||
next(watch_query("1 -a select 1;", cur=cur))
|
||||
with pytest.raises(ProgrammingError):
|
||||
next(watch_query('-c -a select 1;', cur=cur))
|
||||
next(watch_query("-c -a select 1;", cur=cur))
|
||||
|
||||
|
||||
@dbtest
|
||||
@patch('click.clear')
|
||||
@patch("click.clear")
|
||||
def test_watch_query_interval_clear(clear_mock):
|
||||
"""Test `watch` command with interval and clear flag."""
|
||||
|
||||
def test_asserts(gen):
|
||||
clear_mock.reset_mock()
|
||||
start = time()
|
||||
|
|
@ -296,46 +282,32 @@ def test_watch_query_interval_clear(clear_mock):
|
|||
seconds = 1.0
|
||||
watch_query = mycli.packages.special.iocommands.watch_query
|
||||
with db_connection().cursor() as cur:
|
||||
test_asserts(watch_query('{0!s} -c select 1;'.format(seconds),
|
||||
cur=cur))
|
||||
test_asserts(watch_query('-c {0!s} select 1;'.format(seconds),
|
||||
cur=cur))
|
||||
test_asserts(watch_query("{0!s} -c select 1;".format(seconds), cur=cur))
|
||||
test_asserts(watch_query("-c {0!s} select 1;".format(seconds), cur=cur))
|
||||
|
||||
|
||||
def test_split_sql_by_delimiter():
|
||||
for delimiter_str in (';', '$', '😀'):
|
||||
for delimiter_str in (";", "$", "😀"):
|
||||
mycli.packages.special.set_delimiter(delimiter_str)
|
||||
sql_input = "select 1{} select \ufffc2".format(delimiter_str)
|
||||
queries = (
|
||||
"select 1",
|
||||
"select \ufffc2"
|
||||
)
|
||||
for query, parsed_query in zip(
|
||||
queries, mycli.packages.special.split_queries(sql_input)):
|
||||
assert(query == parsed_query)
|
||||
queries = ("select 1", "select \ufffc2")
|
||||
for query, parsed_query in zip(queries, mycli.packages.special.split_queries(sql_input)):
|
||||
assert query == parsed_query
|
||||
|
||||
|
||||
def test_switch_delimiter_within_query():
|
||||
mycli.packages.special.set_delimiter(';')
|
||||
mycli.packages.special.set_delimiter(";")
|
||||
sql_input = "select 1; delimiter $$ select 2 $$ select 3 $$"
|
||||
queries = (
|
||||
"select 1",
|
||||
"delimiter $$ select 2 $$ select 3 $$",
|
||||
"select 2",
|
||||
"select 3"
|
||||
)
|
||||
for query, parsed_query in zip(
|
||||
queries,
|
||||
mycli.packages.special.split_queries(sql_input)):
|
||||
assert(query == parsed_query)
|
||||
queries = ("select 1", "delimiter $$ select 2 $$ select 3 $$", "select 2", "select 3")
|
||||
for query, parsed_query in zip(queries, mycli.packages.special.split_queries(sql_input)):
|
||||
assert query == parsed_query
|
||||
|
||||
|
||||
def test_set_delimiter():
|
||||
|
||||
for delim in ('foo', 'bar'):
|
||||
for delim in ("foo", "bar"):
|
||||
mycli.packages.special.set_delimiter(delim)
|
||||
assert mycli.packages.special.get_current_delimiter() == delim
|
||||
|
||||
|
||||
def teardown_function():
|
||||
mycli.packages.special.set_delimiter(';')
|
||||
mycli.packages.special.set_delimiter(";")
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue