diff options
Diffstat (limited to 'tests/test_main.py')
| -rw-r--r-- | tests/test_main.py | 99 |
1 files changed, 98 insertions, 1 deletions
diff --git a/tests/test_main.py b/tests/test_main.py index cbf20a6..3683d49 100644 --- a/tests/test_main.py +++ b/tests/test_main.py @@ -1,5 +1,8 @@ import os import platform +import re +import tempfile +import datetime from unittest import mock import pytest @@ -11,7 +14,9 @@ except ImportError: from pgcli.main import ( obfuscate_process_password, + duration_in_words, format_output, + notify_callback, PGCli, OutputSettings, COLOR_CODE_REGEX, @@ -297,6 +302,24 @@ def test_i_works(tmpdir, executor): @dbtest +def test_toggle_verbose_errors(executor): + cli = PGCli(pgexecute=executor) + + cli._evaluate_command("\\v on") + assert cli.verbose_errors + output, _ = cli._evaluate_command("SELECT 1/0") + assert "SQLSTATE" in output[0] + + cli._evaluate_command("\\v off") + assert not cli.verbose_errors + output, _ = cli._evaluate_command("SELECT 1/0") + assert "SQLSTATE" not in output[0] + + cli._evaluate_command("\\v") + assert cli.verbose_errors + + +@dbtest def test_echo_works(executor): cli = PGCli(pgexecute=executor) statement = r"\echo asdf" @@ -313,6 +336,34 @@ def test_qecho_works(executor): @dbtest +def test_logfile_works(executor): + with tempfile.TemporaryDirectory() as tmpdir: + log_file = f"{tmpdir}/tempfile.log" + cli = PGCli(pgexecute=executor, log_file=log_file) + statement = r"\qecho hello!" + cli.execute_command(statement) + with open(log_file, "r") as f: + log_contents = f.readlines() + assert datetime.datetime.fromisoformat(log_contents[0].strip()) + assert log_contents[1].strip() == r"\qecho hello!" + assert log_contents[2].strip() == "hello!" + + +@dbtest +def test_logfile_unwriteable_file(executor): + cli = PGCli(pgexecute=executor) + statement = r"\log-file forbidden.log" + with mock.patch("builtins.open") as mock_open: + mock_open.side_effect = PermissionError( + "[Errno 13] Permission denied: 'forbidden.log'" + ) + result = run(executor, statement, pgspecial=cli.pgspecial) + assert result == [ + "[Errno 13] Permission denied: 'forbidden.log'\nLogfile capture disabled" + ] + + +@dbtest def test_watch_works(executor): cli = PGCli(pgexecute=executor) @@ -431,6 +482,7 @@ def test_pg_service_file(tmpdir): "b_host", "5435", "", + notify_callback, application_name="pgcli", ) del os.environ["PGPASSWORD"] @@ -486,5 +538,50 @@ def test_application_name_db_uri(tmpdir): cli = PGCli(pgclirc_file=str(tmpdir.join("rcfile"))) cli.connect_uri("postgres://bar@baz.com/?application_name=cow") mock_pgexecute.assert_called_with( - "bar", "bar", "", "baz.com", "", "", application_name="cow" + "bar", "bar", "", "baz.com", "", "", notify_callback, application_name="cow" ) + + +@pytest.mark.parametrize( + "duration_in_seconds,words", + [ + (0, "0 seconds"), + (0.0009, "0.001 second"), + (0.0005, "0.001 second"), + (0.0004, "0.0 second"), # not perfect, but will do + (0.2, "0.2 second"), + (1, "1 second"), + (1.4, "1 second"), + (2, "2 seconds"), + (3.4, "3 seconds"), + (60, "1 minute"), + (61, "1 minute 1 second"), + (123, "2 minutes 3 seconds"), + (3600, "1 hour"), + (7235, "2 hours 35 seconds"), + (9005, "2 hours 30 minutes 5 seconds"), + (86401, "24 hours 1 second"), + ], +) +def test_duration_in_words(duration_in_seconds, words): + assert duration_in_words(duration_in_seconds) == words + + +@dbtest +def test_notifications(executor): + run(executor, "listen chan1") + + with mock.patch("pgcli.main.click.secho") as mock_secho: + run(executor, "notify chan1, 'testing1'") + mock_secho.assert_called() + arg = mock_secho.call_args_list[0].args[0] + assert re.match( + r'Notification received on channel "chan1" \(PID \d+\):\ntesting1', + arg, + ) + + run(executor, "unlisten chan1") + + with mock.patch("pgcli.main.click.secho") as mock_secho: + run(executor, "notify chan1, 'testing2'") + mock_secho.assert_not_called() |