mirror of
https://github.com/RustPython/RustPython.git
synced 2026-10-28 18:59:04 +08:00
Jeong, YunWon
7c4017b3d8
* Retry select/poll/fcntl on EINTR
select() no longer shares one empty list for r/w/x and now
calls select(2) even with empty sets so the timeout elapses.
poll() rejects type instantiation, treats float timeout as
milliseconds, and raises RuntimeError on concurrent poll().
fcntl, ioctl, flock, and lockf retry after EINTR.
Assisted-by: Grok:grok-4.6
* Add select.kqueue and select.kevent
Expose kqueue(2)/kevent(2) on BSD and macOS, including
control() EINTR retry and marking open kqueue objects
closed after fork.
Assisted-by: Grok:grok-4.6
* Add login_tty and fix termios/syslog/mmap
termios.error is Exception, not OSError.
os.login_tty prepares a tty for a new session.
syslog.openlog encodes ident as UTF-8.
mmap.seek returns the new position.
Assisted-by: Grok:grok-4.6
* Honor mmap trackfd on Unix
When trackfd is false, do not keep a duplicated file
descriptor. size() then fails with EBADF and resize()
raises ValueError.
Assisted-by: Grok:grok-4.6
* address review: portable kevent and kqueue close
Build kevent values by field assignment so FreeBSD's
ext array compiles. Normalize timespec nanoseconds,
keep the kqueue fd live across control(), reinit the
open list lock after fork, make trackfd keyword-only,
and skip login_tty on Solaris/illumos.
Assisted-by: Grok:grok-4.6
(cherry picked from commit 0ba7048ac9)
Assisted-by: Codex:gpt-6
105 lines
3.4 KiB
Python
Vendored
105 lines
3.4 KiB
Python
Vendored
import errno
|
|
import select
|
|
import subprocess
|
|
import sys
|
|
import textwrap
|
|
import unittest
|
|
from test import support
|
|
|
|
support.requires_working_socket(module=True)
|
|
|
|
@unittest.skipIf((sys.platform[:3]=='win'),
|
|
"can't easily test on this system")
|
|
class SelectTestCase(unittest.TestCase):
|
|
|
|
class Nope:
|
|
pass
|
|
|
|
class Almost:
|
|
def fileno(self):
|
|
return 'fileno'
|
|
|
|
def test_error_conditions(self):
|
|
self.assertRaises(TypeError, select.select, 1, 2, 3)
|
|
self.assertRaises(TypeError, select.select, [self.Nope()], [], [])
|
|
self.assertRaises(TypeError, select.select, [self.Almost()], [], [])
|
|
self.assertRaises(TypeError, select.select, [], [], [], "not a number")
|
|
self.assertRaises(ValueError, select.select, [], [], [], -1)
|
|
|
|
# Issue #12367: http://www.freebsd.org/cgi/query-pr.cgi?pr=kern/155606
|
|
@unittest.skipIf(sys.platform.startswith('freebsd'),
|
|
'skip because of a FreeBSD bug: kern/155606')
|
|
def test_errno(self):
|
|
with open(__file__, 'rb') as fp:
|
|
fd = fp.fileno()
|
|
fp.close()
|
|
try:
|
|
select.select([fd], [], [], 0)
|
|
except OSError as err:
|
|
self.assertEqual(err.errno, errno.EBADF)
|
|
else:
|
|
self.fail("exception not raised")
|
|
|
|
def test_returned_list_identity(self):
|
|
# See issue #8329
|
|
r, w, x = select.select([], [], [], 1)
|
|
self.assertIsNot(r, w)
|
|
self.assertIsNot(r, x)
|
|
self.assertIsNot(w, x)
|
|
|
|
@support.requires_fork()
|
|
def test_select(self):
|
|
code = textwrap.dedent('''
|
|
import time
|
|
for i in range(10):
|
|
print("testing...", flush=True)
|
|
time.sleep(0.050)
|
|
''')
|
|
cmd = [sys.executable, '-I', '-c', code]
|
|
with subprocess.Popen(cmd, stdout=subprocess.PIPE) as proc:
|
|
pipe = proc.stdout
|
|
for timeout in (0, 1, 2, 4, 8, 16) + (None,)*10:
|
|
if support.verbose:
|
|
print(f'timeout = {timeout}')
|
|
rfd, wfd, xfd = select.select([pipe], [], [], timeout)
|
|
self.assertEqual(wfd, [])
|
|
self.assertEqual(xfd, [])
|
|
if not rfd:
|
|
continue
|
|
if rfd == [pipe]:
|
|
line = pipe.readline()
|
|
if support.verbose:
|
|
print(repr(line))
|
|
if not line:
|
|
if support.verbose:
|
|
print('EOF')
|
|
break
|
|
continue
|
|
self.fail('Unexpected return values from select():',
|
|
rfd, wfd, xfd)
|
|
|
|
# Issue 16230: Crash on select resized list
|
|
@unittest.skipIf(
|
|
support.is_emscripten, "Emscripten cannot select a fd multiple times."
|
|
)
|
|
def test_select_mutated(self):
|
|
a = []
|
|
class F:
|
|
def fileno(self):
|
|
del a[-1]
|
|
return sys.__stdout__.fileno()
|
|
a[:] = [F()] * 10
|
|
self.assertEqual(select.select([], a, []), ([], a[:5], []))
|
|
|
|
def test_disallow_instantiation(self):
|
|
support.check_disallow_instantiation(self, type(select.poll()))
|
|
|
|
if hasattr(select, 'devpoll'):
|
|
support.check_disallow_instantiation(self, type(select.devpoll()))
|
|
|
|
def tearDownModule():
|
|
support.reap_children()
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|