https://github.com/python/cpython/commit/b71442f44fe9ead71dab9a63f929388cec7e9d46
commit: b71442f44fe9ead71dab9a63f929388cec7e9d46
branch: 3.13
author: Miss Islington (bot) <31488909+miss-isling...@users.noreply.github.com>
committer: serhiy-storchaka <storch...@gmail.com>
date: 2025-04-28T19:10:42Z
summary:
[3.13] gh-132742: Fix newly added tcflush() tests on Android (GH-133070)
(GH-133104)
(cherry picked from commit 25186c2472f6f2ceecb00fcecbce52b924510ed7)
Co-authored-by: Serhiy Storchaka <storch...@gmail.com>
files:
M Lib/test/test_ioctl.py
M Lib/test/test_termios.py
diff --git a/Lib/test/test_ioctl.py b/Lib/test/test_ioctl.py
index a49ff8497d13ca..980a20aadc0bde 100644
--- a/Lib/test/test_ioctl.py
+++ b/Lib/test/test_ioctl.py
@@ -4,7 +4,7 @@
import sys
import threading
import unittest
-from test.support import get_attribute
+from test import support
from test.support import threading_helper
from test.support.import_helper import import_module
fcntl = import_module('fcntl')
@@ -13,7 +13,7 @@
class IoctlTestsTty(unittest.TestCase):
@classmethod
def setUpClass(cls):
- TIOCGPGRP = get_attribute(termios, 'TIOCGPGRP')
+ TIOCGPGRP = support.get_attribute(termios, 'TIOCGPGRP')
try:
tty = open("/dev/tty", "rb")
except OSError:
@@ -143,7 +143,9 @@ def setUp(self):
def test_ioctl_clear_input_or_output(self):
wfd = self.slave_fd
rfd = self.master_fd
- inbuf = sys.platform == 'linux'
+ # The data is buffered in the input buffer on Linux, and in
+ # the output buffer on other platforms.
+ inbuf = sys.platform in ('linux', 'android')
os.write(wfd, b'abcdef')
self.assertEqual(os.read(rfd, 2), b'ab')
@@ -173,20 +175,22 @@ def test_ioctl_suspend_and_resume_output(self):
def writer():
os.write(wfd, b'abc')
- write_suspended.wait()
+ self.assertTrue(write_suspended.wait(support.SHORT_TIMEOUT))
os.write(wfd, b'def')
write_finished.set()
with threading_helper.start_threads([threading.Thread(target=writer)]):
self.assertEqual(os.read(rfd, 3), b'abc')
try:
- fcntl.ioctl(wfd, termios.TCXONC, termios.TCOOFF)
- write_suspended.set()
+ try:
+ fcntl.ioctl(wfd, termios.TCXONC, termios.TCOOFF)
+ finally:
+ write_suspended.set()
self.assertFalse(write_finished.wait(0.5),
'output was not suspended')
finally:
fcntl.ioctl(wfd, termios.TCXONC, termios.TCOON)
- self.assertTrue(write_finished.wait(0.5),
+ self.assertTrue(write_finished.wait(support.SHORT_TIMEOUT),
'output was not resumed')
self.assertEqual(os.read(rfd, 1024), b'def')
diff --git a/Lib/test/test_termios.py b/Lib/test/test_termios.py
index abafb26e7aa1cc..5fd62b30263a37 100644
--- a/Lib/test/test_termios.py
+++ b/Lib/test/test_termios.py
@@ -142,7 +142,9 @@ def test_tcflush_errors(self):
def test_tcflush_clear_input_or_output(self):
wfd = self.fd
rfd = self.master_fd
- inbuf = sys.platform == 'linux'
+ # The data is buffered in the input buffer on Linux, and in
+ # the output buffer on other platforms.
+ inbuf = sys.platform in ('linux', 'android')
os.write(wfd, b'abcdef')
self.assertEqual(os.read(rfd, 2), b'ab')
@@ -187,20 +189,22 @@ def test_tcflow_suspend_and_resume_output(self):
def writer():
os.write(wfd, b'abc')
- write_suspended.wait()
+ self.assertTrue(write_suspended.wait(support.SHORT_TIMEOUT))
os.write(wfd, b'def')
write_finished.set()
with threading_helper.start_threads([threading.Thread(target=writer)]):
self.assertEqual(os.read(rfd, 3), b'abc')
try:
- termios.tcflow(wfd, termios.TCOOFF)
- write_suspended.set()
+ try:
+ termios.tcflow(wfd, termios.TCOOFF)
+ finally:
+ write_suspended.set()
self.assertFalse(write_finished.wait(0.5),
'output was not suspended')
finally:
termios.tcflow(wfd, termios.TCOON)
- self.assertTrue(write_finished.wait(0.5),
+ self.assertTrue(write_finished.wait(support.SHORT_TIMEOUT),
'output was not resumed')
self.assertEqual(os.read(rfd, 1024), b'def')
_______________________________________________
Python-checkins mailing list -- python-checkins@python.org
To unsubscribe send an email to python-checkins-le...@python.org
https://mail.python.org/mailman3/lists/python-checkins.python.org/
Member address: arch...@mail-archive.com
