FM / test

/usr/lib/python3/dist-packages/twisted/logger/test/

/usr/lib/python3/dist-packages/twisted/logger/test

Name	Size	Mode	Actions
__pycache__/	-	0755	rm
test_buffer.py	1642	0644	edit dl rm
test_file.py	5623	0644	edit dl rm
test_filter.py	12010	0644	edit dl rm
test_flatten.py	9137	0644	edit dl rm
test_format.py	22563	0644	edit dl rm
test_global.py	12049	0644	edit dl rm
test_io.py	7212	0644	edit dl rm
test_json.py	18584	0644	edit dl rm
test_legacy.py	14395	0644	edit dl rm
test_levels.py	875	0644	edit dl rm
test_logger.py	7354	0644	edit dl rm
test_observer.py	5253	0644	edit dl rm
test_stdlib.py	8639	0644	edit dl rm
test_util.py	2671	0644	edit dl rm
__init__.py	161	0644	edit dl rm

Edit: /usr/lib/python3/dist-packages/twisted/logger/test/test_buffer.py (1642B)

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

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