Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Catch logger output #111

Closed
ArneBachmannDLR opened this issue Sep 24, 2021 · 8 comments
Closed

Catch logger output #111

ArneBachmannDLR opened this issue Sep 24, 2021 · 8 comments

Comments

@ArneBachmannDLR
Copy link

Whenever I want to catch and check the logger output in my doctests, I use the following approach (not sure if there is a better way, though):

import logging

_log = logging.getLogger(__name__)
debug, info, warn, error = _log.debug, _log.info, _log.warning, _log.error

def funcUnderTest():
  '''
  >>> _log.addHandler(logging.StreamHandler(sys.stdout))
  >>> funcUnderTest()
  Logger output
  '''
  info("Logger output")

Using xdoctest the logger is not caught. Is there a workaround?

@Erotemic
Copy link
Owner

Erotemic commented Sep 24, 2021

The relevant code isthe CaptureStdout class: https://github.com/Erotemic/xdoctest/blob/main/xdoctest/utils/util_stream.py#L108

Currently the way this works is that it stores a reference to current value of sys.stdout, and then overrides it with a TeeStringIO object that it manages. When the context is over it cleans up and restores the value of sys.stdout. This is the way I've seen it done in most other places, so I'm not sure if there is a better option.

My guess it that logging isn't captured because the logging.StreamHandler stores its own reference to sys.stdout, which will not change if sys.stdout itself is overridden.

One workaround could be trying to find all live instances of logging.StreamHandler objects and performing the same surgery on them, but I'm somewhat hesitant to do that. It feels like it would be intrusive to muck with those internals (moreso than we are already doing).

I can think of a workaround that you could implement in your own code base. The code for StreamHandler itself is reasonably concise:

class StreamHandler(Handler):
    """
    A handler class which writes logging records, appropriately formatted,
    to a stream. Note that this class does not close the stream, as
    sys.stdout or sys.stderr may be used.
    """

    terminator = '\n'

    def __init__(self, stream=None):
        """
        Initialize the handler.

        If stream is not specified, sys.stderr is used.
        """
        Handler.__init__(self)
        if stream is None:
            stream = sys.stderr
        self.stream = stream

    def flush(self):
        """
        Flushes the stream.
        """
        self.acquire()
        try:
            if self.stream and hasattr(self.stream, "flush"):
                self.stream.flush()
        finally:
            self.release()

    def emit(self, record):
        """
        Emit a record.

        If a formatter is specified, it is used to format the record.
        The record is then written to the stream with a trailing newline.  If
        exception information is present, it is formatted using
        traceback.print_exception and appended to the stream.  If the stream
        has an 'encoding' attribute, it is used to determine how to do the
        output to the stream.
        """
        try:
            msg = self.format(record)
            stream = self.stream
            # issue 35046: merged two stream.writes into one.
            stream.write(msg + self.terminator)
            self.flush()
        except RecursionError:  # See issue 36272
            raise
        except Exception:
            self.handleError(record)

    def setStream(self, stream):
        """
        Sets the StreamHandler's stream to the specified value,
        if it is different.

        Returns the old stream, if the stream was changed, or None
        if it wasn't.
        """
        if stream is self.stream:
            result = None
        else:
            result = self.stream
            self.acquire()
            try:
                self.flush()
                self.stream = stream
            finally:
                self.release()
        return result

    def __repr__(self):
        level = getLevelName(self.level)
        name = getattr(self.stream, 'name', '')
        #  bpo-36015: name can be an int
        name = str(name)
        if name:
            name += ' '
        return '<%s %s(%s)>' % (self.__class__.__name__, name, level)

You could make a modified version such that instead of accepting a reference directly, it accepts a function that looks up the current value of sys.stdout.

class StreamHandler2(StreamHandler):
    def __init__(self, _getstream=None):
        """
        Initialize the handler.

        If stream is not specified, sys.stderr is used.
        """
        Handler.__init__(self)
        if _getstream is None:
            _getstream = lambda: sys.stderr
        self._getstream = _getstream
        
    @property
    def stream(self):
        return self._getstream()

    def setStream(self, stream):
        raise NotImplementedError

Then use StreamHandler2 instead of StreamHandler.

IMO: I'd probably avoid such a workaround and instead perhaps implement a simple handler designed to just append the message to a list, so then in your test you can just check the value of the logging message you are trying to test in that list. FWIW: I almost never use the got-want syntax when I write my doctests. Instead I try to opt for something that could be ported to a standalone unit test with no additional code. In the case I do want to show the "want" value for documentation purposes. I almost always use the # xdoctest: +IGNORE_WANT directive to disable checking, because string representations are simply too system dependent too often.

@ArneBachmannDLR
Copy link
Author

ArneBachmannDLR commented Sep 30, 2021

Thanks for the elaborate support!
I tried your first approach, but it got stuck with an AttributeError. I then implemented a handler that collects the emitted records and allows checking them later - same problem, with doctest it works, with xdoctest only for some functions. Sometimes Got is empty, not sure what's going on.

class TestStreamHandler(logging.StreamHandler):
  ''' This handler allows checking log messages in doctests. '''
  def __init__(_, _log) -> None: _._records:List = []; super().__init__(); _log.addHandler(_); _._log = _log
  def emit(_, record)   -> None: _._records.append(record)
  def records(_) -> List[Tuple[str,str]]: r =  [(record.levelname, record.getMessage()) for record in _._records]; _.reset(); return r
  def lines(_)   -> None: [print(record.getMessage()) for record in _._records]; _.reset()  # WARN print may use wrong stream!
  def __del__(_) -> None: _._log.removeHandler(_)
  def reset(_): _._records.clear()

@Erotemic
Copy link
Owner

For the attribute error, I think defining a property in a child class when the parent class already has an attribute with that name must have issues. This can be done in an alternate way, and this does seem to work. Here is a MWE:

# -*- coding: utf-8 -*-
"""
demo_usage_with_logger.py

Script to demo a workaround to [Issue111]_.


CommandLine:
    # Run with xdoctest runner
    xdoctest ~/code/xdoctest/dev/demo_usage_with_logger.py

    # Run with pytest runner
    pytest -s --xdoctest --xdoctest-verbose=3 ~/code/xdoctest/dev/demo_usage_with_logger.py

    # Run with builtin main
    python ~/code/xdoctest/dev/demo_usage_with_logger.py

References:
    .. [Issue111] https://github.com/Erotemic/xdoctest/issues/111
"""
import logging
import sys


class StreamHandler2(logging.StreamHandler):
    def __init__(self, _getstream=None):
        """
        Initialize the handler.

        If stream is not specified, sys.stderr is used.
        """
        logging.Handler.__init__(self)
        if _getstream is None:
            _getstream = lambda: sys.stderr  # NOQA
        self._getstream = _getstream
        self.__class__.stream = property(lambda self: self._getstream())

    def setStream(self, stream):
        raise NotImplementedError


handler = StreamHandler2(lambda: sys.stdout)

_log = logging.getLogger('mylog')
_log.setLevel(logging.INFO)
_log.addHandler(handler)

_log.info('hello')
_log.info('hello hello')


def func_with_doctest():
    """
    Example:
        >>> _log.info('greetings from my doctest')
        greetings from my doctest
    """


def main():
    import xdoctest
    xdoctest.doctest_callable(func_with_doctest)


if __name__ == '__main__':
    main()

@ArneBachmannDLR
Copy link
Author

Thanks again. I think the trick you gave me here is initiating the handler outside the doctests, which means there is a finalized instance available to all tests and no need to add/remove handlers dynamically. The drawback of this approach is that the handler is always active, even in production when not testing, which means overhead and possible memory problems. To define/initialize the handler only for testing makes my code more complex (see below).

I should mention that I allow testing a file by running it (as __main__) or by running package/tests.py using DoctestRunners, and now adding the option to run via xdoctest package, and each type of invocation needs extra code to make it possible. Python kinda sucks sometimes...

Here's my setup to make it work (in every file):

from __future__ import annotations

__sut__ = __name__ == '__main__'
if __sut__:
  globals()["__name__"] = "package." + __file__.replace(".py", "")
  import os, sys
  sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

_log = logging.getLogger(__name__); debug, info, warn, error = _log.debug, _log.info, _log.warning, _log.error

... # use testlog in doctests, clean via .reset() and get lines via .lines()

if __sut__ or '--test' in sys.argv:
  from package.tests import TestStreamHandler; testlog = TestStreamHandler(_log)
  logging.basicConfig(level=logging.DEBUG if '--debug' in sys.argv else logging.INFO, stream=sys.stderr, format="%(asctime)-25s %(levelname)-8s %(name)-12s:%(lineno)d | %(message)s")
  import doctest; doctest.testmod(optionflags=doctest.ELLIPSIS)

... and this seems to work fine now.

@Erotemic
Copy link
Owner

Logging in general is super weird because you need to ensure that every entry point correctly initializes the loggers. I've never gotten the hang of it. I just fallback to print statements and a verbosity integer 99% of the time.

FWIW xdoctest does have a --global-exec argument which you can specify code that runs before every doctest. That could call a function that initializes the handler once, and then becomes a noop.

It's also curious that you mention regular doctest does seem to work with logging by default? I took a peek at the code, but I didn't see any mechanism for it. If it works with regular doctest, I do want it to work with xdoctest, so I may take some time to investigate this, but unfortunately, I can't make it a priority.

Random aside, what does __sut__ stand for?

@ArneBachmannDLR
Copy link
Author

system-under-test. Yes, I think if it works with doctest, it should work with xdoctest, but the type of incovation is different, as I don't run from command line, but from code:

from doctest import testmod, ELLIPSIS, IGNORE_EXCEPTION_DETAIL

def runAllDoctests() -> int:
  to_test_files:List[str] = [filename for filename in os.listdir('./package')
                          if filename.endswith('.py')
                         and os.path.isfile('package' + os.sep + filename)
                         and filename[0] != '_'
                         and filename not in {
                            "constants.py",
                            "environment.py",
                           ...
                          }
                            ]  # find testable source files
  to_test_mods:List = [importlib.import_module('package.' + test.replace('.py', '')) for test in to_test_files]
  importlib.invalidate_caches()

  fails = total = 0
  for mod in to_test_mods:
    failed, trials = testmod(mod, optionflags=ELLIPSIS|IGNORE_EXCEPTION_DETAIL)
    print((f"{failed}/{trials} tests failed" if failed > 0 else f"All {trials} tests passed") + f" for {mod.__name__}")
    fails += failed; total += trials
  print(f'{fails}/{total} tests failed.' if fails > 0 else f'All {total} tests passed.')
  return max(1, fails) if total == 0 or fails > 0 else fails

if __name__ == '__main__': sys.exit(runAllDoctests())

To be fair, I might need to replace this code with your variant of invoking xdoctests, and then compare if it works.

@Erotemic
Copy link
Owner

Erotemic commented Oct 1, 2021

The analogous code to doctest.testmod would be xdoctest.doctest_module. It returns a dictionary that contains summary statistics, so it should be fairly straightforward switch between doctest and xdoctest.

I'm also open to adding a xdoctest.testmod function as a wrapper such that xdoctest can mimic the doctest API.

@Erotemic
Copy link
Owner

Closing, reopen if you feel this is still an issue that xdoctest itself should address.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
Labels
None yet
Projects
None yet
Development

No branches or pull requests

2 participants