ยปCore Development>Code coverage>Lib/test/test_timeout.py

Python code coverage for Lib/test/test_timeout.py

#countcontent
1n/a"""Unit tests for socket timeout feature."""
2n/a
3n/aimport functools
4n/aimport unittest
5n/afrom test import support
6n/a
7n/a# This requires the 'network' resource as given on the regrtest command line.
8n/askip_expected = not support.is_resource_enabled('network')
9n/a
10n/aimport time
11n/aimport errno
12n/aimport socket
13n/a
14n/a
15n/a@functools.lru_cache()
16n/adef resolve_address(host, port):
17n/a """Resolve an (host, port) to an address.
18n/a
19n/a We must perform name resolution before timeout tests, otherwise it will be
20n/a performed by connect().
21n/a """
22n/a with support.transient_internet(host):
23n/a return socket.getaddrinfo(host, port, socket.AF_INET,
24n/a socket.SOCK_STREAM)[0][4]
25n/a
26n/a
27n/aclass CreationTestCase(unittest.TestCase):
28n/a """Test case for socket.gettimeout() and socket.settimeout()"""
29n/a
30n/a def setUp(self):
31n/a self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
32n/a
33n/a def tearDown(self):
34n/a self.sock.close()
35n/a
36n/a def testObjectCreation(self):
37n/a # Test Socket creation
38n/a self.assertEqual(self.sock.gettimeout(), None,
39n/a "timeout not disabled by default")
40n/a
41n/a def testFloatReturnValue(self):
42n/a # Test return value of gettimeout()
43n/a self.sock.settimeout(7.345)
44n/a self.assertEqual(self.sock.gettimeout(), 7.345)
45n/a
46n/a self.sock.settimeout(3)
47n/a self.assertEqual(self.sock.gettimeout(), 3)
48n/a
49n/a self.sock.settimeout(None)
50n/a self.assertEqual(self.sock.gettimeout(), None)
51n/a
52n/a def testReturnType(self):
53n/a # Test return type of gettimeout()
54n/a self.sock.settimeout(1)
55n/a self.assertEqual(type(self.sock.gettimeout()), type(1.0))
56n/a
57n/a self.sock.settimeout(3.9)
58n/a self.assertEqual(type(self.sock.gettimeout()), type(1.0))
59n/a
60n/a def testTypeCheck(self):
61n/a # Test type checking by settimeout()
62n/a self.sock.settimeout(0)
63n/a self.sock.settimeout(0)
64n/a self.sock.settimeout(0.0)
65n/a self.sock.settimeout(None)
66n/a self.assertRaises(TypeError, self.sock.settimeout, "")
67n/a self.assertRaises(TypeError, self.sock.settimeout, "")
68n/a self.assertRaises(TypeError, self.sock.settimeout, ())
69n/a self.assertRaises(TypeError, self.sock.settimeout, [])
70n/a self.assertRaises(TypeError, self.sock.settimeout, {})
71n/a self.assertRaises(TypeError, self.sock.settimeout, 0j)
72n/a
73n/a def testRangeCheck(self):
74n/a # Test range checking by settimeout()
75n/a self.assertRaises(ValueError, self.sock.settimeout, -1)
76n/a self.assertRaises(ValueError, self.sock.settimeout, -1)
77n/a self.assertRaises(ValueError, self.sock.settimeout, -1.0)
78n/a
79n/a def testTimeoutThenBlocking(self):
80n/a # Test settimeout() followed by setblocking()
81n/a self.sock.settimeout(10)
82n/a self.sock.setblocking(1)
83n/a self.assertEqual(self.sock.gettimeout(), None)
84n/a self.sock.setblocking(0)
85n/a self.assertEqual(self.sock.gettimeout(), 0.0)
86n/a
87n/a self.sock.settimeout(10)
88n/a self.sock.setblocking(0)
89n/a self.assertEqual(self.sock.gettimeout(), 0.0)
90n/a self.sock.setblocking(1)
91n/a self.assertEqual(self.sock.gettimeout(), None)
92n/a
93n/a def testBlockingThenTimeout(self):
94n/a # Test setblocking() followed by settimeout()
95n/a self.sock.setblocking(0)
96n/a self.sock.settimeout(1)
97n/a self.assertEqual(self.sock.gettimeout(), 1)
98n/a
99n/a self.sock.setblocking(1)
100n/a self.sock.settimeout(1)
101n/a self.assertEqual(self.sock.gettimeout(), 1)
102n/a
103n/a
104n/aclass TimeoutTestCase(unittest.TestCase):
105n/a # There are a number of tests here trying to make sure that an operation
106n/a # doesn't take too much longer than expected. But competing machine
107n/a # activity makes it inevitable that such tests will fail at times.
108n/a # When fuzz was at 1.0, I (tim) routinely saw bogus failures on Win2K
109n/a # and Win98SE. Boosting it to 2.0 helped a lot, but isn't a real
110n/a # solution.
111n/a fuzz = 2.0
112n/a
113n/a localhost = support.HOST
114n/a
115n/a def setUp(self):
116n/a raise NotImplementedError()
117n/a
118n/a tearDown = setUp
119n/a
120n/a def _sock_operation(self, count, timeout, method, *args):
121n/a """
122n/a Test the specified socket method.
123n/a
124n/a The method is run at most `count` times and must raise a socket.timeout
125n/a within `timeout` + self.fuzz seconds.
126n/a """
127n/a self.sock.settimeout(timeout)
128n/a method = getattr(self.sock, method)
129n/a for i in range(count):
130n/a t1 = time.time()
131n/a try:
132n/a method(*args)
133n/a except socket.timeout as e:
134n/a delta = time.time() - t1
135n/a break
136n/a else:
137n/a self.fail('socket.timeout was not raised')
138n/a # These checks should account for timing unprecision
139n/a self.assertLess(delta, timeout + self.fuzz)
140n/a self.assertGreater(delta, timeout - 1.0)
141n/a
142n/a
143n/aclass TCPTimeoutTestCase(TimeoutTestCase):
144n/a """TCP test case for socket.socket() timeout functions"""
145n/a
146n/a def setUp(self):
147n/a self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
148n/a self.addr_remote = resolve_address('www.python.org.', 80)
149n/a
150n/a def tearDown(self):
151n/a self.sock.close()
152n/a
153n/a def testConnectTimeout(self):
154n/a # Testing connect timeout is tricky: we need to have IP connectivity
155n/a # to a host that silently drops our packets. We can't simulate this
156n/a # from Python because it's a function of the underlying TCP/IP stack.
157n/a # So, the following Snakebite host has been defined:
158n/a blackhole = resolve_address('blackhole.snakebite.net', 56666)
159n/a
160n/a # Blackhole has been configured to silently drop any incoming packets.
161n/a # No RSTs (for TCP) or ICMP UNREACH (for UDP/ICMP) will be sent back
162n/a # to hosts that attempt to connect to this address: which is exactly
163n/a # what we need to confidently test connect timeout.
164n/a
165n/a # However, we want to prevent false positives. It's not unreasonable
166n/a # to expect certain hosts may not be able to reach the blackhole, due
167n/a # to firewalling or general network configuration. In order to improve
168n/a # our confidence in testing the blackhole, a corresponding 'whitehole'
169n/a # has also been set up using one port higher:
170n/a whitehole = resolve_address('whitehole.snakebite.net', 56667)
171n/a
172n/a # This address has been configured to immediately drop any incoming
173n/a # packets as well, but it does it respectfully with regards to the
174n/a # incoming protocol. RSTs are sent for TCP packets, and ICMP UNREACH
175n/a # is sent for UDP/ICMP packets. This means our attempts to connect to
176n/a # it should be met immediately with ECONNREFUSED. The test case has
177n/a # been structured around this premise: if we get an ECONNREFUSED from
178n/a # the whitehole, we proceed with testing connect timeout against the
179n/a # blackhole. If we don't, we skip the test (with a message about not
180n/a # getting the required RST from the whitehole within the required
181n/a # timeframe).
182n/a
183n/a # For the records, the whitehole/blackhole configuration has been set
184n/a # up using the 'pf' firewall (available on BSDs), using the following:
185n/a #
186n/a # ext_if="bge0"
187n/a #
188n/a # blackhole_ip="35.8.247.6"
189n/a # whitehole_ip="35.8.247.6"
190n/a # blackhole_port="56666"
191n/a # whitehole_port="56667"
192n/a #
193n/a # block return in log quick on $ext_if proto { tcp udp } \
194n/a # from any to $whitehole_ip port $whitehole_port
195n/a # block drop in log quick on $ext_if proto { tcp udp } \
196n/a # from any to $blackhole_ip port $blackhole_port
197n/a #
198n/a
199n/a skip = True
200n/a sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
201n/a # Use a timeout of 3 seconds. Why 3? Because it's more than 1, and
202n/a # less than 5. i.e. no particular reason. Feel free to tweak it if
203n/a # you feel a different value would be more appropriate.
204n/a timeout = 3
205n/a sock.settimeout(timeout)
206n/a try:
207n/a sock.connect((whitehole))
208n/a except socket.timeout:
209n/a pass
210n/a except OSError as err:
211n/a if err.errno == errno.ECONNREFUSED:
212n/a skip = False
213n/a finally:
214n/a sock.close()
215n/a del sock
216n/a
217n/a if skip:
218n/a self.skipTest(
219n/a "We didn't receive a connection reset (RST) packet from "
220n/a "{}:{} within {} seconds, so we're unable to test connect "
221n/a "timeout against the corresponding {}:{} (which is "
222n/a "configured to silently drop packets)."
223n/a .format(
224n/a whitehole[0],
225n/a whitehole[1],
226n/a timeout,
227n/a blackhole[0],
228n/a blackhole[1],
229n/a )
230n/a )
231n/a
232n/a # All that hard work just to test if connect times out in 0.001s ;-)
233n/a self.addr_remote = blackhole
234n/a with support.transient_internet(self.addr_remote[0]):
235n/a self._sock_operation(1, 0.001, 'connect', self.addr_remote)
236n/a
237n/a def testRecvTimeout(self):
238n/a # Test recv() timeout
239n/a with support.transient_internet(self.addr_remote[0]):
240n/a self.sock.connect(self.addr_remote)
241n/a self._sock_operation(1, 1.5, 'recv', 1024)
242n/a
243n/a def testAcceptTimeout(self):
244n/a # Test accept() timeout
245n/a support.bind_port(self.sock, self.localhost)
246n/a self.sock.listen()
247n/a self._sock_operation(1, 1.5, 'accept')
248n/a
249n/a def testSend(self):
250n/a # Test send() timeout
251n/a with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as serv:
252n/a support.bind_port(serv, self.localhost)
253n/a serv.listen()
254n/a self.sock.connect(serv.getsockname())
255n/a # Send a lot of data in order to bypass buffering in the TCP stack.
256n/a self._sock_operation(100, 1.5, 'send', b"X" * 200000)
257n/a
258n/a def testSendto(self):
259n/a # Test sendto() timeout
260n/a with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as serv:
261n/a support.bind_port(serv, self.localhost)
262n/a serv.listen()
263n/a self.sock.connect(serv.getsockname())
264n/a # The address argument is ignored since we already connected.
265n/a self._sock_operation(100, 1.5, 'sendto', b"X" * 200000,
266n/a serv.getsockname())
267n/a
268n/a def testSendall(self):
269n/a # Test sendall() timeout
270n/a with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as serv:
271n/a support.bind_port(serv, self.localhost)
272n/a serv.listen()
273n/a self.sock.connect(serv.getsockname())
274n/a # Send a lot of data in order to bypass buffering in the TCP stack.
275n/a self._sock_operation(100, 1.5, 'sendall', b"X" * 200000)
276n/a
277n/a
278n/aclass UDPTimeoutTestCase(TimeoutTestCase):
279n/a """UDP test case for socket.socket() timeout functions"""
280n/a
281n/a def setUp(self):
282n/a self.sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
283n/a
284n/a def tearDown(self):
285n/a self.sock.close()
286n/a
287n/a def testRecvfromTimeout(self):
288n/a # Test recvfrom() timeout
289n/a # Prevent "Address already in use" socket exceptions
290n/a support.bind_port(self.sock, self.localhost)
291n/a self._sock_operation(1, 1.5, 'recvfrom', 1024)
292n/a
293n/a
294n/adef test_main():
295n/a support.requires('network')
296n/a support.run_unittest(
297n/a CreationTestCase,
298n/a TCPTimeoutTestCase,
299n/a UDPTimeoutTestCase,
300n/a )
301n/a
302n/aif __name__ == "__main__":
303n/a test_main()