def test_status_iterator(app, status, warning):
    logging.setup(app, status, warning)

    # test for old_status_iterator
    status.truncate(0)
    yields = list(status_iterator(['hello', 'sphinx', 'world'], 'testing ... '))
    output = strip_escseq(status.getvalue())
    assert 'testing ... hello sphinx world \n' in output
    assert yields == ['hello', 'sphinx', 'world']

    # test for status_iterator (verbosity=0)
    status.truncate(0)
    yields = list(status_iterator(['hello', 'sphinx', 'world'], 'testing ... ',
                                  length=3, verbosity=0))
    output = strip_escseq(status.getvalue())
    assert 'testing ... [ 33%] hello                \r' in output
    assert 'testing ... [ 66%] sphinx               \r' in output
    assert 'testing ... [100%] world                \r\n' in output
    assert yields == ['hello', 'sphinx', 'world']

    # test for status_iterator (verbosity=1)
    status.truncate(0)
    yields = list(status_iterator(['hello', 'sphinx', 'world'], 'testing ... ',
                                  length=3, verbosity=1))
    output = strip_escseq(status.getvalue())
    assert 'testing ... [ 33%] hello\n' in output
    assert 'testing ... [ 66%] sphinx\n' in output
    assert 'testing ... [100%] world\n\n' in output
    assert yields == ['hello', 'sphinx', 'world']

def test_warning_location(app, status, warning):
    logging.setup(app, status, warning)
    logger = logging.getLogger(__name__)

    logger.warning('message1', location='index')
    assert 'index.txt: WARNING: message1' in warning.getvalue()

    logger.warning('message2', location=('index', 10))
    assert 'index.txt:10: WARNING: message2' in warning.getvalue()

    logger.warning('message3', location=None)
    assert colorize('darkred', 'WARNING: message3') in warning.getvalue()

    node = nodes.Node()
    node.source, node.line = ('index.txt', 10)
    logger.warning('message4', location=node)
    assert 'index.txt:10: WARNING: message4' in warning.getvalue()

    node.source, node.line = ('index.txt', None)
    logger.warning('message5', location=node)
    assert 'index.txt:: WARNING: message5' in warning.getvalue()

    node.source, node.line = (None, 10)
    logger.warning('message6', location=node)
    assert '<unknown>:10: WARNING: message6' in warning.getvalue()

    node.source, node.line = (None, None)
    logger.warning('message7', location=node)
    assert colorize('darkred', 'WARNING: message7') in warning.getvalue()

def test_suppress_warnings(app, status, warning):
    logging.setup(app, status, warning)
    logger = logging.getLogger(__name__)

    app._warncount = 0  # force reset

    app.config.suppress_warnings = []
    warning.truncate(0)
    logger.warning('message1', type='test', subtype='logging')
    logger.warning('message2', type='test', subtype='crash')
    logger.warning('message3', type='actual', subtype='logging')
    assert 'message1' in warning.getvalue()
    assert 'message2' in warning.getvalue()
    assert 'message3' in warning.getvalue()
    assert app._warncount == 3

    app.config.suppress_warnings = ['test']
    warning.truncate(0)
    logger.warning('message1', type='test', subtype='logging')
    logger.warning('message2', type='test', subtype='crash')
    logger.warning('message3', type='actual', subtype='logging')
    assert 'message1' not in warning.getvalue()
    assert 'message2' not in warning.getvalue()
    assert 'message3' in warning.getvalue()
    assert app._warncount == 4

    app.config.suppress_warnings = ['test.logging']
    warning.truncate(0)
    logger.warning('message1', type='test', subtype='logging')
    logger.warning('message2', type='test', subtype='crash')
    logger.warning('message3', type='actual', subtype='logging')
    assert 'message1' not in warning.getvalue()
    assert 'message2' in warning.getvalue()
    assert 'message3' in warning.getvalue()
    assert app._warncount == 6

def test_info_location(app, status, warning):
    logging.setup(app, status, warning)
    logger = logging.getLogger(__name__)

    logger.info('message1', location='index')
    assert 'index.txt: message1' in status.getvalue()

    logger.info('message2', location=('index', 10))
    assert 'index.txt:10: message2' in status.getvalue()

    logger.info('message3', location=None)
    assert '\nmessage3' in status.getvalue()

    node = nodes.Node()
    node.source, node.line = ('index.txt', 10)
    logger.info('message4', location=node)
    assert 'index.txt:10: message4' in status.getvalue()

    node.source, node.line = ('index.txt', None)
    logger.info('message5', location=node)
    assert 'index.txt:: message5' in status.getvalue()

    node.source, node.line = (None, 10)
    logger.info('message6', location=node)
    assert '<unknown>:10: message6' in status.getvalue()

    node.source, node.line = (None, None)
    logger.info('message7', location=node)
    assert '\nmessage7' in status.getvalue()

def test_new_documenter():
    logging.setup(app, app._status, app._warning)

    class MyDocumenter(ModuleLevelDocumenter):
        objtype = 'integer'
        directivetype = 'data'
        priority = 100

        @classmethod
        def can_document_member(cls, member, membername, isattr, parent):
            return isinstance(member, int)

        def document_members(self, all_members=False):
            return

    add_documenter(MyDocumenter)

    def assert_result_contains(item, objtype, name, **kw):
        app._warning.truncate(0)
        inst = app.registry.documenters[objtype](directive, name)
        inst.generate(**kw)
        # print '\n'.join(directive.result)
        assert app._warning.getvalue() == ''
        assert item in directive.result
        del directive.result[:]

    options.members = ['integer']
    assert_result_contains('.. py:data:: integer', 'module', 'target')

def test_colored_logs(app, status, warning):
    app.verbosity = 2
    logging.setup(app, status, warning)
    logger = logging.getLogger(__name__)

    # default colors
    logger.debug('message1')
    logger.verbose('message2')
    logger.info('message3')
    logger.warning('message4')
    logger.critical('message5')
    logger.error('message6')

    assert colorize('darkgray', 'message1') in status.getvalue()
    assert 'message2\n' in status.getvalue()  # not colored
    assert 'message3\n' in status.getvalue()  # not colored
    assert colorize('darkred', 'WARNING: message4') in warning.getvalue()
    assert 'WARNING: message5\n' in warning.getvalue()  # not colored
    assert 'WARNING: message6\n' in warning.getvalue()  # not colored

    # color specification
    logger.debug('message7', color='white')
    logger.info('message8', color='red')
    assert colorize('white', 'message7') in status.getvalue()
    assert colorize('red', 'message8') in status.getvalue()

def test_add_is_parallel_allowed(app, status, warning):
    logging.setup(app, status, warning)

    assert app.is_parallel_allowed('read') is True
    assert app.is_parallel_allowed('write') is True
    assert warning.getvalue() == ''

    app.setup_extension('read_parallel')
    assert app.is_parallel_allowed('read') is True
    assert app.is_parallel_allowed('write') is True
    assert warning.getvalue() == ''
    app.extensions.pop('read_parallel')

    app.setup_extension('write_parallel')
    assert app.is_parallel_allowed('read') is False
    assert app.is_parallel_allowed('write') is True
    assert ("the write_parallel extension does not declare if it is safe "
            "for parallel reading, assuming it isn't - please ") in warning.getvalue()
    app.extensions.pop('write_parallel')
    warning.truncate(0)  # reset warnings

    app.setup_extension('read_serial')
    assert app.is_parallel_allowed('read') is False
    assert app.is_parallel_allowed('write') is True
    assert warning.getvalue() == ''
    app.extensions.pop('read_serial')

    app.setup_extension('write_serial')
    assert app.is_parallel_allowed('read') is False
    assert app.is_parallel_allowed('write') is False
    assert ("the write_serial extension does not declare if it is safe "
            "for parallel reading, assuming it isn't - please ") in warning.getvalue()
    app.extensions.pop('write_serial')
    warning.truncate(0)  # reset warnings

def test_nonl_info_log(app, status, warning):
    logging.setup(app, status, warning)
    logger = logging.getLogger(__name__)

    logger.info('message1', nonl=True)
    logger.info('message2')
    logger.info('message3')

    assert 'message1message2\nmessage3' in status.getvalue()

def test_warningiserror(app, status, warning):
    logging.setup(app, status, warning)
    logger = logging.getLogger(__name__)

    # if False, warning is not error
    app.warningiserror = False
    logger.warning('message')

    # if True, warning raises SphinxWarning exception
    app.warningiserror = True
    with pytest.raises(SphinxWarning):
        logger.warning('message')

def test_logging_in_ParallelTasks(app, status, warning):
    logging.setup(app, status, warning)
    logger = logging.getLogger(__name__)

    def child_process():
        logger.info('message1')
        logger.warning('message2', location='index')

    tasks = ParallelTasks(1)
    tasks.add_task(child_process)
    tasks.join()
    assert 'message1' in status.getvalue()
    assert 'index.txt: WARNING: message2' in warning.getvalue()

def test_output_with_unencodable_char(app, status, warning):
    class StreamWriter(codecs.StreamWriter):
        def write(self, object):
            self.stream.write(object.encode('cp1252').decode('cp1252'))

    logging.setup(app, StreamWriter(status), warning)
    logger = logging.getLogger(__name__)

    # info with UnicodeEncodeError
    status.truncate(0)
    status.seek(0)
    logger.info(u"unicode \u206d...")
    assert status.getvalue() == "unicode ?...\n"

def test_execfile_python2(capsys, app, status, warning):
    logging.setup(app, status, warning)

    ns = {}
    with tempfile.NamedTemporaryFile() as tmp:
        tmp.write(b'print "hello"\n')
        tmp.flush()
        execfile_(tmp.name, ns)
    msg = (
        'Support for evaluating Python 2 syntax is deprecated '
        'and will be removed in Sphinx 4.0. '
        'Convert %s to Python 3 syntax.\n' % tmp.name)
    assert msg in strip_escseq(warning.getvalue())
    captured = capsys.readouterr()
    assert captured.out == 'hello\n'

def test_pending_warnings(app, status, warning):
    logging.setup(app, status, warning)
    logger = logging.getLogger(__name__)

    logger.warning('message1')
    with logging.pending_warnings():
        # not logged yet (bufferred) in here
        logger.warning('message2')
        logger.warning('message3')
        assert 'WARNING: message1' in warning.getvalue()
        assert 'WARNING: message2' not in warning.getvalue()
        assert 'WARNING: message3' not in warning.getvalue()

    # actually logged as ordered
    assert 'WARNING: message2\nWARNING: message3' in strip_escseq(warning.getvalue())

def test_warningiserror(app, status, warning):
    logging.setup(app, status, warning)
    logger = logging.getLogger(__name__)

    # if False, warning is not error
    app.warningiserror = False
    logger.warning('message')

    # if True, warning raises SphinxWarning exception
    app.warningiserror = True
    with pytest.raises(SphinxWarning):
        logger.warning('message: %s', 'arg')

    # message contains format string (refs: #4070)
    with pytest.raises(SphinxWarning):
        logger.warning('%s')

def test_prefixed_warnings(app, status, warning):
    logging.setup(app, status, warning)
    logger = logging.getLogger(__name__)

    logger.warning('message1')
    with prefixed_warnings('PREFIX:'):
        logger.warning('message2')
        with prefixed_warnings('Another PREFIX:'):
            logger.warning('message3')
        logger.warning('message4')
    logger.warning('message5')

    assert 'WARNING: message1' in warning.getvalue()
    assert 'WARNING: PREFIX: message2' in warning.getvalue()
    assert 'WARNING: Another PREFIX: message3' in warning.getvalue()
    assert 'WARNING: PREFIX: message4' in warning.getvalue()
    assert 'WARNING: message5' in warning.getvalue()

def test_parse_name():
    logging.setup(app, app._status, app._warning)

    def verify(objtype, name, result):
        inst = app.registry.documenters[objtype](directive, name)
        assert inst.parse_name()
        assert (inst.modname, inst.objpath, inst.args, inst.retann) == result

    # for modules
    verify('module', 'test_autodoc', ('test_autodoc', [], None, None))
    verify('module', 'test.test_autodoc', ('test.test_autodoc', [], None, None))
    verify('module', 'test(arg)', ('test', [], 'arg', None))
    assert 'signature arguments' in app._warning.getvalue()

    # for functions/classes
    verify('function', 'test_autodoc.raises',
           ('test_autodoc', ['raises'], None, None))
    verify('function', 'test_autodoc.raises(exc) -> None',
           ('test_autodoc', ['raises'], 'exc', 'None'))
    directive.env.temp_data['autodoc:module'] = 'test_autodoc'
    verify('function', 'raises', ('test_autodoc', ['raises'], None, None))
    del directive.env.temp_data['autodoc:module']
    directive.env.ref_context['py:module'] = 'test_autodoc'
    verify('function', 'raises', ('test_autodoc', ['raises'], None, None))
    verify('class', 'Base', ('test_autodoc', ['Base'], None, None))

    # for members
    directive.env.ref_context['py:module'] = 'foo'
    verify('method', 'util.SphinxTestApp.cleanup',
           ('util', ['SphinxTestApp', 'cleanup'], None, None))
    directive.env.ref_context['py:module'] = 'util'
    directive.env.ref_context['py:class'] = 'Foo'
    directive.env.temp_data['autodoc:class'] = 'SphinxTestApp'
    verify('method', 'cleanup', ('util', ['SphinxTestApp', 'cleanup'], None, None))
    verify('method', 'SphinxTestApp.cleanup',
           ('util', ['SphinxTestApp', 'cleanup'], None, None))

    # and clean up
    del directive.env.ref_context['py:module']
    del directive.env.ref_context['py:class']
    del directive.env.temp_data['autodoc:class']

def test_skip_warningiserror(app, status, warning):
    logging.setup(app, status, warning)
    logger = logging.getLogger(__name__)

    app.warningiserror = True
    with logging.skip_warningiserror():
        logger.warning('message')

    # if False, warning raises SphinxWarning exception
    with pytest.raises(SphinxWarning):
        with logging.skip_warningiserror(False):
            logger.warning('message')

    # It also works during pending_warnings.
    with logging.pending_warnings():
        with logging.skip_warningiserror():
            logger.warning('message')

    with pytest.raises(SphinxWarning):
        with logging.pending_warnings():
            with logging.skip_warningiserror(False):
                logger.warning('message')

def test_info_and_warning(app, status, warning):
    app.verbosity = 2
    logging.setup(app, status, warning)
    logger = logging.getLogger(__name__)

    logger.debug('message1')
    logger.info('message2')
    logger.warning('message3')
    logger.critical('message4')
    logger.error('message5')

    assert 'message1' in status.getvalue()
    assert 'message2' in status.getvalue()
    assert 'message3' not in status.getvalue()
    assert 'message4' not in status.getvalue()
    assert 'message5' not in status.getvalue()

    assert 'message1' not in warning.getvalue()
    assert 'message2' not in warning.getvalue()
    assert 'message3' in warning.getvalue()
    assert 'message4' in warning.getvalue()
    assert 'message5' in warning.getvalue()

def test_verbosity_filter(app, status, warning):
    # verbosity = 0: INFO
    app.verbosity = 0
    logging.setup(app, status, warning)
    logger = logging.getLogger(__name__)

    logger.info('message1')
    logger.verbose('message2')
    logger.debug('message3')

    assert 'message1' in status.getvalue()
    assert 'message2' not in status.getvalue()
    assert 'message3' not in status.getvalue()
    assert 'message4' not in status.getvalue()

    # verbosity = 1: VERBOSE
    app.verbosity = 1
    logging.setup(app, status, warning)
    logger = logging.getLogger(__name__)

    logger.info('message1')
    logger.verbose('message2')
    logger.debug('message3')

    assert 'message1' in status.getvalue()
    assert 'message2' in status.getvalue()
    assert 'message3' not in status.getvalue()
    assert 'message4' not in status.getvalue()

    # verbosity = 2: DEBUG
    app.verbosity = 2
    logging.setup(app, status, warning)
    logger = logging.getLogger(__name__)

    logger.info('message1')
    logger.verbose('message2')
    logger.debug('message3')

    assert 'message1' in status.getvalue()
    assert 'message2' in status.getvalue()
    assert 'message3' in status.getvalue()
    assert 'message4' not in status.getvalue()

def test_generate():
    logging.setup(app, app._status, app._warning)

    def assert_warns(warn_str, objtype, name, **kw):
        inst = app.registry.documenters[objtype](directive, name)
        inst.generate(**kw)
        assert len(directive.result) == 0, directive.result

        assert warn_str in app._warning.getvalue()
        app._warning.truncate(0)

    def assert_works(objtype, name, **kw):
        inst = app.registry.documenters[objtype](directive, name)
        inst.generate(**kw)
        assert directive.result
        # print '\n'.join(directive.result)
        assert app._warning.getvalue() == ''
        del directive.result[:]

    def assert_processes(items, objtype, name, **kw):
        del processed_docstrings[:]
        del processed_signatures[:]
        assert_works(objtype, name, **kw)
        assert set(processed_docstrings) | set(processed_signatures) == set(items)

    def assert_result_contains(item, objtype, name, **kw):
        inst = app.registry.documenters[objtype](directive, name)
        inst.generate(**kw)
        # print '\n'.join(directive.result)
        assert app._warning.getvalue() == ''
        assert item in directive.result
        del directive.result[:]

    def assert_order(items, objtype, name, member_order, **kw):
        inst = app.registry.documenters[objtype](directive, name)
        inst.options.member_order = member_order
        inst.generate(**kw)
        assert app._warning.getvalue() == ''
        items = list(reversed(items))
        lineiter = iter(directive.result)
        # for line in directive.result:
        #     if line.strip():
        #         print repr(line)
        while items:
            item = items.pop()
            for line in lineiter:
                if line == item:
                    break
            else:  # ran out of items!
                assert False, ('item %r not found in result or not in the '
                               ' correct order' % item)
        del directive.result[:]

    options.members = []

    # no module found?
    assert_warns("import for autodocumenting 'foobar'",
                 'function', 'foobar', more_content=None)
    # importing
    assert_warns("failed to import module 'test_foobar'",
                 'module', 'test_foobar', more_content=None)
    # attributes missing
    assert_warns("failed to import function 'foobar' from module 'util'",
                 'function', 'util.foobar', more_content=None)
    # method missing
    assert_warns("failed to import method 'Class.foobar' from module 'target';",
                 'method', 'target.Class.foobar', more_content=None)

    # test auto and given content mixing
    directive.env.ref_context['py:module'] = 'target'
    assert_result_contains('   Function.', 'method', 'Class.meth')
    add_content = ViewList()
    add_content.append('Content.', '', 0)
    assert_result_contains('   Function.', 'method',
                           'Class.meth', more_content=add_content)
    assert_result_contains('   Content.', 'method',
                           'Class.meth', more_content=add_content)

    # test check_module
    inst = FunctionDocumenter(directive, 'add_documenter')
    inst.generate(check_module=True)
    assert len(directive.result) == 0

    # assert that exceptions can be documented
    assert_works('exception', 'target.CustomEx', all_members=True)
    assert_works('exception', 'target.CustomEx')

    # test diverse inclusion settings for members
    should = [('class', 'target.Class')]
    assert_processes(should, 'class', 'Class')
    should.extend([('method', 'target.Class.meth')])
    options.members = ['meth']
    options.exclude_members = set(['excludemeth'])
    assert_processes(should, 'class', 'Class')
    should.extend([('attribute', 'target.Class.prop'),
                   ('attribute', 'target.Class.descr'),
                   ('attribute', 'target.Class.attr'),
                   ('attribute', 'target.Class.docattr'),
                   ('attribute', 'target.Class.udocattr'),
                   ('attribute', 'target.Class.mdocattr'),
#.........这里部分代码省略.........