Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_port_db_uri(tmpdir):
with mock.patch.object(PGCli, "connect") as mock_connect:
cli = PGCli(pgclirc_file=str(tmpdir.join("rcfile")))
cli.connect_uri("postgres://bar:foo@baz.com:2543/testdb")
mock_connect.assert_called_with(
database="testdb", host="baz.com", user="bar", passwd="foo", port="2543"
)
def pgspecial():
return PGCli().pgspecial
def test_multihost_db_uri(tmpdir):
with mock.patch.object(PGCli, "connect") as mock_connect:
cli = PGCli(pgclirc_file=str(tmpdir.join("rcfile")))
cli.connect_uri(
"postgres://bar:foo@baz1.com:2543,baz2.com:2543,baz3.com:2543/testdb"
)
mock_connect.assert_called_with(
database="testdb",
host="baz1.com,baz2.com,baz3.com",
user="bar",
passwd="foo",
port="2543,2543,2543",
)
def test_missing_rc_dir(tmpdir):
rcfile = str(tmpdir.join("subdir").join("rcfile"))
PGCli(pgclirc_file=rcfile)
assert os.path.exists(rcfile)
def test_i_works(tmpdir, executor):
sqlfile = tmpdir.join("test.sql")
sqlfile.write("SELECT NOW()")
rcfile = str(tmpdir.join("rcfile"))
cli = PGCli(pgexecute=executor, pgclirc_file=rcfile)
statement = r"\i {0}".format(sqlfile)
run(executor, statement, pgspecial=cli.pgspecial)
def test_row_limit_on_non_select(over_limit):
cli = PGCli()
stmt = "UPDATE students SET name='Boby'"
result = cli._should_limit_output(stmt, over_limit)
assert result is False
cli = PGCli(row_limit=0)
result = cli._should_limit_output(stmt, over_limit)
assert result is False
def test_quoted_db_uri(tmpdir):
with mock.patch.object(PGCli, "connect") as mock_connect:
cli = PGCli(pgclirc_file=str(tmpdir.join("rcfile")))
cli.connect_uri("postgres://bar%5E:%5Dfoo@baz.com/testdb%5B")
mock_connect.assert_called_with(
database="testdb[", host="baz.com", user="bar^", passwd="]foo"
)
def test_row_limit_with_LIMIT_clause(LIMIT, over_limit):
cli = PGCli(row_limit=LIMIT)
stmt = "SELECT * FROM students LIMIT 1000"
result = cli._should_limit_output(stmt, over_limit)
assert result is False
cli = PGCli(row_limit=0)
result = cli._should_limit_output(stmt, over_limit)
assert result is False
def pset_pager_mocks():
cli = PGCli()
cli.watch_command = None
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
if list_dsn:
try:
cfg = load_config(pgclirc, config_full_path)
for alias in cfg["alias_dsn"]:
click.secho(alias + " : " + cfg["alias_dsn"][alias])
sys.exit(0)
except Exception as err:
click.secho(
"Invalid DSNs found in the config file. "
'Please check the "[alias_dsn]" section in pgclirc.',
err=True,
fg="red",
)
exit(1)
pgcli = PGCli(
prompt_passwd,
never_prompt,
pgclirc_file=pgclirc,
row_limit=row_limit,
single_connection=single_connection,
less_chatty=less_chatty,
prompt=prompt,
prompt_dsn=prompt_dsn,
auto_vertical_output=auto_vertical_output,
warn=warn,
)
# Choose which ever one has a valid value.
if dbname_opt and dbname:
# work as psql: when database is given as option and argument use the argument as user
username = dbname