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_udp_internals.py (4837B)

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

"""
Tests for the internal implementation details of L{twisted.internet.udp}.
"""

import socket

from twisted.internet import udp
from twisted.internet.protocol import DatagramProtocol
from twisted.python.runtime import platformType
from twisted.trial import unittest

if platformType == "win32":
    from errno import WSAEWOULDBLOCK as EWOULDBLOCK  # type: ignore[attr-defined]
else:
    from errno import EWOULDBLOCK

class StringUDPSocket:
    """
    A fake UDP socket object, which returns a fixed sequence of strings and/or
    socket errors.  Useful for testing.

@ivar retvals: A C{list} containing either strings or C{socket.error}s.

@ivar connectedAddr: The address the socket is connected to.
    """

def __init__(self, retvals):
        self.retvals = retvals
        self.connectedAddr = None

def connect(self, addr):
        self.connectedAddr = addr

def recvfrom(self, size):
        """
        Return (or raise) the next value from C{self.retvals}.
        """
        ret = self.retvals.pop(0)
        if isinstance(ret, socket.error):
            raise ret
        return ret, None

class KeepReads(DatagramProtocol):
    """
    Accumulate reads in a list.
    """

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

def datagramReceived(self, data, addr):
        self.reads.append(data)

class ErrorsTests(unittest.SynchronousTestCase):
    """
    Error handling tests for C{udp.Port}.
    """

def test_socketReadNormal(self):
        """
        Socket reads with some good data followed by a socket error which can
        be ignored causes reading to stop, and no log messages to be logged.
        """
        # Add a fake error to the list of ignorables:
        udp._sockErrReadIgnore.append(-7000)
        self.addCleanup(udp._sockErrReadIgnore.remove, -7000)

protocol = KeepReads()
        port = udp.Port(None, protocol)

# Normal result, no errors
        port.socket = StringUDPSocket(
            [b"result", b"123", socket.error(-7000), b"456", socket.error(-7000)]
        )
        port.doRead()
        # Read stops on error:
        self.assertEqual(protocol.reads, [b"result", b"123"])
        port.doRead()
        self.assertEqual(protocol.reads, [b"result", b"123", b"456"])

def test_readImmediateError(self):
        """
        If the socket is unconnected, socket reads with an immediate
        connection refusal are ignored, and reading stops. The protocol's
        C{connectionRefused} method is not called.
        """
        # Add a fake error to the list of those that count as connection
        # refused:
        udp._sockErrReadRefuse.append(-6000)
        self.addCleanup(udp._sockErrReadRefuse.remove, -6000)

protocol = KeepReads()
        # Fail if connectionRefused is called:
        protocol.connectionRefused = lambda: 1 / 0

port = udp.Port(None, protocol)

# Try an immediate "connection refused"
        port.socket = StringUDPSocket(
            [b"a", socket.error(-6000), b"b", socket.error(EWOULDBLOCK)]
        )
        port.doRead()
        # Read stops on error:
        self.assertEqual(protocol.reads, [b"a"])
        # Read again:
        port.doRead()
        self.assertEqual(protocol.reads, [b"a", b"b"])

def test_connectedReadImmediateError(self):
        """
        If the socket connected, socket reads with an immediate
        connection refusal are ignored, and reading stops. The protocol's
        C{connectionRefused} method is called.
        """
        # Add a fake error to the list of those that count as connection
        # refused:
        udp._sockErrReadRefuse.append(-6000)
        self.addCleanup(udp._sockErrReadRefuse.remove, -6000)

protocol = KeepReads()
        refused = []
        protocol.connectionRefused = lambda: refused.append(True)

port = udp.Port(None, protocol)
        port.socket = StringUDPSocket(
            [b"a", socket.error(-6000), b"b", socket.error(EWOULDBLOCK)]
        )
        port.connect("127.0.0.1", 9999)

# Read stops on error:
        port.doRead()
        self.assertEqual(protocol.reads, [b"a"])
        self.assertEqual(refused, [True])

# Read again:
        port.doRead()
        self.assertEqual(protocol.reads, [b"a", b"b"])
        self.assertEqual(refused, [True])

def test_readUnknownError(self):
        """
        Socket reads with an unknown socket error are raised.
        """
        protocol = KeepReads()
        port = udp.Port(None, protocol)

# Some good data, followed by an unknown error
        port.socket = StringUDPSocket([b"good", socket.error(-1337)])
        self.assertRaises(socket.error, port.doRead)
        self.assertEqual(protocol.reads, [b"good"])

cmd: