diff --git a/src/neo4j/debug.py b/src/neo4j/debug.py index efd2b9eaa..bb2a7e47f 100644 --- a/src/neo4j/debug.py +++ b/src/neo4j/debug.py @@ -166,12 +166,14 @@ def watch( self.stop() handler = StreamHandler(out) handler.setFormatter(self.formatter) + handler.setLevel(level) if self._task_info: handler.addFilter(TaskIdFilter()) for logger in self. _loggers: self._handlers[logger.name] = handler logger.addHandler(handler) - logger.setLevel(level) + if logger.getEffectiveLevel() > level: + logger.setLevel(level) def stop(self) -> None: """Disable logging for all loggers.""" diff --git a/tests/unit/common/test_debug.py b/tests/unit/common/test_debug.py index 7454148e8..377efa095 100644 --- a/tests/unit/common/test_debug.py +++ b/tests/unit/common/test_debug.py @@ -43,16 +43,24 @@ def __call__(self, *args: str) -> t.Sequence[t.Any]: @pytest.fixture -def logger_mocker(mocker) -> _TSetupMockProtocol: +def logger_mocker(mocker) -> t.Generator[_TSetupMockProtocol, None, None]: + original_levels = {} + def setup_mock(*logger_names): + nonlocal original_levels + loggers = [logging.getLogger(name) for name in logger_names] for logger in loggers: - logger.addHandler = mocker.Mock() - logger.removeHandler = mocker.Mock() - logger.setLevel = mocker.Mock() + original_levels[logger] = logger.level + mocker.patch.object(logger, "addHandler") + mocker.patch.object(logger, "removeHandler") + mocker.spy(logger, "setLevel") return loggers - return setup_mock + yield setup_mock + + for logger, level in original_levels.items(): + logger.setLevel(level) def test_watch_returns_watcher(logger_mocker) -> None: @@ -106,22 +114,21 @@ def test_watcher_context_manager(mocker) -> None: watcher.stop.assert_called_once() -@pytest.mark.parametrize( - ("default_level", "level", "expected_level"), - ( - (None, None, logging.DEBUG), - (logging.WARNING, None, logging.WARNING), - (logging.WARNING, logging.DEBUG, logging.DEBUG), - (logging.DEBUG, logging.WARNING, logging.WARNING), - (1, None, 1), - (None, 1, 1), - ) +WATCH_ARGS = ( + #default_level, level, expected_level + (None, None, logging.DEBUG), + (logging.WARNING, None, logging.WARNING), + (logging.WARNING, logging.DEBUG, logging.DEBUG), + (logging.DEBUG, logging.WARNING, logging.WARNING), + (logging.INFO, None, logging.INFO), + (logging.INFO, logging.DEBUG, logging.DEBUG), + (logging.DEBUG, logging.INFO, logging.INFO), + (1, None, 1), + (None, 1, 1), ) -def test_watcher_level( - logger_mocker, default_level, level, expected_level -) -> None: - logger_name = "neo4j" - logger = logger_mocker(logger_name)[0] + + +def _setup_watch(logger_name, default_level, level): kwargs = {} if default_level is not None: kwargs["default_level"] = default_level @@ -131,9 +138,73 @@ def test_watcher_level( kwargs["level"] = level watcher.watch(**kwargs) + +@pytest.mark.parametrize( + ("default_level", "level", "expected_level"), + WATCH_ARGS +) +@pytest.mark.parametrize( + "effective_level", + (logging.DEBUG, logging.WARNING, logging.INFO, logging.ERROR) +) +def test_watcher_level( + logger_mocker, default_level, level, expected_level, effective_level, +) -> None: + logger_name = "neo4j" + logger = logger_mocker(logger_name)[0] + logger.setLevel(effective_level) + logger.setLevel.reset_mock() + _setup_watch(logger_name, default_level, level) + (handler,), _ = logger.addHandler.call_args - assert handler.level == logging.NOTSET - logger.setLevel.assert_called_once_with(expected_level) + assert handler.level == expected_level + if effective_level <= expected_level: + logger.setLevel.assert_not_called() + else: + logger.setLevel.assert_called_once_with(expected_level) + + +@pytest.mark.parametrize( + ("default_level1", "level1", "expected_level1"), + WATCH_ARGS +) +@pytest.mark.parametrize( + ("default_level2", "level2", "expected_level2"), + WATCH_ARGS +) +@pytest.mark.parametrize( + "effective_level", + (logging.DEBUG, logging.WARNING, logging.INFO, logging.ERROR) +) +def test_watcher_level_multiple_watchers( + logger_mocker, default_level1, level1, expected_level1, + default_level2, level2, expected_level2, + effective_level, +) -> None: + logger_name = "neo4j" + logger = logger_mocker(logger_name)[0] + logger.setLevel(effective_level) + logger.setLevel.reset_mock() + _setup_watch(logger_name, default_level1, level1) + _setup_watch(logger_name, default_level2, level2) + + assert logger.addHandler.call_count == 2 + (handler1,), _ = logger.addHandler.call_args_list[0] + (handler2,), _ = logger.addHandler.call_args_list[1] + + assert handler1.level == expected_level1 + assert handler2.level == expected_level2 + + expected_logger_level = min(expected_level1, expected_level2) + if effective_level <= expected_logger_level: + logger.setLevel.assert_not_called() + else: + if effective_level > expected_level1 > expected_level2: + assert logger.setLevel.call_count == 2 + else: + assert logger.setLevel.call_count == 1 + (level,), _ = logger.setLevel.call_args_list[-1] + assert level == expected_logger_level custom_log_out = io.StringIO() @@ -220,6 +291,7 @@ def _assert_task_injection( handler_cls_mock = mocker.patch("neo4j.debug.StreamHandler", autospec=True) handler_mock = handler_cls_mock.return_value logger_name = "neo4j" + logger_mocker(logger_name)[0] watcher = neo4j_debug.Watcher(logger_name, colour=colour, thread_info=thread, task_info=task) record_mock = mocker.Mock(spec=logging.LogRecord)