@contextmanager
def swallow_logs(new_level=None, file_=None, name='datalad'):
    'Context manager to consume all logs.\n\n    '
    lgr = logging.getLogger(name)
    old_level = lgr.level
    old_handlers = lgr.handlers

    class StringIOAdapter(object):
        'Little adapter to help getting out values\n\n        And to stay consistent with how swallow_outputs behaves\n        '

        def __init__(self):
            if (file_ is None):
                kw = get_tempfile_kwargs({}, prefix='logs')
                self._out = NamedTemporaryFile(mode='a', delete=False, **kw)
            else:
                out_file = file_
                self._out = open(out_file, 'a')
            self._final_out = None

        def _read(self, h):
            with open(h.name) as f:
                return f.read()

        @property
        def out(self):
            if (self._final_out is not None):
                return self._final_out
            else:
                self._out.flush()
                return self._read(self._out)

        @property
        def lines(self):
            return self.out.split('\n')

        @property
        def handle(self):
            return self._out

        def cleanup(self):
            self._final_out = self.out
            self._out.close()
            out_name = self._out.name
            del self._out
            gc.collect()
            if (not file_):
                rmtemp(out_name)

        def assert_logged(self, msg=None, level=None, regex=True, **kwargs):
            'Provide assertion on whether a msg was logged at a given level\n\n            If neither `msg` nor `level` provided, checks if anything was logged\n            at all.\n\n            Parameters\n            ----------\n            msg: str, optional\n              Message (as a regular expression, if `regex`) to be searched.\n              If no msg provided, checks if anything was logged at a given level.\n            level: str, optional\n              String representing the level to be logged\n            regex: bool, optional\n              If False, regular `assert_in` is used\n            **kwargs: str, optional\n              Passed to `assert_re_in` or `assert_in`\n            '
            from datalad.tests.utils import assert_re_in
            from datalad.tests.utils import assert_in
            if regex:
                match = (('\\[%s\\] ' % level) if level else '\\[\\S+\\] ')
            else:
                match = (('[%s] ' % level) if level else '')
            if msg:
                match += msg
            if match:
                (assert_re_in if regex else assert_in)(match, self.out, **kwargs)
            else:
                assert (not kwargs), 'no kwargs to be passed anywhere'
                assert self.out, 'Nothing was logged!?'
    adapter = StringIOAdapter()
    swallow_handler = logging.StreamHandler(adapter.handle)
    swallow_handler.setFormatter(logging.Formatter('[%(levelname)s] %(message)s'))
    swallow_handler.filters = sum([h.filters for h in old_handlers], [])
    lgr.handlers = [swallow_handler]
    if (old_level < logging.DEBUG):
        lgr.handlers += old_handlers
    if isinstance(new_level, str):
        new_level = getattr(logging, new_level)
    if (new_level is not None):
        lgr.setLevel(new_level)
    try:
        (yield adapter)
    finally:
        lgr.handlers = old_handlers
        lgr.setLevel(old_level)
        adapter.cleanup()
Context manager to consume all logs.
@contextmanager
def swallow_logs(new_level=None, file_=None, name='datalad'):
    '\n\n    '
    lgr = logging.getLogger(name)
    old_level = lgr.level
    old_handlers = lgr.handlers

    class StringIOAdapter(object):
        'Little adapter to help getting out values\n\n        And to stay consistent with how swallow_outputs behaves\n        '

        def __init__(self):
            if (file_ is None):
                kw = get_tempfile_kwargs({}, prefix='logs')
                self._out = NamedTemporaryFile(mode='a', delete=False, **kw)
            else:
                out_file = file_
                self._out = open(out_file, 'a')
            self._final_out = None

        def _read(self, h):
            with open(h.name) as f:
                return f.read()

        @property
        def out(self):
            if (self._final_out is not None):
                return self._final_out
            else:
                self._out.flush()
                return self._read(self._out)

        @property
        def lines(self):
            return self.out.split('\n')

        @property
        def handle(self):
            return self._out

        def cleanup(self):
            self._final_out = self.out
            self._out.close()
            out_name = self._out.name
            del self._out
            gc.collect()
            if (not file_):
                rmtemp(out_name)

        def assert_logged(self, msg=None, level=None, regex=True, **kwargs):
            'Provide assertion on whether a msg was logged at a given level\n\n            If neither `msg` nor `level` provided, checks if anything was logged\n            at all.\n\n            Parameters\n            ----------\n            msg: str, optional\n              Message (as a regular expression, if `regex`) to be searched.\n              If no msg provided, checks if anything was logged at a given level.\n            level: str, optional\n              String representing the level to be logged\n            regex: bool, optional\n              If False, regular `assert_in` is used\n            **kwargs: str, optional\n              Passed to `assert_re_in` or `assert_in`\n            '
            from datalad.tests.utils import assert_re_in
            from datalad.tests.utils import assert_in
            if regex:
                match = (('\\[%s\\] ' % level) if level else '\\[\\S+\\] ')
            else:
                match = (('[%s] ' % level) if level else )
            if msg:
                match += msg
            if match:
                (assert_re_in if regex else assert_in)(match, self.out, **kwargs)
            else:
                assert (not kwargs), 'no kwargs to be passed anywhere'
                assert self.out, 'Nothing was logged!?'
    adapter = StringIOAdapter()
    swallow_handler = logging.StreamHandler(adapter.handle)
    swallow_handler.setFormatter(logging.Formatter('[%(levelname)s] %(message)s'))
    swallow_handler.filters = sum([h.filters for h in old_handlers], [])
    lgr.handlers = [swallow_handler]
    if (old_level < logging.DEBUG):
        lgr.handlers += old_handlers
    if isinstance(new_level, str):
        new_level = getattr(logging, new_level)
    if (new_level is not None):
        lgr.setLevel(new_level)
    try:
        (yield adapter)
    finally:
        lgr.handlers = old_handlers
        lgr.setLevel(old_level)
        adapter.cleanup()