python (3.11.7)
1 import errno
2 import os
3 import select
4 import subprocess
5 import sys
6 import textwrap
7 import unittest
8 from test import support
9
10 support.requires_working_socket(module=True)
11
12 @unittest.skipIf((sys.platform[:3]=='win'),
13 "can't easily test on this system")
14 class ESC[4;38;5;81mSelectTestCase(ESC[4;38;5;149munittestESC[4;38;5;149m.ESC[4;38;5;149mTestCase):
15
16 class ESC[4;38;5;81mNope:
17 pass
18
19 class ESC[4;38;5;81mAlmost:
20 def fileno(self):
21 return 'fileno'
22
23 def test_error_conditions(self):
24 self.assertRaises(TypeError, select.select, 1, 2, 3)
25 self.assertRaises(TypeError, select.select, [self.Nope()], [], [])
26 self.assertRaises(TypeError, select.select, [self.Almost()], [], [])
27 self.assertRaises(TypeError, select.select, [], [], [], "not a number")
28 self.assertRaises(ValueError, select.select, [], [], [], -1)
29
30 # Issue #12367: http://www.freebsd.org/cgi/query-pr.cgi?pr=kern/155606
31 @unittest.skipIf(sys.platform.startswith('freebsd'),
32 'skip because of a FreeBSD bug: kern/155606')
33 def test_errno(self):
34 with open(__file__, 'rb') as fp:
35 fd = fp.fileno()
36 fp.close()
37 try:
38 select.select([fd], [], [], 0)
39 except OSError as err:
40 self.assertEqual(err.errno, errno.EBADF)
41 else:
42 self.fail("exception not raised")
43
44 def test_returned_list_identity(self):
45 # See issue #8329
46 r, w, x = select.select([], [], [], 1)
47 self.assertIsNot(r, w)
48 self.assertIsNot(r, x)
49 self.assertIsNot(w, x)
50
51 @support.requires_fork()
52 def test_select(self):
53 code = textwrap.dedent('''
54 import time
55 for i in range(10):
56 print("testing...", flush=True)
57 time.sleep(0.050)
58 ''')
59 cmd = [sys.executable, '-I', '-c', code]
60 with subprocess.Popen(cmd, stdout=subprocess.PIPE) as proc:
61 pipe = proc.stdout
62 for timeout in (0, 1, 2, 4, 8, 16) + (None,)*10:
63 if support.verbose:
64 print(f'timeout = {timeout}')
65 rfd, wfd, xfd = select.select([pipe], [], [], timeout)
66 self.assertEqual(wfd, [])
67 self.assertEqual(xfd, [])
68 if not rfd:
69 continue
70 if rfd == [pipe]:
71 line = pipe.readline()
72 if support.verbose:
73 print(repr(line))
74 if not line:
75 if support.verbose:
76 print('EOF')
77 break
78 continue
79 self.fail('Unexpected return values from select():',
80 rfd, wfd, xfd)
81
82 # Issue 16230: Crash on select resized list
83 @unittest.skipIf(
84 support.is_emscripten, "Emscripten cannot select a fd multiple times."
85 )
86 def test_select_mutated(self):
87 a = []
88 class ESC[4;38;5;81mF:
89 def fileno(self):
90 del a[-1]
91 return sys.__stdout__.fileno()
92 a[:] = [F()] * 10
93 self.assertEqual(select.select([], a, []), ([], a[:5], []))
94
95 def test_disallow_instantiation(self):
96 support.check_disallow_instantiation(self, type(select.poll()))
97
98 if hasattr(select, 'devpoll'):
99 support.check_disallow_instantiation(self, type(select.devpoll()))
100
101 def tearDownModule():
102 support.reap_children()
103
104 if __name__ == "__main__":
105 unittest.main()
Browse
Usage
Versions