Coverage for python / lsst / daf / butler / tests / cliLogTestBase.py: 31%
148 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-04-28 08:36 +0000
1# This file is part of daf_butler.
2#
3# Developed for the LSST Data Management System.
4# This product includes software developed by the LSST Project
5# (http://www.lsst.org).
6# See the COPYRIGHT file at the top-level directory of this distribution
7# for details of code ownership.
8#
9# This software is dual licensed under the GNU General Public License and also
10# under a 3-clause BSD license. Recipients may choose which of these licenses
11# to use; please see the files gpl-3.0.txt and/or bsd_license.txt,
12# respectively. If you choose the GPL option then the following text applies
13# (but note that there is still no warranty even if you opt for BSD instead):
14#
15# This program is free software: you can redistribute it and/or modify
16# it under the terms of the GNU General Public License as published by
17# the Free Software Foundation, either version 3 of the License, or
18# (at your option) any later version.
19#
20# This program is distributed in the hope that it will be useful,
21# but WITHOUT ANY WARRANTY; without even the implied warranty of
22# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
23# GNU General Public License for more details.
24#
25# You should have received a copy of the GNU General Public License
26# along with this program. If not, see <http://www.gnu.org/licenses/>.
28"""Unit tests for the daf_butler CliLog utility.
30Code is implemented in
31daf_butler but some only runs if lsst.log.Log can be imported so these parts of
32it can't be tested there because daf_butler does not directly depend on
33lsst.log, and only uses it if it has been setup by another package.
34"""
36from __future__ import annotations
38__all__ = ["CliLogTestBase"]
40import logging
41import os
42import re
43import subprocess
44import tempfile
45import unittest.mock
46from collections import namedtuple
47from collections.abc import Callable
48from functools import partial
49from io import StringIO
50from logging import DEBUG, INFO, WARNING
51from typing import TYPE_CHECKING, Any
53import click
55from lsst.daf.butler.cli.butler import UncachedButlerCLI
56from lsst.daf.butler.cli.cliLog import CliLog
57from lsst.daf.butler.cli.opt import (
58 log_file_option,
59 log_label_option,
60 log_level_option,
61 log_tty_option,
62 long_log_option,
63)
64from lsst.daf.butler.cli.utils import LogCliRunner, clickResultMsg, command_test_env
65from lsst.daf.butler.logging import ButlerLogRecords
66from lsst.utils.logging import TRACE
68try:
69 import lsst.log as lsstLog
71 lsstLog_INFO = lsstLog.INFO
72 lsstLog_DEBUG = lsstLog.DEBUG
73 lsstLog_WARN = lsstLog.WARN
74except ModuleNotFoundError:
75 lsstLog = None
76 lsstLog_INFO = 0
77 lsstLog_DEBUG = 0
78 lsstLog_WARN = 0
81@click.command(cls=UncachedButlerCLI)
82@log_level_option()
83@long_log_option()
84@log_file_option()
85@log_tty_option()
86@log_label_option()
87def butlerCli(log_level: str, long_log: bool, log_file: str, log_tty: bool, log_label: str) -> None:
88 """Uncached ButlerCLI.
90 Parameters
91 ----------
92 log_level : `str`
93 The log level to use by default. ``log_level`` is handled by
94 ``get_command`` and ``list_commands``, and is called in
95 one of those functions before this is called.
96 long_log : `bool`
97 Enable extended log output. ``long_log`` is handled by
98 ``setup_logging``.
99 log_file : `str`
100 The log file name.
101 log_tty : `bool`
102 Whether to send logs to standard output.
103 log_label : `str`
104 Log labels.
105 """
106 pass
109@click.command()
110@click.option("--expected-pyroot-level", type=int)
111@click.option("--expected-pylsst-level", type=int)
112@click.option("--expected-pybutler-level", type=int)
113@click.option("--expected-lsstroot-level", type=int)
114@click.option("--expected-lsstbutler-level", type=int)
115@click.option("--expected-lsstx-level", type=int)
116def command_log_settings_test( # numpydoc ignore=PR01
117 expected_pyroot_level: str,
118 expected_pylsst_level: str,
119 expected_pybutler_level: str,
120 expected_lsstroot_level: str,
121 expected_lsstbutler_level: str,
122 expected_lsstx_level: str,
123) -> None:
124 """Test command-line log settings."""
125 LogLevel = namedtuple("LogLevel", ("expected", "actual", "name"))
127 logLevels = [
128 LogLevel(expected_pyroot_level, logging.getLogger().level, "pyRoot"),
129 LogLevel(expected_pylsst_level, logging.getLogger("lsst").getEffectiveLevel(), "pyLsst"),
130 LogLevel(
131 expected_pybutler_level, logging.getLogger("lsst.daf.butler").getEffectiveLevel(), "pyButler"
132 ),
133 LogLevel(expected_lsstx_level, logging.getLogger("lsstx").getEffectiveLevel(), "pyLsstx"),
134 ]
135 if lsstLog is not None:
136 logLevels.extend(
137 [
138 LogLevel(expected_lsstroot_level, lsstLog.getLogger("lsst").getEffectiveLevel(), "lsstRoot"),
139 LogLevel(
140 expected_lsstbutler_level,
141 lsstLog.getLogger("lsst.daf.butler").getEffectiveLevel(),
142 "lsstButler",
143 ),
144 ]
145 )
146 for expected, actual, name in logLevels:
147 if expected != actual:
148 raise (
149 click.ClickException(message=f"expected {name} level to be {expected!r}, actual:{actual!r}")
150 )
153class CliLogTestBase:
154 """Tests log initialization, reset, and setting log levels."""
156 if TYPE_CHECKING:
157 assertEqual: Callable
158 assertIn: Callable
159 assertTrue: Callable
160 assertFalse: Callable
161 assertGreater: Callable
162 subTest: Callable
163 assertNotIn: Callable
165 def setUp(self) -> None:
166 self.runner = LogCliRunner()
168 class PythonLogger:
169 """Keeps track of log level of a component and number of handlers
170 attached to it at the time this object was initialized.
172 Parameters
173 ----------
174 component : `str` or `None`
175 The logger name.
176 """
178 def __init__(self, component: str | None) -> None:
179 self.logger = logging.getLogger(component)
180 self.initialLevel = self.logger.level
182 class LsstLogger:
183 """Keeps track of log level for a component at the time this object was
184 initialized.
186 Parameters
187 ----------
188 component : `str` or `None`
189 The logger name.
190 """
192 def __init__(self, component: str) -> None:
193 if lsstLog:
194 self.logger = lsstLog.getLogger(component)
195 self.initialLevel = self.logger.getLevel()
196 else:
197 self.logger = None
198 self.initialLevel = None
200 def runTest(self, cmd: Callable) -> None:
201 """Test that the log context manager works with the butler cli.
203 Parameters
204 ----------
205 cmd : `~collections.abc.Callable`
206 The command to run.
208 Notes
209 -----
210 Tests that it will initialize the logging system according to cli
211 inputs for the duration of the command execution and resets the logging
212 system to its previous state or expected state when command execution
213 finishes.
214 """
215 pyRoot = self.PythonLogger(None)
216 pyButler = self.PythonLogger("lsst.daf.butler")
217 pyLsstRoot = self.PythonLogger("lsst")
218 lsstRoot = self.LsstLogger("")
219 lsstButler = self.LsstLogger("lsst.daf.butler")
221 with command_test_env(
222 self.runner, "lsst.daf.butler.tests.cliLogTestBase", "command-log-settings-test"
223 ):
224 result = cmd()
225 self.assertEqual(result.exit_code, 0, clickResultMsg(result))
227 # The test environment may have changed the python root logger
228 # so we can not assume it will be WARNING.
229 self.assertEqual(pyRoot.logger.level, logging.getLogger().getEffectiveLevel())
230 self.assertEqual(pyLsstRoot.logger.level, logging.INFO)
231 self.assertEqual(pyButler.logger.level, pyButler.initialLevel)
232 if lsstLog is not None:
233 assert lsstRoot.logger is not None
234 assert lsstButler.logger is not None
235 self.assertEqual(lsstRoot.logger.getLevel(), lsstLog.INFO)
236 # lsstLogLevel can either be the initial level, or uninitialized or
237 # the defined default value.
238 expectedLsstLogLevel = (
239 (lsstButler.initialLevel,)
240 if lsstButler.initialLevel != -1
241 else (-1, CliLog.defaultLsstLogLevel)
242 )
243 self.assertIn(lsstButler.logger.getLevel(), expectedLsstLogLevel)
245 def test_butlerCliLog(self) -> None:
246 """Test that the log context manager works with the butler cli to
247 initialize the logging system according to cli inputs for the duration
248 of the command execution and resets the logging system to its previous
249 state or expected state when command execution finishes.
250 """
251 # Run with two different log level settings.
252 log_levels = (
253 # --log-level / --log-level / expected pyroot, pylsst, pybutler,
254 # lsstroot, lsstbutler, lsstx
255 (
256 "WARNING",
257 "lsst.daf.butler=DEBUG",
258 WARNING,
259 WARNING,
260 DEBUG,
261 lsstLog_WARN,
262 lsstLog_DEBUG,
263 WARNING,
264 ),
265 ("DEBUG", "lsst.daf.butler=TRACE", WARNING, DEBUG, TRACE, lsstLog_DEBUG, lsstLog_DEBUG, WARNING),
266 (".=DEBUG", "lsst.daf.butler=WARNING", DEBUG, INFO, WARNING, lsstLog_INFO, lsstLog_WARN, DEBUG),
267 (".=DEBUG", "DEBUG", DEBUG, DEBUG, DEBUG, lsstLog_DEBUG, lsstLog_DEBUG, DEBUG),
268 (".=DEBUG", "conda=DEBUG", DEBUG, INFO, INFO, lsstLog_INFO, lsstLog_INFO, DEBUG),
269 )
271 self._test_levels(log_levels)
273 # Check that the environment variable can set additional roots.
274 log_levels = (
275 # --log-level / --log-level / expected pyroot, pylsst, pybutler,
276 # lsstroot, lsstbutler, lsstx
277 (
278 "WARNING",
279 "lsst.daf.butler=DEBUG",
280 WARNING,
281 WARNING,
282 DEBUG,
283 lsstLog_WARN,
284 lsstLog_DEBUG,
285 WARNING,
286 ),
287 ("DEBUG", "lsst.daf.butler=TRACE", WARNING, DEBUG, TRACE, lsstLog_DEBUG, lsstLog_DEBUG, DEBUG),
288 (".=DEBUG", "lsst.daf.butler=WARNING", DEBUG, INFO, WARNING, lsstLog_INFO, lsstLog_WARN, INFO),
289 (".=DEBUG", "DEBUG", DEBUG, DEBUG, DEBUG, lsstLog_DEBUG, lsstLog_DEBUG, DEBUG),
290 (".=DEBUG", "conda=DEBUG", DEBUG, INFO, INFO, lsstLog_INFO, lsstLog_INFO, INFO),
291 )
293 with unittest.mock.patch.dict(os.environ, {"DAF_BUTLER_ROOT_LOGGER": "lsstx"}):
294 self._test_levels(log_levels)
296 def _test_levels(self, log_levels: tuple[tuple[str, str, int, int, int, Any, Any, int], ...]) -> None:
297 for level1, level2, x_pyroot, x_pylsst, x_pybutler, x_lsstroot, x_lsstbutler, x_lsstx in log_levels:
298 with self.subTest("Test different log levels", level1=level1, level2=level2):
299 self.runTest(
300 partial(
301 self.runner.invoke,
302 butlerCli,
303 [
304 "--log-level",
305 level1,
306 "--log-level",
307 level2,
308 "command-log-settings-test",
309 "--expected-pyroot-level",
310 x_pyroot,
311 "--expected-pylsst-level",
312 x_pylsst,
313 "--expected-pybutler-level",
314 x_pybutler,
315 "--expected-lsstroot-level",
316 x_lsstroot,
317 "--expected-lsstbutler-level",
318 x_lsstbutler,
319 "--expected-lsstx-level",
320 x_lsstx,
321 ],
322 )
323 )
325 def test_helpLogReset(self) -> None:
326 """Verify that when a command does not execute, like when the help menu
327 is printed instead, that CliLog is still reset.
328 """
329 self.runTest(partial(self.runner.invoke, butlerCli, ["command-log-settings-test", "--help"]))
331 def testLongLog(self) -> None:
332 """Verify the timestamp is in the log messages when the --long-log
333 flag is set.
334 """
335 # When longlog=True, loglines start with the log level and a
336 # timestamp with the following format:
337 # "year-month-day T hour-minute-second.millisecond-zoneoffset"
338 # For example: "DEBUG 2020-10-28T10:20:31-07:00 ...""
339 # The log level name can change, we verify there is an all
340 # caps word there but do not verify the word. We do not verify
341 # the rest of the log string, assume that if the timestamp is
342 # in the string that the rest of the string will appear as
343 # expected.
344 timestampRegex = re.compile(
345 r".*[A-Z]+ [0-9]{4}-[0-9]{2}-[0-9]{2}T[0-9]{2}:[0-9]{2}:[0-9]{2}(.[0-9]{3})"
346 "?([-,+][01][0-9]:[034][05]|Z) .*"
347 )
349 # When longlog=False, log lines start with the module name and
350 # log level, for example:
351 # lsst.daf.butler.config DEBUG: ...
352 modulesRegex = re.compile(r".* ([a-z]+\.)+[a-z]+ [A-Z]+: .*")
354 with self.runner.isolated_filesystem():
355 for longlog in (True, False):
356 # The pytest log handler interferes with the log configuration
357 # settings set up by initLog -- therefore test by using
358 # a subprocess.
359 args: tuple[str, ...]
360 if longlog:
361 args = ("butler", "--log-level", "DEBUG", "--long-log", "create", "here")
362 else:
363 args = ("butler", "--log-level", "DEBUG", "create", "here")
364 result = subprocess.run(args, capture_output=True)
365 # There are cases where the newlines are stripped from the log
366 # output (like in Jenkins), since we can't depend on newlines
367 # in log output they are removed here from test output.
368 output = StringIO(result.stderr.decode().replace("\n", " "))
369 startedWithTimestamp = any(timestampRegex.match(line) for line in output.readlines())
370 output.seek(0)
371 startedWithModule = any(modulesRegex.match(line) for line in output.readlines())
372 if longlog:
373 self.assertTrue(
374 startedWithTimestamp, msg=f"did not find timestamp in: \n{output.getvalue()}"
375 )
376 self.assertFalse(
377 startedWithModule, msg=f"found lines starting with module in: \n{output.getvalue()}"
378 )
379 else:
380 self.assertFalse(startedWithTimestamp, msg=f"found timestamp in: \n{output.getvalue()}")
381 self.assertTrue(
382 startedWithModule,
383 msg=f"did not find lines starting with module in: \n{output.getvalue()}",
384 )
386 def testFileLogging(self) -> None:
387 """Test --log-file option."""
388 with self.runner.isolated_filesystem():
389 for i, suffix in enumerate([".json", ".log"]):
390 # Get a temporary file name and immediately close it
391 fd = tempfile.NamedTemporaryFile(suffix=suffix)
392 filename = fd.name
393 fd.close()
395 args = (
396 "--log-level",
397 "DEBUG",
398 "--log-file",
399 filename,
400 "--log-label",
401 "k1=v1,k2=v2",
402 "--log-label",
403 "k3=v3",
404 "create",
405 f"here{i}",
406 )
408 result = self.runner.invoke(butlerCli, args)
409 self.assertEqual(result.exit_code, 0, clickResultMsg(result))
411 # We have no real control over other packages causing lsst
412 # DEBUG log messages to turn up (for example from misconfigured
413 # CLI plugins). Check for any DEBUG messages.
414 # We know there are at least this number of log
415 # messages issued.
416 min_records = 10
418 if suffix == ".json":
419 records = ButlerLogRecords.from_file(filename)
420 self.assertGreater(len(records), min_records)
421 self.assertEqual(records[0].MDC, dict(K1="v1", K2="v2", K3="v3"))
422 # Find a DEBUG message.
423 self.assertTrue(any(rec.levelname == "DEBUG" for rec in records), records)
424 else:
425 with open(filename) as filed:
426 records_text = filed.readlines()
427 self.assertGreater(len(records_text), min_records) # Counting lines not records.
428 content = "".join(records_text)
429 self.assertFalse(content.startswith("{"), f"Checking for JSON content in {content}")
431 self.assertTrue(any("DEBUG:" in rec for rec in records_text), content)
433 def testLogTty(self) -> None:
434 """Verify that log output to terminal can be suppressed."""
435 with self.runner.isolated_filesystem():
436 for log_tty in (True, False):
437 # The pytest log handler interferes with the log configuration
438 # settings set up by initLog -- therefore test by using
439 # a subprocess.
440 if log_tty:
441 args = ("butler", "--log-level", "DEBUG", "--log-tty", "create", "here")
442 else:
443 args = ("butler", "--log-level", "DEBUG", "--no-log-tty", "create", "here2")
444 result = subprocess.run(args, capture_output=True)
446 output = result.stderr.decode()
447 if log_tty:
448 self.assertIn("DEBUG", output)
449 else:
450 self.assertNotIn("DEBUG", output)
453if __name__ == "__main__":
454 unittest.main()