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	20442	0644	edit dl rm
fakeendpoint.py	1555	0644	edit dl rm
modulehelpers.py	1726	0644	edit dl rm
process_cli.py	664	0644	edit dl rm
process_connectionlost.py	119	0644	edit dl rm
process_gireactornocompat.py	721	0644	edit dl rm
process_helper.py	1081	0644	edit dl rm
reactormixins.py	14054	0644	edit dl rm
test_abstract.py	2032	0644	edit dl rm
test_address.py	8598	0644	edit dl rm
test_asyncioreactor.py	1331	0644	edit dl rm
test_base.py	13369	0644	edit dl rm
test_baseprocess.py	2613	0644	edit dl rm
test_core.py	11262	0644	edit dl rm
test_coroutines.py	1787	0644	edit dl rm
test_default.py	3540	0644	edit dl rm
test_endpoints.py	148723	0644	edit dl rm
test_epollreactor.py	7336	0644	edit dl rm
test_fdset.py	13501	0644	edit dl rm
test_filedescriptor.py	2808	0644	edit dl rm
test_gireactor.py	9042	0644	edit dl rm
test_glibbase.py	2284	0644	edit dl rm
test_inlinecb.py	11599	0644	edit dl rm
test_inotify.py	16686	0644	edit dl rm
test_iocp.py	5170	0644	edit dl rm
test_kqueuereactor.py	1916	0644	edit dl rm
test_main.py	1436	0644	edit dl rm
test_newtls.py	6693	0644	edit dl rm
test_pollingfile.py	1278	0644	edit dl rm
test_posixbase.py	9536	0644	edit dl rm
test_posixprocess.py	11166	0644	edit dl rm
test_process.py	32045	0644	edit dl rm
test_protocol.py	18727	0644	edit dl rm
test_resolver.py	20047	0644	edit dl rm
test_serialport.py	1948	0644	edit dl rm
test_sigchld.py	3935	0644	edit dl rm
test_socket.py	9329	0644	edit dl rm
test_stdio.py	6449	0644	edit dl rm
test_tcp.py	108446	0644	edit dl rm
test_threads.py	8582	0644	edit dl rm
test_time.py	3717	0644	edit dl rm
test_tls.py	13459	0644	edit dl rm
test_udp.py	17140	0644	edit dl rm
test_udp_internals.py	4909	0644	edit dl rm
test_unix.py	35712	0644	edit dl rm
test_win32events.py	6500	0644	edit dl rm
test_win32serialport.py	5282	0644	edit dl rm
_awaittests.py.3only	6180	0644	edit dl rm
_posixifaces.py	4689	0644	edit dl rm
_win32ifaces.py	3883	0644	edit dl rm
_yieldfromtests.py.3only	4338	0644	edit dl rm
__init__.py	112	0644	edit dl rm

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

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

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

from __future__ import division, absolute_import

import socket

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

if platformType == 'win32':
    from errno import WSAEWOULDBLOCK as EWOULDBLOCK
else:
    from errno import EWOULDBLOCK

class StringUDPSocket(object):
    """
    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: