Merging upstream version 1.29.2.
Signed-off-by: Daniel Baumann <daniel@debian.org>
This commit is contained in:
parent
3180ec4213
commit
f15fe4f59f
GPG key ID:
FBB4F0E80A80222F
68 changed files with 3723 additions and 3336 deletions
|
|
@ -13,52 +13,62 @@ from textwrap import dedent
|
|||
from collections import namedtuple
|
||||
|
||||
from tempfile import NamedTemporaryFile
|
||||
from textwrap import dedent
|
||||
|
||||
|
||||
test_dir = os.path.abspath(os.path.dirname(__file__))
|
||||
project_dir = os.path.dirname(test_dir)
|
||||
default_config_file = os.path.join(project_dir, 'test', 'myclirc')
|
||||
login_path_file = os.path.join(test_dir, 'mylogin.cnf')
|
||||
default_config_file = os.path.join(project_dir, "test", "myclirc")
|
||||
login_path_file = os.path.join(test_dir, "mylogin.cnf")
|
||||
|
||||
os.environ['MYSQL_TEST_LOGIN_FILE'] = login_path_file
|
||||
CLI_ARGS = ['--user', USER, '--host', HOST, '--port', PORT,
|
||||
'--password', PASSWORD, '--myclirc', default_config_file,
|
||||
'--defaults-file', default_config_file,
|
||||
'mycli_test_db']
|
||||
os.environ["MYSQL_TEST_LOGIN_FILE"] = login_path_file
|
||||
CLI_ARGS = [
|
||||
"--user",
|
||||
USER,
|
||||
"--host",
|
||||
HOST,
|
||||
"--port",
|
||||
PORT,
|
||||
"--password",
|
||||
PASSWORD,
|
||||
"--myclirc",
|
||||
default_config_file,
|
||||
"--defaults-file",
|
||||
default_config_file,
|
||||
"mycli_test_db",
|
||||
]
|
||||
|
||||
|
||||
@dbtest
|
||||
def test_execute_arg(executor):
|
||||
run(executor, 'create table test (a text)')
|
||||
run(executor, "create table test (a text)")
|
||||
run(executor, 'insert into test values("abc")')
|
||||
|
||||
sql = 'select * from test;'
|
||||
sql = "select * from test;"
|
||||
runner = CliRunner()
|
||||
result = runner.invoke(cli, args=CLI_ARGS + ['-e', sql])
|
||||
result = runner.invoke(cli, args=CLI_ARGS + ["-e", sql])
|
||||
|
||||
assert result.exit_code == 0
|
||||
assert 'abc' in result.output
|
||||
assert "abc" in result.output
|
||||
|
||||
result = runner.invoke(cli, args=CLI_ARGS + ['--execute', sql])
|
||||
result = runner.invoke(cli, args=CLI_ARGS + ["--execute", sql])
|
||||
|
||||
assert result.exit_code == 0
|
||||
assert 'abc' in result.output
|
||||
assert "abc" in result.output
|
||||
|
||||
expected = 'a\nabc\n'
|
||||
expected = "a\nabc\n"
|
||||
|
||||
assert expected in result.output
|
||||
|
||||
|
||||
@dbtest
|
||||
def test_execute_arg_with_table(executor):
|
||||
run(executor, 'create table test (a text)')
|
||||
run(executor, "create table test (a text)")
|
||||
run(executor, 'insert into test values("abc")')
|
||||
|
||||
sql = 'select * from test;'
|
||||
sql = "select * from test;"
|
||||
runner = CliRunner()
|
||||
result = runner.invoke(cli, args=CLI_ARGS + ['-e', sql] + ['--table'])
|
||||
expected = '+-----+\n| a |\n+-----+\n| abc |\n+-----+\n'
|
||||
result = runner.invoke(cli, args=CLI_ARGS + ["-e", sql] + ["--table"])
|
||||
expected = "+-----+\n| a |\n+-----+\n| abc |\n+-----+\n"
|
||||
|
||||
assert result.exit_code == 0
|
||||
assert expected in result.output
|
||||
|
|
@ -66,12 +76,12 @@ def test_execute_arg_with_table(executor):
|
|||
|
||||
@dbtest
|
||||
def test_execute_arg_with_csv(executor):
|
||||
run(executor, 'create table test (a text)')
|
||||
run(executor, "create table test (a text)")
|
||||
run(executor, 'insert into test values("abc")')
|
||||
|
||||
sql = 'select * from test;'
|
||||
sql = "select * from test;"
|
||||
runner = CliRunner()
|
||||
result = runner.invoke(cli, args=CLI_ARGS + ['-e', sql] + ['--csv'])
|
||||
result = runner.invoke(cli, args=CLI_ARGS + ["-e", sql] + ["--csv"])
|
||||
expected = '"a"\n"abc"\n'
|
||||
|
||||
assert result.exit_code == 0
|
||||
|
|
@ -80,35 +90,29 @@ def test_execute_arg_with_csv(executor):
|
|||
|
||||
@dbtest
|
||||
def test_batch_mode(executor):
|
||||
run(executor, '''create table test(a text)''')
|
||||
run(executor, '''insert into test values('abc'), ('def'), ('ghi')''')
|
||||
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;\n" "select * from test limit 1;"
|
||||
|
||||
runner = CliRunner()
|
||||
result = runner.invoke(cli, args=CLI_ARGS, input=sql)
|
||||
|
||||
assert result.exit_code == 0
|
||||
assert 'count(*)\n3\na\nabc\n' in "".join(result.output)
|
||||
assert "count(*)\n3\na\nabc\n" in "".join(result.output)
|
||||
|
||||
|
||||
@dbtest
|
||||
def test_batch_mode_table(executor):
|
||||
run(executor, '''create table test(a text)''')
|
||||
run(executor, '''insert into test values('abc'), ('def'), ('ghi')''')
|
||||
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;\n" "select * from test limit 1;"
|
||||
|
||||
runner = CliRunner()
|
||||
result = runner.invoke(cli, args=CLI_ARGS + ['-t'], input=sql)
|
||||
result = runner.invoke(cli, args=CLI_ARGS + ["-t"], input=sql)
|
||||
|
||||
expected = (dedent("""\
|
||||
expected = dedent("""\
|
||||
+----------+
|
||||
| count(*) |
|
||||
+----------+
|
||||
|
|
@ -118,7 +122,7 @@ def test_batch_mode_table(executor):
|
|||
| a |
|
||||
+-----+
|
||||
| abc |
|
||||
+-----+"""))
|
||||
+-----+""")
|
||||
|
||||
assert result.exit_code == 0
|
||||
assert expected in result.output
|
||||
|
|
@ -126,14 +130,13 @@ def test_batch_mode_table(executor):
|
|||
|
||||
@dbtest
|
||||
def test_batch_mode_csv(executor):
|
||||
run(executor, '''create table test(a text, b text)''')
|
||||
run(executor,
|
||||
'''insert into test (a, b) values('abc', 'de\nf'), ('ghi', 'jkl')''')
|
||||
run(executor, """create table test(a text, b text)""")
|
||||
run(executor, """insert into test (a, b) values('abc', 'de\nf'), ('ghi', 'jkl')""")
|
||||
|
||||
sql = 'select * from test;'
|
||||
sql = "select * from test;"
|
||||
|
||||
runner = CliRunner()
|
||||
result = runner.invoke(cli, args=CLI_ARGS + ['--csv'], input=sql)
|
||||
result = runner.invoke(cli, args=CLI_ARGS + ["--csv"], input=sql)
|
||||
|
||||
expected = '"a","b"\n"abc","de\nf"\n"ghi","jkl"\n'
|
||||
|
||||
|
|
@ -150,15 +153,15 @@ def test_help_strings_end_with_periods():
|
|||
"""Make sure click options have help text that end with a period."""
|
||||
for param in cli.params:
|
||||
if isinstance(param, click.core.Option):
|
||||
assert hasattr(param, 'help')
|
||||
assert param.help.endswith('.')
|
||||
assert hasattr(param, "help")
|
||||
assert param.help.endswith(".")
|
||||
|
||||
|
||||
def test_command_descriptions_end_with_periods():
|
||||
"""Make sure that mycli commands' descriptions end with a period."""
|
||||
MyCli()
|
||||
for _, command in SPECIAL_COMMANDS.items():
|
||||
assert command[3].endswith('.')
|
||||
assert command[3].endswith(".")
|
||||
|
||||
|
||||
def output(monkeypatch, terminal_size, testdata, explicit_pager, expect_pager):
|
||||
|
|
@ -166,23 +169,23 @@ def output(monkeypatch, terminal_size, testdata, explicit_pager, expect_pager):
|
|||
clickoutput = ""
|
||||
m = MyCli(myclirc=default_config_file)
|
||||
|
||||
class TestOutput():
|
||||
class TestOutput:
|
||||
def get_size(self):
|
||||
size = namedtuple('Size', 'rows columns')
|
||||
size = namedtuple("Size", "rows columns")
|
||||
size.columns, size.rows = terminal_size
|
||||
return size
|
||||
|
||||
class TestExecute():
|
||||
host = 'test'
|
||||
user = 'test'
|
||||
dbname = 'test'
|
||||
server_info = ServerInfo.from_version_string('unknown')
|
||||
class TestExecute:
|
||||
host = "test"
|
||||
user = "test"
|
||||
dbname = "test"
|
||||
server_info = ServerInfo.from_version_string("unknown")
|
||||
port = 0
|
||||
|
||||
def server_type(self):
|
||||
return ['test']
|
||||
return ["test"]
|
||||
|
||||
class PromptBuffer():
|
||||
class PromptBuffer:
|
||||
output = TestOutput()
|
||||
|
||||
m.prompt_app = PromptBuffer()
|
||||
|
|
@ -199,8 +202,8 @@ def output(monkeypatch, terminal_size, testdata, explicit_pager, expect_pager):
|
|||
global clickoutput
|
||||
clickoutput += s + "\n"
|
||||
|
||||
monkeypatch.setattr(click, 'echo_via_pager', echo_via_pager)
|
||||
monkeypatch.setattr(click, 'secho', secho)
|
||||
monkeypatch.setattr(click, "echo_via_pager", echo_via_pager)
|
||||
monkeypatch.setattr(click, "secho", secho)
|
||||
m.output(testdata)
|
||||
if clickoutput.endswith("\n"):
|
||||
clickoutput = clickoutput[:-1]
|
||||
|
|
@ -208,59 +211,29 @@ def output(monkeypatch, terminal_size, testdata, explicit_pager, expect_pager):
|
|||
|
||||
|
||||
def test_conditional_pager(monkeypatch):
|
||||
testdata = "Lorem ipsum dolor sit amet consectetur adipiscing elit sed do".split(
|
||||
" ")
|
||||
testdata = "Lorem ipsum dolor sit amet consectetur adipiscing elit sed do".split(" ")
|
||||
# User didn't set pager, output doesn't fit screen -> pager
|
||||
output(
|
||||
monkeypatch,
|
||||
terminal_size=(5, 10),
|
||||
testdata=testdata,
|
||||
explicit_pager=False,
|
||||
expect_pager=True
|
||||
)
|
||||
output(monkeypatch, terminal_size=(5, 10), testdata=testdata, explicit_pager=False, expect_pager=True)
|
||||
# User didn't set pager, output fits screen -> no pager
|
||||
output(
|
||||
monkeypatch,
|
||||
terminal_size=(20, 20),
|
||||
testdata=testdata,
|
||||
explicit_pager=False,
|
||||
expect_pager=False
|
||||
)
|
||||
output(monkeypatch, terminal_size=(20, 20), testdata=testdata, explicit_pager=False, expect_pager=False)
|
||||
# User manually configured pager, output doesn't fit screen -> pager
|
||||
output(
|
||||
monkeypatch,
|
||||
terminal_size=(5, 10),
|
||||
testdata=testdata,
|
||||
explicit_pager=True,
|
||||
expect_pager=True
|
||||
)
|
||||
output(monkeypatch, terminal_size=(5, 10), testdata=testdata, explicit_pager=True, expect_pager=True)
|
||||
# User manually configured pager, output fit screen -> pager
|
||||
output(
|
||||
monkeypatch,
|
||||
terminal_size=(20, 20),
|
||||
testdata=testdata,
|
||||
explicit_pager=True,
|
||||
expect_pager=True
|
||||
)
|
||||
output(monkeypatch, terminal_size=(20, 20), testdata=testdata, explicit_pager=True, expect_pager=True)
|
||||
|
||||
SPECIAL_COMMANDS['nopager'].handler()
|
||||
output(
|
||||
monkeypatch,
|
||||
terminal_size=(5, 10),
|
||||
testdata=testdata,
|
||||
explicit_pager=False,
|
||||
expect_pager=False
|
||||
)
|
||||
SPECIAL_COMMANDS['pager'].handler('')
|
||||
SPECIAL_COMMANDS["nopager"].handler()
|
||||
output(monkeypatch, terminal_size=(5, 10), testdata=testdata, explicit_pager=False, expect_pager=False)
|
||||
SPECIAL_COMMANDS["pager"].handler("")
|
||||
|
||||
|
||||
def test_reserved_space_is_integer(monkeypatch):
|
||||
"""Make sure that reserved space is returned as an integer."""
|
||||
|
||||
def stub_terminal_size():
|
||||
return (5, 5)
|
||||
|
||||
with monkeypatch.context() as m:
|
||||
m.setattr(shutil, 'get_terminal_size', stub_terminal_size)
|
||||
m.setattr(shutil, "get_terminal_size", stub_terminal_size)
|
||||
mycli = MyCli()
|
||||
assert isinstance(mycli.get_reserved_space(), int)
|
||||
|
||||
|
|
@ -268,18 +241,20 @@ def test_reserved_space_is_integer(monkeypatch):
|
|||
def test_list_dsn():
|
||||
runner = CliRunner()
|
||||
# keep Windows from locking the file with delete=False
|
||||
with NamedTemporaryFile(mode="w",delete=False) as myclirc:
|
||||
myclirc.write(dedent("""\
|
||||
with NamedTemporaryFile(mode="w", delete=False) as myclirc:
|
||||
myclirc.write(
|
||||
dedent("""\
|
||||
[alias_dsn]
|
||||
test = mysql://test/test
|
||||
"""))
|
||||
""")
|
||||
)
|
||||
myclirc.flush()
|
||||
args = ['--list-dsn', '--myclirc', myclirc.name]
|
||||
args = ["--list-dsn", "--myclirc", myclirc.name]
|
||||
result = runner.invoke(cli, args=args)
|
||||
assert result.output == "test\n"
|
||||
result = runner.invoke(cli, args=args + ['--verbose'])
|
||||
result = runner.invoke(cli, args=args + ["--verbose"])
|
||||
assert result.output == "test : mysql://test/test\n"
|
||||
|
||||
|
||||
# delete=False means we should try to clean up
|
||||
try:
|
||||
if os.path.exists(myclirc.name):
|
||||
|
|
@ -287,41 +262,41 @@ def test_list_dsn():
|
|||
except Exception as e:
|
||||
print(f"An error occurred while attempting to delete the file: {e}")
|
||||
|
||||
|
||||
|
||||
|
||||
def test_prettify_statement():
|
||||
statement = 'SELECT 1'
|
||||
statement = "SELECT 1"
|
||||
m = MyCli()
|
||||
pretty_statement = m.handle_prettify_binding(statement)
|
||||
assert pretty_statement == 'SELECT\n 1;'
|
||||
assert pretty_statement == "SELECT\n 1;"
|
||||
|
||||
|
||||
def test_unprettify_statement():
|
||||
statement = 'SELECT\n 1'
|
||||
statement = "SELECT\n 1"
|
||||
m = MyCli()
|
||||
unpretty_statement = m.handle_unprettify_binding(statement)
|
||||
assert unpretty_statement == 'SELECT 1;'
|
||||
assert unpretty_statement == "SELECT 1;"
|
||||
|
||||
|
||||
def test_list_ssh_config():
|
||||
runner = CliRunner()
|
||||
# keep Windows from locking the file with delete=False
|
||||
with NamedTemporaryFile(mode="w",delete=False) as ssh_config:
|
||||
ssh_config.write(dedent("""\
|
||||
with NamedTemporaryFile(mode="w", delete=False) as ssh_config:
|
||||
ssh_config.write(
|
||||
dedent("""\
|
||||
Host test
|
||||
Hostname test.example.com
|
||||
User joe
|
||||
Port 22222
|
||||
IdentityFile ~/.ssh/gateway
|
||||
"""))
|
||||
""")
|
||||
)
|
||||
ssh_config.flush()
|
||||
args = ['--list-ssh-config', '--ssh-config-path', ssh_config.name]
|
||||
args = ["--list-ssh-config", "--ssh-config-path", ssh_config.name]
|
||||
result = runner.invoke(cli, args=args)
|
||||
assert "test\n" in result.output
|
||||
result = runner.invoke(cli, args=args + ['--verbose'])
|
||||
result = runner.invoke(cli, args=args + ["--verbose"])
|
||||
assert "test : test.example.com\n" in result.output
|
||||
|
||||
|
||||
# delete=False means we should try to clean up
|
||||
try:
|
||||
if os.path.exists(ssh_config.name):
|
||||
|
|
@ -343,7 +318,7 @@ def test_dsn(monkeypatch):
|
|||
pass
|
||||
|
||||
class MockMyCli:
|
||||
config = {'alias_dsn': {}}
|
||||
config = {"alias_dsn": {}}
|
||||
|
||||
def __init__(self, **args):
|
||||
self.logger = Logger()
|
||||
|
|
@ -357,97 +332,109 @@ def test_dsn(monkeypatch):
|
|||
pass
|
||||
|
||||
import mycli.main
|
||||
monkeypatch.setattr(mycli.main, 'MyCli', MockMyCli)
|
||||
|
||||
monkeypatch.setattr(mycli.main, "MyCli", MockMyCli)
|
||||
runner = CliRunner()
|
||||
|
||||
# When a user supplies a DSN as database argument to mycli,
|
||||
# use these values.
|
||||
result = runner.invoke(mycli.main.cli, args=[
|
||||
"mysql://dsn_user:dsn_passwd@dsn_host:1/dsn_database"]
|
||||
)
|
||||
result = runner.invoke(mycli.main.cli, args=["mysql://dsn_user:dsn_passwd@dsn_host:1/dsn_database"])
|
||||
assert result.exit_code == 0, result.output + " " + str(result.exception)
|
||||
assert \
|
||||
MockMyCli.connect_args["user"] == "dsn_user" and \
|
||||
MockMyCli.connect_args["passwd"] == "dsn_passwd" and \
|
||||
MockMyCli.connect_args["host"] == "dsn_host" and \
|
||||
MockMyCli.connect_args["port"] == 1 and \
|
||||
MockMyCli.connect_args["database"] == "dsn_database"
|
||||
assert (
|
||||
MockMyCli.connect_args["user"] == "dsn_user"
|
||||
and MockMyCli.connect_args["passwd"] == "dsn_passwd"
|
||||
and MockMyCli.connect_args["host"] == "dsn_host"
|
||||
and MockMyCli.connect_args["port"] == 1
|
||||
and MockMyCli.connect_args["database"] == "dsn_database"
|
||||
)
|
||||
|
||||
MockMyCli.connect_args = None
|
||||
|
||||
# When a use supplies a DSN as database argument to mycli,
|
||||
# and used command line arguments, use the command line
|
||||
# arguments.
|
||||
result = runner.invoke(mycli.main.cli, args=[
|
||||
"mysql://dsn_user:dsn_passwd@dsn_host:2/dsn_database",
|
||||
"--user", "arg_user",
|
||||
"--password", "arg_password",
|
||||
"--host", "arg_host",
|
||||
"--port", "3",
|
||||
"--database", "arg_database",
|
||||
])
|
||||
result = runner.invoke(
|
||||
mycli.main.cli,
|
||||
args=[
|
||||
"mysql://dsn_user:dsn_passwd@dsn_host:2/dsn_database",
|
||||
"--user",
|
||||
"arg_user",
|
||||
"--password",
|
||||
"arg_password",
|
||||
"--host",
|
||||
"arg_host",
|
||||
"--port",
|
||||
"3",
|
||||
"--database",
|
||||
"arg_database",
|
||||
],
|
||||
)
|
||||
assert result.exit_code == 0, result.output + " " + str(result.exception)
|
||||
assert \
|
||||
MockMyCli.connect_args["user"] == "arg_user" and \
|
||||
MockMyCli.connect_args["passwd"] == "arg_password" and \
|
||||
MockMyCli.connect_args["host"] == "arg_host" and \
|
||||
MockMyCli.connect_args["port"] == 3 and \
|
||||
MockMyCli.connect_args["database"] == "arg_database"
|
||||
assert (
|
||||
MockMyCli.connect_args["user"] == "arg_user"
|
||||
and MockMyCli.connect_args["passwd"] == "arg_password"
|
||||
and MockMyCli.connect_args["host"] == "arg_host"
|
||||
and MockMyCli.connect_args["port"] == 3
|
||||
and MockMyCli.connect_args["database"] == "arg_database"
|
||||
)
|
||||
|
||||
MockMyCli.config = {
|
||||
'alias_dsn': {
|
||||
'test': 'mysql://alias_dsn_user:alias_dsn_passwd@alias_dsn_host:4/alias_dsn_database'
|
||||
}
|
||||
}
|
||||
MockMyCli.config = {"alias_dsn": {"test": "mysql://alias_dsn_user:alias_dsn_passwd@alias_dsn_host:4/alias_dsn_database"}}
|
||||
MockMyCli.connect_args = None
|
||||
|
||||
# When a user uses a DSN from the configuration file (alias_dsn),
|
||||
# use these values.
|
||||
result = runner.invoke(cli, args=['--dsn', 'test'])
|
||||
result = runner.invoke(cli, args=["--dsn", "test"])
|
||||
assert result.exit_code == 0, result.output + " " + str(result.exception)
|
||||
assert \
|
||||
MockMyCli.connect_args["user"] == "alias_dsn_user" and \
|
||||
MockMyCli.connect_args["passwd"] == "alias_dsn_passwd" and \
|
||||
MockMyCli.connect_args["host"] == "alias_dsn_host" and \
|
||||
MockMyCli.connect_args["port"] == 4 and \
|
||||
MockMyCli.connect_args["database"] == "alias_dsn_database"
|
||||
assert (
|
||||
MockMyCli.connect_args["user"] == "alias_dsn_user"
|
||||
and MockMyCli.connect_args["passwd"] == "alias_dsn_passwd"
|
||||
and MockMyCli.connect_args["host"] == "alias_dsn_host"
|
||||
and MockMyCli.connect_args["port"] == 4
|
||||
and MockMyCli.connect_args["database"] == "alias_dsn_database"
|
||||
)
|
||||
|
||||
MockMyCli.config = {
|
||||
'alias_dsn': {
|
||||
'test': 'mysql://alias_dsn_user:alias_dsn_passwd@alias_dsn_host:4/alias_dsn_database'
|
||||
}
|
||||
}
|
||||
MockMyCli.config = {"alias_dsn": {"test": "mysql://alias_dsn_user:alias_dsn_passwd@alias_dsn_host:4/alias_dsn_database"}}
|
||||
MockMyCli.connect_args = None
|
||||
|
||||
# When a user uses a DSN from the configuration file (alias_dsn)
|
||||
# and used command line arguments, use the command line arguments.
|
||||
result = runner.invoke(cli, args=[
|
||||
'--dsn', 'test', '',
|
||||
"--user", "arg_user",
|
||||
"--password", "arg_password",
|
||||
"--host", "arg_host",
|
||||
"--port", "5",
|
||||
"--database", "arg_database",
|
||||
])
|
||||
assert result.exit_code == 0, result.output + " " + str(result.exception)
|
||||
assert \
|
||||
MockMyCli.connect_args["user"] == "arg_user" and \
|
||||
MockMyCli.connect_args["passwd"] == "arg_password" and \
|
||||
MockMyCli.connect_args["host"] == "arg_host" and \
|
||||
MockMyCli.connect_args["port"] == 5 and \
|
||||
MockMyCli.connect_args["database"] == "arg_database"
|
||||
|
||||
# Use a DSN without password
|
||||
result = runner.invoke(mycli.main.cli, args=[
|
||||
"mysql://dsn_user@dsn_host:6/dsn_database"]
|
||||
result = runner.invoke(
|
||||
cli,
|
||||
args=[
|
||||
"--dsn",
|
||||
"test",
|
||||
"",
|
||||
"--user",
|
||||
"arg_user",
|
||||
"--password",
|
||||
"arg_password",
|
||||
"--host",
|
||||
"arg_host",
|
||||
"--port",
|
||||
"5",
|
||||
"--database",
|
||||
"arg_database",
|
||||
],
|
||||
)
|
||||
assert result.exit_code == 0, result.output + " " + str(result.exception)
|
||||
assert \
|
||||
MockMyCli.connect_args["user"] == "dsn_user" and \
|
||||
MockMyCli.connect_args["passwd"] is None and \
|
||||
MockMyCli.connect_args["host"] == "dsn_host" and \
|
||||
MockMyCli.connect_args["port"] == 6 and \
|
||||
MockMyCli.connect_args["database"] == "dsn_database"
|
||||
assert (
|
||||
MockMyCli.connect_args["user"] == "arg_user"
|
||||
and MockMyCli.connect_args["passwd"] == "arg_password"
|
||||
and MockMyCli.connect_args["host"] == "arg_host"
|
||||
and MockMyCli.connect_args["port"] == 5
|
||||
and MockMyCli.connect_args["database"] == "arg_database"
|
||||
)
|
||||
|
||||
# Use a DSN without password
|
||||
result = runner.invoke(mycli.main.cli, args=["mysql://dsn_user@dsn_host:6/dsn_database"])
|
||||
assert result.exit_code == 0, result.output + " " + str(result.exception)
|
||||
assert (
|
||||
MockMyCli.connect_args["user"] == "dsn_user"
|
||||
and MockMyCli.connect_args["passwd"] is None
|
||||
and MockMyCli.connect_args["host"] == "dsn_host"
|
||||
and MockMyCli.connect_args["port"] == 6
|
||||
and MockMyCli.connect_args["database"] == "dsn_database"
|
||||
)
|
||||
|
||||
|
||||
def test_ssh_config(monkeypatch):
|
||||
|
|
@ -463,7 +450,7 @@ def test_ssh_config(monkeypatch):
|
|||
pass
|
||||
|
||||
class MockMyCli:
|
||||
config = {'alias_dsn': {}}
|
||||
config = {"alias_dsn": {}}
|
||||
|
||||
def __init__(self, **args):
|
||||
self.logger = Logger()
|
||||
|
|
@ -477,58 +464,62 @@ def test_ssh_config(monkeypatch):
|
|||
pass
|
||||
|
||||
import mycli.main
|
||||
monkeypatch.setattr(mycli.main, 'MyCli', MockMyCli)
|
||||
|
||||
monkeypatch.setattr(mycli.main, "MyCli", MockMyCli)
|
||||
runner = CliRunner()
|
||||
|
||||
# Setup temporary configuration
|
||||
# keep Windows from locking the file with delete=False
|
||||
with NamedTemporaryFile(mode="w",delete=False) as ssh_config:
|
||||
ssh_config.write(dedent("""\
|
||||
with NamedTemporaryFile(mode="w", delete=False) as ssh_config:
|
||||
ssh_config.write(
|
||||
dedent("""\
|
||||
Host test
|
||||
Hostname test.example.com
|
||||
User joe
|
||||
Port 22222
|
||||
IdentityFile ~/.ssh/gateway
|
||||
"""))
|
||||
""")
|
||||
)
|
||||
ssh_config.flush()
|
||||
|
||||
# When a user supplies a ssh config.
|
||||
result = runner.invoke(mycli.main.cli, args=[
|
||||
"--ssh-config-path",
|
||||
ssh_config.name,
|
||||
"--ssh-config-host",
|
||||
"test"
|
||||
])
|
||||
assert result.exit_code == 0, result.output + \
|
||||
" " + str(result.exception)
|
||||
assert \
|
||||
MockMyCli.connect_args["ssh_user"] == "joe" and \
|
||||
MockMyCli.connect_args["ssh_host"] == "test.example.com" and \
|
||||
MockMyCli.connect_args["ssh_port"] == 22222 and \
|
||||
MockMyCli.connect_args["ssh_key_filename"] == os.path.expanduser(
|
||||
"~") + "/.ssh/gateway"
|
||||
result = runner.invoke(mycli.main.cli, args=["--ssh-config-path", ssh_config.name, "--ssh-config-host", "test"])
|
||||
assert result.exit_code == 0, result.output + " " + str(result.exception)
|
||||
assert (
|
||||
MockMyCli.connect_args["ssh_user"] == "joe"
|
||||
and MockMyCli.connect_args["ssh_host"] == "test.example.com"
|
||||
and MockMyCli.connect_args["ssh_port"] == 22222
|
||||
and MockMyCli.connect_args["ssh_key_filename"] == os.path.expanduser("~") + "/.ssh/gateway"
|
||||
)
|
||||
|
||||
# When a user supplies a ssh config host as argument to mycli,
|
||||
# and used command line arguments, use the command line
|
||||
# arguments.
|
||||
result = runner.invoke(mycli.main.cli, args=[
|
||||
"--ssh-config-path",
|
||||
ssh_config.name,
|
||||
"--ssh-config-host",
|
||||
"test",
|
||||
"--ssh-user", "arg_user",
|
||||
"--ssh-host", "arg_host",
|
||||
"--ssh-port", "3",
|
||||
"--ssh-key-filename", "/path/to/key"
|
||||
])
|
||||
assert result.exit_code == 0, result.output + \
|
||||
" " + str(result.exception)
|
||||
assert \
|
||||
MockMyCli.connect_args["ssh_user"] == "arg_user" and \
|
||||
MockMyCli.connect_args["ssh_host"] == "arg_host" and \
|
||||
MockMyCli.connect_args["ssh_port"] == 3 and \
|
||||
MockMyCli.connect_args["ssh_key_filename"] == "/path/to/key"
|
||||
|
||||
result = runner.invoke(
|
||||
mycli.main.cli,
|
||||
args=[
|
||||
"--ssh-config-path",
|
||||
ssh_config.name,
|
||||
"--ssh-config-host",
|
||||
"test",
|
||||
"--ssh-user",
|
||||
"arg_user",
|
||||
"--ssh-host",
|
||||
"arg_host",
|
||||
"--ssh-port",
|
||||
"3",
|
||||
"--ssh-key-filename",
|
||||
"/path/to/key",
|
||||
],
|
||||
)
|
||||
assert result.exit_code == 0, result.output + " " + str(result.exception)
|
||||
assert (
|
||||
MockMyCli.connect_args["ssh_user"] == "arg_user"
|
||||
and MockMyCli.connect_args["ssh_host"] == "arg_host"
|
||||
and MockMyCli.connect_args["ssh_port"] == 3
|
||||
and MockMyCli.connect_args["ssh_key_filename"] == "/path/to/key"
|
||||
)
|
||||
|
||||
# delete=False means we should try to clean up
|
||||
try:
|
||||
if os.path.exists(ssh_config.name):
|
||||
|
|
@ -542,9 +533,7 @@ def test_init_command_arg(executor):
|
|||
init_command = "set sql_select_limit=1000"
|
||||
sql = 'show variables like "sql_select_limit";'
|
||||
runner = CliRunner()
|
||||
result = runner.invoke(
|
||||
cli, args=CLI_ARGS + ["--init-command", init_command], input=sql
|
||||
)
|
||||
result = runner.invoke(cli, args=CLI_ARGS + ["--init-command", init_command], input=sql)
|
||||
|
||||
expected = "sql_select_limit\t1000\n"
|
||||
assert result.exit_code == 0
|
||||
|
|
@ -553,18 +542,13 @@ 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"'
|
||||
)
|
||||
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"'
|
||||
runner = CliRunner()
|
||||
result = runner.invoke(
|
||||
cli, args=CLI_ARGS + ['--init-command', init_command], input=sql
|
||||
)
|
||||
result = runner.invoke(cli, args=CLI_ARGS + ["--init-command", init_command], input=sql)
|
||||
|
||||
expected_sql_select_limit = 'sql_select_limit\t2000\n'
|
||||
expected_max_join_size = 'max_join_size\t20000\n'
|
||||
expected_sql_select_limit = "sql_select_limit\t2000\n"
|
||||
expected_max_join_size = "max_join_size\t20000\n"
|
||||
|
||||
assert result.exit_code == 0
|
||||
assert expected_sql_select_limit in result.output
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue