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_filedescriptor.py (2761B)

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

"""
Whitebox tests for L{twisted.internet.abstract.FileDescriptor}.
"""

from zope.interface.verify import verifyClass

from twisted.internet.abstract import FileDescriptor
from twisted.internet.interfaces import IPushProducer
from twisted.trial.unittest import SynchronousTestCase

class MemoryFile(FileDescriptor):
    """
    A L{FileDescriptor} customization which writes to a Python list in memory
    with certain limitations.

@ivar _written: A C{list} of C{bytes} which have been accepted as written.

@ivar _freeSpace: A C{int} giving the number of bytes which will be accepted
        by future writes.
    """

connected = True

def __init__(self):
        FileDescriptor.__init__(self, reactor=object())
        self._written = []
        self._freeSpace = 0

def startWriting(self):
        pass

def stopWriting(self):
        pass

def writeSomeData(self, data):
        """
        Copy at most C{self._freeSpace} bytes from C{data} into C{self._written}.

@return: A C{int} indicating how many bytes were copied from C{data}.
        """
        acceptLength = min(self._freeSpace, len(data))
        if acceptLength:
            self._freeSpace -= acceptLength
            self._written.append(data[:acceptLength])
        return acceptLength

class FileDescriptorTests(SynchronousTestCase):
    """
    Tests for L{FileDescriptor}.
    """

def test_writeWithUnicodeRaisesException(self):
        """
        L{FileDescriptor.write} doesn't accept unicode data.
        """
        fileDescriptor = FileDescriptor(reactor=object())
        self.assertRaises(TypeError, fileDescriptor.write, "foo")

def test_writeSequenceWithUnicodeRaisesException(self):
        """
        L{FileDescriptor.writeSequence} doesn't accept unicode data.
        """
        fileDescriptor = FileDescriptor(reactor=object())
        self.assertRaises(
            TypeError, fileDescriptor.writeSequence, [b"foo", "bar", b"baz"]
        )

def test_implementInterfaceIPushProducer(self):
        """
        L{FileDescriptor} should implement L{IPushProducer}.
        """
        self.assertTrue(verifyClass(IPushProducer, FileDescriptor))

class WriteDescriptorTests(SynchronousTestCase):
    """
    Tests for L{FileDescriptor}'s implementation of L{IWriteDescriptor}.
    """

def test_kernelBufferFull(self):
        """
        When L{FileDescriptor.writeSomeData} returns C{0} to indicate no more
        data can be written immediately, L{FileDescriptor.doWrite} returns
        L{None}.
        """
        descriptor = MemoryFile()
        descriptor.write(b"hello, world")
        self.assertIsNone(descriptor.doWrite())

cmd: