Gecko [ dataharbor.africa ]

Name	Size	Permission
?;fixtures	[ DIR ]	drwxr-xr-x
__init__.py	20 B	-rw-r--r--
__init__.pyc	148 B	-rw-r--r--
__init__.pyo	148 B	-rw-r--r--
base.py	33.88 KB	-rw-r--r--
base.pyc	52.56 KB	-rw-r--r--
base.pyo	52.56 KB	-rw-r--r--
test_childutils.py	5.27 KB	-rw-r--r--
test_childutils.pyc	7.63 KB	-rw-r--r--
test_childutils.pyo	7.63 KB	-rw-r--r--
test_confecho.py	540 B	-rw-r--r--
test_confecho.pyc	1.25 KB	-rw-r--r--
test_confecho.pyo	1.25 KB	-rw-r--r--
test_datatypes.py	26.65 KB	-rw-r--r--
test_datatypes.pyc	47.5 KB	-rw-r--r--
test_datatypes.pyo	47.5 KB	-rw-r--r--
test_dispatchers.py	47.68 KB	-rw-r--r--
test_dispatchers.pyc	47.49 KB	-rw-r--r--
test_dispatchers.pyo	47.49 KB	-rw-r--r--
test_events.py	20.35 KB	-rw-r--r--
test_events.pyc	24.96 KB	-rw-r--r--
test_events.pyo	24.96 KB	-rw-r--r--
test_http.py	24.48 KB	-rw-r--r--
test_http.pyc	34.38 KB	-rw-r--r--
test_http.pyo	34.38 KB	-rw-r--r--
test_loggers.py	12.65 KB	-rw-r--r--
test_loggers.pyc	18.18 KB	-rw-r--r--
test_loggers.pyo	18.18 KB	-rw-r--r--
test_options.py	132.88 KB	-rw-r--r--
test_options.pyc	136.01 KB	-rw-r--r--
test_options.pyo	136.01 KB	-rw-r--r--
test_poller.py	16.29 KB	-rw-r--r--
test_poller.pyc	20.74 KB	-rw-r--r--
test_poller.pyo	20.5 KB	-rw-r--r--
test_process.py	91.59 KB	-rw-r--r--
test_process.pyc	93.63 KB	-rw-r--r--
test_process.pyo	93.63 KB	-rw-r--r--
test_rpcinterfaces.py	94.86 KB	-rw-r--r--
test_rpcinterfaces.pyc	80.93 KB	-rw-r--r--
test_rpcinterfaces.pyo	80.93 KB	-rw-r--r--
test_socket_manager.py	7.84 KB	-rw-r--r--
test_socket_manager.pyc	11.29 KB	-rw-r--r--
test_socket_manager.pyo	11.29 KB	-rw-r--r--
test_states.py	2.22 KB	-rw-r--r--
test_states.pyc	4.4 KB	-rw-r--r--
test_states.pyo	4.4 KB	-rw-r--r--
test_supervisorctl.py	66.26 KB	-rw-r--r--
test_supervisorctl.pyc	87.29 KB	-rw-r--r--
test_supervisorctl.pyo	87.29 KB	-rw-r--r--
test_supervisord.py	29.39 KB	-rw-r--r--
test_supervisord.pyc	27.4 KB	-rw-r--r--
test_supervisord.pyo	27.4 KB	-rw-r--r--
test_web.py	6.67 KB	-rw-r--r--
test_web.pyc	9.52 KB	-rw-r--r--
test_web.pyo	9.52 KB	-rw-r--r--
test_xmlrpc.py	34.05 KB	-rw-r--r--
test_xmlrpc.pyc	47.64 KB	-rw-r--r--
test_xmlrpc.pyo	47.64 KB	-rw-r--r--

Code Editor : test_loggers.py

import sys
import unittest
import tempfile
import shutil
import os
import syslog

import mock

from supervisor.tests.base import DummyStream

class LevelTests(unittest.TestCase):
    def test_LOG_LEVELS_BY_NUM_doesnt_include_builtins(self):
        from supervisor import loggers
        for level_name in loggers.LOG_LEVELS_BY_NUM.values():
            self.assertFalse(level_name.startswith('_'))

class HandlerTests:
    def setUp(self):
        self.basedir = tempfile.mkdtemp()
        self.filename = os.path.join(self.basedir, 'thelog')

def tearDown(self):
        try:
            shutil.rmtree(self.basedir)
        except OSError:
            pass

def _makeOne(self, *arg, **kw):
        klass = self._getTargetClass()
        return klass(*arg, **kw)

def _makeLogRecord(self, msg):
        from supervisor import loggers
        record = loggers.LogRecord(level=loggers.LevelsByName.INFO,
                                   msg=msg,
                                   exc_info=None)
        return record

class FileHandlerTests(HandlerTests, unittest.TestCase):
    def _getTargetClass(self):
        from supervisor.loggers import FileHandler
        return FileHandler

def test_ctor(self):
        handler = self._makeOne(self.filename)
        self.assertTrue(os.path.exists(self.filename), self.filename)
        self.assertEqual(handler.mode, 'a')
        self.assertEqual(handler.baseFilename, self.filename)
        self.assertEqual(handler.stream.name, self.filename)

def test_close(self):
        handler = self._makeOne(self.filename)
        handler.stream = DummyStream()
        handler.close()
        self.assertEqual(handler.stream.closed, True)

def test_close_raises(self):
        handler = self._makeOne(self.filename)
        handler.stream = DummyStream(OSError)
        self.assertRaises(OSError, handler.close)
        self.assertEqual(handler.stream.closed, False)

def test_reopen(self):
        handler = self._makeOne(self.filename)
        stream = DummyStream()
        handler.stream = stream
        handler.reopen()
        self.assertEqual(stream.closed, True)
        self.assertEqual(handler.stream.name, self.filename)

def test_reopen_raises(self):
        handler = self._makeOne(self.filename)
        stream = DummyStream()
        handler.stream = stream
        handler.baseFilename = os.path.join(self.basedir, 'notthere', 'a.log')
        self.assertRaises(IOError, handler.reopen)
        self.assertEqual(stream.closed, True)

def test_remove_exists(self):
        handler = self._makeOne(self.filename)
        self.assertTrue(os.path.exists(self.filename), self.filename)
        handler.remove()
        self.assertFalse(os.path.exists(self.filename), self.filename)

def test_remove_doesntexist(self):
        handler = self._makeOne(self.filename)
        os.remove(self.filename)
        self.assertFalse(os.path.exists(self.filename), self.filename)
        handler.remove() # should not raise
        self.assertFalse(os.path.exists(self.filename), self.filename)

def test_remove_raises(self):
        handler = self._makeOne(self.filename)
        os.remove(self.filename)
        os.mkdir(self.filename)
        self.assertTrue(os.path.exists(self.filename), self.filename)
        self.assertRaises(OSError, handler.remove)

def test_emit_ascii_noerror(self):
        handler = self._makeOne(self.filename)
        record = self._makeLogRecord('hello!')
        handler.emit(record)
        content = open(self.filename, 'r').read()
        self.assertEqual(content, 'hello!')

def test_emit_unicode_noerror(self):
        handler = self._makeOne(self.filename)
        record = self._makeLogRecord(u'fi\xed')
        handler.emit(record)
        content = open(self.filename, 'r').read()
        self.assertEqual(content, 'fi\xc3\xad')

def test_emit_error(self):
        handler = self._makeOne(self.filename)
        handler.stream = DummyStream(error=OSError)
        record = self._makeLogRecord('hello!')
        try:
            old_stderr = sys.stderr
            dummy_stderr = DummyStream()
            sys.stderr = dummy_stderr
            handler.emit(record)
        finally:
            sys.stderr = old_stderr

self.assertTrue(dummy_stderr.written.endswith('OSError\n'),
                        dummy_stderr.written)

class RotatingFileHandlerTests(FileHandlerTests):

def _getTargetClass(self):
        from supervisor.loggers import RotatingFileHandler
        return RotatingFileHandler

def test_ctor(self):
        handler = self._makeOne(self.filename)
        self.assertEqual(handler.mode, 'a')
        self.assertEqual(handler.maxBytes, 512*1024*1024)
        self.assertEqual(handler.backupCount, 10)

def test_emit_does_rollover(self):
        handler = self._makeOne(self.filename, maxBytes=10, backupCount=2)
        record = self._makeLogRecord('a' * 4)

handler.emit(record) # 4 bytes
        self.assertFalse(os.path.exists(self.filename + '.1'))
        self.assertFalse(os.path.exists(self.filename + '.2'))

handler.emit(record) # 8 bytes
        self.assertFalse(os.path.exists(self.filename + '.1'))
        self.assertFalse(os.path.exists(self.filename + '.2'))

handler.emit(record) # 12 bytes, do rollover
        self.assertTrue(os.path.exists(self.filename + '.1'))
        self.assertFalse(os.path.exists(self.filename + '.2'))

handler.emit(record) # 16 bytes
        self.assertTrue(os.path.exists(self.filename + '.1'))
        self.assertFalse(os.path.exists(self.filename + '.2'))

handler.emit(record) # 20 bytes
        self.assertTrue(os.path.exists(self.filename + '.1'))
        self.assertFalse(os.path.exists(self.filename + '.2'))

handler.emit(record) # 24 bytes, do rollover
        self.assertTrue(os.path.exists(self.filename + '.1'))
        self.assertTrue(os.path.exists(self.filename + '.2'))

handler.emit(record) # 28 bytes
        self.assertTrue(os.path.exists(self.filename + '.1'))
        self.assertTrue(os.path.exists(self.filename + '.2'))

current = open(self.filename,'r').read()
        self.assertEqual(current, 'a' * 4)
        one = open(self.filename+ '.1','r').read()
        self.assertEqual(one, 'a'*12)
        two = open(self.filename+ '.2','r').read()
        self.assertEqual(two, 'a'*12)

def test_current_logfile_removed(self):
        handler = self._makeOne(self.filename, maxBytes=6, backupCount=1)
        record = self._makeLogRecord('a' * 4)

handler.emit(record) # 4 bytes
        self.assertTrue(os.path.exists(self.filename))
        self.assertFalse(os.path.exists(self.filename + '.1'))

# Someone removes the active log file! :-(
        os.unlink(self.filename)
        self.assertFalse(os.path.exists(self.filename))

handler.emit(record) # 8 bytes, do rollover
        self.assertTrue(os.path.exists(self.filename))
        self.assertFalse(os.path.exists(self.filename + '.1'))

class BoundIOTests(unittest.TestCase):
    def _getTargetClass(self):
        from supervisor.loggers import BoundIO
        return BoundIO

def _makeOne(self, maxbytes, buf=''):
        klass = self._getTargetClass()
        return klass(maxbytes, buf)

def test_write_overflow(self):
        io = self._makeOne(1, 'a')
        io.write('b')
        self.assertEqual(io.buf, 'b')

def test_getvalue(self):
        io = self._makeOne(1, 'a')
        self.assertEqual(io.getvalue(), 'a')

def test_clear(self):
        io = self._makeOne(1, 'a')
        io.clear()
        self.assertEqual(io.buf, '')

def test_close(self):
        io = self._makeOne(1, 'a')
        io.close()
        self.assertEqual(io.buf, '')

class LoggerTests(unittest.TestCase):
    def _getTargetClass(self):
        from supervisor.loggers import Logger
        return Logger

def _makeOne(self, level=None, handlers=None):
        klass = self._getTargetClass()
        return klass(level, handlers)

def test_blather(self):
        from supervisor.loggers import LevelsByName
        handler = DummyHandler(LevelsByName.BLAT)
        logger = self._makeOne(LevelsByName.BLAT, (handler,))
        logger.blather('hello')
        self.assertEqual(len(handler.records), 1)
        logger.level = LevelsByName.TRAC
        logger.blather('hello')
        self.assertEqual(len(handler.records), 1)

def test_trace(self):
        from supervisor.loggers import LevelsByName
        handler = DummyHandler(LevelsByName.TRAC)
        logger = self._makeOne(LevelsByName.TRAC, (handler,))
        logger.trace('hello')
        self.assertEqual(len(handler.records), 1)
        logger.level = LevelsByName.DEBG
        logger.trace('hello')
        self.assertEqual(len(handler.records), 1)

def test_debug(self):
        from supervisor.loggers import LevelsByName
        handler = DummyHandler(LevelsByName.DEBG)
        logger = self._makeOne(LevelsByName.DEBG, (handler,))
        logger.debug('hello')
        self.assertEqual(len(handler.records), 1)
        logger.level = LevelsByName.INFO
        logger.debug('hello')
        self.assertEqual(len(handler.records), 1)

def test_info(self):
        from supervisor.loggers import LevelsByName
        handler = DummyHandler(LevelsByName.INFO)
        logger = self._makeOne(LevelsByName.INFO, (handler,))
        logger.info('hello')
        self.assertEqual(len(handler.records), 1)
        logger.level = LevelsByName.WARN
        logger.info('hello')
        self.assertEqual(len(handler.records), 1)

def test_warn(self):
        from supervisor.loggers import LevelsByName
        handler = DummyHandler(LevelsByName.WARN)
        logger = self._makeOne(LevelsByName.WARN, (handler,))
        logger.warn('hello')
        self.assertEqual(len(handler.records), 1)
        logger.level = LevelsByName.ERRO
        logger.warn('hello')
        self.assertEqual(len(handler.records), 1)

def test_error(self):
        from supervisor.loggers import LevelsByName
        handler = DummyHandler(LevelsByName.ERRO)
        logger = self._makeOne(LevelsByName.ERRO, (handler,))
        logger.error('hello')
        self.assertEqual(len(handler.records), 1)
        logger.level = LevelsByName.CRIT
        logger.error('hello')
        self.assertEqual(len(handler.records), 1)

def test_critical(self):
        from supervisor.loggers import LevelsByName
        handler = DummyHandler(LevelsByName.CRIT)
        logger = self._makeOne(LevelsByName.CRIT, (handler,))
        logger.critical('hello')
        self.assertEqual(len(handler.records), 1)

def test_close(self):
        from supervisor.loggers import LevelsByName
        handler = DummyHandler(LevelsByName.CRIT)
        logger = self._makeOne(LevelsByName.CRIT, (handler,))
        logger.close()
        self.assertEqual(handler.closed, True)

class MockSysLog(mock.Mock):
    def __call__(self, *args, **kwargs):
        message = args[-1]
        if sys.version_info < (3, 0) and isinstance(message, unicode):
            # Python 2.x raises a UnicodeEncodeError when attempting to
            #  transmit unicode characters that don't encode in the
            #  default encoding.
            message.encode()
        super(MockSysLog, self).__call__(*args, **kwargs)

class SyslogHandlerTests(HandlerTests, unittest.TestCase):
    def setUp(self):
        pass

def tearDown(self):
        pass

def _getTargetClass(self):
        return __import__('supervisor.loggers').loggers.SyslogHandler

def _makeOne(self):
        return self._getTargetClass()()

@mock.patch('syslog.syslog', MockSysLog())
    def test_emit_ascii_noerror(self):
        handler = self._makeOne()
        record = self._makeLogRecord('hello!')
        handler.emit(record)
        syslog.syslog.assert_called_with('hello!')

@mock.patch('syslog.syslog', MockSysLog())
    def test_close(self):
        handler = self._makeOne()
        handler.close()  # no-op for syslog

@mock.patch('syslog.syslog', MockSysLog())
    def test_reopen(self):
        handler = self._makeOne()
        handler.reopen()  # no-op for syslog

@mock.patch('syslog.syslog', MockSysLog())
    def test_emit_unicode_noerror(self):
        handler = self._makeOne()
        record = self._makeLogRecord(u'fi\xed')
        handler.emit(record)
        if sys.version_info < (3, 0):
            syslog.syslog.assert_called_with('fi\xc3\xad')
        else:
            syslog.syslog.assert_called_with(u'fi\xed')

class DummyHandler:
    close = False
    def __init__(self, level):
        self.level = level
        self.records = []
    def emit(self, record):
        self.records.append(record)
    def close(self):
        self.closed = True

def test_suite():
    return unittest.findTestCases(sys.modules[__name__])

if __name__ == '__main__':
    unittest.main(defaultTest='test_suite')

${this.title}

Code Editor : test_loggers.py