| Server IP : 74.208.127.88 / Your IP : 216.73.216.126 Web Server : Apache/2.4.41 (Ubuntu) System : Linux ubuntu 5.4.0-163-generic #180-Ubuntu SMP Tue Sep 5 13:21:23 UTC 2023 x86_64 User : www-data ( 33) PHP Version : 7.4.3-4ubuntu2.29 Disable Function : pcntl_alarm,pcntl_fork,pcntl_waitpid,pcntl_wait,pcntl_wifexited,pcntl_wifstopped,pcntl_wifsignaled,pcntl_wifcontinued,pcntl_wexitstatus,pcntl_wtermsig,pcntl_wstopsig,pcntl_signal,pcntl_signal_get_handler,pcntl_signal_dispatch,pcntl_get_last_error,pcntl_strerror,pcntl_sigprocmask,pcntl_sigwaitinfo,pcntl_sigtimedwait,pcntl_exec,pcntl_getpriority,pcntl_setpriority,pcntl_async_signals,pcntl_unshare, MySQL : OFF | cURL : ON | WGET : ON | Perl : ON | Python : OFF | Sudo : ON | Pkexec : ON Directory : /proc/self/root/usr/lib/python3/dist-packages/twisted/logger/test/ |
Upload File : |
# Copyright (c) Twisted Matrix Laboratories.
# See LICENSE for details.
"""
Test cases for L{twisted.logger._observer}.
"""
from zope.interface.verify import verifyObject, BrokenMethodImplementation
from twisted.trial import unittest
from .._logger import Logger
from .._observer import ILogObserver
from .._observer import LogPublisher
class LogPublisherTests(unittest.TestCase):
"""
Tests for L{LogPublisher}.
"""
def test_interface(self):
"""
L{LogPublisher} is an L{ILogObserver}.
"""
publisher = LogPublisher()
try:
verifyObject(ILogObserver, publisher)
except BrokenMethodImplementation as e:
self.fail(e)
def test_observers(self):
"""
L{LogPublisher.observers} returns the observers.
"""
o1 = lambda e: None
o2 = lambda e: None
publisher = LogPublisher(o1, o2)
self.assertEqual(set((o1, o2)), set(publisher._observers))
def test_addObserver(self):
"""
L{LogPublisher.addObserver} adds an observer.
"""
o1 = lambda e: None
o2 = lambda e: None
o3 = lambda e: None
publisher = LogPublisher(o1, o2)
publisher.addObserver(o3)
self.assertEqual(set((o1, o2, o3)), set(publisher._observers))
def test_addObserverNotCallable(self):
"""
L{LogPublisher.addObserver} refuses to add an observer that's
not callable.
"""
publisher = LogPublisher()
self.assertRaises(TypeError, publisher.addObserver, object())
def test_removeObserver(self):
"""
L{LogPublisher.removeObserver} removes an observer.
"""
o1 = lambda e: None
o2 = lambda e: None
o3 = lambda e: None
publisher = LogPublisher(o1, o2, o3)
publisher.removeObserver(o2)
self.assertEqual(set((o1, o3)), set(publisher._observers))
def test_removeObserverNotRegistered(self):
"""
L{LogPublisher.removeObserver} removes an observer that is not
registered.
"""
o1 = lambda e: None
o2 = lambda e: None
o3 = lambda e: None
publisher = LogPublisher(o1, o2)
publisher.removeObserver(o3)
self.assertEqual(set((o1, o2)), set(publisher._observers))
def test_fanOut(self):
"""
L{LogPublisher} calls its observers.
"""
event = dict(foo=1, bar=2)
events1 = []
events2 = []
events3 = []
o1 = lambda e: events1.append(e)
o2 = lambda e: events2.append(e)
o3 = lambda e: events3.append(e)
publisher = LogPublisher(o1, o2, o3)
publisher(event)
self.assertIn(event, events1)
self.assertIn(event, events2)
self.assertIn(event, events3)
def test_observerRaises(self):
"""
Observer raises an exception during fan out: a failure is logged, but
not re-raised. Life goes on.
"""
event = dict(foo=1, bar=2)
exception = RuntimeError("ARGH! EVIL DEATH!")
events = []
def observer(event):
shouldRaise = not events
events.append(event)
if shouldRaise:
raise exception
collector = []
publisher = LogPublisher(observer, collector.append)
publisher(event)
# Verify that the observer saw my event
self.assertIn(event, events)
# Verify that the observer raised my exception
errors = [
e["log_failure"] for e in collector
if "log_failure" in e
]
self.assertEqual(len(errors), 1)
self.assertIs(errors[0].value, exception)
# Make sure the exceptional observer does not receive its own error.
self.assertEqual(len(events), 1)
def test_observerRaisesAndLoggerHatesMe(self):
"""
Observer raises an exception during fan out and the publisher's Logger
pukes when the failure is reported. The exception does not propagate
back to the caller.
"""
event = dict(foo=1, bar=2)
exception = RuntimeError("ARGH! EVIL DEATH!")
def observer(event):
raise RuntimeError("Sad panda")
class GurkLogger(Logger):
def failure(self, *args, **kwargs):
raise exception
publisher = LogPublisher(observer)
publisher.log = GurkLogger()
publisher(event)
# Here, the lack of an exception thus far is a success, of sorts
def test_trace(self):
"""
Tracing keeps track of forwarding to observers.
"""
event = dict(foo=1, bar=2, log_trace=[])
traces = {}
# Copy trace to a tuple; otherwise, both observers will store the same
# mutable list, and we won't be able to see o1's view distinctly.
o1 = lambda e: traces.setdefault(1, tuple(e["log_trace"]))
o2 = lambda e: traces.setdefault(2, tuple(e["log_trace"]))
publisher = LogPublisher(o1, o2)
publisher(event)
self.assertEqual(traces[1], ((publisher, o1),))
self.assertEqual(traces[2], ((publisher, o1), (publisher, o2)))