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_iocp.py (7700B)

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

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

import errno
import sys
import time
from array import array
from socket import AF_INET, AF_INET6, SOCK_STREAM, SOL_SOCKET, socket
from struct import pack
from unittest import skipIf

from zope.interface.verify import verifyClass

from twisted.internet.interfaces import IPushProducer
from twisted.python.log import msg
from twisted.trial.unittest import TestCase

try:
    from twisted.internet.iocpreactor import iocpsupport as _iocp, tcp, udp
    from twisted.internet.iocpreactor.abstract import FileHandle
    from twisted.internet.iocpreactor.const import SO_UPDATE_ACCEPT_CONTEXT
    from twisted.internet.iocpreactor.interfaces import IReadWriteHandle
    from twisted.internet.iocpreactor.reactor import (
        EVENTS_PER_LOOP,
        KEY_NORMAL,
        IOCPReactor,
    )
except ImportError:
    if sys.platform == "win32":
        raise

skip = "This test only applies to IOCPReactor"

try:
    socket(AF_INET6, SOCK_STREAM).close()
except OSError as e:
    ipv6Skip = True
    ipv6SkipReason = str(e)

else:
    ipv6Skip = False
    ipv6SkipReason = ""

class SupportTests(TestCase):
    """
    Tests for L{twisted.internet.iocpreactor.iocpsupport}, low-level reactor
    implementation helpers.
    """

def _acceptAddressTest(self, family, localhost):
        """
        Create a C{SOCK_STREAM} connection to localhost using a socket with an
        address family of C{family} and assert that the result of
        L{iocpsupport.get_accept_addrs} is consistent with the result of
        C{socket.getsockname} and C{socket.getpeername}.

A port starts listening (is bound) at the low-level socket without
        calling accept() yet.
        A client is then connected.
        After the client is connected IOCP accept() is called, which is the
        target of these tests.

Most of the time, the socket is ready instantly, but sometimes
        the socket is not ready right away after calling IOCP accept().
        It should not take more than 5 seconds for a socket to be ready, as
        the client connection is already made over the loopback interface.

These are flaky tests.
        Tweak the failure rate by changing the number of retries and the
        wait/sleep between retries.

If you will need to update the retries to wait more than 5 seconds
        for the port to be available, then there might a bug in the code and
        not the test (or a very, very busy VM running the tests).
        """
        msg(f"family = {family!r}")
        port = socket(family, SOCK_STREAM)
        self.addCleanup(port.close)
        port.bind(("", 0))
        port.listen(1)
        client = socket(family, SOCK_STREAM)
        self.addCleanup(client.close)
        client.setblocking(False)
        try:
            client.connect((localhost, port.getsockname()[1]))
        except OSError as e:
            self.assertIn(e.errno, (errno.EINPROGRESS, errno.EWOULDBLOCK))

server = socket(family, SOCK_STREAM)
        self.addCleanup(server.close)
        buff = array("B", b"\0" * 256)
        self.assertEqual(0, _iocp.accept(port.fileno(), server.fileno(), buff, None))

for attemptsRemaining in reversed(range(5)):
            # Calling setsockopt after _iocp.accept might fail for both IPv4
            # and IPV6 with "[Errno 10057] A request to send or receive ..."
            # This is when ERROR_IO_PENDING is returned and means that the
            # socket is not yet ready and accept will be handled via the
            # callback event.
            # For this test, with the purpose of keeping the test simple,
            # we don't implement the event callback.
            # The event callback functionality is tested via the high level
            # tests for general reactor API.
            # We retry multiple times to cover.
            try:
                server.setsockopt(
                    SOL_SOCKET, SO_UPDATE_ACCEPT_CONTEXT, pack("P", port.fileno())
                )
                break
            except OSError as socketError:
                # getattr is used below to make mypy happy.
                if socketError.errno != getattr(errno, "WSAENOTCONN"):
                    # This is not the expected error so re-raise the error without retrying.
                    raise

# The socket is not yet ready to accept connections,
                # setsockopt fails.
                if attemptsRemaining == 0:
                    # We ran out of retries.
                    raise

# Without a sleep here even retrying 20 times will fail.
            # This should allow other threads to execute and hopefully with the next
            # try setsockopt will succeed.
            time.sleep(0.2)

self.assertEqual(
            (family, client.getpeername()[:2], client.getsockname()[:2]),
            _iocp.get_accept_addrs(server.fileno(), buff),
        )

def test_ipv4AcceptAddress(self):
        """
        L{iocpsupport.get_accept_addrs} returns a three-tuple of address
        information about the socket associated with the file descriptor passed
        to it.  For a connection using IPv4:

- the first element is C{AF_INET}
          - the second element is a two-tuple of a dotted decimal notation IPv4
            address and a port number giving the peer address of the connection
          - the third element is the same type giving the host address of the
            connection
        """
        self._acceptAddressTest(AF_INET, "127.0.0.1")

@skipIf(ipv6Skip, ipv6SkipReason)
    def test_ipv6AcceptAddress(self):
        """
        Like L{test_ipv4AcceptAddress}, but for IPv6 connections.
        In this case:

- the first element is C{AF_INET6}
          - the second element is a two-tuple of a hexadecimal IPv6 address
            literal and a port number giving the peer address of the connection
          - the third element is the same type giving the host address of the
            connection
        """
        self._acceptAddressTest(AF_INET6, "::1")

class IOCPReactorTests(TestCase):
    def test_noPendingTimerEvents(self):
        """
        Test reactor behavior (doIteration) when there are no pending time
        events.
        """
        ir = IOCPReactor()
        ir.wakeUp()
        self.assertFalse(ir.doIteration(None))

def test_reactorInterfaces(self):
        """
        Verify that IOCP socket-representing classes implement IReadWriteHandle
        """
        self.assertTrue(verifyClass(IReadWriteHandle, tcp.Connection))
        self.assertTrue(verifyClass(IReadWriteHandle, udp.Port))

def test_fileHandleInterfaces(self):
        """
        Verify that L{Filehandle} implements L{IPushProducer}.
        """
        self.assertTrue(verifyClass(IPushProducer, FileHandle))

def test_maxEventsPerIteration(self):
        """
        Verify that we don't lose an event when more than EVENTS_PER_LOOP
        events occur in the same reactor iteration
        """

class FakeFD:
            counter = 0

def logPrefix(self):
                return "FakeFD"

def cb(self, rc, bytes, evt):
                self.counter += 1

ir = IOCPReactor()
        fd = FakeFD()
        event = _iocp.Event(fd.cb, fd)
        for _ in range(EVENTS_PER_LOOP + 1):
            ir.port.postEvent(0, KEY_NORMAL, event)
        ir.doIteration(None)
        self.assertEqual(fd.counter, EVENTS_PER_LOOP)
        ir.doIteration(0)
        self.assertEqual(fd.counter, EVENTS_PER_LOOP + 1)

cmd: