2016-02-03 23:45:36 +03:00
|
|
|
import os
|
2015-10-23 12:49:20 +03:00
|
|
|
import platform
|
2016-04-27 18:58:16 +03:00
|
|
|
import mock
|
2016-02-03 23:45:36 +03:00
|
|
|
|
|
|
|
import pytest
|
2015-10-27 01:00:42 +03:00
|
|
|
try:
|
|
|
|
import setproctitle
|
|
|
|
except ImportError:
|
|
|
|
setproctitle = None
|
2016-02-02 21:56:20 +03:00
|
|
|
|
2017-01-16 07:27:19 +03:00
|
|
|
from pgcli.main import (
|
|
|
|
obfuscate_process_password, format_output, PGCli, OutputSettings
|
|
|
|
)
|
2016-02-02 21:56:20 +03:00
|
|
|
from utils import dbtest, run
|
2015-10-03 21:41:31 +03:00
|
|
|
|
|
|
|
|
2015-10-23 12:49:20 +03:00
|
|
|
@pytest.mark.skipif(platform.system() == 'Windows',
|
|
|
|
reason='Not applicable in windows')
|
2015-10-27 01:00:42 +03:00
|
|
|
@pytest.mark.skipif(not setproctitle,
|
|
|
|
reason='setproctitle not available')
|
2015-10-19 20:40:50 +03:00
|
|
|
def test_obfuscate_process_password():
|
2015-10-18 23:55:56 +03:00
|
|
|
original_title = setproctitle.getproctitle()
|
|
|
|
|
|
|
|
setproctitle.setproctitle("pgcli user=root password=secret host=localhost")
|
2015-10-19 20:40:50 +03:00
|
|
|
obfuscate_process_password()
|
2015-10-18 23:55:56 +03:00
|
|
|
title = setproctitle.getproctitle()
|
|
|
|
expected = "pgcli user=root password=xxxx host=localhost"
|
|
|
|
assert title == expected
|
|
|
|
|
2015-10-19 20:40:50 +03:00
|
|
|
setproctitle.setproctitle("pgcli user=root password=top secret host=localhost")
|
|
|
|
obfuscate_process_password()
|
|
|
|
title = setproctitle.getproctitle()
|
|
|
|
expected = "pgcli user=root password=xxxx host=localhost"
|
|
|
|
assert title == expected
|
|
|
|
|
|
|
|
setproctitle.setproctitle("pgcli user=root password=top secret")
|
|
|
|
obfuscate_process_password()
|
|
|
|
title = setproctitle.getproctitle()
|
|
|
|
expected = "pgcli user=root password=xxxx"
|
|
|
|
assert title == expected
|
|
|
|
|
2015-10-18 23:55:56 +03:00
|
|
|
setproctitle.setproctitle("pgcli postgres://root:secret@localhost/db")
|
2015-10-19 20:40:50 +03:00
|
|
|
obfuscate_process_password()
|
2015-10-18 23:55:56 +03:00
|
|
|
title = setproctitle.getproctitle()
|
|
|
|
expected = "pgcli postgres://root:xxxx@localhost/db"
|
|
|
|
assert title == expected
|
|
|
|
|
|
|
|
setproctitle.setproctitle(original_title)
|
2015-11-22 16:46:06 +03:00
|
|
|
|
2016-09-27 13:43:05 +03:00
|
|
|
|
2015-11-22 16:46:06 +03:00
|
|
|
def test_format_output():
|
2017-01-16 07:27:19 +03:00
|
|
|
settings = OutputSettings(table_format='psql', dcmlfmt='d', floatfmt='g')
|
2015-11-22 16:46:06 +03:00
|
|
|
results = format_output('Title', [('abc', 'def')], ['head1', 'head2'],
|
2017-01-16 07:27:19 +03:00
|
|
|
'test status', settings)
|
2015-11-22 16:46:06 +03:00
|
|
|
expected = ['Title', '+---------+---------+\n| head1 | head2 |\n|---------+---------|\n| abc | def |\n+---------+---------+', 'test status']
|
|
|
|
assert results == expected
|
|
|
|
|
|
|
|
def test_format_output_auto_expand():
|
2017-01-16 07:27:19 +03:00
|
|
|
settings = OutputSettings(table_format='psql', dcmlfmt='d', floatfmt='g', max_width=100)
|
2015-11-22 16:46:06 +03:00
|
|
|
table_results = format_output('Title', [('abc', 'def')],
|
2017-01-16 07:27:19 +03:00
|
|
|
['head1', 'head2'], 'test status', settings)
|
2015-11-22 16:46:06 +03:00
|
|
|
table = ['Title', '+---------+---------+\n| head1 | head2 |\n|---------+---------|\n| abc | def |\n+---------+---------+', 'test status']
|
|
|
|
assert table_results == table
|
|
|
|
expanded_results = format_output('Title', [('abc', 'def')],
|
2017-01-16 07:27:19 +03:00
|
|
|
['head1', 'head2'], 'test status', settings._replace(max_width=1))
|
2015-11-22 16:46:06 +03:00
|
|
|
expanded = ['Title', u'-[ RECORD 0 ]-------------------------\nhead1 | abc\nhead2 | def\n', 'test status']
|
|
|
|
assert expanded_results == expanded
|
2016-02-02 21:56:20 +03:00
|
|
|
|
|
|
|
|
|
|
|
@dbtest
|
|
|
|
def test_i_works(tmpdir, executor):
|
|
|
|
sqlfile = tmpdir.join("test.sql")
|
|
|
|
sqlfile.write("SELECT NOW()")
|
2016-02-03 23:45:36 +03:00
|
|
|
rcfile = str(tmpdir.join("rcfile"))
|
2016-02-02 21:56:20 +03:00
|
|
|
cli = PGCli(
|
|
|
|
pgexecute=executor,
|
2016-02-03 23:45:36 +03:00
|
|
|
pgclirc_file=rcfile,
|
2016-02-02 21:56:20 +03:00
|
|
|
)
|
2016-02-04 01:31:10 +03:00
|
|
|
statement = r"\i {0}".format(sqlfile)
|
2016-02-02 21:56:20 +03:00
|
|
|
run(executor, statement, pgspecial=cli.pgspecial)
|
2016-02-03 23:45:36 +03:00
|
|
|
|
|
|
|
|
|
|
|
def test_missing_rc_dir(tmpdir):
|
|
|
|
rcfile = str(tmpdir.join("subdir").join("rcfile"))
|
|
|
|
|
2016-02-03 23:58:32 +03:00
|
|
|
PGCli(pgclirc_file=rcfile)
|
2016-02-03 23:45:36 +03:00
|
|
|
assert os.path.exists(rcfile)
|
2016-04-27 18:58:16 +03:00
|
|
|
|
|
|
|
|
|
|
|
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('testdb[', 'baz.com', 'bar^', None, ']foo')
|
2016-06-29 20:58:17 +03:00
|
|
|
|
|
|
|
|
|
|
|
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('testdb', 'baz.com', 'bar', '2543', 'foo')
|