Server : nginx/1.18.0 System : Linux localhost 6.14.3-x86_64-linode168 #1 SMP PREEMPT_DYNAMIC Mon Apr 21 19:47:55 EDT 2025 x86_64 User : www-data ( 33) PHP Version : 8.0.16 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, Directory :  /usr/lib/python3/dist-packages/twisted/logger/test/

Upload File : current_dir [ Writeable ] document_root [ Writeable ]

# Copyright (c) Twisted Matrix Laboratories.
# See LICENSE for details.

"""
Test cases for L{twisted.logger._buffer}.
"""

from zope.interface.verify import verifyObject, BrokenMethodImplementation

from twisted.trial import unittest

from .._observer import ILogObserver
from .._buffer import LimitedHistoryLogObserver



class LimitedHistoryLogObserverTests(unittest.TestCase):
    """
    Tests for L{LimitedHistoryLogObserver}.
    """

    def test_interface(self):
        """
        L{LimitedHistoryLogObserver} provides L{ILogObserver}.
        """
        observer = LimitedHistoryLogObserver(0)
        try:
            verifyObject(ILogObserver, observer)
        except BrokenMethodImplementation as e:
            self.fail(e)


    def test_order(self):
        """
        L{LimitedHistoryLogObserver} saves history in the order it is received.
        """
        size = 4
        events = [dict(n=n) for n in range(size//2)]
        observer = LimitedHistoryLogObserver(size)

        for event in events:
            observer(event)

        outEvents = []
        observer.replayTo(outEvents.append)
        self.assertEqual(events, outEvents)


    def test_limit(self):
        """
        When more events than a L{LimitedHistoryLogObserver}'s maximum size are
        buffered, older events will be dropped.
        """
        size = 4
        events = [dict(n=n) for n in range(size*2)]
        observer = LimitedHistoryLogObserver(size)

        for event in events:
            observer(event)
        outEvents = []
        observer.replayTo(outEvents.append)
        self.assertEqual(events[-size:], outEvents)