# 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')])
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 |
|