404

[ Avaa Bypassed ]




Upload:

Command:

botdev@18.222.57.238: ~ $
# Copyright (c) Twisted Matrix Laboratories.
# See LICENSE for details.

from twisted.trial.unittest import TestCase
from twisted.python.failure import Failure

try:
    import syslog as stdsyslog
except ImportError:
    stdsyslog = None
else:
    from twisted.python import syslog



class SyslogObserverTests(TestCase):
    """
    Tests for L{SyslogObserver} which sends Twisted log events to the syslog.
    """
    events = None

    if stdsyslog is None:
        skip = "syslog is not supported on this platform"

    def setUp(self):
        self.patch(syslog.SyslogObserver, 'openlog', self.openlog)
        self.patch(syslog.SyslogObserver, 'syslog', self.syslog)
        self.observer = syslog.SyslogObserver('SyslogObserverTests')


    def openlog(self, prefix, options, facility):
        self.logOpened = (prefix, options, facility)
        self.events = []


    def syslog(self, options, message):
        self.events.append((options, message))


    def test_emitWithoutMessage(self):
        """
        L{SyslogObserver.emit} ignores events with an empty value for the
        C{'message'} key.
        """
        self.observer.emit({'message': (), 'isError': False, 'system': '-'})
        self.assertEqual(self.events, [])


    def test_emitCustomPriority(self):
        """
        L{SyslogObserver.emit} uses the value of the C{'syslogPriority'} as the
        syslog priority, if that key is present in the event dictionary.
        """
        self.observer.emit({
                'message': ('hello, world',), 'isError': False, 'system': '-',
                'syslogPriority': stdsyslog.LOG_DEBUG})
        self.assertEqual(
            self.events,
            [(stdsyslog.LOG_DEBUG, '[-] hello, world')])


    def test_emitErrorPriority(self):
        """
        L{SyslogObserver.emit} uses C{LOG_ALERT} if the event represents an
        error.
        """
        self.observer.emit({
                'message': ('hello, world',), 'isError': True, 'system': '-',
                'failure': Failure(Exception("foo"))})
        self.assertEqual(
            self.events,
            [(stdsyslog.LOG_ALERT, '[-] hello, world')])


    def test_emitCustomPriorityOverridesError(self):
        """
        L{SyslogObserver.emit} uses the value of the C{'syslogPriority'} key if
        it is specified even if the event dictionary represents an error.
        """
        self.observer.emit({
                'message': ('hello, world',), 'isError': True, 'system': '-',
                'syslogPriority': stdsyslog.LOG_NOTICE,
                'failure': Failure(Exception("bar"))})
        self.assertEqual(
            self.events,
            [(stdsyslog.LOG_NOTICE, '[-] hello, world')])


    def test_emitCustomFacility(self):
        """
        L{SyslogObserver.emit} uses the value of the C{'syslogPriority'} as the
        syslog priority, if that key is present in the event dictionary.
        """
        self.observer.emit({
                'message': ('hello, world',), 'isError': False, 'system': '-',
                'syslogFacility': stdsyslog.LOG_CRON})
        self.assertEqual(
            self.events,
            [(stdsyslog.LOG_INFO | stdsyslog.LOG_CRON, '[-] hello, world')])


    def test_emitCustomSystem(self):
        """
        L{SyslogObserver.emit} uses the value of the C{'system'} key to prefix
        the logged message.
        """
        self.observer.emit({'message': ('hello, world',), 'isError': False,
            'system': 'nonDefaultSystem'})
        self.assertEqual(
            self.events,
            [(stdsyslog.LOG_INFO, "[nonDefaultSystem] hello, world")])


    def test_emitMessage(self):
        """
        L{SyslogObserver.emit} logs the value of the C{'message'} key of the
        event dictionary it is passed to the syslog.
        """
        self.observer.emit({
                'message': ('hello, world',), 'isError': False,
                'system': '-'})
        self.assertEqual(
            self.events,
            [(stdsyslog.LOG_INFO, "[-] hello, world")])


    def test_emitMultilineMessage(self):
        """
        Each line of a multiline message is emitted separately to the syslog.
        """
        self.observer.emit({
                'message': ('hello,\nworld',), 'isError': False,
                'system': '-'})
        self.assertEqual(
            self.events,
            [(stdsyslog.LOG_INFO, '[-] hello,'),
             (stdsyslog.LOG_INFO, '[-] \tworld')])


    def test_emitStripsTrailingEmptyLines(self):
        """
        Trailing empty lines of a multiline message are omitted from the
        messages sent to the syslog.
        """
        self.observer.emit({
                'message': ('hello,\nworld\n\n',), 'isError': False,
                'system': '-'})
        self.assertEqual(
            self.events,
            [(stdsyslog.LOG_INFO, '[-] hello,'),
             (stdsyslog.LOG_INFO, '[-] \tworld')])

Filemanager

Name Type Size Permission Actions
__pycache__ Folder 0755
__init__.py File 42 B 0644
_deprecatetests.py.3only File 1.77 KB 0644
deprecatedattributes.py File 571 B 0644
modules_helpers.py File 1.57 KB 0644
pullpipe.py File 1.19 KB 0644
test_appdirs.py File 1.06 KB 0644
test_components.py File 25.37 KB 0644
test_constants.py File 37.22 KB 0644
test_deprecate.py File 38.02 KB 0644
test_dist3.py File 1.81 KB 0644
test_fakepwd.py File 13.78 KB 0644
test_htmlizer.py File 1.24 KB 0644
test_inotify.py File 3.55 KB 0644
test_release.py File 40.08 KB 0644
test_runtime.py File 7.74 KB 0644
test_sendmsg.py File 25.25 KB 0644
test_setup.py File 11.78 KB 0644
test_shellcomp.py File 21.32 KB 0644
test_syslog.py File 4.83 KB 0644
test_systemd.py File 6.25 KB 0644
test_textattributes.py File 712 B 0644
test_tzhelper.py File 3.89 KB 0644
test_url.py File 29.23 KB 0644
test_urlpath.py File 10.09 KB 0644
test_util.py File 35.55 KB 0644
test_versions.py File 5.27 KB 0644
test_zippath.py File 3.39 KB 0644
test_zipstream.py File 11.88 KB 0644