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_socket.py (9443B)

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

"""
Tests for implementations of L{IReactorSocket}.

Generally only tests for failure cases are found here.  Success cases for
this interface are tested elsewhere.  For example, the success case for
I{AF_INET} is in L{twisted.internet.test.test_tcp}, since that case should
behave exactly the same as L{IReactorTCP.listenTCP}.
"""

import errno
import socket

from zope.interface import verify

from twisted.internet.error import UnsupportedAddressFamily
from twisted.internet.interfaces import IReactorSocket
from twisted.internet.protocol import DatagramProtocol, ServerFactory
from twisted.internet.test.reactormixins import ReactorBuilder, needsRunningReactor
from twisted.python.log import err
from twisted.python.runtime import platform

class IReactorSocketVerificationTestsBuilder(ReactorBuilder):
    """
    Builder for testing L{IReactorSocket} implementations for required
    methods and method signatures.

L{ReactorBuilder} already runs L{IReactorSocket.providedBy} to
    ensure that these tests will only be run on reactor classes that
    claim to implement L{IReactorSocket}.

These tests ensure that reactors which claim to provide the
    L{IReactorSocket} interface actually have all the required methods
    and that those methods have the expected number of arguments.

These tests will be skipped for reactors which do not claim to
    provide L{IReactorSocket}.
    """

requiredInterfaces = [IReactorSocket]

def test_provider(self):
        """
        The reactor instance returned by C{buildReactor} provides
        L{IReactorSocket}.
        """
        reactor = self.buildReactor()
        self.assertTrue(verify.verifyObject(IReactorSocket, reactor))

class AdoptStreamPortErrorsTestsBuilder(ReactorBuilder):
    """
    Builder for testing L{IReactorSocket.adoptStreamPort} implementations.

Generally only tests for failure cases are found here.  Success cases for
    this interface are tested elsewhere.  For example, the success case for
    I{AF_INET} is in L{twisted.internet.test.test_tcp}, since that case should
    behave exactly the same as L{IReactorTCP.listenTCP}.
    """

requiredInterfaces = [IReactorSocket]

def test_invalidDescriptor(self):
        """
        An implementation of L{IReactorSocket.adoptStreamPort} raises
        L{socket.error} if passed an integer which is not associated with a
        socket.
        """
        reactor = self.buildReactor()

probe = socket.socket()
        fileno = probe.fileno()
        probe.close()

exc = self.assertRaises(
            socket.error,
            reactor.adoptStreamPort,
            fileno,
            socket.AF_INET,
            ServerFactory(),
        )
        if platform.isWindows():
            self.assertEqual(exc.args[0], errno.WSAENOTSOCK)
        else:
            self.assertEqual(exc.args[0], errno.EBADF)

def test_invalidAddressFamily(self):
        """
        An implementation of L{IReactorSocket.adoptStreamPort} raises
        L{UnsupportedAddressFamily} if passed an address family it does not
        support.
        """
        reactor = self.buildReactor()

port = socket.socket()
        port.bind(("127.0.0.1", 0))
        port.listen(1)
        self.addCleanup(port.close)

arbitrary = 2 ** 16 + 7

self.assertRaises(
            UnsupportedAddressFamily,
            reactor.adoptStreamPort,
            port.fileno(),
            arbitrary,
            ServerFactory(),
        )

def test_stopOnlyCloses(self):
        """
        When the L{IListeningPort} returned by
        L{IReactorSocket.adoptStreamPort} is stopped using
        C{stopListening}, the underlying socket is closed but not
        shutdown.  This allows another process which still has a
        reference to it to continue accepting connections over it.
        """
        reactor = self.buildReactor()

portSocket = socket.socket()
        self.addCleanup(portSocket.close)

portSocket.bind(("127.0.0.1", 0))
        portSocket.listen(1)
        portSocket.setblocking(False)

# The file descriptor is duplicated by adoptStreamPort
        port = reactor.adoptStreamPort(
            portSocket.fileno(), portSocket.family, ServerFactory()
        )
        d = port.stopListening()

def stopped(ignored):
            # Should still be possible to accept a connection on
            # portSocket.  If it was shutdown, the exception would be
            # EINVAL instead.
            exc = self.assertRaises(socket.error, portSocket.accept)
            if platform.isWindows():
                self.assertEqual(exc.args[0], errno.WSAEWOULDBLOCK)
            else:
                self.assertEqual(exc.args[0], errno.EAGAIN)

d.addCallback(stopped)
        d.addErrback(err, "Failed to accept on original port.")

needsRunningReactor(
            reactor, lambda: d.addCallback(lambda ignored: reactor.stop())
        )

reactor.run()

class AdoptStreamConnectionErrorsTestsBuilder(ReactorBuilder):
    """
    Builder for testing L{IReactorSocket.adoptStreamConnection}
    implementations.

Generally only tests for failure cases are found here.  Success cases for
    this interface are tested elsewhere.  For example, the success case for
    I{AF_INET} is in L{twisted.internet.test.test_tcp}, since that case should
    behave exactly the same as L{IReactorTCP.listenTCP}.
    """

requiredInterfaces = [IReactorSocket]

def test_invalidAddressFamily(self):
        """
        An implementation of L{IReactorSocket.adoptStreamConnection} raises
        L{UnsupportedAddressFamily} if passed an address family it does not
        support.
        """
        reactor = self.buildReactor()

connection = socket.socket()
        self.addCleanup(connection.close)

arbitrary = 2 ** 16 + 7

self.assertRaises(
            UnsupportedAddressFamily,
            reactor.adoptStreamConnection,
            connection.fileno(),
            arbitrary,
            ServerFactory(),
        )

class AdoptDatagramPortErrorsTestsBuilder(ReactorBuilder):
    """
    Builder for testing L{IReactorSocket.adoptDatagramPort} implementations.
    """

requiredInterfaces = [IReactorSocket]

def test_invalidDescriptor(self):
        """
        An implementation of L{IReactorSocket.adoptDatagramPort} raises
        L{socket.error} if passed an integer which is not associated with a
        socket.
        """
        reactor = self.buildReactor()

probe = socket.socket()
        fileno = probe.fileno()
        probe.close()

exc = self.assertRaises(
            socket.error,
            reactor.adoptDatagramPort,
            fileno,
            socket.AF_INET,
            DatagramProtocol(),
        )
        if platform.isWindows():
            self.assertEqual(exc.args[0], errno.WSAENOTSOCK)
        else:
            self.assertEqual(exc.args[0], errno.EBADF)

def test_invalidAddressFamily(self):
        """
        An implementation of L{IReactorSocket.adoptDatagramPort} raises
        L{UnsupportedAddressFamily} if passed an address family it does not
        support.
        """
        reactor = self.buildReactor()

port = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        self.addCleanup(port.close)

arbitrary = 2 ** 16 + 7

self.assertRaises(
            UnsupportedAddressFamily,
            reactor.adoptDatagramPort,
            port.fileno(),
            arbitrary,
            DatagramProtocol(),
        )

def test_stopOnlyCloses(self):
        """
        When the L{IListeningPort} returned by
        L{IReactorSocket.adoptDatagramPort} is stopped using
        C{stopListening}, the underlying socket is closed but not
        shutdown.  This allows another process which still has a
        reference to it to continue reading and writing to it.
        """
        reactor = self.buildReactor()

portSocket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        self.addCleanup(portSocket.close)

portSocket.bind(("127.0.0.1", 0))
        portSocket.setblocking(False)

# The file descriptor is duplicated by adoptDatagramPort
        port = reactor.adoptDatagramPort(
            portSocket.fileno(), portSocket.family, DatagramProtocol()
        )
        d = port.stopListening()

def stopped(ignored):
            # Should still be possible to recv on portSocket.  If
            # it was shutdown, the exception would be EINVAL instead.
            exc = self.assertRaises(socket.error, portSocket.recvfrom, 1)
            if platform.isWindows():
                self.assertEqual(exc.args[0], errno.WSAEWOULDBLOCK)
            else:
                self.assertEqual(exc.args[0], errno.EAGAIN)

d.addCallback(stopped)
        d.addErrback(err, "Failed to read on original port.")

needsRunningReactor(
            reactor, lambda: d.addCallback(lambda ignored: reactor.stop())
        )

reactor.run()

globals().update(IReactorSocketVerificationTestsBuilder.makeTestCaseClasses())
globals().update(AdoptStreamPortErrorsTestsBuilder.makeTestCaseClasses())
globals().update(AdoptStreamConnectionErrorsTestsBuilder.makeTestCaseClasses())
globals().update(AdoptDatagramPortErrorsTestsBuilder.makeTestCaseClasses())

cmd: