From afa07ba7941f0ff4a110fc5b3d32139a812da755 Mon Sep 17 00:00:00 2001 From: Serhiy Storchaka Date: Mon, 28 Apr 2025 21:41:09 +0300 Subject: [PATCH 1/2] gh-132742: Fix newly added tcflush() tests on Android (GH-133070) (cherry picked from commit 25186c2472f6f2ceecb00fcecbce52b924510ed7) Co-authored-by: Serhiy Storchaka --- Lib/test/test_ioctl.py | 21 +++++++++++++-------- Lib/test/test_termios.py | 17 +++++++++++------ 2 files changed, 24 insertions(+), 14 deletions(-) diff --git a/Lib/test/test_ioctl.py b/Lib/test/test_ioctl.py index a49ff8497d13ca..62b37cbd4cf939 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 abafb26e7aa1cc..9a37d43aaecd68 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') @@ -178,7 +180,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 @@ -187,20 +190,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') From 8cef3ad8f206875ffcc9fbe8a854ff1c7d1ad3b5 Mon Sep 17 00:00:00 2001 From: Serhiy Storchaka Date: Mon, 28 Apr 2025 21:46:22 +0300 Subject: [PATCH 2/2] Fix using nonexisting helpers. --- Lib/test/test_ioctl.py | 3 +-- Lib/test/test_termios.py | 3 +-- 2 files changed, 2 insertions(+), 4 deletions(-) diff --git a/Lib/test/test_ioctl.py b/Lib/test/test_ioctl.py index 62b37cbd4cf939..980a20aadc0bde 100644 --- a/Lib/test/test_ioctl.py +++ b/Lib/test/test_ioctl.py @@ -165,8 +165,7 @@ def test_ioctl_clear_input_or_output(self): os.write(wfd, b'ABCDEF') self.assertEqual(os.read(rfd, 1024), b'ABCDEF') - @support.skip_android_selinux('tcflow') - @unittest.skipUnless(sys.platform in ('linux', 'android'), 'only works on Linux') + @unittest.skipUnless(sys.platform == 'linux', 'only works on Linux') @unittest.skipUnless(hasattr(termios, 'TCXONC'), 'requires termios.TCXONC') def test_ioctl_suspend_and_resume_output(self): wfd = self.slave_fd diff --git a/Lib/test/test_termios.py b/Lib/test/test_termios.py index 9a37d43aaecd68..5fd62b30263a37 100644 --- a/Lib/test/test_termios.py +++ b/Lib/test/test_termios.py @@ -180,8 +180,7 @@ def test_tcflow_errors(self): self.assertRaises(TypeError, termios.tcflow, object(), termios.TCOON) self.assertRaises(TypeError, termios.tcflow, self.fd) - @support.skip_android_selinux('tcflow') - @unittest.skipUnless(sys.platform in ('linux', 'android'), 'only works on Linux') + @unittest.skipUnless(sys.platform == 'linux', 'only works on Linux') def test_tcflow_suspend_and_resume_output(self): wfd = self.fd rfd = self.master_fd