gh-132742: Fix newly added tcflush() tests on Android (GH-133070)

diff --git a/Lib/test/test_ioctl.py b/Lib/test/test_ioctl.py
index 9fd3b6196b81c2eddb709d95ae7df4ae396a5ce8..7a986048bdac2ac813193c585df56c24e4460e69 100644 (file)
--- a/Lib/test/test_ioctl.py
+++ b/Lib/test/test_ioctl.py
@@ -4,7 +4,7 @@ import struct
 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 @@ termios = import_module('termios')
 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 @@ class IoctlTestsPty(unittest.TestCase):
     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 @@ class IoctlTestsPty(unittest.TestCase):
         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 @@ class IoctlTestsPty(unittest.TestCase):
 
         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 18965f09a2569efd626094edc5af7dea2c1aa960..e5d11cf84d2a66e0eb6ef7f1c18085178e70f4b6 100644 (file)
--- a/Lib/test/test_termios.py
+++ b/Lib/test/test_termios.py
@@ -152,7 +152,9 @@ class TestFunctions(unittest.TestCase):
     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 @@ class TestFunctions(unittest.TestCase):
         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 @@ class TestFunctions(unittest.TestCase):
 
         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')