FM / test

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

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

Name	Size	Mode	Actions
fake_CAs/	-	0755	rm
__pycache__/	-	0755	rm
connectionmixins.py	20184	0644	edit dl rm
fakeendpoint.py	1663	0644	edit dl rm
modulehelpers.py	1672	0644	edit dl rm
process_cli.py	547	0644	edit dl rm
process_connectionlost.py	126	0644	edit dl rm
process_gireactornocompat.py	816	0644	edit dl rm
process_helper.py	1282	0644	edit dl rm
reactormixins.py	14609	0644	edit dl rm
test_abstract.py	1976	0644	edit dl rm
test_address.py	8266	0644	edit dl rm
test_asyncioreactor.py	9327	0644	edit dl rm
test_base.py	13684	0644	edit dl rm
test_baseprocess.py	2588	0644	edit dl rm
test_core.py	11125	0644	edit dl rm
test_default.py	3419	0644	edit dl rm
test_defer_await.py	6514	0644	edit dl rm
test_defer_yieldfrom.py	4848	0644	edit dl rm
test_endpoints.py	148720	0644	edit dl rm
test_epollreactor.py	7322	0644	edit dl rm
test_error.py	1090	0644	edit dl rm
test_fdset.py	13497	0644	edit dl rm
test_filedescriptor.py	2761	0644	edit dl rm
test_gireactor.py	6516	0644	edit dl rm
test_glibbase.py	2217	0644	edit dl rm
test_inlinecb.py	11482	0644	edit dl rm
test_inotify.py	18562	0644	edit dl rm
test_iocp.py	7700	0644	edit dl rm
test_kqueuereactor.py	1853	0644	edit dl rm
test_main.py	1331	0644	edit dl rm
test_newtls.py	6532	0644	edit dl rm
test_pollingfile.py	1297	0644	edit dl rm
test_posixbase.py	9718	0644	edit dl rm
test_posixprocess.py	11203	0644	edit dl rm
test_process.py	36732	0644	edit dl rm
test_protocol.py	18520	0644	edit dl rm
test_resolver.py	19610	0644	edit dl rm
test_serialport.py	1988	0644	edit dl rm
test_sigchld.py	3922	0644	edit dl rm
test_socket.py	9443	0644	edit dl rm
test_stdio.py	6597	0644	edit dl rm
test_tcp.py	106621	0644	edit dl rm
test_testing.py	16239	0644	edit dl rm
test_threads.py	8734	0644	edit dl rm
test_time.py	3758	0644	edit dl rm
test_tls.py	13349	0644	edit dl rm
test_udp.py	16881	0644	edit dl rm
test_udp_internals.py	4837	0644	edit dl rm
test_unix.py	34854	0644	edit dl rm
test_win32events.py	6473	0644	edit dl rm
test_win32serialport.py	5330	0644	edit dl rm
_posixifaces.py	4391	0644	edit dl rm
_win32ifaces.py	3975	0644	edit dl rm
__init__.py	112	0644	edit dl rm

Edit: /usr/lib/python3/dist-packages/twisted/internet/test/test_epollreactor.py (7322B)

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

"""
Tests for L{twisted.internet.epollreactor}.
"""

from unittest import skipIf

from twisted.internet.error import ConnectionDone
from twisted.internet.posixbase import _ContinuousPolling
from twisted.internet.task import Clock
from twisted.trial.unittest import TestCase

try:
    from twisted.internet import epollreactor
except ImportError:
    epollreactor = None  # type: ignore[assignment]

class Descriptor:
    """
    Records reads and writes, as if it were a C{FileDescriptor}.
    """

def __init__(self):
        self.events = []

def fileno(self):
        return 1

def doRead(self):
        self.events.append("read")

def doWrite(self):
        self.events.append("write")

def connectionLost(self, reason):
        reason.trap(ConnectionDone)
        self.events.append("lost")

@skipIf(not epollreactor, "epoll not supported in this environment.")
class ContinuousPollingTests(TestCase):
    """
    L{_ContinuousPolling} can be used to read and write from C{FileDescriptor}
    objects.
    """

def test_addReader(self):
        """
        Adding a reader when there was previously no reader starts up a
        C{LoopingCall}.
        """
        poller = _ContinuousPolling(Clock())
        self.assertIsNone(poller._loop)
        reader = object()
        self.assertFalse(poller.isReading(reader))
        poller.addReader(reader)
        self.assertIsNotNone(poller._loop)
        self.assertTrue(poller._loop.running)
        self.assertIs(poller._loop.clock, poller._reactor)
        self.assertTrue(poller.isReading(reader))

def test_addWriter(self):
        """
        Adding a writer when there was previously no writer starts up a
        C{LoopingCall}.
        """
        poller = _ContinuousPolling(Clock())
        self.assertIsNone(poller._loop)
        writer = object()
        self.assertFalse(poller.isWriting(writer))
        poller.addWriter(writer)
        self.assertIsNotNone(poller._loop)
        self.assertTrue(poller._loop.running)
        self.assertIs(poller._loop.clock, poller._reactor)
        self.assertTrue(poller.isWriting(writer))

def test_removeReader(self):
        """
        Removing a reader stops the C{LoopingCall}.
        """
        poller = _ContinuousPolling(Clock())
        reader = object()
        poller.addReader(reader)
        poller.removeReader(reader)
        self.assertIsNone(poller._loop)
        self.assertEqual(poller._reactor.getDelayedCalls(), [])
        self.assertFalse(poller.isReading(reader))

def test_removeWriter(self):
        """
        Removing a writer stops the C{LoopingCall}.
        """
        poller = _ContinuousPolling(Clock())
        writer = object()
        poller.addWriter(writer)
        poller.removeWriter(writer)
        self.assertIsNone(poller._loop)
        self.assertEqual(poller._reactor.getDelayedCalls(), [])
        self.assertFalse(poller.isWriting(writer))

def test_removeUnknown(self):
        """
        Removing unknown readers and writers silently does nothing.
        """
        poller = _ContinuousPolling(Clock())
        poller.removeWriter(object())
        poller.removeReader(object())

def test_multipleReadersAndWriters(self):
        """
        Adding multiple readers and writers results in a single
        C{LoopingCall}.
        """
        poller = _ContinuousPolling(Clock())
        writer = object()
        poller.addWriter(writer)
        self.assertIsNotNone(poller._loop)
        poller.addWriter(object())
        self.assertIsNotNone(poller._loop)
        poller.addReader(object())
        self.assertIsNotNone(poller._loop)
        poller.addReader(object())
        poller.removeWriter(writer)
        self.assertIsNotNone(poller._loop)
        self.assertTrue(poller._loop.running)
        self.assertEqual(len(poller._reactor.getDelayedCalls()), 1)

def test_readerPolling(self):
        """
        Adding a reader causes its C{doRead} to be called every 1
        milliseconds.
        """
        reactor = Clock()
        poller = _ContinuousPolling(reactor)
        desc = Descriptor()
        poller.addReader(desc)
        self.assertEqual(desc.events, [])
        reactor.advance(0.00001)
        self.assertEqual(desc.events, ["read"])
        reactor.advance(0.00001)
        self.assertEqual(desc.events, ["read", "read"])
        reactor.advance(0.00001)
        self.assertEqual(desc.events, ["read", "read", "read"])

def test_writerPolling(self):
        """
        Adding a writer causes its C{doWrite} to be called every 1
        milliseconds.
        """
        reactor = Clock()
        poller = _ContinuousPolling(reactor)
        desc = Descriptor()
        poller.addWriter(desc)
        self.assertEqual(desc.events, [])
        reactor.advance(0.001)
        self.assertEqual(desc.events, ["write"])
        reactor.advance(0.001)
        self.assertEqual(desc.events, ["write", "write"])
        reactor.advance(0.001)
        self.assertEqual(desc.events, ["write", "write", "write"])

def test_connectionLostOnRead(self):
        """
        If a C{doRead} returns a value indicating disconnection,
        C{connectionLost} is called on it.
        """
        reactor = Clock()
        poller = _ContinuousPolling(reactor)
        desc = Descriptor()
        desc.doRead = lambda: ConnectionDone()
        poller.addReader(desc)
        self.assertEqual(desc.events, [])
        reactor.advance(0.001)
        self.assertEqual(desc.events, ["lost"])

def test_connectionLostOnWrite(self):
        """
        If a C{doWrite} returns a value indicating disconnection,
        C{connectionLost} is called on it.
        """
        reactor = Clock()
        poller = _ContinuousPolling(reactor)
        desc = Descriptor()
        desc.doWrite = lambda: ConnectionDone()
        poller.addWriter(desc)
        self.assertEqual(desc.events, [])
        reactor.advance(0.001)
        self.assertEqual(desc.events, ["lost"])

def test_removeAll(self):
        """
        L{_ContinuousPolling.removeAll} removes all descriptors and returns
        the readers and writers.
        """
        poller = _ContinuousPolling(Clock())
        reader = object()
        writer = object()
        both = object()
        poller.addReader(reader)
        poller.addReader(both)
        poller.addWriter(writer)
        poller.addWriter(both)
        removed = poller.removeAll()
        self.assertEqual(poller.getReaders(), [])
        self.assertEqual(poller.getWriters(), [])
        self.assertEqual(len(removed), 3)
        self.assertEqual(set(removed), {reader, writer, both})

def test_getReaders(self):
        """
        L{_ContinuousPolling.getReaders} returns a list of the read
        descriptors.
        """
        poller = _ContinuousPolling(Clock())
        reader = object()
        poller.addReader(reader)
        self.assertIn(reader, poller.getReaders())

def test_getWriters(self):
        """
        L{_ContinuousPolling.getWriters} returns a list of the write
        descriptors.
        """
        poller = _ContinuousPolling(Clock())
        writer = object()
        poller.addWriter(writer)
        self.assertIn(writer, poller.getWriters())

cmd: