1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
|
import errno, socket
from twisted.python.log import err
from twisted.internet.interfaces import IReactorSocket
from twisted.internet.error import UnsupportedAddressFamily
from twisted.internet.protocol import ServerFactory
from twisted.internet.test.reactormixins import (
ReactorBuilder, needsRunningReactor)
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())
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.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.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)
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())
globals().update(AdoptStreamPortErrorsTestsBuilder.makeTestCaseClasses())
globals().update(AdoptStreamConnectionErrorsTestsBuilder.makeTestCaseClasses())
|
