test for setting log level

Signed-off-by: Jakub Jelen <jjelen@redhat.com>

Author: Jakuje

tests/conftest.py

@@ -16,11 +16,23 @@
 from pylibsshext.logging import ANSIBLE_PYLIBSSH_TRACE
 from pylibsshext.session import Session
 
+from pylibsshext import __libssh_version__
+
 
 _DIR_PRIV_RW_OWNER = 0o700
 _FILE_PRIV_RW_OWNER = 0o600
 
 
+@pytest.fixture
+def libssh_version_tuple():
+    """
+    Get the libssh version as a tuple of (major, minor, patch).
+
+    This is useful for simple comparison and conditioning tests based on this value.
+    """
+    return tuple(map(int, (__libssh_version__.split('.'))))
+
+
 @pytest.fixture
 def free_port_num():
     """Detect a free port number using a temporary ephemeral port.

tests/integration/logging_test.py

@@ -0,0 +1,155 @@
+"""Tests suite for logging."""
+
+import contextlib as _ctx
+import logging
+
+import pytest
+
+from pylibsshext.errors import LibsshSessionException
+from pylibsshext.logging import ANSIBLE_PYLIBSSH_NOLOG, ANSIBLE_PYLIBSSH_TRACE
+from pylibsshext.session import Session
+
+
+LOCALHOST = '127.0.0.1'
+BAD_LOG_LEVEL = 99
+
+
+@pytest.fixture(autouse=True)
+def _only_capture_project_logs(caplog: pytest.LogCaptureFixture) -> None:
+    """Hide unrelated log entries from capture."""
+    caplog.set_level(ANSIBLE_PYLIBSSH_NOLOG + 1)  # root logger
+    caplog.set_level(
+        logging.NOTSET,
+        logger='ansible-pylibssh',
+    )  # don't block things early
+    caplog.clear()
+
+
+def test_session_log_level_warning(
+    caplog: pytest.LogCaptureFixture,
+    free_port_num: int,
+    libssh_version_tuple,
+) -> None:
+    """
+    Test setting the log level to WARNING.
+
+    It should show "Connection refused" on libssh 0.11.0 and newer.
+    But no debug/ainfo/trace messages.
+    """
+    ssh_session = Session()
+    # keeping these in sync makes sure the log levels are not "lost"
+    # in the processing. Technically, setting the log level in libssh
+    # is not needed as the default is logging everything but filtering
+    # is done on the python side
+    ssh_session.set_log_level(logging.WARNING)
+    caplog.set_level(logging.WARNING)
+
+    # the connection will fail but first log lands before that
+    with _ctx.suppress(LibsshSessionException):
+        ssh_session.connect(host=LOCALHOST, port=free_port_num)
+
+    log_records = caplog.records
+
+    # This message is available only in libssh 0.11.0 and newer
+    expected_substring = 'ssh_client_connection_callback: Connection refused'
+    if libssh_version_tuple >= (0, 11, 0):
+        assert any(
+            record.levelname == 'WARNING' and expected_substring in record.msg
+            for record in log_records
+        )
+
+    # No INFO and higher log messages should show up at this log level
+    assert all(record.levelname != 'INFO' for record in log_records)
+    assert all(record.levelname != 'DEBUG' for record in log_records)
+    assert all(record.levelname != 'TRACE' for record in log_records)
+
+
+def test_session_log_level_debug(
+    caplog: pytest.LogCaptureFixture,
+    free_port_num: int,
+) -> None:
+    """
+    Test setting the log level to DEBUG.
+
+    It should reveal copyright information. But no trace messages.
+    """
+    ssh_session = Session()
+    ssh_session.set_log_level(logging.DEBUG)
+    caplog.set_level(logging.DEBUG)
+
+    # the connection will fail but first log lands before that
+    with _ctx.suppress(LibsshSessionException):
+        ssh_session.connect(host=LOCALHOST, port=free_port_num)
+
+    log_records = caplog.records
+
+    expected_copyright_substring = 'and libssh contributors.'
+    # This log message is shown at different log levels
+    # in different libssh versions. Changed at 657d9143d1 (before 0.11.0)
+    # but backported to some RHEL9 versions so matching on the libssh version
+    # is not reliable.
+    assert any(
+        record.levelname in {'DEBUG', 'INFO'}
+        and expected_copyright_substring in record.msg
+        for record in log_records
+    )
+
+    # No TRACE log messages should show up at this log level
+    assert all(record.levelname != 'TRACE' for record in log_records)
+
+
+def test_session_log_level_no_log(
+    caplog: pytest.LogCaptureFixture,
+    free_port_num: int,
+) -> None:
+    """Test setting the log level to NOLOG should be quiet."""
+    ssh_session = Session()
+    ssh_session.set_log_level(ANSIBLE_PYLIBSSH_NOLOG)
+    caplog.set_level(ANSIBLE_PYLIBSSH_NOLOG)
+
+    # the connection will fail but first log lands before that
+    with _ctx.suppress(LibsshSessionException):
+        ssh_session.connect(host=LOCALHOST, port=free_port_num)
+
+    assert not caplog.records
+
+
+def test_session_log_level_trace(
+    caplog: pytest.LogCaptureFixture,
+    free_port_num: int,
+) -> None:
+    """Test setting the log level to TRACE should provide all of the logs."""
+    ssh_session = Session()
+    ssh_session.set_log_level(ANSIBLE_PYLIBSSH_TRACE)
+    caplog.set_level(ANSIBLE_PYLIBSSH_TRACE)
+
+    with _ctx.suppress(LibsshSessionException):
+        ssh_session.connect(host=LOCALHOST, port=free_port_num)
+
+    expected_poll_message = 'ssh_socket_pollcallback: Poll callback on socket'
+    assert expected_poll_message in caplog.text
+
+
+def test_session_log_level_not_set(
+    caplog: pytest.LogCaptureFixture,
+    free_port_num: int,
+) -> None:
+    """Test setting the log level to NOTSET = all messages should be logged too."""
+    ssh_session = Session()
+    ssh_session.set_log_level(logging.NOTSET)
+    caplog.set_level(logging.NOTSET)
+
+    with _ctx.suppress(LibsshSessionException):
+        ssh_session.connect(host=LOCALHOST, port=free_port_num)
+
+    expected_poll_message = 'ssh_socket_pollcallback: Poll callback on socket'
+    assert expected_poll_message in caplog.text
+
+
+def test_session_log_level_bad():
+    """Test that setting the log level to an unsupported value is rejected."""
+    ssh_session = Session()
+
+    error_msg = r'^Invalid log level \[99\]$'
+    with pytest.raises(LibsshSessionException, match=error_msg):
+        ssh_session.set_log_level(BAD_LOG_LEVEL)