Apache/2.4.7 (Ubuntu) Linux sman1baleendah 3.13.0-24-generic #46-Ubuntu SMP Thu Apr 10 19:11:08 UTC 2014 x86_64 uid=33(www-data) gid=33(www-data) groups=33(www-data) safemode : OFF MySQL: ON | Perl: ON | cURL: OFF | WGet: ON > / usr / lib / python2.7 / dist-packages / twisted / internet / test / | server ip : 104.21.89.46 your ip : 172.70.80.170 H O M E |
Filename | /usr/lib/python2.7/dist-packages/twisted/internet/test/test_udp_internals.py |
Size | 4.88 kb |
Permission | rw-r--r-- |
Owner | root : root |
Create time | 27-Apr-2025 09:56 |
Last modified | 29-Sep-2012 03:05 |
Last accessed | 07-Jul-2025 07:08 |
Actions | edit | rename | delete | download (gzip) |
View | text | code | image |
# 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
from errno import WSAECONNREFUSED as ECONNREFUSED
else:
from errno import EWOULDBLOCK
from errno import ECONNREFUSED
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 ErrorsTestCase(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"])
# 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
from errno import WSAECONNREFUSED as ECONNREFUSED
else:
from errno import EWOULDBLOCK
from errno import ECONNREFUSED
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 ErrorsTestCase(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"])