text: # -*- coding: utf-8 -*-
  import sys
  import logging
  import unittest
  
  from testfixtures import LogCapture
  from twisted.python.failure import Failure
  
  from scrapy.utils.log import (failure_to_exc_info, TopLevelFormatter,
                                LogCounterHandler, StreamLogger)
  from scrapy.utils.test import get_crawler
  from scrapy.extensions import telnet
  
  
  class FailureToExcInfoTest(unittest.TestCase):
  
      def test_failure(self):
          try:
              0/0
          except ZeroDivisionError:
              exc_info = sys.exc_info()
              failure = Failure()
  
          self.assertTupleEqual(exc_info, failure_to_exc_info(failure))
  
      def test_non_failure(self):
          self.assertIsNone(failure_to_exc_info('test'))
  
  
  class TopLevelFormatterTest(unittest.TestCase):
  
      def setUp(self):
          self.handler = LogCapture()
          self.handler.addFilter(TopLevelFormatter(['test']))
  
      def test_top_level_logger(self):
          logger = logging.getLogger('test')
          with self.handler as l:
              logger.warning('test log msg')
  
          l.check(('test', 'WARNING', 'test log msg'))
  
      def test_children_logger(self):
          logger = logging.getLogger('test.test1')
          with self.handler as l:
              logger.warning('test log msg')
  
          l.check(('test', 'WARNING', 'test log msg'))
  
      def test_overlapping_name_logger(self):
          logger = logging.getLogger('test2')
          with self.handler as l:
              logger.warning('test log msg')
  
          l.check(('test2', 'WARNING', 'test log msg'))
  
      def test_different_name_logger(self):
          logger = logging.getLogger('different')
          with self.handler as l:
              logger.warning('test log msg')
  
          l.check(('different', 'WARNING', 'test log msg'))
  
  
  class LogCounterHandlerTest(unittest.TestCase):
  
      def setUp(self):
          settings = {'LOG_LEVEL': 'WARNING'}
          if not telnet.TWISTED_CONCH_AVAILABLE:
              # disable it to avoid the extra warning
              settings['TELNETCONSOLE_ENABLED'] = False
          self.logger = logging.getLogger('test')
          self.logger.setLevel(logging.NOTSET)
          self.logger.propagate = False
          self.crawler = get_crawler(settings_dict=settings)
          self.handler = LogCounterHandler(self.crawler)
          self.logger.addHandler(self.handler)
  
      def tearDown(self):
          self.logger.propagate = True
          self.logger.removeHandler(self.handler)
  
      def test_init(self):
          self.assertIsNone(self.crawler.stats.get_value('log_count/DEBUG'))
          self.assertIsNone(self.crawler.stats.get_value('log_count/INFO'))
          self.assertIsNone(self.crawler.stats.get_value('log_count/WARNING'))
          self.assertIsNone(self.crawler.stats.get_value('log_count/ERROR'))
          self.assertIsNone(self.crawler.stats.get_value('log_count/CRITICAL'))
  
      def test_accepted_level(self):
          self.logger.error('test log msg')
          self.assertEqual(self.crawler.stats.get_value('log_count/ERROR'), 1)
  
      def test_filtered_out_level(self):
          self.logger.debug('test log msg')
          self.assertIsNone(self.crawler.stats.get_value('log_count/INFO'))
  
  
  class StreamLoggerTest(unittest.TestCase):
  
      def setUp(self):
          self.stdout = sys.stdout
          logger = logging.getLogger('test')
          logger.setLevel(logging.WARNING)
          sys.stdout = StreamLogger(logger, logging.ERROR)
  
      def tearDown(self):
          sys.stdout = self.stdout
  
      def test_redirect(self):
          with LogCapture() as l:
              print('test log msg')
          l.check(('test', 'ERROR', 'test log msg'))
  