Merging upstream version 4.1.0.

Signed-off-by: Daniel Baumann <daniel@debian.org>

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,
@@ -296,6 +301,24 @@ def test_i_works(tmpdir, executor):
     run(executor, statement, pgspecial=cli.pgspecial)
 
 
+@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)
@@ -312,6 +335,34 @@ def test_qecho_works(executor):
     assert result == ["asdf"]
 
 
+@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()