https://github.com/python/cpython/commit/25186c2472f6f2ceecb00fcecbce52b924510ed7
commit: 25186c2472f6f2ceecb00fcecbce52b924510ed7
branch: main
author: Serhiy Storchaka <storch...@gmail.com>
committer: serhiy-storchaka <storch...@gmail.com>
date: 2025-04-28T21:41:09+03:00
summary:
gh-132742: Fix newly added tcflush() tests on Android (GH-133070)
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 9fd3b6196b81c2..7a986048bdac2a 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')
@@ -163,7 +165,8 @@ def test_ioctl_clear_input_or_output(self):
os.write(wfd, b'ABCDEF')
self.assertEqual(os.read(rfd, 1024), b'ABCDEF')
- @unittest.skipUnless(sys.platform == 'linux', 'only works on Linux')
+ @support.skip_android_selinux('tcflow')
+ @unittest.skipUnless(sys.platform in ('linux', 'android'), 'only works on
Linux')
@unittest.skipUnless(hasattr(termios, 'TCXONC'), 'requires termios.TCXONC')
def test_ioctl_suspend_and_resume_output(self):
wfd = self.slave_fd
@@ -173,20 +176,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 18965f09a2569e..e5d11cf84d2a66 100644
--- a/Lib/test/test_termios.py
+++ b/Lib/test/test_termios.py
@@ -152,7 +152,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')
@@ -190,7 +192,8 @@ def test_tcflow_errors(self):
self.assertRaises(TypeError, termios.tcflow, object(), termios.TCOON)
self.assertRaises(TypeError, termios.tcflow, self.fd)
- @unittest.skipUnless(sys.platform == 'linux', 'only works on Linux')
+ @support.skip_android_selinux('tcflow')
+ @unittest.skipUnless(sys.platform in ('linux', 'android'), 'only works on
Linux')
def test_tcflow_suspend_and_resume_output(self):
wfd = self.fd
rfd = self.master_fd
@@ -199,20 +202,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
